这个开源项目叫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 这个开源项目是这种习惯的产物而这个习惯本身可能比代码更值得保留。如果你也想做类似的项目我的建议是先从一个最小但能跑通的流程开始然后不断给流程加需求让需求帮你发现底层知识盲区再带着具体问题去查去问去验证。这条路看起来绕但最后无论是技术理解、代码质量还是对问题的判断力都会上一个明显的台阶。