Explicación del código abierto de OpenManus e información sobre su arquitectura de agente de inteligencia artificial

Apertura: El incendio de Manus y la ruptura de OpenManus

Una de las grandes novedades que se están produciendo últimamente en los círculos de la IA es que la Manus Manus, el agente de inteligencia artificial, ha llamado mucho la atención por sus potentes funciones y su uso flexible. En pocas palabras, Manus es como un asistente todoterreno, ya sea para programar, buscar información, procesar documentos o navegar por Internet, puede ayudarte a conseguirlo.

Sin embargo, no es tan fácil utilizar Manus, hay que tener un código de invitación. Esto ha impedido a muchos desarrolladores e investigadores utilizar Manus. El equipo MetaGPT formado por @mannaandpoem, @XiangJinyu, @MoshiQAQ y @didiforgithub invirtió sólo 3 horas en crear un proyecto de código abierto llamado OpenManus. Ahora puedes experimentar la potencia de Manus sin necesidad de un código de invitación. Lo que es aún más emocionante es que OpenManus es un proyecto de código abierto, lo que significa que puedes modificarlo y ampliarlo para adaptarlo a tus necesidades.

La aparición de OpenManus no sólo ofrece a más personas la oportunidad de experimentar el encanto del Agente AI, sino que también inyecta nueva vitalidad al desarrollo del Agente AI. Para los que nos dedicamos a la tecnología, OpenManus no es sólo una buena herramienta, sino también un excelente recurso de aprendizaje. Mediante el estudio de su código, podemos obtener una comprensión más profunda del diseño del marco y los detalles de implementación del Agente AI.

Marco de agentes de IA: filosofía de diseño de OpenManus

La estructura del código de OpenManus es muy clara y adopta un diseño modular, que es como construir bloques para combinar juntos distintos módulos funcionales. La ventaja de este diseño es que el código es muy reutilizable y extensible, y las responsabilidades de cada módulo están claras.

Los componentes principales de OpenManus son

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

Componentes del LLM: El cerebro del agente

Si comparamos un agente con una persona, entonces el LLM (Large Language Model) es el cerebro del agente, responsable de entender las órdenes del usuario, generar respuestas y tomar decisiones. Es responsable de entender los comandos del usuario, generar respuestas y tomar decisiones. OpenManus encapsula la interacción con el LLM a través de la clase 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
)

La clase LLM proporciona dos métodos principales:

• ask:: Enviar una solicitud de diálogo general
• ask_tool:: Enviar peticiones con llamadas a herramientas

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 memoria: memoria del agente

El componente Memoria es como un cuaderno de notas para el Agente, responsable de registrar y gestionar el historial de diálogo del Agente. Con la Memoria, el Agente puede recordar lo que ha dicho antes y mantener la coherencia del 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

El componente Memoria es el componente central del Agente, y se accede a él a través de la función update_memory para añadir un nuevo mensaje:

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 Herramientas: Caja de herramientas del agente

El componente Herramientas es el puente entre el Agente y el mundo exterior, y OpenManus implementa un sistema de herramientas flexible que permite al Agente invocar una variedad de herramientas para realizar tareas.

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

Los resultados de la ejecución de la herramienta están representados por el icono ToolResult para representarlo:

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)

OpenManus proporciona una serie de herramientas 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 planificación: capacidades de planificación del Agente

El componente Planificación es la clave de la capacidad de OpenManus para gestionar tareas complejas. Permite a los Agentes planificar y desglosar tareas complejas en pequeñas tareas paso a paso para luego completarlas una a una.

El componente de planificación consta de dos partes principales:

1. PlanningTool: Ofrece funciones de creación, actualización y seguimiento de planes.
2. PlanningAgent: Uso PlanningTool para realizar la planificación y ejecución de tareas.

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 La metodología básica incluye:

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 flujo: Capacidades de colaboración para agentes

La función del componente Flujo es coordinar a varios Agentes que trabajan juntos para realizar tareas más complejas.

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 es una implementación concreta de Flow para planificar y ejecutar tareas:

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

Implementación de OpenManus como agente: una arquitectura por capas

El Agente de OpenManus utiliza una arquitectura jerárquica, construida capa a capa desde la funcionalidad básica hasta las aplicaciones especializadas. Las ventajas de este diseño son la elevada reutilización del código, la gran escalabilidad y la claridad de las responsabilidades en cada nivel.

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

BaseAgent: la base de bases

BaseAgent es la base de todo el framework y define las propiedades y métodos principales de un 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: el agente pensante

ReActAgent Se ha realizado un modelo "pensar-actuar", dividiendo el proceso de ejecución del Agente en dos 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 con capacidad para herramientas

ToolCallAgent Añade al Agente la posibilidad de utilizar herramientas:

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 planifica.

PlanningAgent Se logró la planificación de tareas y el seguimiento de su ejecución:

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 Omnipotente

Manus es el agente central de OpenManus, que integra herramientas y capacidades para gestionar una amplia variedad de tareas:

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: Pautas de comportamiento para los agentes

El Prompt juega un papel vital en la construcción de un sistema de Agente, actuando como un manual de instrucciones que le dice al Agente qué hacer.

System Prompt: Definir el papel de un agente

El indicador del sistema establece las funciones básicas y las pautas de comportamiento del 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."

El Prompt indica al Agente que es un asistente de IA polivalente que puede utilizar diversas herramientas para satisfacer la petición del usuario.

Planning Prompt: guía al Agente en la planificación

La instrucción de planificación indica al agente cómo dividir tareas complejas en tareas más pequeñas con un plan de ejecución:

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

Esta pregunta le dice al Agente que es un experto en planificación y que necesita utilizar el comando planning herramientas para crear, actualizar y hacer un seguimiento de los planes.

Indicación de uso de la herramienta: Indica al Agente cómo utilizar la herramienta.

El indicador de uso de herramientas describe detalladamente las funciones y los escenarios de uso de cada herramienta para ayudar al Agente a elegir la herramienta adecuada:

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

Dynamic Prompt: más flexibilidad para los agentes

Los avisos en OpenManus no sólo pueden ser estáticos, también pueden generarse dinámicamente. Por ejemplo, en el PlanningAgent en el Prompt, el sistema añade el estado actual del plan al 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))

Este indicador dinámico permite al Agente tomar decisiones más racionales en función de la situación actual.

Resumen: conclusiones de OpenManus

Analizando el código de OpenManus, podemos resumir varios componentes clave del marco del agente de IA:

1. AgenteDiseño jerárquico de básico a profesional para alcanzar diferentes niveles de competencia.
   • BaseAgent: Proporciona gestión de estado básica y bucles de ejecución.
   • ReActAgentRealización del modelo think-act.
   • ToolCallAgent: Añade funciones de llamada a herramientas.
   • Agente profesional: por ejemplo PlanningAgentySWEAgent responder cantando Manus.
2. LLM: Encapsular la interacción con grandes modelos lingüísticos, proporcionando capacidades de diálogo e invocación de herramientas.
   • Soporte para diálogo normal y llamadas a herramientas.
   • Implementar el mecanismo de reintento y la gestión de errores.
   • Se admite la respuesta de streaming.
3. Memoria: Gestionar la historia y el contexto del diálogo.
   • Almacenar y recuperar mensajes.
   • Mantener el contexto del diálogo.
4. Herramienta: Proporcionar una interfaz para interactuar con el mundo exterior.
   • Abstracción básica de herramientas.
   • Múltiples herramientas especializadas para conseguirlo.
   • Tratamiento de los resultados de las herramientas.
5. Planificación: Permitir la planificación de tareas y el seguimiento de la ejecución.
   • Creación y gestión de planes.
   • Seguimiento del estado de los pasos.
   • Programas de ajuste dinámico.
6. FlujoGestión de la colaboración entre varios agentes.
   • Tareas.
   • Integración de resultados.
   • Control de procesos.
7. Pregunte aGuía el comportamiento y la toma de decisiones del Agente.
   • System Prompt define el rol.
   • Professional Prompt orienta la toma de decisiones.
   • Generación dinámica de avisos.

Con su diseño claro y su código bien estructurado, OpenManus es un ejemplo excelente para aprender sobre implementaciones de agentes de IA. Su diseño modular permite a los desarrolladores ampliar y personalizar fácilmente sus agentes.

OpenManus proporciona un buen punto de partida para los desarrolladores que quieran aprender más sobre los Agentes de IA o construir sus propios sistemas de agentes. Al conocer su arquitectura e implementación, podemos entender mejor cómo funcionan y cómo se diseñan los Agentes de IA.