如果最近在做 Agent 或多轮任务流程你大概率遇到过这一类问题写好的 prompt 单个使用时效果尚可一旦把多个步骤串起来就变成了“一步错步步错”而且你很难判断问题到底出在哪个环节。拆步骤、写中间 prompt、做人工校验这套方案能跑通但维护成本很高尤其是当任务链路变长之后整个系统变成了一张脆弱的管道网络。DSPy 的级联管道Cascading pipelines值得重新审视它把原来靠“人工调 prompt 拼代码”的链式调用变成了可编程、可编译、可自动优化的模块化流程。这篇文章会讲清楚级联管道到底解决了什么问题、和普通链式调用的本质区别在哪、如何实现一个完整的级联示例以及在实际工程中容易踩到哪些坑。1. 这篇文章真正要解决的问题先说判断DSPy 的级联管道并不是什么新概念它本质上就是多阶段推理流程的工程化封装。但它的价值点不在于“把步骤串起来”而在于让每一步的 prompt、输出结构和优化目标都变成可编译、可测试的对象。如果你经历过下面的场景这篇文章就是写给你的你的任务需要多个 LLM 调用协作完成例如先抽取结构化信息再基于抽取结果做判断最后生成自然语言答案。你发现手动拼接 prompt 时中间步骤的输出片段格式不稳定导致下游步骤无法解析你却很难定位是哪一步出的问题。你想尝试自动优化整条链路的 prompt但不知道从哪里开始也不知道如何设计优化信号。你已经用过 LangChain 之类框架的链式 API觉得一切正常但又不确定系统在换了模型或提示词之后是否还能稳定工作。这篇文章会从头构建一个可运行的 DSPy 级联管道示例带你走完从定义模块、串联流程到编译优化、结果验证的完整过程然后重点讨论生产环境中最容易踩坑的几个问题。2. 基础概念与核心原理2.1 DSPy 解决的是什么问题在进入级联管道之前先看一下 DSPy 这个框架的基本定位。DSPy 是由斯坦福 NLP 团队开源的一个编程框架它的核心思想是把 prompt 视为可编程模块而不是“写在字符串里的一段话”。传统开发 LLM 应用时你写 prompt、调用模型、解析输出、再手动调整 prompt在 DSPy 里你定义模块的输入输出签名框架负责生成 prompt、调用模型、解析结果。这种设计带来的直接影响是你可以在不手工修改 prompt 的情况下针对指标自动优化 prompt。比如你希望输出的格式稳定、内容更贴合某个评分标准可以通过编译过程让框架自己调整 prompt 指令。DSPy 的几个核心概念如下。Signature定义模块的输入和输出字段例如input: str - output: str。它是模块的接口也是 prompt 的生成依据。Module一个逻辑单元可以包含多个签名或者多个子模块。它可以嵌套这正好支持构建级联结构。Optimizer优化器负责调整模块内部的 prompt 指令目标是在验证集上最大化评分指标。Metric评分函数告诉优化器“什么样的输出算好”。Trace过程记录用于调试和指标追踪。2.2 级联管道是什么意思级联管道Cascading pipelines指的是多个模块按顺序执行前一个模块的输出作为后一个模块的输入。举一个业务场景。假设用户输入一段技术支持工单先抽取问题类型、影响范围、紧急程度。再判断这个问题是否需要升级处理。最后生成给客户的回复。这就是一个典型的级联管道。每个步骤都有独立职责步骤之间通过结构化数据传递。没有 DSPy 时你可能这样实现手动编写三段 prompt调用三次模型然后辛苦解析中间输出。问题在于中间输出格式稍有偏差下游就解析失败。每次修改提示词可能改善某一步却破坏下一步。整条链路没有统一的验证机制问题定位困难。DSPy 的级联管道把每一步封装成 Module它的输入输出是明确的字段框架自带输出解析能力。你可以针对整个链路定义评分函数再让优化器同时调整所有子模块的 prompt而不需要逐个人工调试。2.3 与传统链式调用的区别有必要对比一下常见的链式调用方式。很多人用 LangChain LCEL 或普通代码把 LLM 调用串起来这只解决了“顺序执行”的问题没有解决“自动优化”的问题。维度传统链式调用DSPy 级联管道中间数据传递手动拼接字符串或简单 JSON通过 Signature 定义结构化输出Prompt 维护人工逐条维护框架按签名生成可自动优化输出解析手动写解析逻辑框架自动根据签名解析系统优化只能逐个调 prompt编译时统一调优所有模块可复现性较依赖提示词文本优化结果可记录和复现当然这不是说传统方式一无是处。如果只是两个简单步骤且模型固定手动链路足够。但一旦链路变长、需要频繁更换模型或系统复杂度上升DSPy 的编程化优势就会体现出来。3. 环境准备与前置条件开始实践之前先确认环境。DSPy 是 Python 包通过 pip 安装即可。建议在虚拟环境中安装避免污染系统级 Python 环境。3.1 安装依赖python -m venv venv source venv/bin/activate # Windows 下为 venv\Scripts\activate pip install dspy这里先只装 dspy 本身方便演示。实际项目中你可能还需要 openai 或其他模型供应商的 SDK不过 dspy 会自动处理部分依赖。如果你已经有 API Key 环境变量例如OPENAI_API_KEY可以直接在代码里使用。3.2 选择语言模型DSPy 支持的模型范围很广从 OpenAI、Anthropic 到本地部署的模型都有适配。本文示例使用 OpenAI 兼容接口的模型配置方式你需要在代码中指定模型名称。import dspy # 使用 OpenAI 兼容接口 lm dspy.OpenAI( modelgpt-4o-mini, api_keyYOUR_API_KEY, # 建议从环境变量读取 api_basehttps://api.openai.com/v1 # 按实际情况调整 ) dspy.configure(lmlm)如果你的项目接入的是国内模型厂商的 OpenAI 兼容接口把api_base换成对应地址即可。关键在于DSPy 不绑定具体模型厂商只要遵循 OpenAI 协议即可。版本方面DSPy 更新较快不同版本的 API 略有差异。如果你想减少兼容性问题安装时建议指定一个稳定版本例如pip install dspy2.5.x具体版本号以官方发布为准。本文代码基于 DSPy 2.x 的通用 API 编写主旨是演示思路而不是绑定某个精确版本。3.3 准备验证数据级联管道的优势建立在“可编译”之上而编译需要样本数据。对于示例来说10 到 20 条带标注的样本就能跑通完整流程。你可以根据业务场景收集真实数据也可以构造几条演示数据。数据集至少应包含原始输入。期望的中间结果。期望的最终结果。中间结果不是必须的但有了它你可以更细粒度地度量每一步的效果。4. 核心流程拆解接下来拆解一个完整的级联管道项目。业务场景如下用户提交一段技术支持工单文本系统先抽取关键信息再基于信息生成回复建议。4.1 定义签名与模块第一步是定义签名。签名的字段描述会影响 prompt 的生成质量所以字段名和 description 要尽量贴近业务语义。技术工单信息抽取的签名定义如下。class TicketInfo(dspy.Signature): 从工单文本中抽取关键信息。 ticket_text: str dspy.InputField(desc用户提交的原始工单内容) category: str dspy.OutputField(desc问题分类可选值为网络故障、权限问题、软件报错、硬件故障、其他) impact_level: str dspy.OutputField(desc影响程度可选值为single_user、team、department、all_users) keywords: list[str] dspy.OutputField(desc工单中涉及的关键实体或功能模块)回复建议生成的签名定义如下。class ReplySuggestion(dspy.Signature): 根据工单信息生成给用户的技术支持回复。 category: str dspy.InputField(desc问题分类) impact_level: str dspy.InputField(desc影响程度) keywords: list[str] dspy.InputField(desc关键实体) ticket_text: str dspy.InputField(desc原始工单内容) reply: str dspy.OutputField(desc给用户的支持回复需包含解决步骤和后续动作)这两个签名分别对应管道中的两个阶段。4.2 构建级联模块有了签名之后需要把它们组合成一个模块。级联的实现方式是在 Module 的forward方法中依次调用子模块。class SupportPipeline(dspy.Module): def __init__(self): self.extract_info dspy.Predict(TicketInfo) self.generate_reply dspy.Predict(ReplySuggestion) def forward(self, ticket_text): info self.extract_info(ticket_textticket_text) reply self.generate_reply( categoryinfo.category, impact_levelinfo.impact_level, keywordsinfo.keywords, ticket_textticket_text ) return dspy.Prediction( categoryinfo.category, impact_levelinfo.impact_level, keywordsinfo.keywords, replyreply.reply )这里的dspy.Predict是最基础的模块不包含内部推理步骤。如果你希望模型在生成输出前先进行思考可以换成dspy.ChainOfThought。注意dspy.Prediction只是结果载体不是强制要求。你也可以直接返回一个字典或其他可序列化对象。不过为了统一管理使用Prediction更规范。4.3 定义评分函数评分函数是编译优化的关键。它告诉优化器当前模块组合的输出是否令人满意。生产环境中评分函数的设计直接影响优化效果。def support_metric(gold, pred, traceNone): score 0.0 # 分类一致 if gold.category pred.category: score 0.3 # 影响程度一致 if gold.impact_level pred.impact_level: score 0.2 # 关键词覆盖计算交集比例 gold_keywords set(gold.keywords) pred_keywords set(pred.keywords) if gold_keywords: overlap len(gold_keywords pred_keywords) / len(gold_keywords) score 0.2 * overlap # 回复长度合理且包含关键实体 if pred.reply and len(pred.reply) 10: score 0.3 return score这个评分函数不是唯一答案。你可以根据自己的业务定义例如引入语义相似度、参考答案匹配度、人工评分模型等。关键原则是评分函数必须能区分好输出和坏输出否则优化器无从优化。4.4 编译与运行编译过程会基于训练样本调整模块内部的 prompt。在 DSPy 中这一步由编译器和优化器共同完成。from dspy.teleprompt import MIPSDv2 # 将训练集定义为 dspy.Example train_set [...] metric support_metric optimizer MIPSDv2(metricmetric, num_threads4) pipeline SupportPipeline() compiled_pipeline optimizer.compile( pipeline, trainsettrain_set, max_bootstrapped_demos4, max_labeled_demos4, max_demo_traces100 )编译完成之后compiled_pipeline就是优化后的可部署模块。注意编译会消耗额外的 token因为你需要在样本上进行多次模型调用。在动手编译之前先确认 API 配额和成本预算。4.5 推理阶段的使用优化后的管道和普通模块的使用方式完全一致。result compiled_pipeline( ticket_text我的账户无法登录提示密码错误但密码是正确的。其他同事正常只有我没有权限。 )从外部看调用方根本不需要知道内部是几级管道。这是级联模块封装的价值之一。5. 完整示例与代码实现为了让上面的流程更具体这里给出一个可运行的完整示例。例子里我用少量内联演示数据你可以直接在 Jupyter Notebook 或脚本中跑通。import dspy from dspy.teleprompt import MIPSDv2 # 步骤 1配置语言模型 lm dspy.OpenAI(modelgpt-4o-mini) dspy.configure(lmlm) # 步骤 2定义签名 class TicketInfo(dspy.Signature): 从工单文本中抽取关键信息。 ticket_text: str dspy.InputField(desc用户提交的原始工单内容) category: str dspy.OutputField(desc问题分类可选网络故障、权限问题、软件报错、硬件故障、其他) impact_level: str dspy.OutputField(desc影响程度可选single_user、team、department、all_users) keywords: list[str] dspy.OutputField(desc工单中涉及的关键实体或功能模块) class ReplySuggestion(dspy.Signature): 根据工单信息生成给用户的技术支持回复。 category: str dspy.InputField(desc问题分类) impact_level: str dspy.InputField(desc影响程度) keywords: list[str] dspy.InputField(desc关键实体) ticket_text: str dspy.InputField(desc原始工单内容) reply: str dspy.OutputField(desc给用户的支持回复需包含解决步骤和后续动作) # 步骤 3构建级联模块 class SupportPipeline(dspy.Module): def __init__(self): self.extract_info dspy.ChainOfThought(TicketInfo) self.generate_reply dspy.ChainOfThought(ReplySuggestion) def forward(self, ticket_text): info self.extract_info(ticket_textticket_text) reply self.generate_reply( categoryinfo.category, impact_levelinfo.impact_level, keywordsinfo.keywords, ticket_textticket_text ) return dspy.Prediction( categoryinfo.category, impact_levelinfo.impact_level, keywordsinfo.keywords, replyreply.reply ) # 步骤 4构造演示数据集 train_set [ dspy.Example( ticket_text我的开发环境数据库无法连接从应用容器内可以连接但从本机连接超时。紧急影响整个测试组。, category网络故障, impact_levelteam, keywords[数据库, 连接超时, 网络策略] ).with_inputs(ticket_text), dspy.Example( ticket_textUI 界面报错 500刷新后恢复但每天都会出现一次。我这边是个人账号不是管理员。, category软件报错, impact_levelsingle_user, keywords[500, UI, bug] ).with_inputs(ticket_text), ] # 步骤 5定义评分函数 def support_metric(gold, pred, traceNone): score 0.0 if gold.category pred.category: score 0.3 if gold.impact_level pred.impact_level: score 0.2 gold_keywords set(gold.keywords) pred_keywords set(pred.keywords) if gold_keywords: overlap len(gold_keywords pred_keywords) / len(gold_keywords) score 0.2 * overlap if pred.reply and len(pred.reply) 10: score 0.3 return score # 步骤 6编译优化 pipeline SupportPipeline() optimizer MIPSDv2(metricsupport_metric, num_threads4) compiled_pipeline optimizer.compile( pipeline, trainsettrain_set, max_bootstrapped_demos4, max_labeled_demos4, max_demo_traces50 ) # 步骤 7推理测试 test_input 我的账户无法登录提示密码错误但密码是正确的。其他同事正常只有我没有权限。 result compiled_pipeline(ticket_texttest_input) print(分类:, result.category) print(影响程度:, result.impact_level) print(关键词:, result.keywords) print(回复:, result.reply)这段代码里有几点需要解释dspy.ChainOfThought会在生成时先输出推理过程再给出最终结果对多步推理任务通常更稳定。with_inputs(ticket_text)说明训练样本中只有ticket_text是输入字段其他字段是标注结果。MIPSDv2 是 DSPy 2.x 的典型优化器适合小规模数据。使用时要确保数据集不小于 2 条否则优化过程可能无样本可学。max_demo_traces决定编译时搜索空间大小太大会显著增加耗时太小则可能找不到好示例。如果你想把管道切换到普通非编译模式不调用compile直接传入pipeline调用即可。这样适合先跑通逻辑再决定是否优化。6. 运行结果与效果验证运行上面的代码后通常会有两类输出编译过程中的日志以及最终预测结果。编译日志里会显示样本数量、每轮的分数变化等信息这部分可以用来判断优化器是否在有效工作。正常输出类似下面这样。分类: 权限问题 影响程度: single_user 关键词: [账户, 密码, 权限] 回复: 您好您的密码错误提示可能与账户权限配置有关。请先确认您使用的工作区是否正确如果确认无误请联系管理员检查您的账号权限并提供登录时间窗口以便进一步排查。6.1 如何判断编译是否成功建议从三个层面验证。第一评分变化。如果编译前后的评分在验证集上有明显提升说明优化器确实学到了有益的 prompt 模式。如果分数没有变化大概率是数据量太少或评分函数区分度不够。第二输出格式稳定性。多次调用编译前后的管道查看中间字段是否保持一致的风格和内容结构。没有编译时keywords字段可能有时返回 list有时返回逗号分隔字符串编译后框架通常能稳定解析为 list。第三单步回放。DSPy 可以保留 trace你可以打开trace查看每一步的输入输出确认故障发生在哪一级。6.2 一个控制变量的对比实验要验证“级联管道 DSPy 优化”是否确实有价值我建议做一个三组对比手工写死 prompt 的普通链式调用。不加编译的 DSPy 级联管道。加编译的 DSPy 级联管道。在相同测试集上比较三项指标字段抽取正确率、回复合理主观分、输出格式异常率。这种对比能帮你判断性能提升到底来自“结构化签名”还是“自动优化器”避免把框架的功劳误认为优化器的功劳。如果你使用的是本地模型编译成本会非常高建议先在小数据集上做 2 到 3 轮测试再决定是否完整编译。7. 常见问题与排查思路级联管道虽然抽象性好但实际运行中仍会遇到不少问题。问题现象可能原因排查方式解决方案编译时提示没有足够的 demo 数据样本太少无法 bootstrap 示例打印训练集长度和字段完整性增加训练样本至少保证每个分支有 2 条以上表示输出字段偶尔返回空字符串模型输出格式不稳定解析失败查看原始 llama 输出或 trace换更强的模型或调整签名 desc增加格式约束编译后分数没有提升评分函数区分度低或数据过少打印编译前后每轮分数检查 metric 逻辑重新设计评分函数加入约束条件中间字段解析为字符串而非 list模型输出与签名类型不一致检查keywords字段的实际类型在签名 desc 中强调“以 JSON 数组形式输出”API 成本超出预期编译时多次调用模型加上 bootstrap 示例会放大调用次数查看调用日志统计 token减小 max_demo_traces限制编译轮数换模型后效果变差prompt 是为原模型优化的转移性有限在目标模型上重新编译生产模型固定后再单独编译一次线程数过高导致限流编译并行调用太多请求观察 API 限流错误调低 num_threads增加重试机制关于限流问题DSPy 编译过程会大量调用模型接口如果你使用的是按调用次数计费或并发受限的 API一定要手动手动控制num_threads。这一步非常容易忽略尤其是项目从单机小 demo 切换到生产环境集群时。8. 最佳实践与工程建议8.1 先跑通最小级联再优化不要一上来就设计五级管道。先定义两个简单的级联模块验证签名是否稳定、输出是否能被下游正确消费。最小链路跑通之后再逐步添加新模块。如果一开始就构建复杂链路问题定位会非常困难。因为你无法确定是模型能力不足、prompt 不合理还是数据结构设计有问题。8.2 在签名描述里写清楚枚举值枚举值字段例如category、impact_level对输出格式至关重要。desc 明确写出所有可选值对稳定解析有明显帮助。反例是只写“给问题分类”模型可能自由发挥。正例是“问题分类可选网络故障、权限问题、软件报错、硬件故障、其他”。这不仅是给模型看也是给后续优化的优化器提供边界。8.3 不要将编译结果视为黑盒编译后的 pipeline 虽然可以直接部署但最好记录训练集样本。评分函数定义。优化器参数。原始模块结构。涉及的语言模型版本。这样当线上效果波动时你可以回溯是哪一部分发生了变化。DSPy 支持序列化但配置这一层需要自己在工程化时补齐。8.4 生产环境建议先关闭编译刚上线阶段可以用一个基线版 pipeline 先跑记录真实请求的分布再决定是否启用在离线数据上训练好的编译版本。原因有两个真实数据分布可能和训练集差异很大提前编译容易过拟合少量样本。编译版本在每次请求时可能额外携带演示样例增加 token 消耗和延迟。等线上流量稳定后再通过灰度或者离线回放评估优化效果是比较稳妥的做法。8.5 用文件管理评分函数和配置如果你的团队多人协作建议把评分函数、模块定义、数据集构建单独拆成文件避免在 IPython Notebook 里堆代码。下面是一个目录参考。project/ ├── pipeline/ │ ├── signatures.py │ ├── modules.py │ ├── metric.py │ └── optimizer.py ├── data/ │ ├── train.jsonl │ └── dev.jsonl ├── configs/ │ └── pipeline_config.yaml └── main.py这样做的好处是当你需要快速迭代某一部分时不需要改动主流程代码。8.6 注意数据隐私与合规边界级联管道会在中间阶段存储结构化信息这些信息往往比原始文本更敏感。如果处理的是用户工单或内部业务数据需要在管道链路中增加脱敏和审计日志。生产环境中建议在进入管道前过滤隐私字段。在日志中只记录中间字段的去标识化版本。对管道访问使用最小权限服务账号。这些边界不是 DSPy 特有的而是任何链接到内部数据的 LLM 应用都必须考虑的基本要求。9. 总结与后续学习方向这篇文章围绕 DSPy 级联管道回答了三个核心问题它解决的是多阶段任务中 prompt 维护和输出稳定性问题它通过 Signature、Module、Optimizer 三个概念把链路变成可编程对象它真正适合的场景是链路较长、需要自动优化、希望输出可靠结构数据的任务。如果看完文章只打算记住几件事我建议记住这三条级联管线的前提是每个中间模块输出结构化数据否则链路越长越脆弱。评分函数决定优化器的上限评分函数设计得粗糙自动优化反而有可能拉低效果。编译不是免费的要多花 token也有过拟合风险生产环境先用基线版本跑数据分析再决定要不要上线编译版本。后续可以深入的方向包括把级联管道接入 RAG 流程、为复杂路由设计多分支级联、在本地模型上测试优化器效果、对比不同优化器对同一链路的收益差异。只要理解了“签名即接口、模块即单元、优化即调参”这套思路你完全可以在自己的业务场景里复制这套方法。建议你现在先把最小级联示例跑通在真实数据上记录编译前后的指标差再决定是否把这一套引入团队项目。实践永远比讨论更能检验一个框架的真实价值。