Código aberto do OpenManus explicado, com insights sobre a arquitetura de agente de IA por trás dele

Abertura: O incêndio em Manus e o colapso do OpenManus

Uma das grandes coisas que têm acontecido nos círculos de IA ultimamente é que a Manus Manus, o agente de IA, atraiu muita atenção com suas funções avançadas e uso flexível. Simplificando, o Manus é como um assistente completo, seja para programar, procurar informações, processar documentos ou navegar na Internet, ele pode ajudá-lo a fazer tudo.

No entanto, não é tão fácil usar o Manus, você precisa ter um código de convite. Isso impediu muitos desenvolvedores e pesquisadores de usar o Manus. A equipe do MetaGPT formada por @mannaandpoem, @XiangJinyu, @MoshiQAQ e @didiforgithub gastou apenas 3 horas para criar um projeto de código aberto chamado OpenManus. Agora você pode experimentar o poder da Manus sem precisar de um código de convite! O que é ainda mais interessante é que o OpenManus é um projeto de código aberto, o que significa que você pode modificá-lo e estendê-lo para atender às suas necessidades!

O surgimento do OpenManus não só dá a mais pessoas a oportunidade de experimentar o charme do AI Agent, mas também injeta nova vitalidade no desenvolvimento do AI Agent. Para nós que estamos envolvidos com tecnologia, o OpenManus não é apenas uma boa ferramenta, mas também um excelente recurso de aprendizado. Ao estudar seu código, podemos obter uma compreensão mais profunda do design da estrutura e dos detalhes de implementação do AI Agent.

Estrutura do agente de IA: filosofia de design do OpenManus

A estrutura de código do OpenManus é muito clara e adota um design modular, que é como blocos de construção para combinar diferentes módulos funcionais. A vantagem desse design é que o código é altamente reutilizável e extensível, e as responsabilidades de cada módulo são claras.

Os componentes principais do OpenManus incluem:

OpenManus
├── Agent (代理层)
│   ├── BaseAgent (基础抽象类)
│   ├── ReActAgent (思考-行动模式)
│   ├── ToolCallAgent (工具调用能力)
│   ├── PlanningAgent (规划能力)
│   ├── SWEAgent (软件工程能力)
│   └── Manus (通用代理)
├── LLM (语言模型层)
├── Memory (记忆层)
├── Tool (工具层)
│   ├── BaseTool (工具基类)
│   ├── PlanningTool (规划工具)
│   ├── PythonExecute (Python 执行)
│   ├── GoogleSearch (搜索工具)
│   ├── BrowserUseTool (浏览器工具)
│   └── ... (其他工具)
├── Flow (工作流层)
│   ├── BaseFlow (基础流程)
│   └── PlanningFlow (规划流程)
└── Prompt (提示词层)

Componentes do LLM: O cérebro do agente

Se compararmos um agente a uma pessoa, então o LLM (Large Language Model) é o cérebro do agente, responsável por entender os comandos do usuário, gerar respostas e tomar decisões. O OpenManus encapsula a interação com o LLM por meio da classe LLM.

class LLM:
_instances: Dict[str, "LLM"] = {}  # 单例模式实现
def __init__(
self, config_name: str = "default", llm_config: Optional[LLMSettings] = None
):
if not hasattr(self, "client"):  # 只初始化一次
llm_config = llm_config or config.llm
llm_config = llm_config.get(config_name, llm_config["default"])
self.model = llm_config.model
self.max_tokens = llm_config.max_tokens
self.temperature = llm_config.temperature
self.client = AsyncOpenAI(
api_key=llm_config.api_key, base_url=llm_config.base_url
)

A classe LLM oferece dois métodos principais:

• ask:: Envio de solicitações gerais de diálogo
• ask_tool:: Enviar solicitações com chamadas de ferramentas

async def ask_tool(
self,
messages: List[Union[dict, Message]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
timeout: int = 60,
tools: Optional[List[dict]] = None,
tool_choice: Literal["none", "auto", "required"] = "auto",
temperature: Optional[float] = None,
**kwargs,
):
# 格式化消息
if system_msgs:
system_msgs = self.format_messages(system_msgs)
messages = system_msgs + self.format_messages(messages)
else:
messages = self.format_messages(messages)
# 发送请求
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature or self.temperature,
max_tokens=self.max_tokens,
tools=tools,
tool_choice=tool_choice,
timeout=timeout,
**kwargs,
)

Componente de memória: memória do agente

O componente Memória é como um caderno para o Agente, responsável por registrar e gerenciar o histórico de diálogo do Agente. Com a Memória, o Agente pode se lembrar do que disse antes e manter a consistência do diálogo.

class Memory(BaseModel):
"""Stores and manages agent's conversation history."""
messages: List[Message] = Field(default_factory=list)
def add_message(self, message: Union[Message, dict]) -> None:
"""Add a message to memory."""
if isinstance(message, dict):
message = Message(**message)
self.messages.append(message)
def get_messages(self) -> List[Message]:
"""Get all messages in memory."""
return self.messages

O componente de memória é o componente principal do agente e é acessado por meio do componente update_memory para adicionar uma nova mensagem:

def update_memory(
self,
role: Literal["user", "system", "assistant", "tool"],
content: str,
**kwargs,
) -> None:
"""Add a message to the agent's memory."""
message_map = {
"user": Message.user_message,
"system": Message.system_message,
"assistant": Message.assistant_message,
"tool": lambda content, **kw: Message.tool_message(content, **kw),
}
if role not in message_map:
raise ValueError(f"Unsupported message role: {role}")
msg_factory = message_map[role]
msg = msg_factory(content, **kwargs) if role == "tool" else msg_factory(content)
self.memory.add_message(msg)

Componente de ferramentas: caixa de ferramentas do agente

O componente Ferramentas é a ponte entre o Agente e o mundo externo, e o OpenManus implementa um sistema de ferramentas flexível que permite que o Agente invoque uma variedade de ferramentas para realizar tarefas.

class BaseTool(ABC, BaseModel):
name: str
description: str
parameters: Optional[dict] = None
async def __call__(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
return await self.execute(**kwargs)
    @abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
def to_param(self) -> Dict:
"""Convert tool to function call format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}

Os resultados da execução da ferramenta são representados pelo ToolResult para representá-la:

class ToolResult(BaseModel):
"""Represents the result of a tool execution."""
output: Any = Field(default=None)
error: Optional[str] = Field(default=None)
system: Optional[str] = Field(default=None)

O OpenManus fornece várias ferramentas integradas, como PlanningTool::

class PlanningTool(BaseTool):
"""
A planning tool that allows the agent to create and manage plans for solving complex tasks.
The tool provides functionality for creating plans, updating plan steps, and tracking progress.
"""
name: str = "planning"
description: str = _PLANNING_TOOL_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"command": {
"description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
"enum": [
"create",
"update",
"list",
"get",
"set_active",
"mark_step",
"delete",
],
"type": "string",
},
# 其他参数...
},
"required": ["command"],
}

Componente de planejamento: recursos de planejamento do agente

O componente Planejamento é a chave para a capacidade do OpenManus de lidar com tarefas complexas. Ele permite que os agentes planejem e dividam tarefas complexas em pequenas tarefas passo a passo e, em seguida, concluam-nas uma a uma.

O componente Planejamento consiste em duas partes principais:

1. PlanningToolCriação de planos: fornece recursos de criação, atualização e rastreamento de planos.
2. PlanningAgent: Uso PlanningTool para realizar o planejamento e a execução de tarefas.

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: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None

PlanningAgent A metodologia principal inclui:

async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

Componente de fluxo: recursos de colaboração para agentes

A função do componente Flow é coordenar vários agentes que trabalham juntos para realizar tarefas mais complexas.

class BaseFlow(BaseModel, ABC):
"""Base class for execution flows supporting multiple agents"""
agents: Dict[str, BaseAgent]
tools: Optional[List] = None
primary_agent_key: Optional[str] = None
    @property
def primary_agent(self) -> Optional[BaseAgent]:
"""Get the primary agent for the flow"""
return self.agents.get(self.primary_agent_key)
    @abstractmethod
async def execute(self, input_text: str) -> str:
"""Execute the flow with given input"""

PlanningFlow é uma implementação concreta do Flow para planejamento e execução de tarefas:

class PlanningFlow(BaseFlow):
"""A flow that manages planning and execution of tasks using agents."""
llm: LLM = Field(default_factory=lambda: LLM())
planning_tool: PlanningTool = Field(default_factory=PlanningTool)
executor_keys: List[str] = Field(default_factory=list)
active_plan_id: str = Field(default_factory=lambda: f"plan_{int(time.time())}")
current_step_index: Optional[int] = None
async def execute(self, input_text: str) -> str:
"""Execute the planning flow with agents."""
try:
# 创建初始计划
if input_text:
await self._create_initial_plan(input_text)
# 执行计划步骤
while await self._has_next_step():
# 获取当前步骤
step_info = await self._get_current_step()
# 选择合适的执行者
executor = self.get_executor(step_info.get("type"))
# 执行步骤
result = await self._execute_step(executor, step_info)
# 更新步骤状态
await self._update_step_status(step_info["index"], "completed")
# 完成计划
return await self._finalize_plan()
except Exception as e:
# 处理异常
return f"Error executing flow: {str(e)}"

Implementação de agente do OpenManus: uma arquitetura em camadas

O agente do OpenManus usa uma arquitetura hierárquica, construída camada por camada, desde a funcionalidade básica até aplicativos especializados. Os benefícios desse design são a alta reutilização do código, a alta escalabilidade e as responsabilidades claras em cada nível.

BaseAgent (抽象基类)
└── ReActAgent (思考-行动模式)
└── ToolCallAgent (工具调用能力)
├── PlanningAgent (规划能力)
├── SWEAgent (软件工程能力)
└── Manus (通用代理)

BaseAgent: a base das bases

BaseAgent é a base de toda a estrutura e define as principais propriedades e métodos de um agente:

class BaseAgent(BaseModel, ABC):
"""Abstract base class for managing agent state and execution."""
# 核心属性
name: str = Field(..., description="Unique name of the agent")
description: Optional[str] = Field(None, description="Optional agent description")
# 提示词
system_prompt: Optional[str] = Field(None, description="System-level instruction prompt")
next_step_prompt: Optional[str] = Field(None, description="Prompt for determining next action")
# 依赖组件
llm: LLM = Field(default_factory=LLM, description="Language model instance")
memory: Memory = Field(default_factory=Memory, description="Agent's memory store")
state: AgentState = Field(default=AgentState.IDLE, description="Current agent state")
# 执行控制
max_steps: int = Field(default=10, description="Maximum steps before termination")
current_step: int = Field(default=0, description="Current step in execution")

ReActAgent: o agente pensante

ReActAgent Um modelo "pensar-agir" foi realizado, dividindo o processo de execução do agente em duas fases:

class ReActAgent(BaseAgent, ABC):
    @abstractmethod
async def think(self) -> bool:
"""Process current state and decide next action"""
    @abstractmethod
async def act(self) -> str:
"""Execute 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()

ToolCallAgent: agente com capacidade para ferramentas

ToolCallAgent Adiciona ao Agente a capacidade de usar ferramentas:

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:
# 获取 LLM 响应和工具选择
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
# 处理响应和工具调用
# ...
async def act(self) -> str:
# 执行工具调用
results = []
for command in self.tool_calls:
result = await self.execute_tool(command)
# 添加工具响应到内存
# ...
results.append(result)
return "nn".join(results)

PlanningAgent: agente que planeja.

PlanningAgent O planejamento de tarefas e o acompanhamento da execução foram alcançados:

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.
"""
# 步骤执行跟踪器
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()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

Manus: Agente onipotente

Manus é o principal agente do OpenManus, que integra ferramentas e recursos para lidar com uma ampla variedade de tarefas:

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

Prompt: Diretrizes comportamentais para agentes

O Prompt desempenha uma função essencial na criação de um sistema de Agente, atuando como um manual de instruções que informa ao Agente o que fazer.

Prompt do sistema: definição da função de um agente

O prompt do sistema define as funções básicas e as diretrizes comportamentais para o agente:

SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user. You have various tools at your disposal that you can call upon to efficiently complete complex requests. Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."

O Prompt informa ao Agente que ele é um assistente de IA multifuncional que pode usar uma variedade de ferramentas para atender à solicitação do usuário.

Prompt de planejamento: orienta o agente no planejamento

O prompt de planejamento informa ao agente como dividir tarefas complexas em tarefas menores com um plano de execução:

PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving complex problems by creating and managing structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create clear, actionable plans with the `planning` tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans dynamically
5. Use `finish` to conclude when the task is complete
Available tools will vary by task but may include:
- `planning`: Create, update, and track plans (commands: create, update, mark_step, etc.)
- `finish`: End the task when complete
Break tasks into logical, sequential steps. Think about dependencies and verification methods.
"""

Esse Prompt informa ao Agent que ele é um especialista em planejamento e precisa usar o planning ferramentas para criar, atualizar e monitorar planos.

Prompt de uso da ferramenta: Informa ao agente como usar a ferramenta.

O prompt de uso da ferramenta descreve em detalhes as funções e os cenários de uso de cada ferramenta para ajudar o agente a escolher a ferramenta certa:

NEXT_STEP_PROMPT = """You can interact with the computer using PythonExecute, save important content and information files through FileSaver, open browsers with BrowserUseTool, and retrieve information using GoogleSearch.
PythonExecute: Execute Python code to interact with the computer system, data processing, automation tasks, etc.
FileSaver: Save files locally, such as txt, py, html, etc.
BrowserUseTool: Open, browse, and use web browsers.If you open a local HTML file, you must provide the absolute path to the file.
GoogleSearch: Perform web information retrieval
Based on user needs, proactively select the most appropriate tool or combination of tools. For complex tasks, you can break down the problem and use different tools step by step to solve it. After using each tool, clearly explain the execution results and suggest the next steps.
"""

Prompt dinâmico: tornando os agentes mais flexíveis

Os prompts no OpenManus não podem ser apenas estáticos, mas também podem ser gerados dinamicamente. Por exemplo, na seção PlanningAgent no Prompt, o sistema adiciona o status do plano atual ao Prompt:

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

Esse Prompt dinâmico permite que o Agente tome decisões mais racionais com base na situação atual.

Resumo: Insights do OpenManus

Ao analisar o código do OpenManus, podemos resumir vários componentes-chave da estrutura do AI Agent:

1. AgenteDesign hierárquico, do básico ao profissional, para atingir diferentes níveis de competência.
   • BaseAgentGerenciamento de estado básico e loops de execução: fornece gerenciamento de estado básico e loops de execução.
   • ReActAgentRealização do modelo think-act.
   • ToolCallAgentAdicionar recursos de chamada de ferramentas.
   • Agente profissional: por exemplo PlanningAgenteSWEAgent responder cantando Manus.
2. LLMEncapsulamento da interação com grandes modelos de linguagem, fornecendo recursos de diálogo e de invocação de ferramentas.
   • Suporte para diálogo normal e chamadas de ferramentas.
   • Implementar mecanismo de repetição e tratamento de erros.
   • Há suporte para resposta de fluxo contínuo.
3. MemóriaGerenciar o histórico e o contexto do diálogo.
   • Armazenar e recuperar mensagens.
   • Manter o contexto do diálogo.
4. FerramentaInterface: Fornece uma interface para interagir com o mundo externo.
   • Abstração básica de ferramentas.
   • Várias ferramentas especializadas para conseguir isso.
   • Processamento de resultados de ferramentas.
5. PlanejamentoPermitir o planejamento de tarefas e o rastreamento da execução.
   • Criação e gerenciamento de planos.
   • Rastreamento do status da etapa.
   • Programas de ajuste dinâmico.
6. FluxoGerenciar a colaboração entre vários agentes.
   • Atribuição de tarefas.
   • Integração de resultados.
   • Controle de processos.
7. PromptGuia o comportamento e a tomada de decisões do agente.
   • O prompt do sistema define a função.
   • O Professional Prompt orienta as decisões.
   • Geração dinâmica de prompts.

Com seu design claro e código bem estruturado, o OpenManus é um excelente exemplo para aprender sobre implementações de agentes de IA. Seu design modular permite que os desenvolvedores ampliem e personalizem facilmente seus agentes.

O OpenManus é um bom ponto de partida para os desenvolvedores que desejam saber mais sobre os agentes de IA ou criar seus próprios sistemas de agentes. Ao aprender sobre sua arquitetura e implementação, podemos entender melhor como os AI Agents funcionam e como são projetados.