企业官网建设流程全解析

如果最近在做 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）指的是多个模块按顺序执行，前一个模块的输出作为后一个模块的输入。

举一个业务场景。假设用户输入一段技术支持工单：

1. 先抽取问题类型、影响范围、紧急程度。
2. 再判断这个问题是否需要升级处理。
3. 最后生成给客户的回复。

这就是一个典型的级联管道。每个步骤都有独立职责，步骤之间通过结构化数据传递。

没有 DSPy 时，你可能这样实现：手动编写三段 prompt，调用三次模型，然后辛苦解析中间输出。问题在于：

• 中间输出格式稍有偏差，下游就解析失败。
• 每次修改提示词，可能改善某一步，却破坏下一步。
• 整条链路没有统一的验证机制，问题定位困难。

DSPy 的级联管道把每一步封装成 Module，它的输入输出是明确的字段，框架自带输出解析能力。你可以针对整个链路定义评分函数，再让优化器同时调整所有子模块的 prompt，而不需要逐个人工调试。

2.3 与传统链式调用的区别

有必要对比一下常见的链式调用方式。很多人用 LangChain LCEL 或普通代码把 LLM 调用串起来，这只解决了“顺序执行”的问题，没有解决“自动优化”的问题。

当然，这不是说传统方式一无是处。如果只是两个简单步骤且模型固定，手动链路足够。但一旦链路变长、需要频繁更换模型或系统复杂度上升，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( model="gpt-4o-mini", api_key="YOUR_API_KEY", # 建议从环境变量读取 api_base="https://api.openai.com/v1" # 按实际情况调整 ) dspy.configure(lm=lm)

如果你的项目接入的是国内模型厂商的 OpenAI 兼容接口，把api_base换成对应地址即可。关键在于，DSPy 不绑定具体模型厂商，只要遵循 OpenAI 协议即可。

版本方面，DSPy 更新较快，不同版本的 API 略有差异。如果你想减少兼容性问题，安装时建议指定一个稳定版本，例如pip install dspy==2.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_text=ticket_text) reply = self.generate_reply( category=info.category, impact_level=info.impact_level, keywords=info.keywords, ticket_text=ticket_text ) return dspy.Prediction( category=info.category, impact_level=info.impact_level, keywords=info.keywords, reply=reply.reply )

这里的dspy.Predict是最基础的模块，不包含内部推理步骤。如果你希望模型在生成输出前先进行思考，可以换成dspy.ChainOfThought。

注意dspy.Prediction只是结果载体，不是强制要求。你也可以直接返回一个字典或其他可序列化对象。不过为了统一管理，使用Prediction更规范。

4.3 定义评分函数

评分函数是编译优化的关键。它告诉优化器，当前模块组合的输出是否令人满意。生产环境中，评分函数的设计直接影响优化效果。

def support_metric(gold, pred, trace=None): 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(metric=metric, num_threads=4) pipeline = SupportPipeline() compiled_pipeline = optimizer.compile( pipeline, trainset=train_set, max_bootstrapped_demos=4, max_labeled_demos=4, max_demo_traces=100 )

编译完成之后，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(model="gpt-4o-mini") dspy.configure(lm=lm) # 步骤 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_text=ticket_text) reply = self.generate_reply( category=info.category, impact_level=info.impact_level, keywords=info.keywords, ticket_text=ticket_text ) return dspy.Prediction( category=info.category, impact_level=info.impact_level, keywords=info.keywords, reply=reply.reply ) # 步骤 4：构造演示数据集 train_set = [ dspy.Example( ticket_text="我的开发环境数据库无法连接，从应用容器内可以连接，但从本机连接超时。紧急，影响整个测试组。", category="网络故障", impact_level="team", keywords=["数据库", "连接超时", "网络策略"] ).with_inputs("ticket_text"), dspy.Example( ticket_text="UI 界面报错 500，刷新后恢复，但每天都会出现一次。我这边是个人账号，不是管理员。", category="软件报错", impact_level="single_user", keywords=["500", "UI", "bug"] ).with_inputs("ticket_text"), ] # 步骤 5：定义评分函数 def support_metric(gold, pred, trace=None): 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(metric=support_metric, num_threads=4) compiled_pipeline = optimizer.compile( pipeline, trainset=train_set, max_bootstrapped_demos=4, max_labeled_demos=4, max_demo_traces=50 ) # 步骤 7：推理测试 test_input = "我的账户无法登录，提示密码错误，但密码是正确的。其他同事正常，只有我没有权限。" result = compiled_pipeline(ticket_text=test_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. 常见问题与排查思路

级联管道虽然抽象性好，但实际运行中仍会遇到不少问题。

关于限流问题，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 流程、为复杂路由设计多分支级联、在本地模型上测试优化器效果、对比不同优化器对同一链路的收益差异。只要理解了“签名即接口、模块即单元、优化即调参”这套思路，你完全可以在自己的业务场景里复制这套方法。

建议你现在先把最小级联示例跑通，在真实数据上记录编译前后的指标差，再决定是否把这一套引入团队项目。实践永远比讨论更能检验一个框架的真实价值。