1. 为什么我要自己造一个提示流编排器去年下半年我接手了一个内部知识库问答项目最初的想法很简单调个大模型接口把用户问题拼上检索到的文档片段返回答案就完事了。结果上线两周就被打脸——用户问“帮我查一下上季度华东区的销售数据然后跟华南区做个对比最后生成一份简报”模型直接懵了。它只能回答“我无法访问实时数据”或者更糟编一串看起来像模像样的假数字。这就是纯对话式 LLM 的天花板它只有一张嘴没有手也没有脚。你问它今天天气它只能根据训练数据里的统计规律瞎猜你让它算个数它可能给你一个自信满满的错误答案。Agent智能体这个概念之所以这两年被反复提及核心就在于给大模型装上“手脚”——让它能调用工具、能分步推理、能根据中间结果决定下一步动作。我试过市面上几个现成的编排框架要么太重改一个节点要翻半天文档要么太黑盒出了问题只能看日志猜。索性自己动手从零搭一套轻量的提示流编排器。这篇文章就是整个过程的复盘包括 Agent 节点的设计思路、Tools 工具调用体系的实现细节、ReAct 循环的落地方式以及我在这个过程中踩过的坑。如果你也在做 Agent 开发或者想理解 Function Calling 到底怎么串起来这篇应该能帮你省下不少试错时间。2. 整体架构设计把复杂问题拆成可编排的节点2.1 核心设计原则节点化、可观测、可替换动手写代码之前我先定了三条规矩后面所有设计都围绕它们展开。第一一切皆节点。不管是调用 LLM、执行工具、做条件判断还是简单的文本处理统统抽象成节点。节点有输入、有输出、有配置节点之间通过边连接。这样做的好处是整个流程可以用一张有向图表示调试的时候能清楚看到数据在哪个节点卡住了。第二可观测优先。Agent 最让人头疼的就是“黑盒感”——它为什么调这个工具为什么循环了五次还不结束我在每个节点执行前后都埋了事件钩子记录输入输出、耗时、token 消耗。这些数据不仅用于调试后面做成本优化也靠它。第三工具可替换。工具层用注册表模式管理每个工具声明自己的名称、描述、参数 schema 和执行函数。想加一个新工具注册进去就行编排层不需要改任何代码。这一点在后期扩展时救了我很多次。2.2 为什么选 ReAct 而不是纯 Function Calling这里要展开说一下。Function Calling是模型厂商提供的能力你给它一组工具定义它返回一个结构化的调用请求。听起来很美好但实际用起来有个问题它是“单轮”的。模型决定调哪个工具你执行完把结果塞回去模型再决定下一步。如果任务需要多步推理你得自己写循环。ReActReasoning Acting的思路更完整让模型在每一步先输出思考过程Thought再决定动作Action然后观察结果Observation循环往复直到得出最终答案。这个“思考-行动-观察”的循环天然适合处理需要多步工具调用的任务。我最终采用的是混合方案底层用 Function Calling 保证工具调用的结构化输出上层用 ReAct 循环驱动多步推理。这样既利用了模型原生的工具调用能力又保留了推理链的灵活性。实测下来对于“查数据→算对比→写简报”这类任务混合方案的成功率比纯 Function Calling 高出不少。2.3 数据流与状态管理整个编排器的数据流是这样的用户输入进入起始节点节点依次执行每个节点可以读取和写入一个共享的上下文对象Context。这个 Context 里存着对话历史、中间结果、工具调用记录、当前步数等。状态管理我用了最朴素的方式——一个可序列化的字典。每次节点执行完把变更写回 Context同时发一个事件通知。为什么不用更复杂的状态机因为 Agent 的执行路径本身就不确定强行套状态机会让代码变得很别扭。朴素的字典反而最灵活配合事件钩子调试信息也够用。注意Context 一定要可序列化。我一开始图省事往里面塞了数据库连接对象结果做快照和回放时直接报错。后来改成只存纯数据需要连接时从外部注入。3. Agent 节点与 Tools 工具调用体系的核心实现3.1 Agent 节点的内部结构拆解一个 Agent 节点本质上是一个“带工具调用能力的 LLM 调用器”。它的执行流程可以拆成五步组装提示词把系统提示、对话历史、可用工具列表拼成模型能理解的格式。调用模型发送请求拿到模型的响应。解析响应判断模型是想调用工具还是想直接回复。执行工具如果是要调工具找到对应工具执行拿到结果。循环或结束把工具结果塞回对话历史回到第 1 步如果模型给出了最终答案节点结束。这五步里第 3 步是最容易出问题的。不同模型厂商返回的工具调用格式不一样有的用tool_calls字段有的用function_call还有的干脆把调用意图写在文本里让你自己解析。我在解析层做了一层适配统一转成内部格式上层逻辑就不用关心底层差异了。3.2 工具注册表的设计与实现工具注册表是整个体系的地基。每个工具需要声明四样东西字段说明示例name工具唯一标识query_sales_datadescription给模型看的说明“查询指定区域和时间的销售数据”parameters参数 JSON Schema{region: string, quarter: string}handler实际执行函数def query_sales_data(region, quarter): ...description 的写法很讲究。我踩过的坑是描述写得太简略模型不知道什么时候该用写得太啰嗦又浪费 token 还容易干扰判断。后来总结出一个模板“做什么 什么时候用 返回什么”。比如“查询销售数据。当用户询问具体区域的业绩数字时使用。返回该区域该季度的销售额和同比增长率。”parameters 用 JSON Schema 定义这是 Function Calling 的标准做法。模型会根据这个 schema 生成参数所以字段名要清晰类型要准确必填项要标出来。我见过有人把参数定义成一个大对象结果模型经常漏填字段排查半天才发现是 schema 没写清楚。3.3 ReAct 循环的落地细节ReAct 循环的伪代码看起来很简单while 步数 最大步数: 响应 调用模型(历史 工具列表) if 响应是最终答案: return 响应 else: 结果 执行工具(响应.工具名, 响应.参数) 历史.append(工具结果)但实际写起来有几个细节必须处理。最大步数限制。这是防止死循环的第一道防线。我设的是 10 步超过就强制结束并返回当前已有信息。为什么是 10因为实测下来绝大多数任务在 5 步内能完成10 步足够覆盖复杂场景再多了大概率是模型卡在某个循环里出不来。重复调用检测。有时候模型会反复调同一个工具、传同样的参数。我在循环里加了一个检测如果连续两次工具调用完全一样就注入一条提示“你已经调用过这个工具并得到了结果请基于已有信息给出答案”强制模型跳出循环。工具执行超时。外部工具可能因为网络或服务问题卡住。每个工具执行都包了超时控制默认 30 秒超时返回错误信息给模型让它决定是重试还是换方案。3.4 提示词模板的工程化处理系统提示词是 Agent 的“人格设定”直接决定它的行为模式。我用的模板大致是这样的结构你是一个可以调用工具的智能助手。 当前可用工具 {tool_descriptions} 请按以下格式回应 - 如果需要调用工具输出工具名称和参数。 - 如果已有足够信息直接给出最终答案。 注意 1. 不要编造工具返回的数据。 2. 如果工具返回错误尝试其他方案或告知用户。 3. 回答要简洁不要重复工具返回的原始数据。这个模板我改了十几版。最早没写“不要编造数据”结果模型在工具返回空结果时自己编了一组数字。后来加了“如果工具返回错误尝试其他方案”模型的重试行为明显更合理了。提示词工程不是一蹴而就的得根据实际 bad case 反复调。4. 完整实操从零跑通一个多工具 Agent4.1 环境准备与依赖选型我用的技术栈比较克制Python 3.11主要是看中它的 asyncio 改进和类型提示支持。FastAPI提供 HTTP 接口方便前端调用和调试。Pydantic做参数校验和序列化和 JSON Schema 天然契合。OpenAI SDK调用模型接口它对新版 Function Calling 支持最好。没有用 LangChain 这类框架原因前面说了——想要完全的控制权。依赖少还有一个好处部署简单一个 Dockerfile 就能打包。4.2 定义第一个工具销售数据查询先定义一个最简单的工具来跑通链路from pydantic import BaseModel, Field class SalesQueryParams(BaseModel): region: str Field(description区域名称如华东、华南) quarter: str Field(description季度格式如2025Q3) def query_sales_data(region: str, quarter: str) - dict: # 实际项目中这里查数据库 mock_data { (华东, 2025Q3): {sales: 1250000, yoy: 0.15}, (华南, 2025Q3): {sales: 980000, yoy: 0.08}, } result mock_data.get((region, quarter)) if not result: return {error: f未找到{region}{quarter}的数据} return result tool_registry.register( namequery_sales_data, description查询指定区域和季度的销售数据。当用户询问业绩数字时使用。返回销售额和同比增长率。, parametersSalesQueryParams, handlerquery_sales_data, )这里用 Pydantic 模型定义参数注册时自动转成 JSON Schema。handler 返回字典编排层会把它序列化成字符串塞回对话历史。4.3 组装 Agent 节点并接入编排流Agent 节点的核心逻辑大概两百行关键部分是这样class AgentNode: def __init__(self, llm_client, tool_registry, max_steps10): self.llm llm_client self.tools tool_registry self.max_steps max_steps async def execute(self, context): messages context.get(messages, []) for step in range(self.max_steps): response await self.llm.chat( messagesmessages, toolsself.tools.get_schemas(), ) if response.has_tool_calls(): for call in response.tool_calls: result await self.tools.execute( call.name, call.arguments ) messages.append({ role: tool, tool_call_id: call.id, content: json.dumps(result), }) else: context[final_answer] response.content return context context[final_answer] 达到最大步数限制未能完成任务 return context这段代码跑起来后我拿“对比华东和华南上季度销售”测试模型的行为是先调query_sales_data(华东, 2025Q3)拿到结果后再调query_sales_data(华南, 2025Q3)最后生成对比文字。整个链路通了。4.4 加入第二个工具验证多工具协作只调一个工具说明不了问题我又加了一个计算工具class CalcParams(BaseModel): expression: str Field(description数学表达式如(1250000-980000)/980000) def calculate(expression: str) - dict: try: # 生产环境要用安全的表达式求值 result eval(expression, {__builtins__: {}}, {}) return {result: result} except Exception as e: return {error: str(e)}然后问 Agent“华东比华南上季度销售额高多少百分比”这次模型先调两次查询工具再调计算工具算增长率最后给出答案。多工具协作跑通了。实操心得工具的 description 里一定要写清楚“什么时候用”。我一开始计算工具的 description 只写了“执行数学计算”结果模型在不需要计算的时候也去调它。改成“当需要对数字进行加减乘除运算时使用不要用于简单的事实查询”之后误调用明显减少。5. 踩坑记录与常见问题排查5.1 模型不调工具直接编答案这是最常见的问题。表现是明明注册了工具模型却直接给一个看起来合理的答案完全不走工具调用。排查下来有三个原因。一是系统提示词没强调“必须使用工具获取数据”二是工具 description 写得太模糊模型不确定该不该用三是模型本身的能力问题某些小参数模型对 Function Calling 的支持就是不稳定。解决办法在系统提示里加一句“对于需要实时数据或精确计算的问题必须调用工具不得凭记忆回答”。同时把工具 description 写具体。如果还不行考虑换一个对工具调用支持更好的模型。5.2 工具参数解析失败模型生成的参数有时候不符合 schema比如该传字符串的传了数字该传数组的传了单个值。我在工具执行前加了一层参数校验和修复def validate_and_fix_params(params, schema): # 类型转换、缺失字段补默认值、多余字段剔除 ...校验失败时不直接报错而是把错误信息返回给模型让它重新生成参数。这样模型有机会自我修正比直接失败体验好很多。5.3 循环不终止前面提过模型可能卡在循环里。除了最大步数和重复检测我还加了一个“进展检测”如果连续三步的工具调用没有产生新的有效信息比如都是返回同样的错误就强制结束。5.4 常见问题速查表问题现象可能原因排查方向模型不调工具提示词未强调/描述模糊检查系统提示和工具 description参数解析失败schema 不清晰/模型能力不足加参数校验和修复层循环不终止缺少步数限制/重复检测加最大步数和重复调用检测工具执行超时外部服务慢/网络问题加超时控制和重试逻辑结果不稳定温度参数过高调低 temperature 到 0.1 以下token 消耗过大历史消息太长做历史截断或摘要压缩5.5 几个容易被忽略的细节工具返回结果要精简。我一开始把数据库查询的完整结果塞回去几百行 JSONtoken 直接爆炸。后来改成只返回关键字段模型反而更容易抓住重点。错误信息要友好。工具执行失败时返回给模型的错误信息要写清楚“什么失败了、可能的原因是什么”而不是直接抛一个堆栈。模型看到“数据库连接超时请稍后重试”比看到“ConnectionError at line 42”更能做出合理决策。日志要记全。每次模型调用、每次工具执行、每个节点的输入输出都要记下来。出问题时这些日志就是唯一的线索。我用的是结构化日志每条记录带 trace_id方便串联。6. 后续可以怎么扩展这套编排器目前跑通的是最核心的链路但离生产可用还有距离。我接下来打算做几件事。一是加并行工具调用。现在工具是一个一个串行执行的如果两个工具之间没有依赖完全可以并行跑能省不少时间。Function Calling 本身支持一次返回多个调用请求编排层改成并发执行就行。二是做工具结果的缓存。同样的查询在短时间内重复执行直接返回缓存结果既省 token 又省时间。缓存 key 用工具名加参数哈希。三是加人工审核节点。对于敏感操作比如写数据库、发邮件在工具执行前插入一个人工确认步骤。这个用编排图的条件边就能实现不需要改核心逻辑。四是做可视化编排界面。现在流程是代码里写死的后面想做成拖拽式的让非技术人员也能配置简单的 Agent 流程。这个工作量不小但价值很大。如果你也在做类似的东西建议先从最小闭环开始——一个模型、一个工具、一个循环跑通了再往上加。我见过太多人一上来就设计复杂的多 Agent 协作架构结果连单工具调用都没调稳。先把地基打牢后面的扩展都是水到渠成的事。