PTC架构：把工具编排从模型循环中剥离，实现稳定可控的LLM Agent

1. 先搞清楚模型循环里的工具编排到底卡在哪这两年做LLM Agent应用但凡涉及工具调用Function Calling / Tool Use绝大多数人一开始的写法都是同一个套路把工具调用直接塞进模型循环里。伪代码长这样while True: response llm.chat(messages) if response.tool_calls: for call in response.tool_calls: result execute_tool(call.name, call.arguments) messages.append(tool_result_message(call, result)) continue else: return response.content这个循环看起来挺自然的模型觉得该调工具就调工具调完把结果喂回去模型再接着想。但你真在业务里跑一阵子就会发现这条路越走越窄。先说最直观的Token消耗完全不可控。每调一次工具就得把工具返回结果拼回对话上下文里再重新发给模型。工具一多、步骤一长上下文轻松破万token。我见过一个做竞品舆情分析的Agent一次完整分析要调搜索、取页面、提取关键信息、对比历史数据一轮下来上下文飙到三万token光调用成本就让人肉疼。这还只是成本问题。更麻烦的是模型的决策不可控。模型在长循环里很容易“绕圈”同一个工具反复调用、明明上一步结果已经足够却还要再查一次、两个工具之间的调用顺序搞反。有一次我调试一个自动化报表Agent模型在循环里把查询工具连续调了六次每次参数都差不多只是因为前一次的结果里有个字段格式不太对它就一直想“再查一次试试”。这种问题在简单demo里很难暴露一旦工具超过五六个、步骤超过三四步模型的“流程感”就会明显下降。还有工程上的痛苦编排逻辑全写在循环里没法独立测试。你想验证“A工具失败后应该走B工具重试”这个逻辑没法单独测必须启动整个LLM链路还得祈祷模型这轮决策是对的。更别提条件分支和并发——模型循环里想实现“如果返回结果超过阈值就并发调三个工具”你得在prompt里描述这个逻辑然后再在循环代码里写一堆if else去承接模型的任意输出稍有不慎就崩。我印象最深的是一次生产事故模型在循环里连续两次调用同一个目标工具第二次因为上游依赖还没就绪直接报错报错信息又被塞回上下文模型为了弥补又调了一次结果数据已经不一致了。整个循环来回折腾了十几轮最后生成了一份完全错误的报表。这些问题的根源其实不在模型而在架构。你把流程控制权交给了每轮都在变化的模型输出等于让一个“每次发挥不太稳定的同事”来当项目总指挥他既要负责具体干活又要负责记住整个工作流程不出乱子才怪。2. PTC的核心思路把编排从“思考循环”里搬出来后来我换了个思路也是这篇文章要聊的PTC方案。PTC是我内部对这套架构的简称全称Pipeline Tool Composition核心就一句话工具编排不写在模型循环里而是下沉到一个独立的运行时中由运行时负责流程控制模型循环只负责单步决策。打个比方你就明白了。以前的做法是你雇了一个特别聪明但容易走神的实习生让他一边回忆公司流程一边干活干着干着还自作主张改流程。PTC的做法是你先把标准作业流程写在SOP手册里这就是“编排”让实习生只回答“当前这一步该选哪个方法”具体先做什么后做什么、失败怎么重试、哪些步骤可以并行全由SOP和监工运行时说了算。这个“搬”字是精髓。搬出模型循环之后三个层面的问题一次性解决。第一流程确定性回来了。工具的调用顺序、依赖关系、重试策略、超时处理全都在运行时里用代码和配置定义好。模型再怎么发挥不稳定运行时都会按图执行不允许模型跳过某个必做步骤也不允许它把两个工具的顺序搞反。这就把“Agent行为”从概率问题变成了确定性问题。第二调试和测试变得正常了。编排图是一份独立的、不依赖模型的数据结构。你可以直接对编排逻辑写单元测试模拟某个工具返回错误验证重试路径是否生效模拟某个节点的输出满足分支条件验证是否走对了下一跳。这些测试不需要启动LLM跑得飞快还能在CI里每天跑。第三成本显著下降。因为流程是运行时控制的模型循环不再需要反复“看着上下文想下一步”。每个决策点是一次独立的小规模LLM调用上下文只需要包含当前节点需要的信息而不是整个历史轨迹。我实测下来同样一个多步调研任务PTC方案的Token消耗大约是传统模型循环写法的三分之一到四分之一。PTC不是什么玄乎的新理论它其实是软件工程里最朴素的“控制反转”思想在LLM Agent场景的一次应用。模型不再是流程的所有者它退化成一颗“决策内核”而运行时成为流程的所有者。3. PTC运行时的架构拆解3.1 核心组件调度器、节点与上下文一个可用的PTC运行时最少需要三个核心组件调度器Scheduler、节点Node、上下文Context。调度器是整个运行时的引擎它维护一个待执行节点的队列或就绪集合按照编排图定义的依赖关系按顺序或并发地触发节点执行。节点是最小的执行单元每个节点内部封装一个具体的工具调用逻辑。上下文则是贯穿整个运行时的数据载体节点的输入从上下文中读取输出也写回上下文节点与节点之间不直接通信只通过上下文交换数据。这其实是把Actor模型和黑板模式揉在了一起。节点像Actor各自独立执行上下文像黑板大家都往上写、从上面读。这样做的好处是解耦任何一个节点都可以被单独替换只要输入输出约定不变编排图不需要改。另外还要有一个注册中心Registry用来维护“工具名 → 节点执行函数”的映射。模型决策层返回的是一个逻辑工具名比如search_web调度器拿到这个名字去注册中心查到对应的执行函数然后调用它。注册中心的存在让“模型的输出”和“实际执行的代码”之间有一层薄薄的隔离你可以随时替换底层实现模型完全无感。3.2 编排图怎么定义有向无环图加条件边编排图我建议用有向无环图DAG来建模节点是工具调用或逻辑单元边是依赖关系。为什么限定无环因为环意味着死循环而在运行时里你并不希望模型或某个异常条件触发无限循环。真出现“循环处理”的需求应该用“自环带计数上限”的结构显式建模而不是让图本身成环。每个节点需要定义几个关键属性name全局唯一的节点名也是模型决策层能看到的名字。inputs声明这个节点需要从上下文的哪些key读数据。outputs声明这个节点会往上下文的哪些key写数据。retry_policy失败重试策略重试次数、退避间隔。timeout单次执行的超时时间避免某个工具卡死拖着整个流程。边可以分成普通依赖边和条件边。普通依赖边就是“A执行完才能执行B”。条件边是“A执行完之后根据A的输出结果或上下文的某个值决定是否执行B”。条件边的判定条件是一个纯函数(ctx) - bool调度器在执行完A后对每条条件边求值为true的边会触发对应的下游节点加入就绪队列。这比状态机要灵活得多。状态机适合“状态少、转移明确”的场景而工具编排往往有十几个节点、几十条依赖关系写状态机你光状态爆炸就够受的。DAG加条件边既能表达“顺序”“并行”“条件跳转”这些常见流控又不需要维护一张大状态表。3.3 与模型循环的接口设计整个PTC架构里最容易设计得别扭的就是模型循环和运行时之间的接口。我的做法是把接口收敛成一个决策点Decision Point。决策点是一个挂在普通节点上的特殊标记表示“这个节点执行完之后需要问一次模型由模型选择下一步走哪条边”。运行时一旦进入决策点会调用DecisionProvider传入当前上下文的摘要信息和可选的边列表模型返回一个选择结果比如“走边edge_2”。接口设计成两个核心方法class DecisionProvider: def decide(self, ctx: Context, choices: list[Edge]) - Edge: 模型从给定的边里选一条走 ...模型在这个接口里只做一件事从有限的候选中选一个。不负责记忆流程不负责构造复杂的计划甚至不需要知道整个编排图长什么样。它只需要根据当前上下文信息做一个选择题。这大大降低了模型出错的空间——选择题显然比开放题稳得多。如果你做的是纯规则场景甚至可以不接模型直接写一个规则决策函数比如“if ctx.get(temperature) 30: return edge_high”。这一层抽象让同一个编排图既能跑LLM决策也能跑规则决策测试的时候尤其方便——你可以用规则决策器跑一遍全流程单测再切到LLM决策器跑线上。4. 从零落地搭建一个PTC运行时4.1 定义节点一个节点就是一个工具我们先从最底层的节点定义开始。我用Python写借助pydantic做数据校验保证节点输入的合法性。每个节点就是一个普通类继承一个BaseNode基类实现run方法。from typing import Any import asyncio, time class BaseNode: name: str base inputs: list[str] [] outputs: list[str] [] retry_policy: dict {max_retries: 2, backoff: 1.0} timeout: float 10.0 async def run(self, ctx: dict[str, Any]) - dict[str, Any]: raise NotImplementedError class SearchWebNode(BaseNode): name search_web inputs [query] outputs [search_results] async def run(self, ctx): q ctx[query] # 这里接真实的搜索API results await async_search(q) return {search_results: results}run的输入是整个上下文字典输出是一个字典调度器会把这个字典合并回上下文。节点之间通过上下文key解耦不直接持有彼此引用。重试和超时不需要写在每个节点里调度器统一处理。这很重要否则每个节点都自己写一遍重试逻辑代码会快速腐烂。调度的职责就是拿到节点 → 按retry_policy做重试 → 按timeout做超时熔断 → 把结果写回上下文。4.2 定义编排图把流程写成人话编排图我用一个PTCGraph类来定义支持链式调用让图的结构能直观地从代码里读出来。做一个“调研报告生成”的场景先搜索再抓取页面然后判断内容量是否足够不够就补充搜索最后生成报告。class PTCGraph: def __init__(self): self.nodes {} self.edges [] def add_node(self, node: BaseNode): self.nodes[node.name] node return self def add_edge(self, src: str, dst: str, conditionNone): self.edges.append({ src: src, dst: dst, condition: condition }) return self def get_out_edges(self, node_name: str) - list[dict]: return [e for e in self.edges if e[src] node_name]用这个Graph类来定义调研流程def build_research_graph(): g PTCGraph() g.add_node(SearchWebNode()) g.add_node(FetchPageNode()) g.add_node(EvaluateSufficiencyNode()) # 这个节点是“决策点” g.add_node(SupplementSearchNode()) g.add_node(GenerateReportNode()) g.add_edge(search_web, fetch_page) g.add_edge(fetch_page, evaluate_sufficiency) # 条件边内容不够才补充搜索 g.add_edge(evaluate_sufficiency, supplement_search, conditionlambda ctx: ctx.get(sufficient) is False) # 条件边内容够了就直接生成报告 g.add_edge(evaluate_sufficiency, generate_report, conditionlambda ctx: ctx.get(sufficient) is True) g.add_edge(supplement_search, generate_report) return g把流程写成人话这是PTC架构里最直观的收益。你不需要看代码逻辑去反推流程读图就是读流程。而且这张图是数据结构不是散落在循环里的逻辑分支你随时可以用它来画可视化。4.3 实现调度器就绪队列加依赖计数调度器是运行时的心脏。我用依赖计数实现每个节点在被加入图时先数清楚有几个前置依赖每有一个前置节点执行完依赖计数减一减到零就进就绪队列。执行完成就继续释放下游依赖。整个过程用asyncio实现天然支持并发。class Scheduler: def __init__(self, graph: PTCGraph): self.graph graph self.dep_count {} self.reverse_deps {} for edge in graph.edges: self.dep_count.setdefault(edge[dst], 0) self.dep_count[edge[dst]] 1 self.reverse_deps.setdefault(edge[src], []).append(edge) async def run(self, initial_ctx: dict, start_node: str): ctx dict(initial_ctx) ready asyncio.Queue() ready.put_nowait(start_node) while not ready.empty(): node_name await ready.get() node self.graph.nodes[node_name] # 执行前先检查条件边的条件是否满足 incoming_edges [e for e in self.graph.edges if e[dst] node_name] skip False for e in incoming_edges: if e.get(condition) and not e[condition](ctx): skip True break if skip: continue # 带重试和超时地执行节点 result await self._execute_with_retry(node, ctx) ctx.update(result) # 释放下游节点 for out_edge in self.graph.get_out_edges(node_name): self.dep_count[out_edge[dst]] - 1 if self.dep_count[out_edge[dst]] 0: await ready.put(out_edge[dst]) # 如果是决策点调用DecisionProvider if isinstance(node, DecisionPointNode): provider self.decision_provider chosen await provider.decide(ctx, self.graph.get_out_edges(node_name)) await ready.put(chosen.dst) return ctx你注意到没有这个调度器本身完全不知道“模型”的存在。它只是在执行一张图碰到决策点就调用一个外部provider。这就是PTC架构的好处——运行时是通用的换成什么模型、甚至换成规则调度器都不需要改。4.4 决策点的实现模型只做选择题决策点节点很轻它本身不执行工具只负责触发一次模型选择。实际实现时我会先让EvaluateSufficiencyNode把“内容摘要”和“当前可选动作”写进上下文然后决策点的run里把这两部分拼进DecisionProvider的输入。class DecisionPointNode(BaseNode): name decision_point inputs [decision_context] outputs [decision_result] async def run(self, ctx): # 实际决策由外部provider完成这里只是个占位 return {decision_result: None}DecisionProvider的内部实现核心是构造一段精简的prompt让模型从候选边里选一条你是一个流程控制器。根据当前状态选择下一个合适的动作。 当前状态摘要 {decision_context} 可选动作 1. supplement_search: 补充搜索更多资料 2. generate_report: 已有信息足够直接生成报告 请只输出动作编号和名称。因为候选通常只有两到五个模型的准确率会显著高于“从一个巨大的工具列表里挑一个”的开放式场景。我把这个结论用在了所有Agent项目里把模型的选择空间压缩到个位数是提升稳定性的第一手段。5. 实际跑通一个PTC流程排查与避坑5.1 单测优先不要直接上LLMPTC架构最大的隐藏福利是你可以先用假决策器做全链路测试。在搭建调试阶段我强烈建议你别接LLM API先用一个规则决策器把整个流程跑通确认节点之间的上下文传递没有bug再切到模型。规则决策器写起来极简单class RuleDecisionProvider: async def decide(self, ctx, choices): if len(ctx.get(search_results, [])) 3: return [c for c in choices if c.dst supplement_search][0] return [c for c in choices if c.dst generate_report][0]用这个跑一遍图确认上下文key对得上、条件边的条件函数能正确从ctx取值然后才接模型。我见过很多团队一上来就跑LLM结果流程一复杂根本分不清是模型决策错了还是编排逻辑错了排查效率非常低。5.2 上下文污染是最隐蔽的坑节点之间通过共享上下文传递数据这带来一个隐患上一步的脏数据会污染下一步的决策。我踩过的一个真实例子是SearchWebNode在一次失败重试中把search_results设成了空列表但重试成功后没有覆盖这个key导致下游GenerateReportNode拿到的既不是空也不是新结果而是一个残缺的中间态。解决办法有两条我都用上了节点输出前统一做一次数据清洗不允许输出None或残缺结构要么是完整结果要么抛异常走重试。上下文增加“版本号”机制每次写入都带上递增版本号决策点读取时校验版本号和上游节点的执行标记。5.3 模型选了不存在的节点怎么办现实中模型偶尔会输出一个完全不在候选列表里的边。我的做法是不去“纠正”模型也不重新问一遍太贵而是增加一个兜底策略默认走第一条可用的条件边。这个策略保证流程不会死锁而且因为候选边本身就是有限的即使走了默认分支通常也不会出大错。兜底策略的优先级排最后任何显式条件都优先于兜底。这相当于给决策层加了“降级开关”线上出问题时可以先顶住流量而不是让整个Agent卡死。5.4 超时设置要分层工具层的超时、节点层的超时、整个编排流程的超时建议三层都设。工具层的超时尤其容易忽略——有些第三方API不稳定偶尔会hang住几十秒。节点超时我一般设成工具超时的1.5倍给重试和上下文写入留点余量。整图超时则根据流程复杂度估算调研类任务我会放到30秒超过直接返回“超时请重试”给用户。5.5 日志里必须有“决策轨迹”排错的时候你需要清晰地知道“模型在哪一步做了什么选择、为什么选它”。我在DecisionProvider里强制记录了三个字段当前上下文摘要的hash、候选列表快照、模型实际选择结果。每次决策产生一条结构化日志就可以复现任何一次线上行为。没有这套日志模型决策类问题你基本只能靠猜。6. 几个实践后的深度思考6.1 模型的角色从“执行者”变成了“评估者”PTC架构推行一段时间后我发现自己对模型的使用方式彻底变了。以前我总是在prompt里详细描述流程你先做什么再做什么如果什么情况就怎么做最后输出什么格式。现在我不这么干了。模型不再需要理解整个流程它只需要在决策点做判断或者在某个节点内做局部的信息提取和生成。这是一个角色转变模型从“流程执行者”变成了“节点内的评估者”。这个转变的收益是巨大的——模型不再需要“记住”流程也就不再可能“忘记”流程。每次决策的上下文都很短模型的准确率反而大幅提升。6.2 编排和决策的边界要画清楚哪些东西放编排图里哪些东西留给模型决策我的经验法则很简单凡是可以用代码可靠描述的逻辑一律放编排图凡是需要“语义理解”的判断才交给模型。比如“根据用户输入决定调哪个工具”——这个判断可能涉及语义理解模型做是合理的。但“结果为空要不要重试”——这个逻辑用代码写就够了不需要让模型参与。把太多琐碎决策交给模型既贵又不稳纯粹是给自己制造排查负担。6.3 图的可视化比你想的值钱因为编排图是纯数据结构只花了半天时间我就给PTC运行时加了一个简单的可视化面板把节点画成方块、边画成箭头执行中的节点高亮完成的节点打勾。效果出乎意料地好。产品同事能看懂Agent在做什么业务方也能直观指出“这一步的顺序不对”。我在之前的项目里遇到的“算法黑盒不透明”的抱怨在可视化面上之后基本消失了。这一点让我意识到Agent工程化走到最后拼的其实不是模型调参技巧而是工程结构的清晰度和可观测性。模型是黑盒但你的系统不该是黑盒。7. 几个典型问题的排查速查表实操中遇到的问题很多我挑几个高频的整理成表格方便对照。现象根本原因排查与解法调度器卡住不往下走依赖计数没减到零可能是条件边导致某节点永远不被触发检查条件函数是否在所有路径上都有返回值增加default_condition兜底节点执行结果反复被覆盖上下游节点对同一上下文key写入规范节点的outputs命名增加命名空间前缀比如search:results模型决策准确率突然下降决策点传入的上下文too大模型被无关信息干扰决策点只传摘要尤其别传原始页面HTML用字数统计和关键词列表替代某个工具偶发超时影响整条链路超时设置不合理工具超时和节点超时太接近工具超时设为节点超时的60%左右留出重试和上下文写入的余量并发执行的节点共享状态错乱两个节点同时往ctx写数据没有锁或版本控制减少共享可变key尽量让每个key单写多读重试导致副作用放大节点内部有非幂等副作用比如发通知、扣款对非幂等操作单独标记idempotentFalse禁止自动重试走人工补偿8. 这套方案的适用边界PTC不是银弹它有明确的适用边界这一点也必须说清楚。如果你的Agent总共只有两三个工具交互路径很短那直接用模型循环写起来更省事引入PTC反而是过度设计。PTC的收益大头来自“工具数量多、流程长、分支多”的场景——这种场景下流程确定性、可测试性、成本控制的需求才会压过模型循环的便利性。另一个适用信号是你已经厌倦了为同一个Agent改prompt里的流程描述。如果你发现自己在反复调整一段话来“教模型走流程”说明流程本身应该交给代码了。模型循环适合探索期PTC适合稳定期——一旦流程的骨架已经清晰迁移到运行时管理是必然趋势。我目前在设计新Agent时已经默认先画编排图再写节点实现最后才接模型。这个顺序本身也是一种思维训练先想清楚流程再考虑模型介入点模型的负担反而变小了系统的稳定性反而变高了。这套架构我已经实际跑了两个多月线上Agent的任务成功率从早期的六成多提升到目前的九成以上排查问题的平均时间从小时级降到了分钟级。如果你也在被Agent流程不可控的问题折磨我建议你试一次把编排搬到运行时——别让模型做它不擅长的事也别让你的流程逻辑散落在循环里不可维护。