从零手写提示流编排器:AI Agent开发中的工程化实录
2026/9/9 11:53:14 网站建设 项目流程

这个开源项目叫PatchCat,定位是“AI 提示流编排器”。说实话,做它的初衷有点“赌气”:我在好几个 AI Agent 项目里写过大量 prompt 调用逻辑,发现真正的复杂度从来不在单次调用的返回结果,而在多个 prompt 之间的流转、重试、分支和变量传递。本着“与其继续在别人的框架里挠头,不如自己把底层原理啃明白”的想法,我决定用“边写边问”的方式,从 0 到 1 造一个开源提示流编排器,这才有了 PatchCat。

如果你正在学 AI 应用开发、想自己实现 Agent 调度逻辑,或者单纯想找一个能拆开看内部实现的开源项目做参考,这篇文章应该能给你一些不一样的思路。我会把“如何边写代码边提出问题、再把问题变成代码改动”这套过程完整复盘一遍,也会把 PatchCat 的核心设计、代码实现和排查过的坑都摆出来。

1. 这个项目到底在做什么:提示流编排器解决的真实痛点

1.1 单次 Prompt 调用很简单,编排才是真难点

先聊聊“提示流”这个概念。在 LLM 应用里,一个稍复杂的功能通常不是调一次大模型就能完成的。比如你想做一个“智能客服”,它大概需要经历这些步骤:先识别用户意图,再从知识库检索相关资料,然后把资料拼成上下文,最后生成回复。如果一次回复不够好,可能还要做一轮自检和改写。

每一步本质上都是一次 prompt 调用。把这些调用当作节点,用边把它们串起来,让前一步的输出能成为后一步的输入,再支持条件分支、循环、重试、并行执行——这个整体结构就是提示流。而负责运行这个结构的系统,就是提示流编排器。

现在市面上有很多现成的编排框架,为什么我还要自己写?因为我想搞清楚的底层问题太多了:节点之间的变量到底是怎么传递的?条件分支是在哪一层判断的?一个节点挂了之后,重试逻辑应该放在执行器里还是节点内部?并行执行的时候,多个分支同时读写同一个变量怎么办?这些问题光读文档是得不到深入答案的,只有把执行引擎亲手写一遍,才能真正理解设计者当初做取舍的原因。

1.2 为什么我要从 0 到 1 写一个自己的编排器,而不是直接搬现成的

说实话,刚开始我也犹豫过。现成的框架功能齐全,社区活跃,直接用确实省事。但我给自己定了一个原则:可以借鉴设计思路,但核心代码必须自己写一遍。理由有三条。

第一,学习效率不一样。直接用现成框架,你学会的是“怎么调用 API”;自己实现一个最小版本,你学到的是“调度器如何遍历图结构”“上下文对象如何隔离变量”“失败重试如何不丢数据”。后者才是真正能迁移到其他场景的能力。

第二,定制空间大。我已经遇到的场景里,有些需求在通用框架里做起来很别扭:比如我需要在某个节点执行完后自动把中间结果写入本地日志;需要按业务字段动态决定下一跳节点,而不是简单按分支条件走。自己写编排器,这些都可以直接作为内置能力。

第三,开源贡献有价值。我特别希望这个项目能成为其他人的“教学源码”。与其只写一篇博客讲道理,不如把能跑的代码开源出来,让大家直接看运行过程。

1.3 项目和“边写边问”方法论的关系

在整个开发过程中,我用的不是传统“先设计再编码”的流程,而是“边写边问”:写一段代码,发现某个概念不明白,马上停下来查证和提问,理解了再继续。这种方式看起来很慢,但实际推进速度非常快,因为每个问题都来自真实的代码上下文,不是凭空想象。

举个例子。写执行器的时候,我一开始天真地认为只要把节点按列表顺序执行就行。结果写到分支的时候,发现“顺序执行”根本不够用,我得让执行器理解一张图。于是我问自己:“图的遍历有几种方式?我应该选深度优先还是广度优先?每个节点需要记录什么状态?”带着这些问题去查资料、做实验,最后才写出了能处理分支和循环的调度器。

这套方法论贯穿了整个项目,下面我详细展开。

2. 干中学的核心:如何让“边问”不流于形式

2.1 把模糊结巴变成能回答的问题

很多人在学习时容易陷入一种状态:感觉哪里没懂,但说不清楚自己到底哪里不懂。这种“模糊结巴”非常消耗时间,因为它无法被检索、无法被验证、无法被解决。

我在写 PatchCat 时,刻意训练自己把模糊感觉转成具体问题。原则很简单:问题里必须包含三个信息——我当前在做什么、我遇到了什么现象、我期望什么结果。比如一开始我不会问“prompt 流怎么设计”,而是问“我有一个节点 A,输出一个字符串;下一个节点 B 想拿这个字符串当输入,在 B 的 prompt 模板里应该用什么语法做变量替换?”这个问题一下子就落到可操作层面了。

你可能会觉得这太琐碎了。但恰恰是这种琐碎的问题,才是驱动的核心。每解决一个,代码就往前推进一小步;十几个小步之后,一个可运行的模块就出现了。

2.2 提问的四个要素:问题、上下文、假设、验证

后来我把提问流程固定成了四步,每次遇到问题都按这个节奏走,效率提升非常明显。

第一步是定义问题。把“我遇到了一个现象”压缩成“我遇到了一个可复现的现象”。比如“节点 B 收到的变量是 None”,比“数据传不对”要好得多。

第二步是准备上下文。我自己动手查源代码、查文档、看报错堆栈,把相关信息整理出来。这个环节不能偷懒,因为问得多不等于学得多,只有自己先搜索过的答案才有记忆深度。

第三步是形成假设。我不只是问“为什么”,而是养成先猜一个原因再验证的习惯。哪怕猜错了也没关系,因为猜的过程强迫我对系统做了一次完整推演。

第四步是验证。把假设变成代码改动或者一次实验,看结果是否和预期一致。如果一致,说明这个知识点真的通了;如果不一致,那说明假设本身有问题,需要回到第二步继续补充上下文。

这套流程本质上不是“提问技巧”,而是一套小循环的科学验证方法。用在写代码上特别合适,因为代码本身就是验证假设的最好工具。

2.3 用“费曼式复盘”检验自己是真懂了还是假懂了

代码跑通不等于理解到位。我有个习惯:每个核心模块完成后,会假装给一个刚入门的同事讲一遍这个模块是怎么工作的。如果讲的过程中出现“这里我也不太确定”或者必须用抽象术语糊弄过去的地方,那这些点就直接标记成“未掌握”,马上回头再研究。

我把这种方式叫“费曼式复盘”。它会逼你用具体的、因果明确的语言描述一个机制,而不是用“底层封装好了”来搪塞。对 PatchCat 来说,我复盘最多的就是“变量作用域”。

写上下文管理模块时,第一版我用了最简单的全局字典,所有节点共享同一个大字典。功能跑通了,但一说到“如果两个并行分支同时往同一个 key 写数据会发生什么”,我就发现自己根本没想清楚。于是我重新设计了作用域系统,这才有后面第三部分的内容。

“费曼式复盘”的建议是:别在脑子里复盘,拿个文档写下来,或者录一段语音自己听。你会发现,说出来和想清楚之间,差距比你想象得大得多。

3. PatchCat 的核心架构拆解与设计取舍

3.1 节点、边、上下文:图执行模型的三件套

PatchCat 的核心模型借鉴了图思想,但又没有引入太过复杂的东西。整体只有三个抽象:节点、边和上下文。

节点是最小执行单元,每个节点封装一段具体的逻辑。在 PatchCat 里,节点通常绑定一个 prompt 模板和一个模型配置,但也可以自定义成纯代码节点,用来做数据处理、条件判断、调用外部 API 等等。

边表示节点之间的依赖关系。一条边从上游节点指向下游节点,含义是“下游节点只有在上游节点成功执行之后才能开始”。没有边的节点是孤立节点,通常会作为入口节点来启动流程。

上下文是贯穿整个流程的数据容器。它同时承担两个职责:一个是存储每个节点的输出结果,另一个是提供节点运行时的参数读取入口。设计上它更像一个命名空间,而不是一个无限膨胀的杂货铺。

为什么不用一个简单的对象数组?因为数组只能表达“顺序”,无法表达“分支”和“汇聚”。图结构表达力更强,所需要的额外成本只是写一个调度器来遍历节点。这个成本非常值得。

3.2 上下文作用域设计:从“全局变量裸奔”到“变量分区管理”

我最开始实现上下文时特别朴素:一个字典,谁都能读谁都能写。结果在做一个多分支流程时出了问题。两个分支节点同时往同一个 key 里写中间结果,后执行完的节点把先执行完的节点数据覆盖了,下游节点拿到的数据完全随机。

这个坑让我重新认真设计变量作用域。现在的实现里,每个节点有一个独立的局部作用域,节点写入的变量默认只在局部可见。如果想跨节点读取,必须显式声明为“输出变量”,由调度器统一收集并写入全局作用域。

这么设计有点像编程语言里的函数参数传递:全局变量不是不能有,但只有通过明确的“返回值”才能传播。这种约束虽然没有全局字典那么灵活,但大大降低了数据冲突的概率。

具体的数据结构也不复杂:全局作用域就是一个字典,局部作用域也是一个字典,每个节点运行时把局部作用域初始化成全局作用域的浅拷贝。浅拷贝有一个潜在问题:如果变量本身是可变对象,子节点修改还是会影响全局对象。所以我在文档里约定,节点之间要传递复杂对象时,尽量用序列化字符串,或者通过专门的数据端口传递。

3.3 执行器与调度器:串行、分支、循环和最大执行步数

执行器是整个编排器的心脏,负责按顺序启动节点、收集结果、判断分支、处理异常。PatchCat 的实现分两层。

最底层是一个“可执行步”循环。调度的基本原理很简单:维护一个待执行节点集合,每轮循环找出所有“依赖已满足”的节点,执行它们,再把它们从待执行集合中移除,并更新依赖关系表。依赖已满足的意思是:该节点的所有上游边都已经产出结果。

这套机制天然支持串行和并行。串行是因为边定义了先后顺序,并行是因为两个没有共同依赖的节点在同一个循环批次里可以同时启动。

条件分支和循环则在节点内部或边条件中处理。PatchCat 支持两种分支方式:一种是在边上面写条件表达式,只有条件为真时,下游节点才会加入待执行集合;另一种是使用专用“分支节点”,它接收入参后返回一个“跳转目标节点 ID”,执行器根据返回值决定往哪个方向走。

循环本质上也是一种分支:让边的目标节点指回前面的节点即可。但为了防止写死循环,我在执行器里加了一个“最大执行步数”的硬性保护。默认最多执行 1000 步,超过直接报错。这个保护在开发阶段救了我好几次。

4. 从零手写:核心代码与流程实现记录

4.1 第一步:先做一个模板渲染器

任何提示流编排器都需要解决最基础的问题:怎么把变量填充进 prompt 模板。我一开始只做了字符串替换,后来发现太脆弱了。如果模板里有多个相同占位符,或者变量是数组、对象,字符串替换就处理不了。

所以第一步就是写一个支持基础语法的模板渲染器。我用的是 Python,最简单可靠的方式是用str.format或者string.Template,但为了兼顾灵活性和可读性,我决定用标准库里的parse方案做简化版,核心代码如下:

import re def render_template(template: str, context: dict) -> str: def replace(match): expr = match.group(1).strip() if expr.startswith("var."): key = expr[4:] if key not in context: raise KeyError(f"变量 {key} 未在上下文中找到") return str(context[key]) return match.group(0) return re.sub(r"\{\{\s*(.*?)\s*\}\}", replace, template)

这个实现很粗糙,但已经足够让节点把{{ var.user_name }}这样的占位符替换成真实值。它逼我思考了一个关键问题:变量名到底应该从哪里取。最终我规定,模板里的变量统一以var.开头,读取上下文里的某个 key。这个约定一直沿用到后面正式实现中。

4.2 第二步:定义节点抽象,让每个 Prompt 变成可复用组件

有了模板渲染器之后,我开始设计节点抽象。我要求节点类必须具备三个能力:声明自己的输入参数、执行时从上下文取参数、执行后把结果写回上下文。

第一个版本长这样:

class BaseNode: def __init__(self, node_id: str, prompt_template: str): self.node_id = node_id self.prompt_template = prompt_template def run(self, ctx: dict) -> str: prompt = render_template(self.prompt_template, ctx) return call_llm(prompt)

看起来很简单对吧?但实际使用中你马上会发现一个问题:不是所有节点都要调大模型。有些节点只是用来做条件判断,有些节点是纯数据变换,比如把列表转成 JSON 字符串。为了让这些节点都能参与编排,我把“执行逻辑”从节点类里彻底抽离出来,采用函数式注册模式。

class Node: def __init__(self, node_id: str, fn, inputs: list[str], outputs: list[str]): self.node_id = node_id self.fn = fn self.inputs = inputs self.outputs = outputs def execute(self, ctx): kwargs = {name: ctx.get(name) for name in self.inputs} result = self.fn(**kwargs) for name in self.outputs: ctx[name] = result

这样定义的话,一个调大模型的节点可以被写成一个函数,一个数据处理节点也可以被写成一个函数,PMC 统一按“输入 → 处理 → 输出”的规范来执行它们。这个设计让我在后面扩展新类型节点时,基本不用改动执行器代码。

4.3 第三步:实现一个最小执行器,跑通第一个 Prompt Flow

节点有了,接下来就是把它们串成流并执行。我实现了一个最小执行器,核心逻辑是维护“入度表”和“待执行集合”。

from collections import deque def execute_graph(entrypoints, edges, nodes): indegree = {node_id: 0 for node_id in nodes} adj = {node_id: [] for node_id in nodes} for source, target in edges: adj[source].append(target) indegree[target] += 1 ready = deque(n for n in entrypoints if indegree[n] == 0) ctx = {} executed = set() while ready: node_id = ready.popleft() if node_id in executed: continue node = nodes[node_id] node.execute(ctx) executed.add(node_id) for next_node in adj[node_id]: indegree[next_node] -= 1 if indegree[next_node] == 0: ready.append(next_node) return ctx

这个执行器的逻辑很简单,但它已经具备“按依赖自动调度”的能力。我把第一个能跑通的 Prompt Flow 定义为三个节点:第一个节点负责把用户问题翻译成英文,第二个节点负责用英文生成一个技术方案,第三个节点负责把方案写成通俗的讲解。运行成功后,那种“自己实现了一个复杂系统”的满足感非常强。

当然,这个版本只是骨架。现在的 PatchCat 在这个基础上增加了错误重试、节点超时、分支判断、并行执行和运行日志等能力,但这套最小的核心逻辑一直没变。

5. 开发过程中踩过的坑和排查思路

5.1 模板变量渲染失效:先查占位符,再查上下文

在开发早期,我经常遇到明明上下文里已经有值,但 prompt 渲染出来还是原样占位符的问题。排查下来,原因往往是模板里写成了{{var.user_name}}而没有留空格,或者反过来写了{{ var.user_name }}。我的正则默认支持可选的空格,但如果你在模板里手写了特殊字符,比如{{ user.name }}里面带点号,就会和var.前缀规则冲突。

这类问题最直接的排查方法就是单独写一个测试:给定固定上下文,调用render_template,打印渲染前后的模板内容。测试通过后再把怀疑范围移动到节点的 inputs 声明是否完整。

5.2 流程死循环:日志追踪 + 最大步数兜底

我在实现循环逻辑时写过一次死循环:一个分支节点在条件不满足时返回了当前节点本身作为下一跳,结果执行器不断循环同一对节点,直到资源耗尽。当时还在调试,一运行就卡死,半天没反应过来。

后来我吸取了两条教训。第一条,开发环境里一定要把“最大执行步数”设置得小一点,比如 50 步,方便快速暴露异常;第二条,每一步执行都要带 trace_id 写日志,至少记录当前节点 ID、输入参数摘要、输出结果摘要。这样一旦出现循环,扫一眼日志就知道在哪两个节点之间反复横跳。

5.3 模型输出不稳定:从不信任 LLM 的 JSON 开始

做提示流编排器,最难的就是不确定性的输入输出。我最初有一个节点,让模型“返回一个 JSON”,结果模型经常在 JSON 外面包代码块说明,或者多打一个逗号,导致解析失败。

现在的处理策略是:第一,prompt 里要求模型只输出 JSON 结构,不要任何解释;第二,代码里不直接用 JSON 解析,而是先尝试去掉首尾的代码块标记,再解析;第三,如果解析还是失败,把原始输出记录到日志,同时把该节点标记为失败并触发重试。宁可多花一次调用,也不要在下游节点里用坏数据。

这条经验用到所有涉及模型结构化输出的场景都适用,甚至可以说是不信任 LLM 的前提。

5.4 并行执行时的变量污染:作用域隔离才是根本

我前面提到过全局字典导致的数据覆盖问题。这里再说一个更隐蔽的情况:即使我改成了局部作用域是全局的浅拷贝,在并行分支里,如果节点 A 修改了某个 list 类型变量的元素,节点 B 读取同一 list 时依然会读到修改后的版本,因为它们共享同一个对象引用。

针对这个问题,我在变量传递规则里做了严格约定:跨节点传递时,默认只支持 JSON 可序列化的值;如果一定要传对象,必须通过专门的输入输出端口声明,并且节点内部不要原地修改对象。更好的方案是让每个节点在处理前先深拷贝一次敏感数据,牺牲一点性能换取确定性。

这个坑告诉我,作用域隔离不只是“字典分层”这么简单,还涉及 Python 对象引用语义。要真正把并行编排做好,数据不可变性和隔离策略必须提前设计好。

5.5 我对“边写边问”这个方式的最后一层理解

代码写到后来,我发现技术本身反而没那么难了,真正难的是不断追问并且不满足于“能跑就行”的态度。每次我发现一个变量作用域的漏洞,就会反问自己:其他地方有没有类似的隐患?每次修复一个解析失败的问题,我都会再看看日志格式是不是足够暴露错误信息。

“边写边问”到后期变成了一种思维习惯:不只问“这个怎么写”,更多地问“这个为什么要这样写”“这个如果换一个场景还会成立吗”。PatchCat 这个开源项目是这种习惯的产物,而这个习惯本身,可能比代码更值得保留。

如果你也想做类似的项目,我的建议是:先从一个最小但能跑通的流程开始,然后不断给流程加需求,让需求帮你发现底层知识盲区,再带着具体问题去查去问去验证。这条路看起来绕,但最后无论是技术理解、代码质量还是对问题的判断力,都会上一个明显的台阶。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询