OpenManus Architecture

class ReActAgent(BaseAgent, ABC):
    @abstractmethod
    async def think(self) -> bool:
        """Process the current state and decide the next action."""

    @abstractmethod
    async def act(self) -> str:
        """Execute the decided actions."""

    async def step(self) -> str:
        """Execute a single step: think and act."""
        should_act = await self.think()
        if not should_act:
            return "Thinking complete - no action needed"
        return await self.act()

class ToolCallAgent(ReActAgent):
    """Base agent class for handling tool/function calls with enhanced abstraction"""

    available_tools: ToolCollection = ToolCollection(
        CreateChatCompletion(), Terminate()
    )
    tool_choices: Literal["none", "auto", "required"] = "auto"

    async def think(self) -> bool:
        # Get the LLM response and tool selection
        response = await self.llm.ask_tool(
            messages=self.messages,
            system_msgs=[Message.system_message(self.system_prompt)]
            if self.system_prompt
            else None,
            tools=self.available_tools.to_params(),
            tool_choice=self.tool_choices,
        )
        self.tool_calls = response.tool_calls

        # Process the response and tool calls
        # ...

    async def act(self) -> str:
        # Execute tool calls
        results = []
        for command in self.tool_calls:
            result = await self.execute_tool(command)
            # Add tool response to memory
            # ...
            results.append(result)

        return "\n\n".join(results)

class PlanningAgent(ToolCallAgent):
    """
    An agent that creates and manages plans to solve tasks.
    This agent uses a planning tool to create and manage structured plans,
    and tracks progress through individual steps until task completion.
    """
    name: str = "planning"
    description: str = "An agent that creates and manages plans to solve tasks"
    system_prompt: str = PLANNING_SYSTEM_PROMPT
    next_step_prompt: str = NEXT_STEP_PROMPT
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(PlanningTool(), Terminate())
    )

    # Step execution tracker
    step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
    current_step_index: Optional[int] = None

    async def think(self) -> bool:
        """Decide the next action based on plan status."""
        prompt = (
            f"CURRENT PLAN STATUS:\n{await self.get_plan()}\n\n{self.next_step_prompt}"
            if self.active_plan_id
            else self.next_step_prompt
        )
        self.messages.append(Message.user_message(prompt))

        # Get the current step index
        self.current_step_index = await self._get_current_step_index()
        result = await super().think()
        # Associate tool calls with the current step
        if result and self.tool_calls:
            # ...association logic...
        return result

class Manus(ToolCallAgent):
    """
    A versatile general-purpose agent that uses planning to solve various tasks.
    This agent extends PlanningAgent with a comprehensive set of tools and capabilities,
    including Python execution, web browsing, file operations, and information retrieval
    to handle a wide range of user requests.
    """
    name: str = "manus"
    description: str = "A versatile general-purpose agent"
    system_prompt: str = SYSTEM_PROMPT
    next_step_prompt: str = NEXT_STEP_PROMPT
    available_tools: ToolCollection = Field(
        default_factory=lambda: ToolCollection(
            PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
        )
    )