2026/8/30 15:20:00

手写最小ReAct Agent:AI Agent开发从原理到工程落地

手写最小ReAct Agent:AI Agent开发从原理到工程落地 最近在调研 Agent 生产落地的过程中正好赶上 AGNTCon 阿姆斯特丹九月阵容公布的消息。近几年 Agent 相关的技术会议越来越多讨论焦点也从“怎么做一个能聊天的 Demo”逐渐转向“怎么让 Agent 在线上稳定跑起来”。这篇文章不打算做新闻搬运而是借这个行业节点把 AI Agent 开发从概念、原理、代码到工程落地完整梳理一遍。文中会给出一个不依赖第三方框架的最小 ReAct Agent 完整代码可以直接复制运行也会补充接入真实大模型的替换思路、生产环境常见问题和最佳实践。无论你是刚接触 Agent 的初学者还是正在构建企业级 Agent 平台的开发者都能从里面找到可复用的内容。1. AGNTCon 与 AI Agent 技术背景1.1 AGNTCon 是什么AGNTCon 是聚焦 AI Agent 的技术会议阿姆斯特丹站安排在九月目前 speaker lineup 已经对外公布。对开发者来说这类会议的真正价值不是追星而是观察行业里最活跃的工程实践和踩坑经验。AI Agent 从 2023 年开始快速升温到 2025 年已经不只是一个实验室概念而是大量出现在客服、运维、数据分析、内网知识问答等真实业务场景里。不同团队对 Agent 的理解并不完全一致。有的团队把所有基于大模型的自动化流程都叫 Agent有的团队则认为只有具备“感知—决策—行动”闭环的系统才算 Agent。这个差异导致了很多沟通成本。所以我在写这类技术内容时会先把概念边界讲清楚再展开具体实现。1.2 AI Agent 解决什么问题大语言模型本身只能做一件事根据输入生成文本。它没有手不能真正操作数据库、调用接口、发送邮件也不能感知外部世界的变化。Agent 做的事情就是给模型装上“手”和“眼睛”。从产品视角看Agent 解决的核心问题是“从对话到行动”。普通 Chatbot 回答完问题就结束了Agent 则在回答之前或回答之后可以主动查询数据、调用工具、基于返回值继续推理直到完成一个具体任务。举个例子普通 Chatbot用户问“北京今天适合穿什么”模型直接根据训练数据或上下文生成一段建议。Agent用户问同样的问题Agent 先调用天气工具拿到北京实时温度、风力、空气质量再结合这些数据生成穿衣建议。这个差异看起来不大但用户体验完全不同。前者是“猜”后者是“查完再答”。1.3 掌握 Agent 需要理解哪些基础概念在进入代码之前先明确几个高频词术语通俗解释Tool / Tool Calling给模型提供的可调用能力比如查询天气、查数据库、发邮件Memory记忆Agent 在对话中保留的信息分短期记忆和长期记忆Orchestration编排控制 Agent 何时调用工具、如何切换策略的逻辑ReAct 循环一种经典 Agent 工作模式思考Reason 行动Act不断交替后面几节会围绕这些概念展开。2. AI Agent 技术栈全景2.1 Agent 的四个核心层不同框架对 Agent 的分层不完全一样但大多数实现都可以归为四个层模型层是 Agent 的大脑负责理解用户意图、生成决策和最终回复。现在主流模型都支持 Function Calling也就是让模型以结构化 JSON 输出“我想调用哪个工具、参数是什么”。记忆层负责保存对话历史、业务数据和长期偏好。简单场景下把最近几轮对话拼进 Prompt 就够了复杂场景可能需要向量数据库做语义检索或者用外部存储保存用户画像。工具层是 Agent 的手可以是普通 HTTP API、数据库查询函数、代码解释器也可以是内部系统 RPA 脚本。工具层的设计质量直接决定 Agent 的能力边界。编排层是 Agent 的逻辑中枢负责决定“下一步该干什么”。它可能是简单的 if-else也可能是复杂的规划器。2.2 主流 Agent 框架对比现在市面上 Agent 框架非常多这里列几个主流方向的适用场景框架特点适合场景LangChain生态丰富组件多文档全快速验证 Agent 原型做 RAG 类应用LlamaIndex偏数据检索和理解知识库问答、文档分析AutoGen强调多 Agent 对话协作研究复杂任务拆解、多角色协作CrewAI角色化 Agent 编排使用简单业务角色模拟比如分析师执行者Semantic Kernel微软出品面向企业集成与 .NET / Java 技术栈结合的场景自研轻量框架自己控制循环和 Prompt深度定制、生产环境要求可控这里要特别提醒框架更新速度很快接口经常变化。我不建议在文章里写死某个框架的 API 版本因为读者看到文章时可能已经过期。选型时以官方文档和 GitHub 最新 release 为准。2.3 选型建议我的经验是做 Demo 可以随便选一个顺手的框架但做生产系统要谨慎评估三点可控性框架封装的层级越深排查问题越困难。依赖成本引入一个重型框架往往要连带引入一堆间接依赖。升级风险框架大版本升级可能破坏现有 Agent 行为。如果你对 Agent 的理解还不够深建议先不用任何框架自己实现一个最小循环。理解原理之后再引入框架会清楚很多。这一篇的第四部分就会带大家走一遍这个过程。3. 核心原理拆解Agent 是怎么工作的3.1 ReAct 模式ReAct 是 Agent 领域最经典的模式名字来自 Reason Act。整个流程可以拆成四步循环思考Thought模型根据当前上下文分析用户要什么、还缺什么信息。行动Action模型决定调用哪个工具并给出参数。观察Observation系统执行工具把结果返回给模型。重复或回答模型判断信息是否足够不够则继续思考足够则输出最终答案。用文字表示就是下面这个流程用户问题 ↓ Thought: 我需要先获取天气信息 ↓ Action: get_weather(city北京) ↓ Observation: 北京当前温度 24℃天气晴 ↓ Thought: 信息已足够可以回答 ↓ Final Answer: 北京今天适合穿薄外套这个循环看起来简单却是很多 Agent 框架最底层的骨架。3.2 Plan-and-Execute 模式ReAct 适合单步决策但遇到复杂任务时模型每走一步都要重新思考效率低且容易失控。Plan-and-Execute 的思路是先让模型生成一个完整计划再逐步执行用户问题: 帮我写一份简单的竞品分析报告 计划: 1. 搜索竞品 A 的最新动态 2. 搜索竞品 B 的最新动态 3. 对比核心功能差异 4. 生成报告 执行: 第 1 步 → 第 2 步 → ...这种模式的优点是任务结构清晰适合流程稳定的场景缺点是如果计划本身就错了后面所有步骤都会跟着错。3.3 Multi-Agent 协作多 Agent 协作是当前比较热门的方向。核心思想是让多个拥有不同角色、不同 Prompt、不同工具权限的 Agent 分工协作。比如一个负责搜索资料一个负责汇总分析一个负责最终审核。多 Agent 能解决单 Agent 上下文过长、角色职责混乱的问题但也引入了新的复杂度Agent 之间如何通信、如何避免互相覆盖状态、如何分配任务、如何保证整体不跑偏。建议团队在单 Agent 都还没跑稳的时候先不要上多 Agent 架构。3.4 上下文与记忆设计Agent 的上下文窗口是有限的你不能把所有历史都塞给模型。常见做法是短期记忆保留最近 N 轮对话控制在 token 预算内。摘要记忆定期把旧对话摘要成一段压缩文本。检索记忆通过向量检索找到与当前问题最相关的历史记录。对于工具调用类 Agent还需要把“已经调用过哪些工具、结果是什么”拼进上下文避免模型重复调用同一个工具。3.5 工具调用稳定性的关键工具调用是 Agent 最容易出问题的地方。你能想到的主要坑点包括模型输出了不存在的工具名。参数格式不符合工具要求。参数缺失、重复、类型不对。工具执行报错后Agent 没有处理错误而是继续乱调。这些问题需要在 Agent 循环里做防御性解析和错误处理。代码示例里我会用一个工具解析函数演示如何处理这些情况。4. 完整实战案例从零实现一个最小 Agent4.1 案例目标这一节我们实现一个不依赖任何第三方 Agent 框架的最小 ReAct Agent。它能做到用户询问当前时间时调用 get_time 工具。用户询问城市天气时调用 get_weather 工具。工具执行结果会返回给模型由模型生成最终回答。为了让代码可以离线直接运行我会先用一个 MockLLM 模拟模型输出然后给出接入真实大模型的替换方式。4.2 环境准备操作系统Windows / macOS / Linux 都可以。Python 版本3.9 及以上。依赖运行 Mock 版本只需要 Python 标准库接入真实大模型时需要requests库。安装 requests 的命令pip install requests示例项目结构如下agent-demo/ └── agent_demo.py为了让教程更清晰我把全部代码放在一个文件里方便直接复制运行。生产项目中建议按模块拆分。4.3 完整实现代码文件路径agent-demo/agent_demo.py 最小 ReAct Agent 演示 功能 1. 使用 Tool 类封装工具 2. 使用 Agent 类实现“思考→行动→观察”循环 3. MockLLM 模拟大模型输出方便离线运行 from typing import Callable, Dict, List import datetime class Tool: 工具封装一个工具包含名称、描述、参数说明和实际执行函数。 def __init__( self, name: str, description: str, func: Callable[[str], str], parameters_desc: str , ): self.name name self.description description self.parameters_desc parameters_desc self.func func def run(self, argument: str) - str: try: return str(self.func(argument)) except Exception as exc: return f工具执行失败: {exc} def prompt(self) - str: return f- {self.name}: {self.description}参数说明{self.parameters_desc} def get_time(_: str) - str: 获取当前时间。 return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_weather(city: str) - str: 查询城市天气模拟数据。 mock_data { 北京: {温度: 24, 天气: 晴}, 上海: {温度: 28, 天气: 多云}, 广州: {温度: 31, 天气: 阵雨}, } city city.strip() if city not in mock_data: return f没有查询到 {city} 的天气数据 info mock_data[city] return f{city}当前温度 {info[温度]}℃天气{info[天气]} class MockLLM: 模拟大模型输出。 为了教学演示这里通过简单的关键词判断来决定模型下一步动作。 真实项目中请将此类替换为对大模型服务的调用。 def generate(self, system_prompt: str, user_prompt: str) - str: # 如果上下文中已经有工具观察结果直接基于观察结果生成最终答案 if Observation: in user_prompt: lines user_prompt.splitlines() observation for line in lines: if line.startswith(Observation:): observation line[len(Observation:) :].strip() return fFinal Answer: {observation} # 从用户问题中提取原始问题 if 用户问题: in user_prompt: user_query user_prompt.split(用户问题: , 1)[1].strip() else: user_query user_prompt # 根据问题关键词决定调用哪个工具 if (现在 in user_query and 时间 in user_query) or 几点 in user_query: return Action: get_time\nAction Input: if 天气 in user_query: for city in [北京, 上海, 广州]: if city in user_query: return fAction: get_weather\nAction Input: {city} return Final Answer: 我没有理解你的需求请提供更明确的信息。 class Agent: 最小 ReAct Agent 主循环 def __init__(self, tools: List[Tool], llm, max_steps: int 5): self.tools {tool.name: tool for tool in tools} self.llm llm self.max_steps max_steps def run(self, user_query: str) - str: # 系统提示词告诉模型可以使用哪些工具以及输出格式 system_prompt ( 你是一个智能体可以调用以下工具来回答问题\n \n.join(tool.prompt() for tool in self.tools.values()) \n当需要调用工具时请严格按照以下两行格式输出\n Action: 工具名\n Action Input: 参数\n 当信息足够时直接输出Final Answer: 你的回答 ) messages [ (assistant, f用户问题: {user_query}), ] for step in range(1, self.max_steps 1): print(f\n Step {step} ) # 拼接对话历史 user_prompt \n.join( f{role}: {content} for role, content in messages ) response self.llm.generate(system_prompt, user_prompt) print(f模型输出:\n{response}) # 如果模型给出最终答案结束循环 if response.startswith(Final Answer:): return response[len(Final Answer:) :].strip() # 如果模型要求调用工具 if response.startswith(Action:): lines response.splitlines() action_name lines[0].replace(Action:, ).strip() action_input for line in lines[1:]: if line.startswith(Action Input:): action_input line.replace(Action Input:, ).strip() if action_name not in self.tools: return f错误未知工具 {action_name} tool_result self.tools[action_name].run(action_input) print(f工具 {action_name} 返回: {tool_result}) # 把模型决策和工具结果追加到对话历史进入下一轮循环 messages.append((assistant, response)) messages.append((user, fObservation: {tool_result})) else: return f错误模型输出格式不正确 - {response} return 达到最大步数仍未得到最终答案。 if __name__ __main__: tools [ Tool(get_time, 获取当前时间, get_time, 无参数), Tool( get_weather, 查询城市天气, get_weather, 城市名称例如 北京, ), ] agent Agent(toolstools, llmMockLLM(), max_steps5) print(测试 1查询时间) print(最终结果:, agent.run(现在几点了)) print(\n测试 2查询天气) print(最终结果:, agent.run(北京天气怎么样))4.4 运行与验证在项目目录下执行python agent_demo.py预期输出结果类似测试 1查询时间 Step 1 模型输出: Action: get_time Action Input: 工具 get_time 返回: 2025-06-10 14:30:00 Step 2 模型输出: Final Answer: 2025-06-10 14:30:00 最终结果: 2025-06-10 14:30:00 测试 2查询天气 Step 1 模型输出: Action: get_weather Action Input: 北京 工具 get_weather 返回: 北京当前温度 24℃天气晴 Step 2 模型输出: Final Answer: 北京当前温度 24℃天气晴 最终结果: 北京当前温度 24℃天气晴过程说明Agent 先让模型输出一个 Action系统执行工具拿到 Observation再把 Observation 拼入上下文模型在下一轮直接输出 Final Answer。这个循环就是 ReAct 最小实现。4.5 接入真实大模型的替换思路上面的 MockLLM 只是为了演示原理。实际项目中你需要把它替换为对真实模型服务的调用。这里给出一个兼容 OpenAI Chat Completions 格式的替换示例import requests class RemoteLLM: 以 OpenAI Chat Completions 兼容接口为例请根据实际服务商调整。 def __init__(self, api_base: str, api_key: str, model: str): self.api_base api_base.rstrip(/) self.api_key api_key self.model model def generate(self, system_prompt: str, user_prompt: str) - str: url f{self.api_base}/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: [ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], temperature: 0, } resp requests.post(url, jsonpayload, headersheaders, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content]使用时把MockLLM()替换为RemoteLLM(...)即可。需要注意三个问题不同模型服务商的接口格式不一定完全相同字段名可能有差异。如果你的模型不支持严格的 JSON 或结构化输出需要在系统提示词中强调输出格式并在代码里做容错解析。真实模型在工具名、参数格式上更容易出错建议在 Agent 循环中增加参数校验和重试机制。5. 从 Demo 到生产工程化要点5.1 可观测性Demo 里打印几条日志没问题生产环境必须建立完整的可观测体系。Agent 的排错难度远高于普通接口因为同一个问题模型可能今天能处理、明天就处理不了。建议至少记录以下信息用户原始输入。每次模型输出的完整内容。调用的工具名称、参数、耗时、返回结果。每轮循环的 token 消耗。最终是否成功以及失败原因。有了这些日志你才能在线上问题发生时回放 Agent 的完整决策过程。5.2 评估体系Agent 没有传统软件那样的确定性输出所以必须建立评估集。常见做法是准备一批典型用户问题给每个问题标注期望结果或期望调用的工具序列然后定期跑回归。评估维度可以包括任务完成率是否在有限步内得到最终答案。工具调用准确率是否调用了正确的工具和参数。回答质量最终答案是否准确、完整。成本指标平均每任务消耗多少 token、多少次工具调用。建议把评估脚本接入 CI每次修改 Prompt 或工具逻辑后自动跑一遍。5.3 成本与限流Agent 和普通接口最大的区别是一次用户请求可能触发多轮模型调用。如果控制不好一个小功能也可能烧掉大量 token。建议从三个方向控制成本限制最大循环步数避免死循环。对单用户请求做频率限制和配额控制。对工具调用做超时控制和并发限制。在代码示例里max_steps5就是最基础的防护。5.4 安全边界Agent 有了工具调用能力就意味着它可以直接影响业务系统。这里必须强调权限最小化原则给 Agent 的 API Key 权限要尽量小不能使用管理员账号。涉及删除、修改、转账、发布等敏感操作必须增加人工审批环节。工具返回的数据可能包含敏感信息日志中要做好脱敏。模型可能被提示注入攻击工具参数要做好校验和过滤。安全是 Agent 生产落地最重要的一条线不能为了体验牺牲控制。6. 常见问题与排查思路问题现象常见原因解决思路Agent 陷入死循环反复调用同一个工具模型每次拿到相同 Observation无法判断信息已足够设置最大步数在上下文中增加“你已经观察到的信息”使用更严格的结束条件工具名/参数格式解析失败模型没有严格按照格式输出使用结构化输出模式增强系统提示词代码中做容错解析工具执行报错后 Agent 继续乱调工具异常信息没有拼入上下文或模型忽略了异常执行异常时返回明确 Observation要求模型换一种方式处理上下文过长导致超限对话历史和工具结果过多做历史摘要、裁剪早期轮次、控制工具返回长度模型回答与工具结果不一致Prompt 中缺少约束模型凭训练记忆回答强制要求模型“只能基于 Observation 回答”不要自由发挥线上偶发失败但本地正常外部接口超时、限流、数据缺失增加重试和超时策略完善日志和监控下面展开几个高频问题的具体排查方式。死循环问题如果 Agent 在线上出现死循环先看日志中连续几步的 Observation 是否完全相同。如果工具返回结果稳定但模型还在反复调用说明模型没有意识到“信息已经够了”。可以在系统提示词中增加一句“当某个问题已经得到 Observation 结果时直接基于该结果输出 Final Answer。”模型输出格式不稳定不同模型对格式的遵从程度差异很大。建议使用支持结构化输出或 Function Calling 的模型接口让模型以 JSON 形式返回工具调用。如果只能用普通文本输出代码里要兼容多种格式并记录格式异常样本定期分析优化 Prompt。上下文超长工具返回结果可能非常大。比如数据库查询返回几百行记录直接塞进 Prompt 既浪费 token 又容易超限。建议在工具层做结果裁剪只把关键字段或前 N 行返回给模型完整结果通过另一个查询接口获取。7. 最佳实践与工程建议7.1 架构设计Agent 服务不要和业务系统耦合太深。建议拆分成以下模块Agent 核心服务负责对话编排和工具调用逻辑。工具网关统一管理工具注册、鉴权、限流、超时。模型网关统一接入不同大模型支持模型路由和降级。评估与监控平台负责日志、评估、告警。这样做的好处是工具变更不需要改动 Agent 核心代码模型切换也不需要业务方感知。7.2 Prompt 与工具设计工具数量不要一开始就铺开每增加一个工具Agent 的决策空间就变大一点出错概率也随之增加。工具描述要写清楚“什么时候用、什么时候不用”比堆功能列表更有效。参数设计遵循简单原则能传一个字符串就不要拆成多个复杂对象。Prompt 版本要纳管每次修改记录 diff方便回滚。7.3 部署与运维Agent 服务的部署和普通服务没有本质区别但更强调版本控制。模型的 Prompt、工具列表、模型版本、代码版本四个维度任意一个变化都会改变 Agent 行为。建议发布时把四者作为一个整体版本记录。线上切换模型时一定要先在灰度环境跑评估集确认效果不低于当前版本再放量。7.4 数据与隐私Agent 处理的数据往往比普通接口更多、更敏感因为上下文里可能包含用户输入、工具返回、历史记录。在数据接入前就要规划好哪些字段可以做工具参数。哪些数据必须脱敏后才能传给模型。日志中哪些内容不能落盘。第三方模型服务是否符合公司数据安全要求。8. 总结与学习路线这篇文章从 AGNTCon 阿姆斯特丹九月阵容公布这个行业动态出发完整梳理了 AI Agent 的核心概念、技术栈、ReAct 原理并手写了一个最小可运行的 Agent。通过示例代码你应该能理解 Agent 循环的本质模型负责决策系统负责执行和观察循环反复直到完成任务。下一步的学习路线建议按这个顺序推进先跑通本文的最简 ReAct 代码改改工具函数增加一个新工具。把 MockLLM 替换成真实大模型体验真实模型输出带来的不稳定问题。学习 Function Calling 的标准用法减少格式解析错误。加入 RAG 能力让 Agent 能访问私有知识库。再做多 Agent 协作但前提是单 Agent 已经足够稳定。生产环境优先关注评估和安全两条线。不要急着把 Agent 抛给所有用户先用小流量验证效果再逐步扩大范围。AI Agent 的发展速度很快会议动态和框架版本都在持续变化具体议程和工具细节记得以官方信息为准。如果你也想动手实践建议从今天这份最小代码开始边跑边改收获会非常大。