Manus逆向工程：用Python拆解AI智能体的ReAct与Plan-Execute

1. 从 Manus 日志里能拆出什么ReAct 与 Plan-Execute 的混合骨架Manus 这类 AI 智能体最值得研究的地方不是它“能做什么”而是它“怎么想”。我把它公开的行为日志逐条拆开看发现它并不是单纯跑一个 ReAct 循环也不是一次性把计划列完就闷头执行而是把 Plan-Execute 和 ReAct 叠在一起用高层用计划-执行管方向底层用思考-行动-观察管细节。这个结构用 Python 复现出来并不复杂难的是把两条路线的边界划清楚。先说结论Plan-Execute 负责“把模糊需求拆成宏观步骤”ReAct 负责“在单个步骤里反复和环境交互直到搞定”。比如“总结 DeepSeek 训练加速技巧”这个请求Plan-Execute 层会拆成搜索、浏览、整理、验证、写报告五步而“浏览某个 URL 并提取关键信息”这一步内部会跑一个 Thought → Action → Observation 的小循环可能反复打开多个页面、写多次 todo.md 才收敛。适合谁看已经会用 Python 调大模型 API、想自己搭一个能规划又能调工具的智能体骨架的开发者。如果你只想要一个现成产品这篇不适合如果你想搞清楚“思考”和“行动”到底怎么交替下面这套骨架可以直接跑。我试过把两条路线硬塞进一个 while 循环里结果日志乱成一团后来才想明白Plan-Execute 是外层 forReAct 是内层 while两者职责不同不能混。下面按这个思路一步步搭。2. 前置准备用 TaoToken 统一 Key 接入别在多个平台间来回切复现智能体最烦的不是写循环是模型接入。ReAct 循环里每一步都要调 LLM如果规划用一个平台、思考用另一个平台Key 管理和计费会把人逼疯。我的做法是用 TaoToken 做统一入口一个 Key 覆盖规划、思考、工具参数生成这几类调用。TaoToken 在这里的角色是模型调用的统一网关官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。你需要在控制台创建一个 API Key然后把它写进环境变量代码里只认一个 base_url 和一个 key切换模型时改 model 字段就行不用动调用逻辑。具体操作进控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建 Key然后在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 复制出来。接入细节如果卡住直接看接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 base_url 和鉴权头的写法。注意Key 只放环境变量别硬编码进 settings.json 或 config.toml 后提交到仓库。我见过有人把 Key 写进配置文件推到公开仓库几分钟就被刷爆额度。这一步做完后面所有 LLM 调用都走同一个客户端ReAct 循环里换模型只改一个字符串。3. 可复制配置骨架settings.json 与 config.toml配置分两层settings.json 管运行时参数模型、超时、迭代上限config.toml 管工具白名单和路径。分开的好处是调参不用改代码工具权限也能单独审。先看 settings.json{ llm: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, planner_model: claude-sonnet-4-5, react_model: claude-sonnet-4-5, timeout_seconds: 60, max_retries: 2 }, agent: { max_plan_steps: 8, max_react_iterations: 6, scratchpad_file: todo.md, log_level: INFO } }planner_model 和 react_model 分开配是因为规划需要更强的长上下文推理ReAct 单步思考可以换更便宜的模型压成本。max_react_iterations 是防止死循环的保险丝设 6 意味着单个子目标最多思考-行动 6 轮超了就强制跳出并记录。再看 config.toml[tools.browser] enabled true allowed_actions [search, open_url, extract_text] timeout_seconds 30 [tools.python_interpreter] enabled true allowed_actions [shell, read_file, write_file, list_files] workdir ./workspace allow_shell true [memory] scratchpad_path ./workspace/todo.md artifact_dir ./workspace/artifactsallow_shell true 是双刃剑本地跑通链路时方便但生产环境建议关掉只留 read_file/write_file。workdir 限定在 ./workspace避免工具误操作到项目根目录。提示config.toml 里的 allowed_actions 是白名单LLM 生成的 action 如果不在列表里直接拒绝并作为 Observation 返回“工具不可用”这样 ReAct 循环会自己换策略而不是崩掉。4. 用 Python 复现思考-行动循环主循环与工具层核心就两个类Agent 主循环和 Tool 执行器。主循环负责 Plan-Execute 的外层 for 和 ReAct 的内层 while工具层负责把 LLM 生成的 action 翻译成真实调用。先写 LLM 客户端统一走 TaoTokenimport os, json, requests class LLMClient: def __init__(self, cfg): self.base_url cfg[llm][base_url] self.api_key os.environ[cfg[llm][api_key_env]] self.timeout cfg[llm][timeout_seconds] def chat(self, model, messages, temperature0.2): resp requests.post( f{self.base_url}/v1/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, json{ model: model, messages: messages, temperature: temperature, }, timeoutself.timeout, ) resp.raise_for_status() return resp.json()[choices][0][message][content]再写工具层把 browser 和 python_interpreter 的动作分发出去import subprocess, pathlib class ToolBox: def __init__(self, cfg): self.cfg cfg self.workdir pathlib.Path(cfg[tools][python_interpreter][workdir]) self.workdir.mkdir(parentsTrue, exist_okTrue) def execute(self, tool_name, action, params): if tool_name browser: return self._browser(action, params) if tool_name python_interpreter: return self._python(action, params) return fError: unknown tool {tool_name} def _browser(self, action, params): if action search: return f[mock] search results for: {params.get(query)} if action open_url: return f[mock] page content of {params.get(url)} if action extract_text: return f[mock] extracted text: {params.get(content, )[:200]} return Error: browser action not allowed def _python(self, action, params): if action write_file: p self.workdir / params[path] p.parent.mkdir(parentsTrue, exist_okTrue) p.write_text(params[content], encodingutf-8) return fFile written: {p} if action read_file: p self.workdir / params[path] return p.read_text(encodingutf-8) if p.exists() else Error: file not found if action shell: r subprocess.run( params[command], shellTrue, cwdself.workdir, capture_outputTrue, textTrue, timeout30, ) return frc{r.returncode}\nstdout{r.stdout}\nstderr{r.stderr} return Error: python action not allowed主循环把两层拼起来关键是内层 while 的退出条件LLM 返回 finish_sub_goal 或达到迭代上限class Agent: def __init__(self, cfg, llm, toolbox): self.cfg cfg self.llm llm self.toolbox toolbox self.scratchpad [] def run(self, user_request): plan self._make_plan(user_request) for step in plan: self._run_react(step) def _make_plan(self, user_request): prompt ( f把任务拆成不超过{self.cfg[agent][max_plan_steps]}个宏观步骤 f每行一个不要编号。任务{user_request} ) raw self.llm.chat(self.cfg[llm][planner_model], [{role: user, content: prompt}]) return [s.strip() for s in raw.splitlines() if s.strip()] def _run_react(self, sub_goal): for i in range(self.cfg[agent][max_react_iterations]): thought self.llm.chat( self.cfg[llm][react_model], [{role: user, content: f子目标{sub_goal}\n历史{self.scratchpad[-3:]}\n f输出JSON{{\thought\:\...\,\tool\:\...\,\action\:\...\,\params\:{{}}}}}], ) step json.loads(thought) if step.get(tool) finish: self.scratchpad.append(f[DONE] {sub_goal}) return obs self.toolbox.execute(step[tool], step[action], step.get(params, {})) self.scratchpad.append(fThought: {step[thought]}\nAction: {step[action]}\nObservation: {obs})跑起来后日志里会看到 Thought 和 Action 交替出现这正是 ReAct 循环在工作的证据。5. 验证请求本地跑通一次任务规划与工具调用链路配置和代码就位后用一个小任务验证整条链路。任务选“在 workspace 下创建 notes.md 并写入三条训练加速技巧”这样能同时触发 write_file 和 shell 两个工具。启动脚本import json, tomllib from agent import Agent, LLMClient, ToolBox with open(settings.json, encodingutf-8) as f: cfg json.load(f) with open(config.toml, rb) as f: cfg.update(tomllib.load(f)) agent Agent(cfg, LLMClient(cfg), ToolBox(cfg)) agent.run(在 workspace 下创建 notes.md写入三条 DeepSeek 训练加速技巧)预期日志形态INFO plan step 1: 创建 notes.md 文件 INFO Thought: 需要先确认文件不存在再写入内容 INFO Action: write_file pathnotes.md INFO Observation: File written: workspace/notes.md INFO Thought: 文件已创建子目标完成 INFO [DONE] 创建 notes.md 文件看到 Thought 和 Action 交替、Observation 紧跟 Action说明 ReAct 循环正常。如果日志里只有 Action 没有 Thought多半是 prompt 里没强制要求输出 thought 字段LLM 偷懒了。验证成功后可以换更复杂的任务比如“搜索三个网页并汇总成 report.md”观察 Plan-Execute 是否拆出多个步骤、每个步骤内部是否跑了多轮 ReAct。这一步跑通骨架就算立住了。6. 本篇常见错排查报错一401 Unauthorized。九成是环境变量没生效。检查echo $TAOTOKEN_API_KEY是否有值settings.json 里的 api_key_env 名字是否和实际环境变量一致。别把 Key 直接写进 json那样换机器就失效。报错二JSONDecodeError。LLM 返回的 thought 不是合法 JSON常见于模型在 JSON 前后加了说明文字。解决方式是在 prompt 里加“只输出 JSON不要任何解释”或者在解析前用正则截取第一个{到最后一个}。报错三ReAct 循环跑满 max_react_iterations 还没结束。说明子目标太大或工具返回的 Observation 信息量不够。把子目标拆细或者在 Observation 里带上更多上下文比如文件当前内容让 LLM 有足够信息判断下一步。报错四shell 命令在 workdir 找不到文件。ToolBox 里 subprocess 的 cwd 设成了 workdir但 LLM 生成的命令可能带绝对路径。统一约定所有路径相对 workdir或者在执行前做一次路径归一化。报错五规划步骤为空。planner_model 返回了空行或纯编号。在 _make_plan 里加一层过滤去掉纯数字和空行再判断列表长度为 0 就重试一次。排查顺序建议先确认 Key 和 base_url再确认 JSON 解析最后看循环退出条件。大部分问题出在前两步。7. 下一步把骨架接到真实模型对话与长期编码场景骨架跑通后下一步是把它接到真实场景。如果你只是想验证模型在 ReAct 循环里的表现可以直接用模型对话 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 手动试几轮 Thought-Action 提示观察不同模型的输出稳定性再决定 planner 和 react 分别用哪个模型。如果你打算把这个骨架用于长期编码或 Agent 任务比如让它自己读代码、改文件、跑测试那单次调用成本会累积得很快。这种情况更适合用 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 把规划、思考、工具参数生成这几类高频调用统一管理避免每个环节单独计费。接入过程中如果遇到鉴权或端点问题回到接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 对照一遍请求头格式多数报错都能定位。Key 管理仍在 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 建议给智能体单独建一个 Key方便按项目追踪用量。最后留一个实用技巧在 scratchpad 里给每条 Observation 打上时间戳和工具名回看日志时能一眼看出哪一步耗时最长、哪个工具调用最频繁。这个习惯比任何性能分析工具都直接。