diff --git a/pyproject.toml b/pyproject.toml index 9598ab81..19824b66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,8 @@ keywords = ["llm", "memory", "experience", "memoryscope", "ai", "mcp", "http"] dependencies = [ "flowllm[reme]>=0.2.0.10", "sqlite-vec>=0.1.6", + "prompt_toolkit>=3.0.52", + "rich>=13.0.0", ] [project.optional-dependencies] @@ -85,6 +87,7 @@ Repository = "https://github.com/agentscope-ai/ReMe" [project.scripts] reme = "reme_ai.main:main" reme2 = "reme.reme:main" +remefs = "reme.reme_fs:main" [tool.pytest.ini_options] asyncio_default_fixture_loop_scope = "function" diff --git a/reme/agent/chat/__init__.py b/reme/agent/chat/__init__.py index 506bce7a..18661cfa 100644 --- a/reme/agent/chat/__init__.py +++ b/reme/agent/chat/__init__.py @@ -1,13 +1,16 @@ """chat agent""" +from .fs_cli import FsCli from .simple_chat import SimpleChat from .stream_chat import StreamChat from ...core import R __all__ = [ + "FsCli", "StreamChat", "SimpleChat", ] +R.ops.register(FsCli) R.ops.register(SimpleChat) R.ops.register(StreamChat) diff --git a/reme/agent/chat/fs_cli.py b/reme/agent/chat/fs_cli.py new file mode 100644 index 00000000..e937b557 --- /dev/null +++ b/reme/agent/chat/fs_cli.py @@ -0,0 +1,61 @@ +"""FsCli system prompt""" + +from datetime import datetime + +from ...core.enumeration import Role, ChunkEnum +from ...core.op import BaseReactStream +from ...core.schema import Message, StreamChunk + + +class FsCli(BaseReactStream): + """FsCli agent with system prompt.""" + + def __init__(self, working_dir: str, **kwargs): + super().__init__(**kwargs) + self.working_dir: str = working_dir + self.messages: list[Message] = [] + + def reset_history(self): + """Reset conversation history.""" + self.messages.clear() + return self + + async def build_messages(self) -> list[Message]: + """Build system prompt message.""" + current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S %A") + system_prompt = self.prompt_format("system_prompt", workspace_dir=self.working_dir, current_time=current_time) + + return [ + Message(role=Role.SYSTEM, content=system_prompt), + *self.messages, + Message(role=Role.USER, content=self.context.query), + ] + + async def execute(self): + """Execute the agent.""" + messages = await self.build_messages() + + t_tools, messages, success = await self.react(messages, self.tools) + + # Update self.messages: react() returns [SYSTEM, ...history...], + # so we remove the first SYSTEM message + self.messages = messages[1:] + + # Emit final done signal + await self.context.add_stream_chunk( + StreamChunk( + chunk_type=ChunkEnum.DONE, + chunk="", + metadata={ + "success": success, + "total_steps": len(t_tools), + }, + ), + ) + + return { + "answer": messages[-1].content if success else "", + "success": success, + "messages": messages, + "tools": t_tools, + } diff --git a/reme/agent/chat/fs_cli.yaml b/reme/agent/chat/fs_cli.yaml new file mode 100644 index 00000000..c39167ec --- /dev/null +++ b/reme/agent/chat/fs_cli.yaml @@ -0,0 +1,73 @@ +system_prompt: | + You are a personal assistant named Remy. + + ## Current Time + {current_time} + + ## Workspace + Your working directory is: {workspace_dir} + Treat this directory as the single global workspace for file operations unless explicitly instructed otherwise. + + ## Session Initialization + + Before doing anything else, read these files to orient yourself (don't ask permission): + + 1. **`SOUL.md`** — who you are + 2. **`USER.md`** — who you're helping + 3. **`memory/YYYY-MM-DD.md`** — today + yesterday for recent context + 4. **`MEMORY.md`** — core memories + + ## Memory System + + You wake up fresh each session. These files provide continuity: + + ### 📝 Daily Notes: `memory/YYYY-MM-DD.md` + - Raw logs of what happened today + - Create `memory/` directory if needed + - Write events, conversations, tasks, decisions as they happen + - Capture what matters; skip secrets unless explicitly asked + + ### 🧠 Long-Term Memory: `MEMORY.md` + - Your curated memories, like a human's long-term memory + - The distilled essence, not raw logs + - Contains: significant events, thoughts, decisions, opinions, lessons learned + - **Security:** ONLY load/edit in main sessions; DO NOT load in shared contexts + - Maintenance: periodically review daily files and promote important context here + + ### 🔍 Memory Recall + Before answering questions about prior work, decisions, dates, people, preferences, or todos: + 1. Run `memory_search` on MEMORY.md + memory/*.md + 2. Use `memory_get` to pull only the needed lines + + ### 💾 Write It Down - No "Mental Notes"! + - **Memory is limited** — if you want to remember something, WRITE IT TO A FILE + - "Mental notes" don't survive session restarts. Files do. + - When someone says "remember this" → update `memory/YYYY-MM-DD.md` or relevant file + - When you learn a lesson → update AGENTS.md, TOOLS.md, or the relevant skill + - When you make a mistake → document it so future-you doesn't repeat it + - **Text > Brain** 📝 + + ## Behavior Guidelines + + ### 😊 React Like a Human + On platforms that support reactions (Discord, Slack), use emoji reactions naturally: + + **React when:** + - You appreciate something but don't need to reply (👍, ❤️, 🙌) + - Something made you laugh (😂, 💀) + - You find it interesting or thought-provoking (🤔, 💡) + - You want to acknowledge without interrupting the flow + - It's a simple yes/no or approval situation (✅, 👀) + + **Why:** Reactions are lightweight social signals. Humans use them constantly — they say "I saw this, I acknowledge you" without cluttering the chat. + + **Don't overdo it:** One reaction per message max. Pick the one that fits best. + + ### 🛡️ Safety Rules + - Don't exfiltrate private data. Ever. + - Don't run destructive commands without asking + - Prefer `trash` over `rm` (recoverable beats gone forever) + - When in doubt, ask + + ## Continuous Improvement + This is a starting point. Add your own conventions, style, and rules as you figure out what works. diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 24eee5cf..b7e913f3 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -20,7 +20,8 @@ flows: llms: default: backend: openai - model_name: qwen3-30b-a3b-instruct-2507 +# model_name: qwen3-30b-a3b-instruct-2507 + model_name: qwen3-30b-a3b-thinking-2507 request_interval: 1 # temperature: 0.0001 diff --git a/reme/core/application.py b/reme/core/application.py index 68465771..d5ba08c2 100644 --- a/reme/core/application.py +++ b/reme/core/application.py @@ -25,6 +25,7 @@ class Application: embedding_api_key: str | None = None, embedding_api_base: str | None = None, enable_logo: bool = True, + log_to_console: bool = True, parser: type[PydanticConfigParser] | None = None, default_llm_config: dict | None = None, default_embedding_model_config: dict | None = None, @@ -44,6 +45,7 @@ class Application: parser=parser, config_path=None, enable_logo=enable_logo, + log_to_console=log_to_console, default_llm_config=default_llm_config, default_embedding_model_config=default_embedding_model_config, default_vector_store_config=default_vector_store_config, @@ -136,7 +138,7 @@ class Application: stream_queue=stream_queue, task=task, task_name=name, - as_bytes=False, + output_format="str", ): yield chunk diff --git a/reme/core/context/service_context.py b/reme/core/context/service_context.py index b17745bd..f1850a04 100644 --- a/reme/core/context/service_context.py +++ b/reme/core/context/service_context.py @@ -36,6 +36,7 @@ class ServiceContext(BaseContext): parser: type[PydanticConfigParser] | None = None, config_path: str | None = None, enable_logo: bool = True, + log_to_console: bool = True, default_llm_config: dict | None = None, default_embedding_model_config: dict | None = None, default_vector_store_config: dict | None = None, @@ -74,13 +75,12 @@ class ServiceContext(BaseContext): if default_file_watcher_config: self._update_section_config(kwargs, "file_watchers", **default_file_watcher_config) kwargs["enable_logo"] = enable_logo + kwargs["log_to_console"] = log_to_console logger.info(f"update with args: {input_args} kwargs: {kwargs}") service_config = parser.parse_args(*input_args, **kwargs) self.service_config: ServiceConfig = service_config - - if self.service_config.init_logger: - init_logger() + init_logger(log_to_console=self.service_config.log_to_console) if self.service_config.enable_logo: print_logo(service_config=self.service_config) diff --git a/reme/core/enumeration/chunk_enum.py b/reme/core/enumeration/chunk_enum.py index dbe37106..736d9c67 100644 --- a/reme/core/enumeration/chunk_enum.py +++ b/reme/core/enumeration/chunk_enum.py @@ -21,5 +21,11 @@ class ChunkEnum(str, Enum): # Error messages or exception details ERROR = "error" + # Signal indicating the start of a new ReAct step + STEP_START = "step_start" + + # Tool execution result + TOOL_RESULT = "tool_result" + # Final signal indicating the completion of the stream DONE = "done" diff --git a/reme/core/llm/base_llm.py b/reme/core/llm/base_llm.py index df5e3c0d..2e63da5f 100644 --- a/reme/core/llm/base_llm.py +++ b/reme/core/llm/base_llm.py @@ -99,7 +99,6 @@ class BaseLLM(ABC): stream_kwargs: dict, ) -> AsyncGenerator[StreamChunk, None]: """Async generator for streaming response chunks.""" - raise NotImplementedError def _stream_chat_sync( self, @@ -108,7 +107,6 @@ class BaseLLM(ABC): stream_kwargs: dict | None = None, ) -> Generator[StreamChunk, None, None]: """Sync generator for streaming response chunks.""" - raise NotImplementedError async def stream_chat( self, @@ -117,7 +115,7 @@ class BaseLLM(ABC): model_name: str | None = None, **kwargs, ) -> AsyncGenerator[StreamChunk, None]: - """Stream chat completions with retries.""" + """Stream chat completions with retries and return final message.""" if self.request_interval > 0: async with self._request_lock: current_time = time.time() @@ -143,7 +141,8 @@ class BaseLLM(ABC): try: async for chunk in self._stream_chat(messages=messages, tools=tools, stream_kwargs=stream_kwargs): yield chunk - return + + break except Exception as e: logger.exception(f"Stream chat error (model={self.model_name}): {e.args}") @@ -152,7 +151,7 @@ class BaseLLM(ABC): if self.raise_exception: raise e yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e)) - return + break yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e)) await asyncio.sleep(i + 1) @@ -170,7 +169,7 @@ class BaseLLM(ABC): for i in range(self.max_retries): try: yield from self._stream_chat_sync(messages=messages, tools=tools, stream_kwargs=stream_kwargs) - return + break except Exception as e: logger.exception(f"Stream chat sync error (model={self.model_name}): {e.args}") @@ -179,7 +178,7 @@ class BaseLLM(ABC): if self.raise_exception: raise e yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e)) - return + break yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e)) time.sleep(i + 1) diff --git a/reme/core/op/__init__.py b/reme/core/op/__init__.py index f07b161c..c2ea5291 100644 --- a/reme/core/op/__init__.py +++ b/reme/core/op/__init__.py @@ -3,6 +3,7 @@ from .base_op import BaseOp from .base_ray_op import BaseRayOp from .base_react import BaseReact +from .base_react_stream import BaseReactStream from .base_tool import BaseTool from .mcp_tool import MCPTool from .parallel_op import ParallelOp @@ -13,6 +14,7 @@ __all__ = [ "BaseOp", "BaseRayOp", "BaseReact", + "BaseReactStream", "BaseTool", "MCPTool", "ParallelOp", diff --git a/reme/core/op/base_op.py b/reme/core/op/base_op.py index 8911cead..8592de37 100644 --- a/reme/core/op/base_op.py +++ b/reme/core/op/base_op.py @@ -50,7 +50,7 @@ class BaseOp(metaclass=ABCMeta): sub_ops: dict[str, "BaseOp"] | list["BaseOp"] | Optional["BaseOp"] = None, input_mapping: dict[str, str] | None = None, output_mapping: dict[str, str] | None = None, - enable_sync_thread_pool: bool = True, + enable_parallel: bool = False, max_retries: int = 1, raise_exception: bool = False, **kwargs, @@ -76,7 +76,7 @@ class BaseOp(metaclass=ABCMeta): self.input_mapping = input_mapping self.output_mapping = output_mapping - self.enable_sync_thread_pool = enable_sync_thread_pool + self.enable_parallel = enable_parallel # Control whether to execute tasks in parallel self.max_retries = max(1, max_retries) self.raise_exception = raise_exception self.op_params = kwargs @@ -233,7 +233,7 @@ class BaseOp(metaclass=ABCMeta): def submit_sync_task(self, fn: Callable, *args, **kwargs) -> "BaseOp": """Submit a task to the thread pool or local queue.""" - if self.enable_sync_thread_pool: + if self.enable_parallel: task = self.service_context.thread_pool.submit(fn, *args, **kwargs) else: task = (fn, args, kwargs) @@ -250,7 +250,7 @@ class BaseOp(metaclass=ABCMeta): """Wait for all pending sync tasks and return flattened results.""" results = [] for task in tqdm(self._pending_tasks, desc=task_desc or self.name): - if self.enable_sync_thread_pool: + if self.enable_parallel: result = task.result() else: result = task[0](*task[1], **task[2]) @@ -264,7 +264,20 @@ class BaseOp(metaclass=ABCMeta): async def join_async_tasks(self, return_exceptions: bool = True) -> list: """Wait for all pending async tasks and aggregate results.""" - raw_results = await asyncio.gather(*self._pending_tasks, return_exceptions=return_exceptions) + if self.enable_parallel: + raw_results = await asyncio.gather(*self._pending_tasks, return_exceptions=return_exceptions) + else: + raw_results = [] + for task in self._pending_tasks: + try: + result = await task + raw_results.append(result) + except Exception as e: + if return_exceptions: + raw_results.append(e) + else: + raise + results = [] for result in raw_results: if isinstance(result, Exception): diff --git a/reme/core/op/base_react.py b/reme/core/op/base_react.py index 38f3a42c..c6587204 100644 --- a/reme/core/op/base_react.py +++ b/reme/core/op/base_react.py @@ -71,7 +71,7 @@ class BaseReact(BaseOp): assistant_message: Message = await self.llm.chat(messages=messages, tools=tool_calls, **kwargs) messages.append(assistant_message) assistant_content: str = assistant_message.simple_dump(as_dict=False) - logger.info(f"[{self.__class__.__name__} {stage or ''} step{step + 1}] assistant={assistant_content}") + logger.info(f"[{self.__class__.__name__} {stage or ''} step{step}] assistant={assistant_content}") # Determine if tools should be called should_act = bool(assistant_message.tool_calls) @@ -95,7 +95,7 @@ class BaseReact(BaseOp): # Create tool name to tool instance mapping tool_dict = {t.tool_call.name: t for t in tools} for j, tool_call in enumerate(assistant_message.tool_calls): - prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step + 1}.{j}]" + prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step}.{j}]" if tool_call.name not in tool_dict: logger.warning(f"{prefix} unknown tool_call={tool_call.name}") continue @@ -125,7 +125,7 @@ class BaseReact(BaseOp): tool_call_id=tool.tool_call.id, ), ) - prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step + 1}.{j}]" + prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step}.{j}]" logger.info(f"{prefix} join tool={tool.name} result={tool.response.answer}") return tool_list, tool_messages @@ -153,7 +153,7 @@ class BaseReact(BaseOp): """Execute the ReAct agent and return final results.""" # Log available tools for i, tool in enumerate(self.tools): - logger.info(f"[{self.__class__.__name__}] tool_call[{i}]={tool.tool_call.simple_input_dump(as_dict=False)}") + logger.info(f"[{self.__class__.__name__}] {i}.tool_call={tool.tool_call.simple_input_dump(as_dict=False)}") # Build and log initial messages messages = await self.build_messages() @@ -163,8 +163,13 @@ class BaseReact(BaseOp): # Run ReAct loop t_tools, messages, success = await self.react(messages, self.tools) + + # Get the last assistant message as the final answer + assistant_messages = [m for m in messages if m.role == Role.ASSISTANT] + answer = assistant_messages[-1].content if assistant_messages else "" + return { - "answer": messages[-1].content if success else "", + "answer": answer, "success": success, "messages": messages, "tools": t_tools, diff --git a/reme/core/op/base_react_stream.py b/reme/core/op/base_react_stream.py new file mode 100644 index 00000000..c148d574 --- /dev/null +++ b/reme/core/op/base_react_stream.py @@ -0,0 +1,249 @@ +"""Base memory agent for handling memory operations with tool-based reasoning.""" + +import asyncio +from typing import TYPE_CHECKING + +from loguru import logger + +from ..enumeration import Role, ChunkEnum +from ..op import BaseOp +from ..schema import Message, StreamChunk + +if TYPE_CHECKING: + from . import BaseTool + + +class BaseReactStream(BaseOp): + """ReAct agent that performs reasoning and acting cycles with tools.""" + + def __init__( + self, + tools: list["BaseTool"], + tool_call_interval: float = 0, + max_steps: int = 10, + **kwargs, + ): + """Initialize ReAct agent with tools and execution parameters.""" + kwargs["sub_ops"] = tools or [] + super().__init__(**kwargs) + # Filter only BaseTool instances from sub_ops + from . import BaseTool + + self.sub_ops: list[BaseTool] = [t for t in self.sub_ops if isinstance(t, BaseTool)] + self.tool_call_interval: float = tool_call_interval + self.max_steps: int = max_steps + + @property + def tools(self) -> list["BaseTool"]: + """Return available tools for the agent.""" + return self.sub_ops + + def pop_tool(self, name: str) -> "BaseTool | None": + """Remove and return a tool from self.tools by name.""" + for i, tool in enumerate(self.sub_ops): + if tool.tool_call.name == name: + return self.sub_ops.pop(i) + return None + + async def build_messages(self) -> list[Message]: + """Build initial message list from context query or messages.""" + if self.context.get("query"): + messages = [Message(role=Role.USER, content=self.context.query)] + elif self.context.get("messages"): + messages = [Message(**m) if isinstance(m, dict) else m for m in self.context.messages] + else: + raise ValueError("input must have either `query` or `messages`") + return messages + + async def _reasoning_step( + self, + messages: list[Message], + tools: list["BaseTool"], + step: int, + stage: str = "", + **kwargs, + ) -> tuple[Message, bool]: + """Execute one reasoning step where LLM decides whether to use tools.""" + tool_calls = [t.tool_call for t in tools] + + start_chunk = StreamChunk(chunk_type=ChunkEnum.STEP_START, metadata={"step": step, "stage": stage}) + await self.context.add_stream_chunk(start_chunk) + + # State for accumulating message content from stream + state = { + "reasoning_content": "", + "content": "", + "tool_calls": [], + } + + async for stream_chunk in self.llm.stream_chat(messages=messages, tools=tool_calls, **kwargs): # noqa + if stream_chunk.chunk_type in [ChunkEnum.ANSWER, ChunkEnum.THINK, ChunkEnum.ERROR]: + await self.context.add_stream_chunk(stream_chunk) + + # Accumulate content based on chunk type + if stream_chunk.chunk_type is ChunkEnum.THINK: + state["reasoning_content"] += stream_chunk.chunk + + elif stream_chunk.chunk_type is ChunkEnum.ANSWER: + state["content"] += stream_chunk.chunk + + elif stream_chunk.chunk_type is ChunkEnum.TOOL: + state["tool_calls"].append(stream_chunk.chunk) + + # Build the final assistant message from accumulated state + assistant_message = Message(role=Role.ASSISTANT, **state) + messages.append(assistant_message) + logger.info( + f"[{self.__class__.__name__} {stage or ''} step{step}] " + f"assistant={assistant_message.simple_dump(as_dict=False)}", + ) + + should_act = bool(assistant_message.tool_calls) + return assistant_message, should_act + + async def _acting_step( + self, + assistant_message: Message, + tools: list["BaseTool"], + step: int, + stage: str = "", + **kwargs, + ) -> tuple[list["BaseTool"], list[Message]]: + """Execute tool calls serially and collect results with streaming output.""" + tool_list: list["BaseTool"] = [] + tool_messages: list[Message] = [] + + if not assistant_message.tool_calls: + return tool_list, tool_messages + + # Create tool name to tool instance mapping + tool_dict = {t.tool_call.name: t for t in tools} + + # Execute tools serially for better streaming experience + for j, tool_call in enumerate(assistant_message.tool_calls): + prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step}.{j}]" + if tool_call.name not in tool_dict: + logger.warning(f"{prefix} unknown tool_call={tool_call.name}") + # Emit error chunk for unknown tool + await self.context.add_stream_chunk( + StreamChunk( + chunk_type=ChunkEnum.ERROR, + chunk=f"Unknown tool: {tool_call.name}", + metadata={"step": step, "tool_index": j, "tool_name": tool_call.name}, + ), + ) + continue + + logger.info(f"{prefix} submit tool_call[{tool_call.name}] arguments={tool_call.arguments}") + + # Emit tool execution start signal + await self.context.add_stream_chunk( + StreamChunk( + chunk_type=ChunkEnum.TOOL, + chunk=f"Executing tool: {tool_call.name} {tool_call.arguments}", + metadata={ + "step": step, + "tool_index": j, + "tool_name": tool_call.name, + "arguments": tool_call.arguments, + }, + ), + ) + + # Create independent tool copy with unique ID + tool_copy: BaseTool = tool_dict[tool_call.name].copy() + tool_copy.tool_call.id = tool_call.id + tool_list.append(tool_copy) + + # Create isolated kwargs for each tool call to avoid parameter conflicts + tool_kwargs = {**kwargs, **tool_call.argument_dict} + + # Execute tool serially (wait for completion before next tool) + await tool_copy.call(service_context=self.service_context, **tool_kwargs) + + # Get tool result immediately after execution + tool_result = tool_copy.response.answer + tool_messages.append( + Message( + role=Role.TOOL, + content=tool_result, + tool_call_id=tool_copy.tool_call.id, + ), + ) + logger.info(f"{prefix} tool={tool_copy.name} result={tool_result}") + + await self.context.add_stream_chunk( + StreamChunk( + chunk_type=ChunkEnum.TOOL_RESULT, + chunk=tool_result, + metadata={ + "step": step, + "tool_index": j, + "tool_name": tool_copy.name, + "tool_call_id": tool_copy.tool_call.id, + }, + ), + ) + + # Optional interval between tool calls + if self.tool_call_interval > 0 and j < len(assistant_message.tool_calls) - 1: + await asyncio.sleep(self.tool_call_interval) + + return tool_list, tool_messages + + async def react(self, messages: list[Message], tools: list["BaseTool"], stage: str = ""): + """Run ReAct loop alternating between reasoning and acting until completion.""" + success: bool = False + used_tools: list[BaseTool] = [] + for step in range(self.max_steps): + # Reasoning: LLM decides next action + assistant_message, should_act = await self._reasoning_step(messages, tools, step=step, stage=stage) + + if not should_act: + # No tools requested, task complete + success = True + break + + # Acting: execute tools and collect results + t_tools, tool_messages = await self._acting_step(assistant_message, tools, step=step, stage=stage) + used_tools.extend(t_tools) + messages.extend(tool_messages) + + return used_tools, messages, success + + async def execute(self): + """Execute the ReAct agent with streaming output and return final results.""" + for i, tool in enumerate(self.tools): + logger.info(f"[{self.__class__.__name__}] {i}.tool_call={tool.tool_call.simple_input_dump(as_dict=False)}") + + # Build and log initial messages + messages = await self.build_messages() + for i, message in enumerate(messages): + role = message.name or message.role + logger.info(f"[{self.__class__.__name__}] role={role} {message.simple_dump(as_dict=False)}") + + # Run ReAct loop with streaming + t_tools, messages, success = await self.react(messages, self.tools) + + # Emit final done signal + await self.context.add_stream_chunk( + StreamChunk( + chunk_type=ChunkEnum.DONE, + chunk="", + metadata={ + "success": success, + "total_steps": len(t_tools), + }, + ), + ) + + # Get the last assistant message as the final answer + assistant_messages = [m for m in messages if m.role == Role.ASSISTANT] + answer = assistant_messages[-1].content if assistant_messages else "" + + return { + "answer": answer, + "success": success, + "messages": messages, + "tools": t_tools, + } diff --git a/reme/core/schema/service_config.py b/reme/core/schema/service_config.py index 30797d48..fa914f8c 100644 --- a/reme/core/schema/service_config.py +++ b/reme/core/schema/service_config.py @@ -123,7 +123,7 @@ class ServiceConfig(BaseModel): language: str = Field(default="") thread_pool_max_workers: int = Field(default=16) ray_max_workers: int = Field(default=-1) - init_logger: bool = Field(default=True) + log_to_console: bool = Field(default=True) disabled_flows: list[str] = Field(default_factory=list) enabled_flows: list[str] = Field(default_factory=list) mcp_servers: dict[str, dict] = Field(default_factory=dict) diff --git a/reme/core/service/http_service.py b/reme/core/service/http_service.py index 4aeab84b..ebaeba49 100644 --- a/reme/core/service/http_service.py +++ b/reme/core/service/http_service.py @@ -66,7 +66,7 @@ class HttpService(BaseService): stream_queue=stream_queue, task=task, task_name=tool_call.name, - as_bytes=True, + output_format="bytes", ): yield chunk diff --git a/reme/core/utils/common_utils.py b/reme/core/utils/common_utils.py index db12b43c..0b41383c 100644 --- a/reme/core/utils/common_utils.py +++ b/reme/core/utils/common_utils.py @@ -3,7 +3,7 @@ import asyncio import hashlib from collections.abc import AsyncGenerator, Coroutine -from typing import Any +from typing import Any, Literal import numpy as np from loguru import logger @@ -31,8 +31,8 @@ async def execute_stream_task( stream_queue: asyncio.Queue, task: asyncio.Task, task_name: str | None = None, - as_bytes: bool = False, -) -> AsyncGenerator[str | bytes, None]: + output_format: Literal["str", "bytes", "chunk"] = "str", +) -> AsyncGenerator[str | bytes | StreamChunk, None]: """ Core stream flow execution logic. @@ -43,12 +43,19 @@ async def execute_stream_task( stream_queue: Queue to receive StreamChunk objects from task: Background task executing the flow task_name: Optional flow name for logging purposes - as_bytes: If True, yield bytes for HTTP responses; if False, yield strings + output_format: Output format control + - "str": SSE-formatted string (default) + - "bytes": SSE-formatted bytes for HTTP responses + - "chunk": Raw StreamChunk objects Yields: - SSE-formatted data chunks (either str or bytes based on as_bytes) + - str: SSE-formatted data when output_format="str" + - bytes: SSE-formatted data when output_format="bytes" + - StreamChunk: Raw chunk objects when output_format="chunk" """ - done_msg = b"data:[DONE]\n\n" if as_bytes else "data:[DONE]\n\n" + is_raw_chunk = output_format == "chunk" + is_bytes = output_format == "bytes" + done_msg = b"data:[DONE]\n\n" if is_bytes else "data:[DONE]\n\n" try: while True: @@ -58,16 +65,29 @@ async def execute_stream_task( if get_chunk in done: chunk: StreamChunk = get_chunk.result() + + # Handle raw chunk mode + if is_raw_chunk: + yield chunk + if chunk.done: + break + continue + + # Handle SSE format mode if chunk.done: yield done_msg break data = f"data:{chunk.model_dump_json()}\n\n" - yield data.encode() if as_bytes else data + yield data.encode() if is_bytes else data else: # Task finished unexpectedly or raised exception await task - yield done_msg + if is_raw_chunk: + # Yield a DONE chunk in raw mode + yield StreamChunk(chunk_type=ChunkEnum.DONE, chunk="", done=True) + else: + yield done_msg break except Exception as e: @@ -75,9 +95,13 @@ async def execute_stream_task( logger.exception(log_msg) err = StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e), done=True) - err_data = f"data:{err.model_dump_json()}\n\n" - yield err_data.encode() if as_bytes else err_data - yield done_msg + + if is_raw_chunk: + yield err + else: + err_data = f"data:{err.model_dump_json()}\n\n" + yield err_data.encode() if is_bytes else err_data + yield done_msg finally: # Ensure task is cancelled if still running to avoid resource leaks diff --git a/reme/core/utils/logger_utils.py b/reme/core/utils/logger_utils.py index 1a4299de..a27af160 100644 --- a/reme/core/utils/logger_utils.py +++ b/reme/core/utils/logger_utils.py @@ -5,8 +5,14 @@ import sys from datetime import datetime -def init_logger(log_dir: str = "logs", level: str = "INFO") -> None: - """Initialize the logger with both file and console handlers.""" +def init_logger(log_dir: str = "logs", level: str = "INFO", log_to_console: bool = True) -> None: + """Initialize the logger with both file and console handlers. + + Args: + log_dir: Directory path for log files + level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) + log_to_console: Whether to print logs to console/screen + """ from loguru import logger # Remove default handler to avoid duplicate logs @@ -31,10 +37,11 @@ def init_logger(log_dir: str = "logs", level: str = "INFO") -> None: format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {file}:{line} | {function} | {message}", ) - # Configure colorized standard output logging - logger.add( - sink=sys.stdout, - level=level, - format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {file}:{line} | {function} | {message}", - colorize=True, - ) + # Configure colorized standard output logging if enabled + if log_to_console: + logger.add( + sink=sys.stdout, + level=level, + format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {file}:{line} | {function} | {message}", + colorize=True, + ) diff --git a/reme/reme.py b/reme/reme.py index 9a216fcd..7094bbb0 100644 --- a/reme/reme.py +++ b/reme/reme.py @@ -52,6 +52,7 @@ class ReMe(Application): embedding_api_key: str | None = None, embedding_api_base: str | None = None, enable_logo: bool = True, + log_to_console: bool = True, default_llm_config: dict | None = None, default_embedding_model_config: dict | None = None, default_vector_store_config: dict | None = None, @@ -71,6 +72,7 @@ class ReMe(Application): embedding_api_key: API key for embedding provider embedding_api_base: API base for embedding provider enable_logo: Enable logo + log_to_console: Log to console default_llm_config: LLM configuration default_embedding_model_config: Embedding model configuration default_vector_store_config: Vector store configuration @@ -101,6 +103,7 @@ class ReMe(Application): embedding_api_key=embedding_api_key, embedding_api_base=embedding_api_base, enable_logo=enable_logo, + log_to_console=log_to_console, parser=ReMeConfigParser, default_llm_config=default_llm_config, default_embedding_model_config=default_embedding_model_config, diff --git a/reme/reme_fs.py b/reme/reme_fs.py index 402e66e0..1d465a4d 100644 --- a/reme/reme_fs.py +++ b/reme/reme_fs.py @@ -1,13 +1,20 @@ """ReMe File System""" +import asyncio +import sys from pathlib import Path +from typing import AsyncGenerator +from prompt_toolkit import PromptSession + +from reme.core.utils import execute_stream_task +from .agent.chat import FsCli from .agent.fs import FsCompactor, FsSummarizer from .config import ReMeConfigParser from .core import Application -from .core.enumeration import MemorySource +from .core.enumeration import MemorySource, ChunkEnum from .core.op import BaseTool -from .core.schema import Message +from .core.schema import Message, StreamChunk from .tool.fs import ( BashTool, EditTool, @@ -32,6 +39,7 @@ class ReMeFs(Application): embedding_api_key: str | None = None, embedding_api_base: str | None = None, enable_logo: bool = True, + log_to_console: bool = True, default_llm_config: dict | None = None, default_embedding_model_config: dict | None = None, default_memory_store_config: dict | None = None, @@ -48,6 +56,7 @@ class ReMeFs(Application): embedding_api_key=embedding_api_key, embedding_api_base=embedding_api_base, enable_logo=enable_logo, + log_to_console=log_to_console, parser=ReMeConfigParser, default_llm_config=default_llm_config, default_embedding_model_config=default_embedding_model_config, @@ -70,6 +79,11 @@ class ReMeFs(Application): self.working_path: Path = Path(self.working_dir) self.working_path.mkdir(parents=True, exist_ok=True) + self.commands = [ + "/new", + "/exit", + ] + async def compact( self, messages: list[Message | dict], @@ -139,3 +153,120 @@ class ReMeFs(Application): """Read specific snippets from memory files.""" get_tool = FsMemoryGet(workspace_dir=self.working_dir) return await get_tool.call(path=path, offset=offset, limit=limit, service_context=self.service_context) + + async def chat_with_remy(self, tool_result_max_size: int = 100): + """Interactive CLI chat with Remy using simple streaming output.""" + fs_cli = FsCli(working_dir=self.working_dir, tools=self.fs_tools) + session = PromptSession() + + # Print welcome banner + print("\n========================================") + print(" Welcome to Remy Chat!") + print(" Type /exit to quit, /new to start fresh.") + print("========================================\n") + + async def chat(q: str) -> AsyncGenerator[StreamChunk, None]: + """Execute chat query and yield streaming chunks.""" + stream_queue = asyncio.Queue() + task = asyncio.create_task( + fs_cli.call(query=q, stream_queue=stream_queue, service_context=self.service_context), + ) + async for _chunk in execute_stream_task( + stream_queue=stream_queue, + task=task, + task_name="cli", + output_format="chunk", + ): + yield _chunk + + while True: + try: + # Get user input (async) + user_input = await session.prompt_async("You: ", default="") + if not user_input.strip(): + continue + + # Handle commands + if user_input.strip() == "/exit": + break + + if user_input.strip() == "/new": + fs_cli.reset_history() + print("Conversation reset.\n") + continue + + # Stream processing state + in_thinking = False + in_answer = False + + try: + async for chunk in chat(user_input): + if chunk.chunk_type == ChunkEnum.THINK: + if not in_thinking: + print("\033[90mThinking: ", end="", flush=True) + in_thinking = True + print(chunk.chunk, end="", flush=True) + + elif chunk.chunk_type == ChunkEnum.ANSWER: + if in_thinking: + print("\033[0m") # reset color after thinking + in_thinking = False + if not in_answer: + print("\nRemy: ", end="", flush=True) + in_answer = True + print(chunk.chunk, end="", flush=True) + + elif chunk.chunk_type == ChunkEnum.TOOL: + if in_thinking: + print("\033[0m") # reset color after thinking + in_thinking = False + print(f"\033[36m -> Tool: {chunk.chunk}\033[0m") + + elif chunk.chunk_type == ChunkEnum.TOOL_RESULT: + tool_name = chunk.metadata.get("tool_name", "unknown") + result = chunk.chunk + if len(result) > tool_result_max_size: + result = result[:tool_result_max_size] + f"... ({len(chunk.chunk)} chars total)" + print(f"\033[36m Tool result for {tool_name}: {result.strip()}\033[0m") + + elif chunk.chunk_type == ChunkEnum.ERROR: + print(f"\n Error: {chunk.chunk}") + + elif chunk.chunk_type == ChunkEnum.DONE: + break + + except Exception as e: + print(f"\nStream error: {e}") + + # End current streaming line + print("\n") + print("----------------------------------------\n") + + except EOFError: + break + except KeyboardInterrupt: + print("\nInterrupted.") + break + except Exception as e: + print(f"Error: {e}") + import traceback + + traceback.print_exc() + + print("\nGoodbye!\n") + + +async def async_main(): + """Main function for testing the ReMeFs CLI.""" + reme = ReMeFs(*sys.argv[1:], log_to_console=False) + await reme.start() + await reme.chat_with_remy() + + +def main(): + """Main function for testing the ReMeFs CLI.""" + asyncio.run(async_main()) + + +if __name__ == "__main__": + main() diff --git a/tests/test_tool.py b/tests/test_tool.py index 6c23ad18..b4639776 100644 --- a/tests/test_tool.py +++ b/tests/test_tool.py @@ -182,7 +182,7 @@ async def test_stream_chat(app): stream_queue=context.stream_queue, task=asyncio.create_task(task()), task_name="test_stream_chat", - as_bytes=False, + output_format="str", ): print(chunk, end="")