很多人一听到 Agent,第一反应就是打开 LangChain、LlamaIndex 或者 AutoGPT 的教程,直接往上套框架。但我觉得,真正理解 Agent 最好的方式,恰恰是从零手写一个最原始的 ReAct 循环。这个项目就是这么来的:不依赖任何框架,不依赖模型的 Function Calling 能力,甚至你用一个本地部署的 7B 模型都能跑通。整套逻辑说到底就一句话——用 Prompt 约定格式,让模型自己输出推理和行动指令,代码去解析这些指令并执行工具,再把结果反馈回去,直到模型给出最终答案。
我把这套东西完整手写了一遍,跑通之后最大的感受是:框架帮你省掉的,恰恰是技术里最有价值的一部分。你搞清楚了 ReAct 循环的每一步,再去用任何框架都不会觉得黑盒。这篇文章把我手写原生 ReAct Agent 循环的全部思路、代码样板、踩坑记录都整理出来,给那些想彻底搞懂 Agent 本质的人参考。
1. 为什么要把 ReAct Agent 循环从零手写?
1.1 框架是捷径,但不是理解之路
先说一个经常被误解的概念:ReAct 这里的 "React" 和前端圈那个 React 框架完全没有关系。它是 Reasoning + Acting 的缩写,意思是“让模型交替进行推理和行动”,先想一步,做一步,观察结果,再想下一步,直到完成任务。很多 Agent 框架的底层,本质上都是这个循环,只不过包了一层又一层:工具有 Plugin、有工具包、有回调机制、有记忆模块……你用起来很方便,但一旦遇到问题,定位起来就特别痛苦。
我最初也被框架吸引过,用了几次之后发现,表面上我是在写业务逻辑,实际上我连每次调用模型传了什么 Prompt 都不知道。一个 Agent 跑出奇怪结果,你根本说不清是模型的推理问题,是 Prompt 问题,还是框架里某个默认参数在捣乱。手写一遍之后,所有变量都在自己手里,思路瞬间清晰。
另外从工程角度讲,框架也有它的锁定风险。你今天用了某个框架的 Agent API,明天框架升级接口变了,你的代码就要跟着改。而手写 ReAct 循环就是普通的 Python 函数加 HTTP 调用,模型接口只要还是 OpenAI 兼容协议,你就能一直用。这是我看重它的原因。
1.2 纯 Prompt 驱动的核心:让模型按剧本输出
所谓“纯 Prompt 驱动”,指的是我们完全不依赖模型提供的 Function Calling 或 Tool Use 原生能力,而是把“可用工具列表”和“输出格式要求”写进 System Prompt,让模型按照我们规定的文本结构输出。
比如我们规定模型只能输出这样一组字段:
Thought: 你的推理过程 Action: 工具名称,或者 final_answer Action Input: 传给工具的参数,JSON 格式模型本身并不知道什么 Action、Observation 这些概念,它只是根据 Prompt 里的规则,模仿出一个“思考并行动”的过程。而我们拿到模型输出后,用正则表达式或者 JSON 解析器把 Action 和 Action Input 提取出来,去调用真实的工具函数,拿到结果后再把它组织成 Observation 文本,拼进对话历史,继续让模型推理。
这种方式的通用性极强。不管是 GPT-4 这种大模型,还是本地跑的量化小模型,也不管是 OpenRouter 还是各类国产模型接口,只要它能正常对话、能遵循基本指令,这个循环就能跑。对我来说,这意味着不需要为每一个模型适配特殊的 API 参数。
1.3 不依赖原生 Function Calling 的真实原因
很多人会问:现在主流模型都支持 Function Calling 了,为什么还要费劲用 Prompt 手写?我遇到的情况很现实:第一,并不是所有模型都支持 Function Calling,尤其是本地部署的开源模型,除了少数几款其他都支持得不算好,经常出现工具参数格式错误、幻觉工具调用的问题。第二,Function Calling 的返回结构各家都不一样,如果你要做一个跨模型的 Agent,就得写很多适配层。第三,有些场景下模型服务是别人提供的,只开放了聊天接口,根本没有原生 FC 可用。
用纯 Prompt 驱动,等于把“工具调用的解释权”拿回到自己手里。模型所有输出都是文本,我们想怎么解析就怎么解析,想兼容什么模型就兼容什么模型,只要模型会聊天,这个 Agent 核心循环就能稳定运行。虽然不如原生 FC 的格式那么严格,但配合严格的 Prompt 和容错解析,实践中已经能达到相当可靠的程度。
2. 手写前的必备认知:ReAct 循环到底在循环什么?
2.1 从 Thought 到 Observation 的闭环
手写 ReAct Agent,第一步要理解这个循环的核心数据流。一次完整的 Agent 运行,实际上是一连串“模型回复 + 工具结果”交替拼接的对话。
它的闭环是这样的:
- 用户输入一个问题,拼在 System Prompt 后面,发给模型。
- 模型输出一段文本,里面包含 Thought、Action、Action Input。
- 代码解析出 Action,如果不是 final_answer,就去查找并执行对应的工具。
- 工具返回结果,代码把它包装成 Observation 追加到对话历史。
- 带着 Observation 再次调用模型,模型继续输出。
- 直到模型输出 Action 为 final_answer,终止循环,返回最终答案。
这里最关键的一点是:模型从来没有真正“调用”工具,它只是输出了一段“我想调用某工具”的文本,真正执行工具的是我们的代码。这就是 Agent 和单纯聊天最大的区别——模型有了干预外部环境的能力,而这个能力其实是我们代码给的。
这个闭环看起来简单,但真把它拆开你才会发现,Agent 的稳定性很大程度上取决于对模型输出的解析质量和对话历史的组织方式。这两块才是手写 Agent 里最有挑战的地方。
2.2 System Prompt 是这台机器的控制面板
整个 Agent 的行为都受 System Prompt 约束,所以它必须设计得非常精确。我的 System Prompt 通常包含这几块内容:
- 角色设定:告诉模型它是一个能使用工具的智能助理,不是单纯聊天。
- 工具列表:把所有工具的名称、用途、参数格式详细列出,让模型知道什么场景该选什么工具。
- 输出格式规则:严格说明必须按 Thought/Action/Action Input 的格式输出,不允许用自然语言直接回答最终结果。
- 终止条件:明确当模型能回答用户问题时,Action 必须是 final_answer,并且 Action Input 里放最终回答。
- 异常提示:告诉模型如果工具调用失败,要基于 Observation 中的错误信息重新思考,不要重复同样的错误。
有一段我常用的核心 Prompt 是这样的:
你是一个智能代理,可以调用外部工具来获得信息并解决问题。 你有以下工具可供使用: - calculator: 计算数学表达式,输入格式为 JSON: {"expression": "1+2"} - search: 搜索互联网获取最新信息,输入格式为 JSON: {"query": "关键词"} 请严格按照以下格式回复: Thought: 你当前的思考过程 Action: 工具名称,或 final_answer Action Input: 工具的 JSON 参数 如果已经足够回答用户问题,Action 必须为 final_answer,Action Input 为直接给出的最终答案。你可能会觉得这段 Prompt 很简短,但它实际效果比我用过的不少框架默认 Prompt 都好。原因很简单:规则清晰、格式唯一、没有多余的扮演性描述。
2.3 选型思考:格式约定用什么最稳定?
手写 ReAct 最核心的决策是“模型输出格式”。常见的方案有三种:
一种是“双行式”,即用Thought:、Action:、Action Input:作为单独行输出。优点是直观,正则解析容易,模型相对容易学会,也很容易人工调试。
另一种是“JSON 块式”,让模型输出一个 JSON,比如{"thought": "...", "action": "...", "action_input": {...}}。优点是机器解析容易,可以支持嵌套结构,适合参数复杂的工具。但缺点是小模型经常生成非法 JSON,容错成本高。
还有一种是“XML 标签式”,类似<thought>...<action>...,解析也方便,但会让输出长度变大,不利于上下文积累。
我实际跑下来,最稳的是“双行式 + 简单的 JSON action input”。首先用Action:行确定工具名,再用Action Input:行放 JSON 参数。这样即使模型把格式写错,我们也能从两行文本里拿到大部分有效信息,不会因为一个括号错了就全盘失败。而 Action Input 里的 JSON,我会做二次容错,比如尝试直接json.loads,失败后用正则提取大括号内容再试一次。
3. 核心实现:一套可直接修改的最小循环代码
3.1 代码骨架与主循环
我用 Python 实现了整个循环,零第三方框架依赖,唯一的依赖是用来调用模型的 OpenAI 兼容客户端(openai库或requests都行)。下面是完整的最小循环代码,我直接贴出来,后面会逐块解释。
import json import re from openai import OpenAI class ReActAgent: def __init__(self, model_name, system_prompt, tools, max_iterations=8, api_base=None, api_key="EMPTY"): self.model_name = model_name self.system_prompt = system_prompt self.tools = tools # dict: name -> {"description": str, "function": callable} self.max_iterations = max_iterations self.client = OpenAI(base_url=api_base, api_key=api_key) def run(self, user_query): messages = [{"role": "system", "content": self.system_prompt}] messages.append({"role": "user", "content": f"用户问题:{user_query}"}) for step in range(self.max_iterations): response = self.client.chat.completions.create( model=self.model_name, messages=messages, temperature=0.0 ) content = response.choices[0].message.content print(f"--- Step {step} ---") print(content) action, action_input = self._parse_action(content) if action == "final_answer": return action_input if action is None: # 解析失败:把错误信息反馈给模型,让它修正 messages.append({"role": "assistant", "content": content}) messages.append({"role": "user", "content": "你的输出不符合格式要求,请严格按照 Thought/Action/Action Input 格式重新输出,Action 必须是工具名或 final_answer。"}) continue if action not in self.tools: messages.append({"role": "assistant", "content": content}) messages.append({"role": "user", "content": f"错误:工具 '{action}' 不存在。可选工具为:{list(self.tools.keys())}。请重新选择。" }) continue tool_func = self.tools[action]["function"] try: observation = tool_func(**action_input) if isinstance(action_input, dict) else tool_func(action_input) obs_text = f"Observation: {json.dumps(observation, ensure_ascii=False)}" except Exception as e: obs_text = f"Observation: 工具执行出错:{str(e)}。请修正参数后重试。" messages.append({"role": "assistant", "content": content}) messages.append({"role": "user", "content": obs_text}) return "达到最大迭代次数,任务未完成。" def _parse_action(self, content): # 先尝试按 classical 格式解析 thought_match = re.search(r"Thought:(.*?)(?=\nAction:|\Z)", content, re.DOTALL) action_match = re.search(r"Action:\s*(\w+)", content) input_match = re.search(r"Action Input:\s*(.+?)(?=\n\s*(?:Thought|Action|Observation)|\Z)", content, re.DOTALL) if not action_match: # 尝试 JSON 格式 try: data = json.loads(content) return data.get("action"), data.get("action_input") except Exception: return None, None action = action_match.group(1) action_input = None if input_match: raw_input = input_match.group(1).strip() # 尝试解析为 JSON,失败则作为字符串 try: action_input = json.loads(raw_input) except json.JSONDecodeError: # 如果是以 Action Input 开头,尝试提取大括号 brace_match = re.search(r"\{.*\}", raw_input, re.DOTALL) if brace_match: try: action_input = json.loads(brace_match.group()) except json.JSONDecodeError: action_input = raw_input else: action_input = raw_input return action, action_input这段代码已经是一个能跑的原型了。可以看到,主循环非常直白:发消息、拿输出、解析动作、执行工具、追加 Observation、再来一轮。所有状态都在messages列表里,没有任何隐藏的缓存或者复杂的回调。
3.2 工具注册与调度
工具注册我用的是一个字典,每个工具包含一个描述和一个真实的 Python 函数。字典的 key 就是模型输出里 Action 字段需要用到的名字。这样写的好处是扩展新工具只需要往字典里加一项,不用改主循环逻辑。
假设我们要给 Agent 配一个计算器和一个获取当前时间的工具:
def calculator(expression: str): """简易计算器,仅用于示例,生产环境请用 ast 或 eval 白名单。""" try: return eval(expression) # 仅演示用,生产请谨慎 except Exception as e: return f"计算失败: {e}" def get_current_time(): from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S") tools = { "calculator": { "description": "计算数学表达式,入参 expression 是字符串形式,例如 '1+2'。", "function": calculator }, "get_current_time": { "description": "获取当前时间,无需参数。", "function": get_current_time } }这里有几个细节值得注意。第一,工具名一定要用英文且不要带空格,比如calculator、search,因为模型的输出格式正则用了\w+匹配,空格会破坏解析。第二,工具描述一定要把参数格式写清楚,例如expression需要是什么类型、什么含义,模型才能正确生成 JSON。第三,如果工具不需要参数,描述里要明确说“无需参数”,否则模型可能会自作主张塞一个空 JSON,虽然不影响解析,但某些实现会报错。
系统 Prompt 里的工具列表我通常会根据工具字典自动生成,避免手工维护两份内容。有个小技巧:把每个工具定义里的description和实际函数签名合并,用inspect.signature自动提取参数名,再让模型按这些参数名输出 JSON,解析效果会好很多。
3.3 模型输出解析与容错
解析模块是整个 Agent 里最容易出问题的地方,因为模型输出永远不按你的预期来。我见过迭代轮数无数次的模型把Action:写成Action:(全角冒号),也见过把Action Input里的 JSON 中括号配错。所以解析必须有多层容错。
我当前使用的解析逻辑分两个阶段:
第一阶段,先按经典的正则提取Thought:、Action:、Action Input:。用正则re.search(r"Action:\s*(\w+)", content)提取动作名,用re.search配合re.DOTALL提取 Action Input 里一直到下一个Thought:或Action:或结尾的内容。
第二阶段,如果第一阶段的 Action 没匹配到,就尝试把整个内容当作 JSON 解析。这个情况偶尔发生在模型天马行空直接输出了一个 JSON 对象的时候。
在 Action Input 的解析上,我会先尝试json.loads,如果失败,再尝试从原始字符串里找大括号段。因为模型经常会在 JSON 前后加解释性文本,比如Action Input: 我来看看,这个表达式是 {"expression": "1+2"}。。遇到这种情况,直接json.loads会失败,用re.search(r"\{.*\}", raw_input, re.DOTALL)再提一次,成功率会高很多。
另一个值得注意的容错点是,当解析完全失败时,我没有直接终止 Agent,而是把“格式错误”作为一个新的用户消息拼回去,要求模型重新输出。这个做法让小模型的最终成功率提升了不少,因为小模型在对话中会逐渐学会格式。
3.4 上下文管理策略
手写 ReAct 遇到的另一个大问题是上下文膨胀。每轮循环你都要把模型的 Thought 和 Observation 放回 messages,而 Observation 可能是很长的工具返回结果(比如搜索返回几十条文本)。几轮迭代下来,Token 消耗会很恐怖,而且模型会被越来越长的历史干扰,忘记最初的用户问题。
我的解决办法是设置“保留窗口”。不是无脑把所有历史都塞回去,而是只保留最近 N 轮原始对话,加上最初的问题。具体实现可以在加入 Observation 之前,对 messages 做一次裁剪。
def _trim_messages(self, messages, keep_rounds=6): # messages 结构: [system, user_query, assistant, observation, assistant, observation, ...] # 保留 system 和第一条 user query fixed = messages[:2] recent = messages[2:] if len(recent) > keep_rounds * 2: recent = recent[-keep_rounds * 2:] return fixed + recent这里keep_rounds表示保留几轮“助手回复 + 用户 Observation”对。对大多数任务,保留最近 4~6 轮就够了。如果确实需要长期记忆,可以单独用一个 summary 消息,每次循环都用模型对历史做一次压缩摘要,再把摘要放到 messages 最前面。但对一个最小实现来说,保留窗口方案最简单也最有效。
要注意的是,裁剪历史会丢失一些早期 Observation,所以如果模型在最后需要某个早期工具的结果,它可能拿不到。解决方法是告诉模型“如果你需要更早的信息,可以重新调用工具获取”,这样它就不会因为历史被裁剪而卡住。
4. 实操中的常见翻车现场与排查方案
4.1 模型就是不按格式输出怎么办
这是最常遇到的问题,尤其是小模型或者用中文 Prompt 的时候。常见症状是模型直接输出“我可以帮你算一下”而不输出 Action。我试过很多方案,最终有效的是两条:
第一,把 Prompt 中的“必须使用工具”改成非常消极的表述,比如“你能调用工具,请严格按照格式输出,否则任务失败”。正面指令有时被模型当成建议,但“否则失败”这种强约束更容易被遵循。
第二,当解析失败时,不要简单地重试同一个 Prompt,而是要明确告诉模型它哪里错了。例如:
你的输出缺少 Action 字段。请重新生成: Thought: 你的思考 Action: 工具名 Action Input: JSON把错误反馈加上之后,模型下轮输出大概率能纠正。我还在 Prompt 里加过 few-shot 示例,效果也不错,但会增加 Token 消耗。我的经验是,先用强约束语句,再配上错误反馈重试,基本能覆盖 90% 的失格式情况。
4.2 Agent 陷入死循环,怎么优雅终止
手写 Agent 比框架还容易死循环,因为很多模型的推理能力一般,会反复调用同一个工具。我曾经遇到过模型在一个计算器工具上来回使用了八轮,每次结果都一样,就是不输出 final_answer。
处理思路有两个。第一是设置最大轮数,比如我默认max_iterations=8,到达上限直接返回“未完成”。第二是设置重复检测,如果模型连续三轮调用同一个工具且传入参数完全一样,我就强制终止并提示模型“你已重复调用,请直接给出最终答案”。
重复检测的实现很简单,在循环里维护一个 previous_actions 列表:
if action in previous_actions and action_input in previous_actions[action]: messages.append({"role": "user", "content": "你刚刚已经调用过相同参数的工具,结果没有变化,请停止重复,直接输出 final_answer。"})这个小技巧能显著降低无意义的循环。另外,在 System Prompt 里也最好写一句“如果工具结果已经足够回答用户问题,请立即输出 final_answer,不要继续调用工具”,这能从源头上减少死循环的概率。
4.3 上下文越跑越长,效果衰减明显
当工具的 Observation 很长时,比如搜索返回了 5000 字的文本,整个 messages 会变得非常重。模型在长上下文里的表现会急剧下降,经常忘记一开始的用户问题,开始胡说八道。
除了前面说的保留窗口方案,对单个长 Observation 做截断也很重要。我通常在追加 Observation 之前,先对结果做一次长度限制,比如只保留前 500 个字符,超出部分用“(结果过长,已截断)”标记。这样能保证上下文规模可控。如果工具结果本身是结构化的,也可以让工具直接返回摘要,比如搜索工具只返回标题和第一段内容。
另一个实用技巧是:在每次模型输出 Thought 的时候,把用户问题原样再拼一次到最近的用户消息里。比如:
用户问题:计算 123 乘以 456 的结果 最近的观察结果:... 请继续推理,不要忘记你最初的任务。这种做法虽然笨,但对小模型特别有效,相当于不断提醒模型别跑偏。
4.4 工具调用的参数解析和错误处理
Action Input 的 JSON 解析失败是手写 Agent 里最常见的工程问题。除了做多层正则提取外,我还会在工具函数签名上做宽容处理。比如calculator接收expression参数,如果模型传了{"expr": "1+2"}而不是{"expression": "1+2"},就会执行失败。
我的处理方式是写一个参数归一化的小函数:取出 action_input 里的所有值,如果工具函数只有一个参数且 action_input 是字符串,就直接传字符串;如果是 dict,就尝试匹配函数签名里的参数名,匹配不到就用**kwargs尝试。灵活一点能让 Agent 少因为参数名问题失败很多次。
工具执行异常处理上,不要直接让整个 Agent 崩溃。我之前在工具函数里不加 try/except,导致一次搜索 API 超时直接抛异常让整个循环退出。后来改成在工具调用外层统一捕获异常,并把错误信息变成 Observation 反馈给模型,模型往往能自动调整参数再试一次。这个改动让 Agent 的健壮性上了一个大台阶。
5. 从最小实现到扩展实战的几点心得
最后再聊几句扩展方向。这套最小 ReAct 循环虽然简单,但你在做实操时会发现,它的可扩展性其实超乎想象。想升级的时候,可以从这几个方向发展:
工具侧:我后来加了搜索、读写文件、调用外部 API 的工具,每个工具的注册方式都一样,就是往字典里加一个函数。注意读写文件等工具要加权限校验,不能让模型随便删改系统文件。
模型侧:因为我用的是 OpenAI 兼容接口,所以本地部署的 vLLM、Ollama 服务都能直接替换 base_url。小模型建议把 temperature 设成 0,并且加大 max_tokens,因为 ReAct 格式输出需要较长文本。
循环侧:可以加入计划模式,让模型先输出一个 Plan 再执行多步行动,变成“Plan-and-Execute”风格;也可以加入反思机制,让模型在最终答案前自己检查一遍。
我个人在实际操作中的体会是,手写 ReAct Agent 最大的价值不是得到一个多强大的系统,而是让你真的理解了 Agent 的工作方式和各类框架的设计取舍。后续遇到再复杂的调度逻辑,你都能很快看出它本质上还是“模型输出指令,代码执行工具,结果反馈模型”这个闭环。只要把闭环的稳定性做好,Agent 能力就稳了一半。
如果让我重新走一遍这个过程,我会更早把“上下文压缩”“重复调用检测”“格式错误反馈”这三个模块加上,它们的收益比任何花哨的功能都高。这套代码现在依然在我的项目里服役,生产环境下也验证过稳定性。如果你正在被各种 Agent 框架搞得一头雾水,不妨像我一样,找个周末把这段循环写一遍,跑几个真实工具,你会看到完全不一样的世界。