简介:这份源码资源面向AI智能体设计与软件开发学习者,聚焦ReAct(推理-行动)与Plan-and-Execute(规划-执行)两种主流运行模式的对比分析,帮助读者理解二者在核心思想、工作流程、优劣势及适用场景上的差异。ReAct强调边推理边行动,灵活性强、可即时调整,适合动态变化的任务环境;Plan-and-Execute则先规划后执行,全局视野好、结构清晰,适合目标明确、步骤清晰的任务。资源包共3个文件,包含1个inscode工程配置、1个html页面及1个gitignore忽略规则文件,压缩包约10KB,体积轻量便于快速查阅与本地运行。目前已有58人学习下载。读者可从中获取两种模式的完整源码示例与对比说明,并参考如何根据项目需求选择合适模式,或结合使用二者实现更灵活的策略,适合智能体开发入门与方案选型参考。
1. 从一次 Agent 跑飞说起:这份源码到底解决了什么
凌晨两点,某开发者在群里发了一张截图:同一个任务,Agent 在本地跑得好好的,换到另一台机器上就开始无限循环调用工具,token 烧掉一大半还没收敛。这不是模型的问题,是运行模式没选对。很多人拿到 Agent 源码第一反应是改 prompt、换模型,但真正决定一个 Agent 能不能稳定干活的,是它背后的运行模式——是单轮直出、ReAct 循环、Plan-and-Execute,还是多 Agent 编排。这份《Agent运行模式对比[源码]》把几种主流运行模式拆成了可运行的代码包,每种模式独立成模块,方便你横向对比同一任务在不同模式下的行为差异。它适合两类人:一类是刚接触 agent 开发、想知道 ReAct 和 Plan-Execute 到底差在哪的新手;另一类是已经在做 agent 框架与编排、需要一套可复现基线来验证自己设计的熟手。下面我按自己拆包的习惯,把这份源码从结构到跑通再到踩坑讲一遍。
2. 拆开代码包:目录结构与四种运行模式的实现差异
拿到一个源码包,我一般先不急着跑,而是先看目录怎么分的。目录结构往往比 README 更能说明作者的意图——哪些是公共抽象、哪些是模式特有逻辑、哪些是给对比实验用的脚手架,一眼就能看出来。
2.1 目录分层与公共抽象层
这份源码的顶层大致分成三块:core/、modes/、examples/。core/放的是与具体运行模式无关的东西,比如 LLM 调用封装、工具注册表、消息历史管理、token 计数。modes/下每个子目录对应一种运行模式,各自实现自己的主循环。examples/里是几个可以直接跑的任务脚本,用来做横向对比。
我习惯先看core/里的抽象基类,因为它定义了所有模式必须遵守的接口。常见做法是定义一个BaseAgent,里面有一个run(task)方法作为入口,内部再拆成think()、act()、observe()三个钩子。不同模式的区别,本质上就是这三个钩子的调用顺序和循环条件不同。
# core/base_agent.py class BaseAgent: def __init__(self, llm, tools, max_steps=10): self.llm = llm self.tools = tools # 工具注册表,dict: name -> callable self.max_steps = max_steps # 硬性步数上限,防止无限循环 self.history = [] # 消息历史,所有模式共用 def run(self, task: str): raise NotImplementedError def think(self, task): # 把 task + history 拼成 prompt,交给 LLM raise NotImplementedError def act(self, decision): # 根据 LLM 输出决定调用哪个工具 raise NotImplementedError这段代码的关键在max_steps和history两个参数。max_steps是所有模式共享的兜底机制,我见过太多 agent 因为没有硬上限而在工具调用里打转。history则是模式差异的载体——ReAct 每步都往 history 里追加 thought/action/observation,而 Plan-and-Execute 只在计划阶段和最终汇总阶段用 history,中间执行步骤相对独立。
2.2 ReAct 模式:思考与行动交替的主循环
ReAct 是最常见的运行模式,核心就一句话:让模型在每一步先输出一段思考,再输出一个动作,执行完把结果喂回去,继续下一步。它的实现难点不在循环本身,而在怎么让模型稳定地按格式输出。
# modes/react/agent.py import re class ReActAgent(BaseAgent): def run(self, task): self.history.append({"role": "user", "content": task}) for step in range(self.max_steps): raw = self.think(task) # LLM 返回原始文本 thought, action = self.parse(raw) # 解析 Thought / Action if action is None: # 没有动作,视为最终答案 return thought obs = self.act(action) # 执行工具 self.history.append({"role": "assistant", "content": raw}) self.history.append({"role": "tool", "content": obs}) return "达到最大步数,未收敛" def parse(self, raw): # 约定格式:Thought: ... \n Action: tool_name(args) m = re.search(r"Action:\s*(\w+)\((.*?)\)", raw, re.S) if not m: return raw, None return raw, (m.group(1), m.group(2))这里的parse是血泪经验最多的地方。模型经常不按格式输出,比如把 Action 写成Action: search不带括号,或者在 Thought 里混入换行导致正则匹配失败。我一般会在 parse 失败时做一次重试,把格式要求再强调一遍,而不是直接抛异常。max_steps设成 10 是个经验值,简单任务 3 到 5 步就够,复杂任务超过 10 步基本说明任务拆解有问题,继续跑也是浪费 token。
2.3 Plan-and-Execute 模式:先规划再执行的两阶段结构
Plan-and-Execute 把一次任务拆成两个阶段:先让模型产出一个步骤列表,再逐步执行。它的好处是全局视野更强,不会像 ReAct 那样走一步看一步容易跑偏;代价是计划一旦错了,后面全错,而且计划阶段本身要消耗一次完整的 LLM 调用。
# modes/plan_execute/agent.py class PlanExecuteAgent(BaseAgent): def run(self, task): plan = self.make_plan(task) # 第一阶段:产出步骤列表 results = [] for i, step in enumerate(plan): # 每个步骤独立执行,带上原始任务和已完成结果作为上下文 ctx = f"任务:{task}\n已完成:{results}\n当前步骤:{step}" out = self.execute_step(ctx) results.append(out) return self.summarize(task, results) # 第二阶段:汇总 def make_plan(self, task): prompt = f"把下面任务拆成不超过5个步骤,每行一个:\n{task}" raw = self.llm(prompt) return [line.strip() for line in raw.split("\n") if line.strip()]make_plan里我强制了「不超过 5 个步骤」,这是防止模型把计划写得过细导致执行阶段调用次数爆炸。execute_step的上下文里带上「已完成」的结果,是为了让后续步骤能用到前面的产出,否则各步骤之间就是孤立的。这个模式适合任务边界清晰、步骤之间依赖不强的场景,比如批量数据处理、多文档摘要。
2.4 多 Agent 编排模式:角色分工与消息传递
多 Agent 模式在这份源码里是最复杂的一块,它把任务分给不同角色的 agent,通过消息传递协作。常见做法是设一个协调者(coordinator)加若干执行者(worker),协调者负责拆任务和汇总,执行者各自带不同的工具集。
# modes/multi_agent/agent.py class Coordinator: def __init__(self, workers: dict): self.workers = workers # {"researcher": agent, "coder": agent} def run(self, task): subtasks = self.decompose(task) # 拆成 {role: subtask} outputs = {} for role, sub in subtasks.items(): if role in self.workers: outputs[role] = self.workers[role].run(sub) return self.merge(task, outputs)这里的关键参数是workers的划分粒度。我见过有人把 worker 拆得太细,结果协调者光在路由上就耗掉大量 token;也见过拆得太粗,等于没分工。经验是 2 到 4 个 worker 比较合适,每个 worker 的工具集要有明显区分,否则不如用单 agent。
3. 跑通第一个对比实验:环境、配置与执行
拆完结构,下一步是让它跑起来。这一章我按「装依赖 → 配参数 → 跑对比 → 看输出」的顺序走,每一步都给出可抄的命令和配置。
3.1 环境准备与依赖安装
这份源码是纯 Python 实现,没有重框架依赖,常见做法是建一个独立虚拟环境再装。我一般用 venv,避免和系统里的包打架。
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txtrequirements.txt里通常只有 LLM SDK、requests、pydantic这几样。如果你的环境里已经有这些包,注意版本冲突——尤其是 pydantic,v1 和 v2 的 API 差异很大,源码里如果用model_validate就是 v2,用parse_obj就是 v1。装完先跑一次pip check,把冲突列出来再决定要不要降级。
3.2 配置文件与关键参数
源码一般会带一个config.yaml或.env.example,需要你填 LLM 的接入信息。这里我不写具体厂商,只说参数结构。
# config.yaml llm: model: "your-model-name" temperature: 0.0 # 对比实验建议设 0,减少随机性 max_tokens: 1024 agent: max_steps: 10 # 所有模式的硬性步数上限 tool_timeout: 30 # 单个工具调用超时,秒 retry_on_parse_error: 2 # 解析失败重试次数temperature设 0 是做对比实验的前提,否则同一任务两次跑的结果不一样,没法归因。tool_timeout是我强烈建议加的,工具调用卡死比模型跑飞更常见,尤其是涉及网络请求的工具。retry_on_parse_error设 2 意味着解析失败最多重试两次,再多就是浪费。
3.3 用同一任务跑四种模式
examples/里一般会有一个对比脚本,把同一个任务分别喂给四种模式。如果没有,可以自己写一个。
# examples/compare.py from modes.react.agent import ReActAgent from modes.plan_execute.agent import PlanExecuteAgent from modes.multi_agent.agent import Coordinator from core.llm import build_llm from core.tools import build_tools task = "查一下当前目录下有多少个 py 文件,并统计总行数" llm = build_llm() tools = build_tools() agents = { "react": ReActAgent(llm, tools, max_steps=10), "plan_execute": PlanExecuteAgent(llm, tools, max_steps=10), } for name, agent in agents.items(): print(f"=== {name} ===") print(agent.run(task)) print()跑之前先确认build_tools()里注册的工具确实包含文件操作类工具,否则任务会卡在「找不到工具」上。跑的时候建议开两个终端,一个跑脚本,一个看日志,日志里重点看每步的 token 消耗和工具调用次数。
3.4 对比输出的三个观察维度
跑完四种模式,不要只看最终答案对不对,那太粗了。我一般从三个维度看:
| 维度 | 看什么 | 说明 |
|---|---|---|
| 步数 | 实际循环了几轮 | 步数越多,token 消耗越大 |
| 工具调用次数 | 调了几次工具、有没有重复调用 | 重复调用是 ReAct 常见问题 |
| 收敛性 | 是否在 max_steps 内给出答案 | 没收敛说明模式或任务不匹配 |
这三个维度比最终答案更能说明问题。比如 ReAct 在简单任务上步数少但容易重复调用工具,Plan-and-Execute 步数固定但计划阶段可能就偏了。多 Agent 模式则要看协调者拆的任务是否合理,拆错了后面全白搭。
4. 避坑与排查:四种模式最容易翻车的地方
这一章是我拆包过程中踩过的坑,按「现象 → 原因 → 解决」写。每一条都是实际跑出来的,不是理论推演。
4.1 ReAct 陷入工具重复调用
现象:Agent 连续多步调用同一个工具,参数几乎一样,history 越来越长但任务没进展。
原因:模型在 observation 里没看到「已经查过了」的信号,或者工具返回的结果格式太长,模型没抓住关键信息。常见于搜索类工具,返回一大段文本,模型每次都要重新理解。
解决:在工具返回结果里加一个简短的状态摘要,比如「已查询 X,结果前 200 字如下」。同时在 prompt 里明确「如果上一步已经调用过某工具且结果可用,不要重复调用」。max_steps设小一点也能强制中断,但那是兜底不是解法。
4.2 Plan-and-Execute 计划阶段产出空步骤
现象:make_plan返回的列表里混入空行或纯标点,执行阶段直接报错或跳过。
原因:模型输出的步骤列表格式不稳定,有时用「1. 2. 3.」有时用「- 」,有时末尾多一个空行。split("\n")之后没过滤。
解决:在make_plan里加过滤,去掉空行和长度小于 3 的行,同时用正则把序号前缀剥掉。更稳的做法是让模型输出 JSON 数组,虽然多花一点 token,但解析可靠得多。
4.3 多 Agent 协调者路由错误
现象:协调者把「写代码」的子任务分给了「研究员」角色,执行者拿着不匹配的工具集干不了活。
原因:decompose阶段模型对角色能力的理解和你定义的不一致。你定义 researcher 只有搜索工具,但模型觉得研究员也能写代码。
解决:在decompose的 prompt 里把每个 worker 的能力和工具集明确列出来,让模型从固定角色名里选,而不是自由发挥。同时加一个校验,如果返回的角色名不在workers里,直接重试。
4.4 工具超时导致整个循环卡死
现象:某个工具调用卡住不返回,整个 agent 停在那里,日志也不更新。
原因:工具内部有网络请求或文件 IO,没有设超时。Python 默认的 socket 超时很长,文件锁也可能一直等。
解决:在act里包一层超时控制,用concurrent.futures或信号量都行。tool_timeout设 30 秒是经验值,超过 30 秒还没返回的工具,要么是网络问题要么是逻辑死锁,继续等没意义。
4.5 对比实验里 temperature 没归零
现象:同一任务同一模式跑两次,结果差异很大,没法判断是模式差异还是随机性。
原因:temperature没设 0,或者设了但 LLM 接入层有默认值覆盖。
解决:在build_llm里显式传temperature=0,并且在日志里打印实际生效的参数。有些 SDK 的默认 temperature 不是 0,配置文件里写了也可能被代码里的默认值覆盖,这个坑我踩过不止一次。
5. 进阶技巧:把运行模式对比做成可复现的基线
跑通一次对比不难,难的是让对比结果可复现、可归因。这一章讲两个我常用的技巧,一个是固定随机种子和记录完整轨迹,另一个是用最小任务集做回归。
5.1 记录完整轨迹与 token 消耗
光看最终答案不够,要把每一步的 thought、action、observation 和 token 数都记下来。我一般会在BaseAgent里加一个trace列表,每步追加一条结构化记录。
# core/base_agent.py 里加 import time def _record(self, step, thought, action, obs, tokens): self.trace.append({ "step": step, "thought": thought[:200], # 截断,避免日志爆炸 "action": action, "obs_len": len(obs) if obs else 0, "tokens": tokens, "ts": time.time(), })thought截断到 200 字是为了日志可读,完整内容在 history 里还有。tokens需要 LLM 接入层返回用量信息,大部分 SDK 都支持。有了 trace,跑完直接导出成 JSON,用 pandas 一分析就能看出哪种模式在哪个步骤开始发散。
5.2 用最小任务集做回归对比
单次对比说服力有限,我一般会准备一个 5 到 10 条的最小任务集,覆盖不同类型:纯问答、单工具调用、多工具串联、需要规划的长任务。每次改完 prompt 或换模型,都跑一遍这个任务集,看四种模式的通过率和平均步数有没有退化。
| 任务类型 | 示例 | 主要考察 |
|---|---|---|
| 纯问答 | 解释一个概念 | 不调工具时的直出能力 |
| 单工具 | 查一个文件的行数 | 工具调用格式稳定性 |
| 多工具串联 | 查文件再统计再排序 | 多步依赖处理 |
| 长任务 | 批量处理多个文件 | 规划与收敛性 |
这个任务集不用大,关键是稳定。我一般把它固化成examples/regression.py,每次跑完输出一张对比表。从那以后我每次改 agent 逻辑,都强制走一遍这个回归,不然很容易改好一个模式跑坏另一个。希望这份源码和上面的拆解,能帮你少走一点我走过的弯路。
本文还有配套的精品资源,点击获取