feat(chat): add FsCli chat agent with streaming capabilities

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"

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)

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,
+        }

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.

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
 

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
 

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)

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"

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)

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",

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):

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,

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,
+        }

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)

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
 

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

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,
+        )

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,

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()

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="")