ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Agentic设计模式实战:从ReAct到Multi-Agent的工程落地指南

Agentic设计模式实战:从ReAct到Multi-Agent的工程落地指南 简介《Agentic Design Patterns: A Hands-On Guide to Building Intelligent Systems》英文原版PDF面向具备一定人工智能、机器学习或软件工程背景的研发人员、技术负责人及AI产品经理。书中重点讲解ParallelAgent与SequentialAgent的并行化执行、短期上下文与长期记忆管理、Human-in-the-Loop人机协同、RAG知识检索增强、任务优先级排序、多代理协作、评估监控机制以及推理引擎内部工作机制并结合Google ADK、LangChain、LangGraph等工具给出大量可运行代码示例。资源共1个文件类型为PDF压缩包约17.56MB便于携带与离线阅读目前已有570人学习/下载。通过研读本书读者可系统掌握智能代理系统的高效架构方法理解流程编排与并行化对效率的提升学会在复杂任务中应用记忆管理和上下文工程进而设计具备人类监督与反馈的安全可控AI系统并构建支持自我验证与合同式交互的高可信度智能体。1. Agentic Design Patterns为什么让模型干活和让模型问答是两套工程帮我写一段 Python 代码和帮我去这个网页抓取数据、清洗、画图并把结论发给同事——同样是用大语言模型后者需要的不是一条更长的 prompt而是一整套工程化的设计模式。这些模式有一个共同的名字Agentic Design Patterns。它们是构建智能系统的可复用骨架解决的是怎么让模型在未知环境里自主决策、调用工具、从失败中恢复这个核心问题。本文从最小可运行的代码入手拆解 ReAct、Reflexion、Plan-and-Execute 和 Multi-Agent 四种模式覆盖选型、参数、踩坑与线上评估让读到的工程师能一天跑通原型、两周打磨成可上线系统。2. Agentic 系统的基础组件模型、工具与记忆的三角关系2.1 模型选型原生函数调用比让模型输出 JSON靠谱得多所有 Agent 循环的本质都是模型决定下一步做什么、调用什么工具、以什么参数调用。这个决定的质量上限直接决定了整套智能系统的能力上限。我一开始做 Agent 时走过弯路——在 system prompt 里写请以 JSON 格式输出你的下一步动作然后模型偶尔会在 JSON 外面多包一层 markdown 代码块标记偶尔把布尔值写成字符串解析层越补越脏。后来切到模型的原生函数调用function calling / tool use这类玄学问题基本绝迹因为输出格式的约束在解码阶段就强制生效了。以 Claude 的 API 为例工具定义长这样tools [ { name: search_web, description: 在互联网上搜索信息返回排名前 k 条结果每条包含标题、URL 和摘要, input_schema: { type: object, properties: { query: { type: string, description: 搜索关键词尽量精简为 2-5 个词 }, k: { type: integer, description: 返回结果条数默认 5最大不超过 10 } }, required: [query] } } ]这段代码里最关键的不是 schema 的类型约束而是 description 字段。模型决定调用哪个工具、填什么参数读的就是这段描述。描述写得太抽象模型就会填出离谱的参数。我在一个搜索工具上做过对比描述从一句话扩写成三句补充结果包含哪些字段k 的含义超时会怎样工具选择的准确率从 82% 提到 91%。参数方面k 默认 5 就好——搜索类工具一次带回太多结果上下文里会塞满无关信息token 成本翻倍但收益趋近于零。2.2 工具执行层模型只负责说要做什么代码负责真的去做工具调用里最容易踩的坑是让模型直接去执行代码或操作数据库。这里有个安全原则必须立住模型永远不直接执行任何东西它只生成一个调用意图tool_use 结构化对象真正执行的是你写的代码。执行完再把结果转成文本、塞回对话上下文。def execute_tool(tool_name: str, arguments: dict) - str: 执行工具调用并返回文本结果。 无论工具原本返回什么类型都统一转成字符串。 LLM 上下文是纯文本结果越简洁、结构化越清晰越好。 if tool_name search_web: results search_web(arguments[query], arguments.get(k, 5)) lines [f{i1}. {r[title]} | {r[url]} | {r[snippet]} for i, r in enumerate(results)] return \n.join(lines) elif tool_name run_sql: results run_sql(arguments[query]) # 强制截断只返回前 50 行超过后明确告知 truncated results[:50] extra if len(results) 50 else f\n-- 已截断共 {len(results)} 行 return str(truncated) extra else: raise ValueError(f未知工具: {tool_name})这里有两个容易翻车的地方。第一执行结果必须截断。一个不带 LIMIT 的 SQL 查询可能返回几万行直接塞进上下文会把窗口撑爆后面所有推理都会退化。第二工具的错误信息要作为正常结果返回而不是抛异常让整个循环崩掉。让模型看到查询失败: 列 user_id 不存在这样的报错文本它能自己修正 SQL 再试一次而如果你抛异常模型就完全不知道发生了什么。我一般把错误文本拼接成工具执行失败: 原始错误的格式返回模型把它当作普通上下文处理成功率明显更高。2.3 记忆分层工作记忆与长期记忆不能混在一个列表里Agent 的记忆不是单个概念至少得拆成两层。工作记忆是当前对话的完整消息列表每次请求都带上它决定模型记得什么代价是 token 线性增长。长期记忆放在外部存储里——向量数据库、键值库都行——每次对话前检索最相关的片段注入上下文。很多 Agent 项目做到一半发现上下文越来越长、费用越来越贵就是因为没有分层。class Memory: def __init__(self, vector_store): self.messages [] # 工作记忆完整对话消息 self.vector_store vector_store # 长期记忆的外部存储 def add_to_context(self, user_message: str, assistant_message: str): self.messages.append({role: user, content: user_message}) self.messages.append({role: assistant, content: assistant_message}) def retrieve_relevant(self, query: str, top_k: int 5) - str: 从长期记忆中检索与当前问题相关的历史片段 docs self.vector_store.similarity_search(query, ktop_k) return \n---\n.join(doc.page_content for doc in docs) def build_prompt(self, user_query: str) - list: memory_context self.retrieve_relevant(user_query) system_prompt ( 你是一个智能客服助手。以下是该用户的历史相关信息\n f{memory_context}\n 结合这些信息回答当前问题与问题无关的历史信息不要主动提及。 ) return [{role: system, content: system_prompt}] \ self.messages[-20:] \ [{role: user, content: user_query}]参数说明self.messages[-20:] 是一个滑动窗口只保留最近 20 条消息作为工作记忆。这个数值是在一个客服场景里试出来的少于 10 条用户中途改需求时模型会失忆前后回答矛盾多于 40 条token 成本翻倍但回答质量几乎不涨。长期记忆的 top_k 我设在 5检索出的片段本身会占几百 token太大的 top_k 容易把不相关的历史带进来造成幻觉传导。3. 四种核心设计模式原理、实现与适用边界3.1 ReAct 模式推理与行动的交替循环ReActReasoning Acting是 Agent 最底层的循环结构模型先想——分析当前状态、决定下一步再做——调用工具拿到新信息然后基于新信息继续想。这个交替过程让模型能处理需要多步搜索、多步计算的复杂任务而不是一次问答就结束。它也是最容易手写实现的一种不需要额外框架。def react_loop(task: str, max_steps: int 8): messages [ {role: system, content: 你是一个能调用工具的智能体。 逐步完成任务。每次回复先输出 thought简短推理再输出 action。 当任务完成时回复 final_answer 并提供最终结果。}, {role: user, content: task}, ] for step in range(max_steps): response llm_call(messages, toolstools) if response.stop_reason tool_use: tool_use response.content[-1] # 结构化字段不是自由文本 result execute_tool(tool_use.name, tool_use.input) messages.append({role: assistant, content: response.content}) messages.append({role: user, content: f工具 {tool_use.name} 返回:\n{result}}) else: return response.content return 已达最大步数任务可能未完成逻辑说明核心分支是 response.stop_reason tool_use。Claude、GPT 系列在模型决定调用工具时API 会返回结构化的 tool_use 对象而不会把调用意图混在普通文本里。你要做的只是把这个对象解析出来、执行、回填结果。注意 messages 列表里 assistant 的回复要原样追加一旦漏了这步模型会丢失自己上一轮说了什么的上下文。max_steps 是保险丝——防止 Agent 在某个任务上无限循环烧完预算。参数说明max_steps 我通常设在 58。查一下今天的天气这种一步任务两步就完成这家公司去年的营收和利润率是多少这种需要搜索、定位财报、读取数字的深链任务8 步是合理的上限。设得太大模型会在复杂任务中迷失方向反复调用同一个工具。设得太小任务还没走完就被强制终止用户体验是问着问着就没下文了。3.2 Reflexion 模式把每一次失败变成下一次的输入ReAct 最大的问题是一次性——这轮失败了下轮从零开始同样的错误会重复犯。Reflexion 模式在循环外面包了一层反思逻辑任务失败后先让模型自己分析为什么失败、下次要怎么改把这段反思文本存起来下一次尝试时注入 system prompt。相当于给 Agent 加了经验积累机制。def reflexion_loop(task: str, max_attempts: int 3): memory [] # 反思记录每次尝试后追加 for attempt in range(max_attempts): reflexion_prompt if memory: reflexion_prompt ( 以下是之前尝试的失败分析。这次执行必须避免同样的错误\n \n.join(f[第{i1}次] {m} for i, m in enumerate(memory)) ) result react_loop(task, max_steps6, extra_systemreflexion_prompt) if evaluate_success(result, task): return result failure_analysis llm_call([ {role: system, content: 分析上述任务失败的原因。 用一句话指出最关键的错误再用一句话给出具体改进建议。}, {role: user, content: f原始任务: {task}\n本轮结果: {result}}, ]) memory.append(f失败分析: {failure_analysis}) return result # 最后一次结果逻辑说明evaluate_success 是这个模式的灵魂。简单场景可以用规则判断比如结果里是否包含 2024 年营收复杂场景只能再调一次模型来评判。Reflexion 的成本比 ReAct 高每一次失败都要多一次反思调用加上原本的重试消耗所以 max_attempts 设为 3 是我实践下来成本和质量之间的平衡点——第五次尝试的边际收益已经很低了。参数说明把反思内容注入 system prompt 而不是追加到对话末尾是个刻意的选择。反思属于全局行为指导模型在每一步决策时都应该看到它如果放在对话末尾它只影响最后一步前面的步骤仍然在重复犯错。还要在反思文本前加第 N 次的前缀避免模型把旧尝试的失败经验误认为当前对话的一部分。3.3 Plan-and-Execute 模式先规划再动手不要走一步看一步ReAct 在复杂任务上有个致命弱点没有全局视角容易被中途的某个中间结果带偏。比如任务分析这份销售数据的季度趋势模型可能在第一步看到某个异常数值后就开始深挖忘了真正要交付的是趋势报告。Plan-and-Execute 把过程拆成两段先让模型一次性生成完整计划然后逐条执行、每条执行后检查是否需要修订计划。def plan_and_execute(task: str): plan llm_call( f为以下任务生成执行计划。要求\n f1. 每一步必须是可验证的具体动作\n f2. 标注每步需要调用的工具\n f3. 控制在 5 步以内\n任务: {task}, json_modeTrue # 输出结构化计划列表 ) for i, step in enumerate(plan[steps]): step_result react_loop(step[instruction], max_steps4) # 执行完每步后让模型判断计划是否需要调整 adjustment llm_call( f原计划: {plan}\n已完成步骤: {i1}/{len(plan[steps])}\n f步骤结果: {step_result}\n f基于此后续步骤需要调整吗如需要请给出修订后的完整计划否则回复 继续。 ) if adjustment.strip() ! 继续: plan parse_adjusted_plan(adjustment) return finalize(plan, results_accumulator)逻辑说明注意这里的每一步内部用的仍然是小型的 ReAct 循环。Plan-and-Execute 和纯 ReAct 的关键区别是决策粒度后者每步都重新考虑下一步做什么前者只在计划的节点上做调整。这带来两个实际好处——token 消耗更低不用每步都重新审视全局行为更可预测计划列表摆在眼前你随时能看出模型哪里想偏了。但代价是灵活性下降如果任务本身的步骤数就不确定计划模式会显得僵硬。选型建议任务有明显的多步骤结构、步骤间依赖清楚比如爬取→清洗→汇总果断选 Plan-and-Execute。任务本身是开放探索型的比如帮我调研这个行业里值得关注的五家创业公司用 ReAct 更合适因为搜着搜着方向就会变。关于 5 步这个限制我的经验是超过 5 步的计划模型在生成时就会开始丢步骤执行时更是频繁调整。真遇到超过 5 步的任务宁可拆成两层计划。3.4 Multi-Agent 模式多个角色分工协作的架构与陷阱多 Agent 模式是把不同系统提示词、不同工具集的多个 Agent 组合在一起。最常见的拓扑是编排者-执行者一个 Orchestrator Agent 负责任务理解、拆解和结果汇总多个 Worker Agent 各自处理一个子任务。这种架构在直觉上很吸引人——让每个 Agent 专注一件事在实际部署中翻车的比例却相当高。做好它需要非常严格地控制边界。def orchestrator(task: str): subtasks llm_call( f将任务拆解为 2-3 个子任务。每个子任务必须\n f- 能独立完成不依赖其他子任务的中间结果\n f- 明确标注需要的工具\n任务: {task}, json_modeTrue ) results {} for subtask in subtasks[subtasks]: worker Agent( system_prompt( f你是负责以下子任务的专家: {subtask[description]}\n 只完成分配给这个子任务的工作不要试图解决整个问题。 ), toolsfilter_tools(subtask[required_tools]) # 只给该子任务需要的工具 ) results[subtask[id]] worker.run(subtask[instruction]) return llm_call(f汇总以下子任务结果生成最终交付\n{results})逻辑说明最重要的一条设计原则是限制每个 Worker 的工具集。不要图省事把所有工具都传给所有 Agent——模型倾向于调用顺手的工具而不是正确的工具。一个负责文本摘要的 Worker 如果手里有搜索工具它在拿不准数据时会去搜索而不是老老实实做摘要输出就不可控了。第二个关键点编排者给 Worker 的指令必须自包含。Worker 看不到整个对话历史它只看到编排者传来的那段 instruction所以指令里必须写清背景、输入、期望输出格式。多 Agent 不是银弹。两个 Worker 之间如果存在隐式依赖——比如第二个任务需要第一个任务的输出来决定怎么做——编排者必须显式地把第一个结果拼进第二个指令里。这个拼装逻辑写起来很繁琐步骤一多容易漏。我的建议先想清楚子任务之间到底是顺序、并行还是交叉交叉依赖的任务不要硬拆留在单 Agent 里做完。多 Agent 适合的是每个子任务需要不同的上下文和工具的场景而不是为了架构表演拆任务。4. 模式选型先给任务做个体检再决定用哪种 Agentic 架构4.1 五个选型维度复杂度、出错成本、预算、延迟、可解释性面对一个具体需求我习惯先拉一张选型表逐项过。这比凭感觉拍脑袋选模式可靠得多维度关键问题倾向模式任务复杂度需要几步步骤之间是顺序还是交叉步骤少用 ReAct步骤明确用 Plan-and-Execute出错成本错了能否自动修正修正代价多大高可靠要求用 ReflexionToken 预算单次任务能接受多少 token 消耗预算紧用 Plan-and-Execute延迟要求用户能等多久需要多快的首个 tokenPlan-and-Execute 可并行步骤延迟更低可解释性需要向用户展示为什么这么回答吗Plan-and-Execute 天然有计划列表可展示这张表的关键是倾向而不是决定。比如一个任务步骤多但延迟要求高Plan-and-Execute 能让部分步骤并行执行就是对工具层的要求更高——得保证各步骤之间没有数据依赖。4.2 什么时候不该用 Agent规则化任务的止损线先泼盆冷水不是所有任务都需要 Agent 架构。如果任务是固定流程、输入输出结构确定、判断逻辑可以用 if-else 写清楚那普通代码比 Agent 快 10 倍、便宜 100 倍。用不用 Agent的判断依据不是这个需求听起来智能而是是否存在需要实时决策的开放环节。适合上 Agent 的场景多跳问答A 公司并购了 B 公司B 公司的 CTO 现在在哪儿任职、需要动态决定工具组合的分析任务评估这批客户数据里的异常行为、用户意图不固定的人机对话。不适合的场景固定告警触发if-else 足够、批量格式转换Python 脚本足够、有标准查询模板的报表参数化 SQL 足够。强行上 Agent 的后果是系统不可预测——同样输入今天可能跑通明天模型换了个版本就中途翻车运维成本远超它省下的那点智能化。4.3 混合模式是生产常态单模式只存在于教学示例中在实际生产系统里纯粹用单一种模式的项目非常少见。外层用 Plan-and-Execute 控制全局节奏内层步骤用 ReAct 处理开放探索遇到失败用 Reflexion 重试——这是我参与过的数据分析智能体的标准三层结构class DataAnalysisAgent: def run(self, query: str): plan self.planner.plan(query) # 第一层计划拆解 results {} for step in plan: if step[type] query_db: results[step[id]] self.sql_agent.run(step[instruction]) # 第二层sql_agent 内部是 ReAct允许它自己调 schema、改 SQL elif step[type] visualize: results[step[id]] self.chart_agent.run(step[instruction]) if not self.validator.validate(results): # 第三层失败反思 for retry in range(2): results self.retry_agent.improve(results, self.validator.last_error) if self.validator.validate(results): break return self.responder.finalize(results)参数说明这个混合架构里每个子 Agent 的工具边界是最重要的配置。sql_agent 只能调用 query_db 和 describe_schema 两个工具chart_agent 只能调用绘图工具。边界越清晰出错时越容易定位——数据不对立刻知道是 SQL 层还是图表层的问题不用在整条链路上翻日志猜原因。retry 次数我设 2 次超过 2 次仍未通过验证就直接返回当前结果并标注需人工复核不要无止境地重试烧钱。5. Agent 系统避坑指南五个高频故障与排查思路5.1 上下文窗口被撑爆模型突然变笨先查输入里塞了什么现象系统跑了一段时间后Agent 的回答越来越短、越来越泛甚至直接在日志里报 context length exceeded。用户反馈它好像失忆了。原因工具返回结果没有截断记忆滑动窗口没实现。SQL 查询返回上万行、网页爬取把全文塞进去、搜索工具一次带回来 20 条链接——上下文里 90% 是噪声真正的问题指令被淹没在中间模型既找不到重点也记不住开头。解决给每个工具的执行结果设置硬性截断超过 N 行/字符就省略并标注工作记忆用滑动窗口限制长度大段文本网页全文、完整财报先存到外部存储上下文里只放摘要或检索片段。给 execute_tool 加一个统一的 max_length 参数所有工具的输出在返回前都过一遍这个截断是最快的止血方案。5.2 Agent 陷入死循环同一个工具调用几十遍参数都不带变的现象日志里看到 Agent 反复执行同一个搜索或查询参数一模一样结果一模一样但模型就是不进入下一步。原因最常见的是模型没从工具结果里拿到新东西。工具结果被截断得太狠或者格式混乱导致模型没看懂它判断这次搜索没成功于是盲目重试。第二个常见原因是工具执行异常但错误信息没有正确返回模型看到的还是成功的假象不知道换一条路。解决工具出错时把错误拼成文本正常返回搜索失败: API rate limit exceeded而不是抛异常让循环崩溃在 max_steps 之外再加一道同工具连续调用次数的约束——连续调用同一工具超过 2 次强制模型改写策略。这两个约束加完之后死循环问题基本绝迹。5.3 工具调用格式不稳定模型不按函数调用协议出牌现象模型没有返回结构化的 tool_use 对象而是在文本里写我想调用一下搜索工具关键词是……你这边解析器完全拿它没办法。原因模型服务端没启用函数调用能力或者用的模型版本太旧、本身就不支持工具调用。另一个可能你在 messages 里混用了不支持 tool use 的旧模型同一个请求在切换模型后行为直接变了。解决优先选原生支持工具调用的模型Claude 3.5 及以上、GPT-4 系列、Qwen 2.5 的工具调用版本。如果业务必须用不支持工具调用的开源模型可以在 system prompt 里给一个严格的输出模板——你只能输出 JSON格式为 {thought:...,tool_name:...,arguments:{...}}——并加一个输出解析兜底解析失败时不要无限重试直接返回需要人工介入。这条兜底的目的是控制损失而不是修复模型。5.4 幻觉在多轮中传导错一次就错一路Reflexion 也会放大现象Agent 在第一步编造了一个不存在的公司名或数据后续所有推理都基于这个错误假设展开最后的结果错得离谱。更隐蔽的是Reflexion 模式把这次错误反思记了下来下一次正确回答反而被旧的错误经验污染。原因模型在需要信息但工具没返回结果的时候倾向于用内部知识补齐空缺而补齐的内容可能就是幻觉。这个问题在多 Agent 协作中更严重——一个 Worker 的输出成为另一个 Worker 的输入错误信号逐级放大。Reflexion 的污染则是因为反思记录没有区分这次失败的具体原因和全局正确做法。解决在 system prompt 里明确写一条工具优先级规则——不知道的信息必须调用 search_web 获取不得自行假设。这条规则要放在 prompt 靠前的位置。对 Reflexion给每条反思加上尝试序号和时间戳注入时明确标注以下是历史经验仅用于参考以当前任务上下文为准。这条标注虽然简单实测能显著减少错误经验覆盖本轮新认知的问题。5.5 可观测性黑洞Agent 为什么这么回答日志里根本找不到线索现象线上出了个用户投诉你打开日志发现只有最终几行输出中间每一步的推理、工具调用参数、中间结果全部没有记录只能靠猜。原因Agent 系统本质上是非确定性的每个步骤的输入输出都直接影响最终结果但很多人仍然像调普通 API 一样只记录请求和响应——丢了中间态。排查问题时没有过程数据等于在黑匣子里修系统。解决在 ReAct 循环的每轮迭代里把模型看到了什么、调了什么工具、拿到了什么结果按结构化 JSON 打到日志系统每条日志带 request_id 用于关联整条链。我一般用专门的 Agent 追踪工具LangSmith 或自己封装 tracing 中间件每个 span 记录完整上下文。这个改造做在所有优化之前——没有过程日志后面所有的评估和迭代都是盲人摸象越优化越玄学。6. 落地评估从能跑到可靠的三个验证技巧6.1 先造 30 条样本的评估集再谈效果优化我接手过的每个 Agent 项目第一周做的事不是调 prompt而是花半天构造一个 30 条样本的评估集。这个评估集不需要完美但要覆盖三种情况正常任务 10 条、边界输入 10 条空输入、超长输入、歧义指令、缺失关键信息、预期失败 10 条任务本身不可完成Agent 应该识别出来并说做不了而不是硬编一个答案。每条样本标注期望答案以及更重要的期望过程——不调用不该调的工具、不编造工具结果、不在不必要的外部搜索上浪费时间。6.2 把过程指标纳入回归别只看最终答案只盯着最终答案准确率会漏掉大量问题。我给每个 Agent 系统配三个过程指标工具调用成功率执行成功次数除以总调用次数、平均步数超过上限说明模型在做无效探索、上下文溢出率。这三个指标放进每次迭代的回归报告。有一次我扩写了工具描述最终答案准确率没变化但工具调用成功率从 91% 升到 96%平均步数从 6 降到 4——改进真实存在只是被最终准确率掩盖了。不盯过程指标这种正向改动很容易被误判为无效。6.3 给 Agent 装一个人肉后门灰度期手动干预钩子最后一条务实的建议Agent 必须能在任意步骤被人工截断。每次工具调用前加一个回调钩子允许操作员检查参数、修改参数或者直接终止任务。这个钩子在灰度期尤其重要——系统上线初期你还没摸清行为边界有人能踩刹车比任何测试集都管用。def react_loop_with_hook(task: str, on_tool_useNone): # on_tool_use(tool_name, args) - (action, modified_args) # action: allow / modify / abort for step in range(max_steps): response llm_call(messages, toolstools) if response.stop_reason tool_use: tool_use response.content[-1] if on_tool_use: action, new_args on_tool_use(tool_use.name, tool_use.input) if action abort: return 已由人工终止 if action modify: tool_use.input new_args result execute_tool(tool_use.name, tool_use.input) messages.append({role: user, content: f工具返回:\n{result}})我的习惯是每次上线新系统都安排一周陪跑每天抽 10 条真实用户请求人工核对每一步工具调用是否合理把不合理案例追加进评估集第二天立刻迭代。这一周积累的边界案例比之后跑一个月的自动化回归更有价值——因为它们来自真实用户不是自己拍脑袋造出来的。这套从模式拆解、代码实现到评估迭代的流程是我踩了不少坑之后固定下来的做法希望帮到你。本文还有配套的精品资源点击获取
返回列表