OpenManus├── app/│   ├── agent/                 # Agent层（核心）│   │   ├── base.py           # BaseAgent：状态管理、执行循环│   │   ├── react.py          # ReActAgent：思考-行动模式│   │   ├── toolcall.py       # ToolCallAgent：工具调用能力│   │   ├── planning.py       # PlanningAgent：任务规划能力│   │   ├── swe_agent.py      # SWEAgent：软件工程专用│   │   └── manus.py          # Manus：通用全能Agent│   ││   ├── flow/                  # 工作流层（多Agent编排）│   │   ├── base.py           # BaseFlow：流程抽象基类│   │   ├── planning.py       # PlanningFlow：规划执行流程│   │   └── flow_factory.py   # 流程工厂│   ││   ├── tool/                  # 工具层（外部交互）│   │   ├── base.py           # BaseTool：工具基类│   │   ├── planning_tool.py  # 规划工具（创建/更新计划）│   │   ├── python_execute.py # Python代码执行│   │   ├── google_search.py  # 谷歌搜索│   │   ├── browser_use_tool.py # 浏览器自动化│   │   ├── file_saver.py     # 文件保存│   │   └── terminate.py      # 终止执行│   ││   ├── llm/                   # LLM层（模型交互）│   │   └── llm.py            # 支持OpenAI/Claude/Azure等│   ││   ├── memory/                # 记忆层（上下文管理）│   │   └── ...│   ││   └── prompt/                # 提示词层│       ├── planning.py       # Planning Agent专用Prompt│       └── manus.py          # Manus Agent Prompt│├── config/                    # 配置目录│   └── config.toml           # LLM API配置│├── main.py                   # 入口：基础Manus Agent└── run_flow.py               # 入口：高级Planning Flow

# 1. BaseAgent：最基础抽象class BaseAgent(ABC):    """    提供Agent基本生命周期管理：    - 状态机（IDLE/RUNNING/FINISHED/ERROR）    - 记忆管理（Message历史）    - 执行循环（Run Loop）    - 最大步数限制（max_steps）    """    name: str = "base"    description: str = "Base Agent"    state: AgentState = AgentState.IDLE    memory: Memory = Field(default_factory=Memory)    max_steps: int = 10    async def run(self, request: str) -> str:        """主入口：初始化 → 执行循环 → 返回结果"""        self._initialize(request)        return await self._run_loop()    async def _run_loop(self) -> str:        """核心循环：直到完成或达到max_steps"""        while self.state == AgentState.RUNNING and self.current_step < self.max_steps:            self.current_step += 1            step_result = await self.step()        return self._get_result()# 2. ReActAgent：引入思考-行动模式class ReActAgent(BaseAgent, ABC):    """    ReAct (Reasoning + Acting) 模式实现：    - Think：分析当前状态，决定下一步行动    - Act：执行具体行动    """    @abstractmethod    async def think(self) -> bool:        """思考阶段：返回True表示需要执行行动"""        pass    @abstractmethod    async def act(self) -> str:        """行动阶段：执行决策并返回结果"""        pass    async def step(self) -> str:        """单步执行：Think → Act"""        should_act = await self.think()        if not should_act:            return "思考完成 - 无需行动"        return await self.act()# 3. ToolCallAgent：添加工具调用能力class ToolCallAgent(ReActAgent):    """    工具调用增强：    - 可用工具集合（ToolCollection）    - 工具选择逻辑（LLM决定调用哪个工具）    - 参数解析与执行    """    available_tools: ToolCollection = Field(default_factory=ToolCollection)    tool_calls: List[ToolCall] = Field(default_factory=list)    async def think(self) -> bool:        """调用LLM生成工具调用决策"""        # 构造Prompt包含工具描述        prompt = self._build_system_prompt()        # 调用LLM获取工具调用请求        response = await self.llm.ask_tool(            messages=self.memory.messages,            tools=self.available_tools.to_params(),            system_msgs=[Message.system_message(prompt)]        )        # 解析工具调用        self.tool_calls = self._parse_tool_calls(response)        return len(self.tool_calls) > 0    async def act(self) -> str:        """执行工具调用"""        results = []        for tool_call in self.tool_calls:            result = await self.execute_tool(tool_call)            results.append(result)            self.memory.add_message(Message.tool_message(result, tool_call.id))        return "\n".join(results)# 4. Manus：具体业务实现class Manus(ToolCallAgent):    """    全能通用Agent：    - 集成所有基础工具    - 定义系统角色Prompt    - 支持Python执行、搜索、浏览器、文件操作    """    name: str = "Manus"    description: str = "A versatile agent that can solve various tasks"    system_prompt: str = SYSTEM_PROMPT  # "You are OpenManus, an all-capable AI assistant..."    next_step_prompt: str = NEXT_STEP_PROMPT    # 绑定具体工具实例    available_tools: ToolCollection = Field(        default_factory=lambda: ToolCollection(            PythonExecute(),      # Python代码执行            GoogleSearch(),       # 谷歌搜索            BrowserUseTool(),     # 浏览器控制            FileSaver(),          # 文件保存            Terminate()           # 终止执行        )    )    max_steps: int = 20  # 最大20步

class PlanningAgent(ToolCallAgent):    """    规划型Agent：    1. 使用PlanningTool创建结构化计划    2. 跟踪每步执行状态    3. 根据执行反馈动态调整计划    """    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    active_plan_id: Optional[str] = None    async def think(self) -> bool:        """基于计划状态的思考"""        # 获取当前计划状态文本        plan_status = await self.get_plan()        # 构造上下文丰富的Prompt        prompt = f"""CURRENT PLAN STATUS:{plan_status}YOUR CURRENT TASK:You are now working on step {self.current_step_index}: "{step_text}"Please execute this step using the appropriate tools."""        self.memory.add_message(Message.user_message(prompt))        # 调用父类思考逻辑        return await super().think()    async def create_initial_plan(self, request: str) -> str:        """创建初始执行计划"""        # 调用PlanningTool的create命令        result = await self.available_tools.execute(            "planning",            command="create",            plan_id=self.active_plan_id,            title="Task Plan",            steps=self._decompose_task(request)        )        return result

class PlanningTool(BaseTool):    """    计划管理工具：    - create: 创建新计划    - update: 更新计划内容      - mark_step: 标记步骤状态（in_progress/completed/blocked）    - get: 获取计划详情    """    name: str = "planning"    description: str = "Create and manage structured plans"    async def execute(self, command: str, **kwargs) -> ToolResult:        if command == "create":            return self._create_plan(**kwargs)        elif command == "mark_step":            return self._update_step_status(**kwargs)        # ...    def _create_plan(self, plan_id: str, title: str, steps: List[str]) -> ToolResult:        plan = {            "plan_id": plan_id,            "title": title,            "steps": [                {"index": i, "text": step, "status": "pending"}                 for i, step in enumerate(steps)            ],            "created_at": time.time()        }        self.plans[plan_id] = plan        return ToolResult(output=f"Plan created with {len(steps)} steps")

class PlanningFlow(BaseFlow):    """    规划流程：    1. 使用PlanningTool分解任务为步骤    2. 按顺序执行每步    3. 可动态选择执行Agent（目前仅支持单Agent）    """    agents: Dict[str, BaseAgent]  # 注册的Agent池    planning_tool: PlanningTool    executor_keys: List[str]      # 执行者ID列表    current_step_index: int = 0    async def execute(self, input_text: str) -> str:        """主执行流程"""        try:            # 阶段1：创建初始计划            if input_text:                await self._create_initial_plan(input_text)            # 阶段2：按步骤执行            while await self._has_next_step():                step_info = await self._get_current_step()                # 选择执行者（目前固定使用Manus，可扩展为动态选择）                executor = self.get_executor(step_info.get("type"))                # 生成步骤专用Prompt                step_prompt = f"""CURRENT PLAN STATUS:{plan_status}YOUR CURRENT TASK:You are now working on step {self.current_step_index}: "{step_text}""""                # 执行步骤                result = await executor.run(step_prompt)                # 更新步骤状态                await self._update_step_status(step_info["index"], "completed")            # 阶段3：完成计划            return await self._finalize_plan()        except Exception as e:            return f"Error executing flow: {str(e)}"

class BaseTool(BaseModel, ABC):    """工具基类：所有工具必须实现"""    name: str    description: str    parameters: dict = Field(default_factory=dict)    @abstractmethod    async def execute(self, **kwargs) -> ToolResult:        """执行工具逻辑"""        passclass ToolCollection:    """工具集合：管理多个工具"""    def __init__(self, *tools: BaseTool):        self.tools = {tool.name: tool for tool in tools}    def to_params(self) -> List[dict]:        """转换为LLM可用的函数描述格式（OpenAI Function Calling）"""        return [{            "type": "function",            "function": {                "name": tool.name,                "description": tool.description,                "parameters": tool.parameters            }        } for tool in self.tools.values()]    async def execute(self, name: str, **params) -> ToolResult:        """根据名称执行指定工具"""        if name not in self.tools:            return ToolResult(error=f"Tool {name} not found")        return await self.tools[name].execute(**params)

class LLM(BaseModel):    """    LLM封装：    - 支持OpenAI、Claude、Azure等    - 实现重试机制、错误处理    - 支持流式响应    """    model: str = "gpt-4o"    base_url: str = "https://api.openai.com/v1"    api_key: str = ""    max_tokens: int = 4096    temperature: float = 0.0    async def ask_tool(self, messages: List[Message], tools: List[dict], **kwargs) -> dict:        """        支持Function Calling的对话        返回LLM决策，可能包含tool_calls        """        response = await self.client.chat.completions.create(            model=self.model,            messages=messages,            tools=tools,            tool_choice="auto",            **kwargs        )        return response.choices[0].message    async def ask(self, messages: List[Message], **kwargs) -> str:        """普通对话，返回文本"""        # ...

用户输入: "分析这个网站的SEO并提供报告"    ↓Manus.run(prompt)    ↓初始化记忆: 添加System Prompt + User Prompt    ↓进入Run Loop（最多20步）    │    ├─→ Step 1:    │    ├─ think(): 调用LLM，返回需要GoogleSearch    │    ├─ act(): 执行GoogleSearch，获取网站信息    │    └─ 保存结果到Memory    │    ├─→ Step 2:    │    ├─ think(): 分析结果，决定使用BrowserUseTool访问网站    │    ├─ act(): 浏览器抓取网站内容    │    └─ 保存结果    │    ├─→ Step 3:    │    ├─ think(): 分析内容，决定用PythonExecute分析SEO指标    │    ├─ act(): 执行Python代码计算SEO分数    │    └─ 保存结果    │    ├─→ Step 4:    │    ├─ think(): 决定使用FileSaver保存报告    │    ├─ act(): 保存SEO报告到本地文件    │    └─ 调用Terminate结束    │    ↓返回结果摘要

用户输入: "帮我规划一次日本7日游，预算1万元"    ↓PlanningFlow.execute()    │    ├─→ 阶段1：创建计划    │    └─ PlanningTool.create → 生成步骤：    │       1. 搜索日本7日游攻略    │       2. 查询机票价格（出发地-东京/大阪）    │       3. 查询酒店价格（东京/京都/大阪）    │       4. 规划每日行程    │       5. 计算总预算    │       6. 生成详细旅行计划文档    │    ├─→ 阶段2：按步执行    │    │    │    ├─→ Step 1: 搜索攻略    │    │    └─ Manus.run("CURRENT PLAN STATUS: ... YOUR CURRENT TASK: 搜索日本7日游攻略")    │    │       └─ 使用GoogleSearch + BrowserUseTool收集信息    │    │    │    ├─→ Step 2: 查询机票    │    │    └─ Manus.run("... YOUR CURRENT TASK: 查询机票价格...")    │    │       └─ 浏览器访问携程/去哪儿查询价格    │    │    │    ├─→ Step 3-6: 类似执行...    │    │    │    └─→ 每步完成后更新计划状态（mark_step completed）    │    └─→ 阶段3：汇总结果         └─ 整合所有步骤输出，生成最终旅行计划

# 方法1：使用Conda（传统方式）conda create -n open_manus python=3.12conda activate open_manusgit clone https://github.com/mannaandpoem/OpenManus.gitcd OpenManuspip install -r requirements.txt# 方法2：使用uv（推荐，更快）curl -LsSf https://astral.sh/uv/install.sh | shgit clone https://github.com/mannaandpoem/OpenManus.gitcd OpenManusuv venv --python 3.12source .venv/bin/activate  # Windows: .venv\Scripts\activateuv pip install -r requirements.txt# 安装浏览器自动化（可选）playwright install

# config/config.toml[llm]model = "gpt-4o"base_url = "https://api.openai.com/v1"api_key = "sk-..."  # 替换为你的API Keymax_tokens = 4096temperature = 0.0# 可选：Vision模型配置（用于图像理解）[llm.vision]model = "gpt-4o"base_url = "https://api.openai.com/v1"api_key = "sk-..."

# app/tool/my_custom_tool.pyfrom app.tool.base import BaseTool, ToolResultclass DatabaseQueryTool(BaseTool):    """    自定义数据库查询工具示例    """    name: str = "database_query"    description: str = "Execute SQL queries on a database"    # 定义JSON Schema参数格式（供LLM理解）    parameters: dict = {        "type": "object",        "properties": {            "query": {                "type": "string",                "description": "SQL query to execute"            },            "connection_string": {                "type": "string",                 "description": "Database connection string"            }        },        "required": ["query"]    }    async def execute(self, query: str, connection_string: str = "default", **kwargs) -> ToolResult:        """执行工具逻辑"""        try:            # 这里实现具体逻辑            import sqlite3            conn = sqlite3.connect(connection_string)            cursor = conn.cursor()            cursor.execute(query)            result = cursor.fetchall()            conn.close()            return ToolResult(output=str(result))        except Exception as e:            return ToolResult(error=f"Database error: {str(e)}")# 注册到Manus Agent# app/agent/manus.pyfrom app.tool.my_custom_tool import DatabaseQueryToolavailable_tools: ToolCollection = Field(    default_factory=lambda: ToolCollection(        PythonExecute(),        GoogleSearch(),        BrowserUseTool(),        FileSaver(),        DatabaseQueryTool(),  # 添加新工具        Terminate()    ))

# app/agent/data_analysis_agent.pyfrom app.agent.toolcall import ToolCallAgentfrom app.tool import ToolCollectionfrom app.tool.python_execute import PythonExecutefrom app.tool.file_saver import FileSaverfrom pydantic import Fieldclass DataAnalysisAgent(ToolCallAgent):    """    数据分析专用Agent：    - 强化Python数据处理能力    - 集成matplotlib/seaborn可视化    - 自动保存分析图表    """    name: str = "data_analyst"    description: str = "Specialized agent for data analysis and visualization"    system_prompt: str = """You are a Data Analysis Expert.Your capabilities:1. Load and clean data using Python (pandas)2. Perform statistical analysis3. Create visualizations (matplotlib/seaborn/plotly)4. Generate comprehensive analysis reportsAlways save your visualizations and final report using FileSaver."""    # 仅保留必要工具    available_tools: ToolCollection = Field(        default_factory=lambda: ToolCollection(            PythonExecute(),  # 核心：Python执行            FileSaver(),      # 保存结果            Terminate()        )    )    max_steps: int = 30  # 数据分析允许更多步骤# 在PlanningFlow中使用专用Agent# run_custom_flow.pyfrom app.agent.data_analysis_agent import DataAnalysisAgentfrom app.agent.manus import Manusfrom app.flow.planning import PlanningFlowasync def run_data_analysis_task():    agents = {        "manus": Manus(),                    # 通用任务        "analyst": DataAnalysisAgent(),      # 数据分析专用    }    flow = PlanningFlow(        agents=agents,        executor_keys=["analyst"]  # 指定使用DataAnalysisAgent    )    result = await flow.execute("分析sales.csv的销售趋势并生成图表")

# app/tool/custom_flow.pyfrom app.flow.base import BaseFlowfrom typing import Dictclass ApprovalFlow(BaseFlow):    """    带人工审批的自定义流程：    1. Agent生成方案    2. 等待人类审批    3. 根据审批结果执行或修改    """    async def execute(self, input_text: str) -> str:        # 步骤1：规划Agent生成方案        planner = self.agents["planner"]        plan = await planner.run(f"Create a plan for: {input_text}")        # 步骤2：等待人工确认（伪代码）        human_approval = await self.wait_for_human_input(            f"请审阅以下方案：\n{plan}\n[确认/修改/取消]"        )        if human_approval == "确认":            # 步骤3：执行Agent实施方案            executor = self.agents["executor"]            result = await executor.run(f"Execute plan: {plan}")            return result        elif human_approval == "修改":            # 重新规划            return await self.execute(input_text + " (需要修改)")        else:            return "任务已取消"

# app/memory/vector_memory.pyfrom app.memory.base import Memoryimport sqlite_vec  # 向量SQLiteclass VectorMemory(Memory):    """    增强记忆：使用向量数据库存储和检索    支持语义搜索历史对话    """    def __init__(self, db_path: str = "memory.db"):        super().__init__()        self.db = sqlite_vec.connect(db_path)        self._init_tables()    async def add_message(self, message: Message):        # 存储原始消息        super().add_message(message)        # 向量化并存储（用于语义检索）        embedding = await self.get_embedding(message.content)        self.db.execute(            "INSERT INTO vec_memory(content, embedding, timestamp) VALUES (?, ?, ?)",            (message.content, embedding, time.time())        )    async def retrieve_relevant(self, query: str, top_k: int = 3) -> List[str]:        """语义检索相关历史记忆"""        query_vec = await self.get_embedding(query)        results = self.db.execute(            "SELECT content FROM vec_memory ORDER BY vec_distance(embedding, ?) LIMIT ?",            (query_vec, top_k)        )        return [r[0] for r in results]