Agent

.env.example

@@ -73,6 +73,8 @@ AGENT_MAX_STEPS=30
 AGENT_DEFAULT_ENABLED=true
 # Agent mode availability (renamed to align with other FEATURE_* flags)
 FEATURE_AGENT_MODE_AVAILABLE=true
+# Agent loop strategy: react (structured reasoning) or think-act (faster, concise)
+AGENT_LOOP_STRATEGY=think-act
 # (Adjust above to stage rollouts. For a bare-bones chat set them all to false.)
 
 APP_LOG_DIR=/workspaces/atlas-ui-3-11/logs

CLAUDE.md

@@ -14,7 +14,9 @@ Atlas UI 3 is a full-stack LLM chat interface with Model Context Protocol (MCP)
 
 # Style note
 
-No Emojis should ever be added in this repo. If you find one, then remove it. 
+No Emojis should ever be added in this repo. If you find one, then remove it.
+
+**File Naming**: Do not use generic names like `main.py`, `cli.py`, `utils.py`, or `helpers.py`. Use descriptive names that reflect the file's purpose (e.g., `chat_service.py`, `mcp_tool_manager.py`, `websocket_handler.py`). Exception: top-level entry points like `backend/main.py` are acceptable. 
 
 # Tests
 
@@ -166,9 +168,10 @@ backend/
 
 1. **Protocol-Based Dependency Injection**: Uses Python `Protocol` (structural subtyping) instead of ABC inheritance for loose coupling
 
-2. **Agent Loop Strategy Pattern**: Two implementations selectable via `APP_AGENT_LOOP_STRATEGY`:
-   - `ReActAgentLoop`: Reasoning-Act (faster, better for tools)
-   - `ThinkActAgentLoop`: Extended thinking (slower, complex reasoning)
+2. **Agent Loop Strategy Pattern**: Three implementations selectable via `APP_AGENT_LOOP_STRATEGY`:
+   - `react`: Reason-Act-Observe cycle (structured reasoning)
+   - `think-act`: Extended thinking (slower, complex reasoning)
+   - `act`: Pure action loop (fastest, minimal overhead)
 
 3. **MCP Transport Auto-Detection**: Automatically detects stdio, HTTP, or SSE based on config
 
@@ -232,9 +235,10 @@ MCP servers defined in `config/defaults/mcp.json`. The backend:
 4. Supports group-based access control
 
 ### Agent Modes
-Two agent loop strategies implement different reasoning patterns:
-- **ReAct** (`backend/application/chat/agent/react_agent_loop.py`): Fast iteration, good for tool-heavy tasks
-- **Think-Act** (`backend/application/chat/agent/think_act_agent_loop.py`): Deep reasoning, slower but more thoughtful
+Three agent loop strategies implement different reasoning patterns:
+- **ReAct** (`backend/application/chat/agent/react_loop.py`): Reason-Act-Observe cycle, good for tool-heavy tasks with structured reasoning
+- **Think-Act** (`backend/application/chat/agent/think_act_loop.py`): Deep reasoning with explicit thinking steps, slower but more thoughtful
+- **Act** (`backend/application/chat/agent/act_loop.py`): Pure action loop without explicit reasoning steps, fastest with minimal overhead. LLM calls tools directly and signals completion via the "finished" tool
 
 ### File Storage
 S3-compatible storage via `backend/modules/file_storage/s3_client.py`:

backend/application/chat/agent/__init__.py

@@ -3,3 +3,4 @@
 from .protocols import AgentLoopProtocol, AgentContext, AgentEvent, AgentResult, AgentEventHandler
 from .react_loop import ReActAgentLoop
 from .think_act_loop import ThinkActAgentLoop
+from .factory import AgentLoopFactory

backend/application/chat/agent/act_loop.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+from interfaces.llm import LLMProtocol
+from interfaces.tools import ToolManagerProtocol
+from modules.prompts.prompt_provider import PromptProvider
+
+from .protocols import AgentContext, AgentEvent, AgentEventHandler, AgentLoopProtocol, AgentResult
+from ..utilities import error_utils, tool_utils
+
+
+class ActAgentLoop(AgentLoopProtocol):
+    """Pure action agent loop - just execute tools in a loop until done.
+
+    No explicit reasoning or observation steps. The LLM directly decides which
+    tools to call and when to finish. Fastest strategy with minimal overhead.
+
+    Exit conditions:
+    - LLM calls the "finished" tool with a final_answer
+    - No tool calls returned (LLM provides text response)
+    - Max steps reached
+    """
+
+    def __init__(
+        self,
+        *,
+        llm: LLMProtocol,
+        tool_manager: Optional[ToolManagerProtocol],
+        prompt_provider: Optional[PromptProvider],
+        connection: Any = None,
+    ) -> None:
+        self.llm = llm
+        self.tool_manager = tool_manager
+        self.prompt_provider = prompt_provider
+        self.connection = connection
+
+    def _extract_finished_args(self, tool_calls: List[Dict[str, Any]]) -> Optional[str]:
+        """Extract final_answer from finished tool call if present."""
+        try:
+            for tc in tool_calls:
+                f = tc.get("function") if isinstance(tc, dict) else None
+                if f and f.get("name") == "finished":
+                    raw_args = f.get("arguments")
+                    if isinstance(raw_args, str):
+                        try:
+                            args = json.loads(raw_args)
+                            return args.get("final_answer")
+                        except Exception:
+                            return None
+                    if isinstance(raw_args, dict):
+                        return raw_args.get("final_answer")
+            return None
+        except Exception:
+            return None
+
+    async def run(
+        self,
+        *,
+        model: str,
+        messages: List[Dict[str, Any]],
+        context: AgentContext,
+        selected_tools: Optional[List[str]],
+        data_sources: Optional[List[str]],
+        max_steps: int,
+        temperature: float,
+        event_handler: AgentEventHandler,
+    ) -> AgentResult:
+        await event_handler(AgentEvent(type="agent_start", payload={"max_steps": max_steps, "strategy": "act"}))
+
+        steps = 0
+        final_answer: Optional[str] = None
+
+        # Define the "finished" control tool
+        finished_tool_schema = {
+            "type": "function",
+            "function": {
+                "name": "finished",
+                "description": "Call this when you have completed the task and are ready to provide a final answer to the user.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "final_answer": {
+                            "type": "string",
+                            "description": "The final response to provide to the user",
+                        },
+                    },
+                    "required": ["final_answer"],
+                    "additionalProperties": False,
+                },
+            },
+        }
+
+        while steps < max_steps and final_answer is None:
+            steps += 1
+            await event_handler(AgentEvent(type="agent_turn_start", payload={"step": steps}))
+
+            # Build tools schema: user tools + finished tool
+            tools_schema: List[Dict[str, Any]] = [finished_tool_schema]
+            if selected_tools and self.tool_manager:
+                user_tools = await error_utils.safe_get_tools_schema(self.tool_manager, selected_tools)
+                tools_schema.extend(user_tools)
+
+            # Call LLM with tools - using "required" to force tool calling during Act phase
+            # The LiteLLM caller has fallback logic to "auto" if "required" is not supported
+            if data_sources and context.user_email:
+                llm_response = await self.llm.call_with_rag_and_tools(
+                    model, messages, data_sources, tools_schema, context.user_email, "required", temperature=temperature
+                )
+            else:
+                llm_response = await self.llm.call_with_tools(
+                    model, messages, tools_schema, "required", temperature=temperature
+                )
+
+            # Process response
+            if llm_response.has_tool_calls():
+                tool_calls = llm_response.tool_calls or []
+
+                # Check if finished tool was called
+                final_answer = self._extract_finished_args(tool_calls)
+                if final_answer:
+                    break
+
+                # Execute first non-finished tool call
+                first_call = None
+                for tc in tool_calls:
+                    f = tc.get("function") if isinstance(tc, dict) else None
+                    if f and f.get("name") != "finished":
+                        first_call = tc
+                        break
+
+                if first_call is None:
+                    # Only finished tool or no valid tools
+                    final_answer = llm_response.content or "Task completed."
+                    break
+
+                # Execute the tool
+                messages.append({
+                    "role": "assistant",
+                    "content": llm_response.content,
+                    "tool_calls": [first_call],
+                })
+
+                result = await tool_utils.execute_single_tool(
+                    tool_call=first_call,
+                    session_context={
+                        "session_id": context.session_id,
+                        "user_email": context.user_email,
+                        "files": context.files,
+                    },
+                    tool_manager=self.tool_manager,
+                    update_callback=(self.connection.send_json if self.connection else None),
+                )
+
+                messages.append({
+                    "role": "tool",
+                    "content": result.content,
+                    "tool_call_id": result.tool_call_id,
+                })
+
+                # Emit tool results for artifact ingestion
+                await event_handler(AgentEvent(type="agent_tool_results", payload={"results": [result]}))
+            else:
+                # No tool calls - treat content as final answer
+                final_answer = llm_response.content or "Task completed."
+                break
+
+        # Fallback if no final answer after max steps
+        if not final_answer:
+            final_answer = await self.llm.call_plain(model, messages, temperature=temperature)
+
+        await event_handler(AgentEvent(type="agent_completion", payload={"steps": steps}))
+        return AgentResult(final_answer=final_answer, steps=steps, metadata={"agent_mode": True, "strategy": "act"})

backend/application/chat/agent/factory.py

@@ -0,0 +1,135 @@
+"""Factory for creating agent loop instances based on strategy."""
+
+import logging
+from typing import Optional
+
+from interfaces.llm import LLMProtocol
+from interfaces.tools import ToolManagerProtocol
+from interfaces.transport import ChatConnectionProtocol
+from modules.prompts.prompt_provider import PromptProvider
+
+from .protocols import AgentLoopProtocol
+from .react_loop import ReActAgentLoop
+from .think_act_loop import ThinkActAgentLoop
+from .act_loop import ActAgentLoop
+
+logger = logging.getLogger(__name__)
+
+
+class AgentLoopFactory:
+    """
+    Factory for creating agent loop instances.
+
+    This factory pattern allows for easy addition of new agent loop strategies
+    without modifying existing code. Simply add a new strategy to the registry.
+    """
+
+    def __init__(
+        self,
+        llm: LLMProtocol,
+        tool_manager: Optional[ToolManagerProtocol] = None,
+        prompt_provider: Optional[PromptProvider] = None,
+        connection: Optional[ChatConnectionProtocol] = None,
+    ):
+        """
+        Initialize factory with shared dependencies.
+
+        Args:
+            llm: LLM protocol implementation
+            tool_manager: Optional tool manager
+            prompt_provider: Optional prompt provider
+            connection: Optional connection for sending updates
+        """
+        self.llm = llm
+        self.tool_manager = tool_manager
+        self.prompt_provider = prompt_provider
+        self.connection = connection
+
+        # Registry of available strategies
+        self._strategy_registry = {
+            "react": ReActAgentLoop,
+            "think-act": ThinkActAgentLoop,
+            "think_act": ThinkActAgentLoop,
+            "thinkact": ThinkActAgentLoop,
+            "act": ActAgentLoop,
+        }
+
+        # Cache of instantiated loops for performance
+        self._loop_cache: dict[str, AgentLoopProtocol] = {}
+
+    def create(self, strategy: str = "think-act") -> AgentLoopProtocol:
+        """
+        Create an agent loop instance for the given strategy.
+
+        Args:
+            strategy: Strategy name (react, think-act, act, etc.)
+
+        Returns:
+            AgentLoopProtocol instance
+
+        Note:
+            If the strategy is not recognized, falls back to 'react' with a warning.
+        """
+        strategy_normalized = strategy.lower().strip()
+
+        # Check cache first
+        if strategy_normalized in self._loop_cache:
+            logger.info(f"Using agent loop strategy: {strategy_normalized}")
+            return self._loop_cache[strategy_normalized]
+
+        # Look up strategy in registry
+        loop_class = self._strategy_registry.get(strategy_normalized)
+
+        if loop_class is None:
+            logger.warning(
+                f"Unknown agent loop strategy '{strategy}', falling back to 'react'"
+            )
+            loop_class = self._strategy_registry["react"]
+            strategy_normalized = "react"
+
+        # Instantiate the loop
+        loop_instance = loop_class(
+            llm=self.llm,
+            tool_manager=self.tool_manager,
+            prompt_provider=self.prompt_provider,
+            connection=self.connection,
+        )
+
+        # Cache for future use
+        self._loop_cache[strategy_normalized] = loop_instance
+
+        logger.info(f"Created and using agent loop strategy: {strategy_normalized}")
+        return loop_instance
+
+    def get_available_strategies(self) -> list[str]:
+        """
+        Get list of available strategy names.
+
+        Returns:
+            List of strategy identifiers
+        """
+        # Return unique strategy names (deduplicated)
+        unique_strategies = set()
+        for strategy in self._strategy_registry.keys():
+            # Normalize to primary name
+            if strategy in ("react",):
+                unique_strategies.add("react")
+            elif strategy in ("think-act", "think_act", "thinkact"):
+                unique_strategies.add("think-act")
+            elif strategy in ("act",):
+                unique_strategies.add("act")
+        return sorted(unique_strategies)
+
+    def register_strategy(self, name: str, loop_class: type[AgentLoopProtocol]) -> None:
+        """
+        Register a new agent loop strategy.
+
+        This allows for dynamic extension of available strategies.
+
+        Args:
+            name: Strategy identifier
+            loop_class: Agent loop class to instantiate
+        """
+        name_normalized = name.lower().strip()
+        self._strategy_registry[name_normalized] = loop_class
+        logger.info(f"Registered new agent loop strategy: {name_normalized}")

backend/application/chat/agent/react_loop.py

@@ -100,7 +100,7 @@ async def run(
         event_handler: AgentEventHandler,
     ) -> AgentResult:
         # Agent start
-        await event_handler(AgentEvent(type="agent_start", payload={"max_steps": max_steps}))
+        await event_handler(AgentEvent(type="agent_start", payload={"max_steps": max_steps, "strategy": "react"}))
 
         steps = 0
         final_response: Optional[str] = None
@@ -210,14 +210,16 @@ async def run(
                 tools_schema = await error_utils.safe_get_tools_schema(self.tool_manager, selected_tools)
 
             tool_results: List[ToolResult] = []
+            # Use "required" to force tool calling during Act phase
+            # The LiteLLM caller has fallback logic to "auto" if "required" is not supported
             if tools_schema:
                 if data_sources and context.user_email:
                     llm_response = await self.llm.call_with_rag_and_tools(
-                        model, messages, data_sources, tools_schema, context.user_email, "auto", temperature=temperature
+                        model, messages, data_sources, tools_schema, context.user_email, "required", temperature=temperature
                     )
                 else:
                     llm_response = await self.llm.call_with_tools(
-                        model, messages, tools_schema, "auto", temperature=temperature
+                        model, messages, tools_schema, "required", temperature=temperature
                     )
 
                 if llm_response.has_tool_calls():