1. 什么是 DSPy 的 Pipeline 与 Compiler——一个实战派的深度拆解你写完第一个dspy.predict()调用模型返回了还算靠谱的答案心里一松AI 编程不过如此。但第二天产品提了个需求“用户上传一份合同 PDF先定位‘违约责任’章节再提取其中所有金额条款最后比对是否符合公司风控阈值不符合的标红并生成简要风险提示”——这已经不是单次 prompt 能扛住的事了。它需要步骤编排、中间状态管理、失败回退、结果校验、甚至根据上一步输出动态决定下一步走哪条分支。这时候你手里的dspy.Module就像一把瑞士军刀里只拧开了螺丝刀而真正需要的是整套可调度、可验证、可重编译的“智能流水线”。这就是 DSPy 的Pipeline管道和Compiler编译器真正发力的地方。它们不是语法糖也不是封装层而是把“让大模型稳定可靠地完成复杂任务”这件事从工程直觉升维成可建模、可验证、可自动优化的系统工程。Pipeline 是你定义的逻辑蓝图它明确声明“我要做什么”比如“先检索→再推理→最后格式化”但它不关心“具体用哪个模型、哪个 prompt、哪个 retrieval 方法最稳”。Compiler 才是那个拿着蓝图去工厂实地踩点、试产、调参、换模具、最终交付一条良品率 99.2% 的产线的人。它负责把你的高层意图编译成一套在当前数据集、当前模型、当前硬件约束下实测性能最优、鲁棒性最强、成本效益比最高的具体执行方案。我第一次在客户现场部署一个三步合同审查 Pipeline 时原始 hand-crafted 版本在 12% 的样本上会因检索失败直接崩掉错误信息还全是KeyError: section_text这种让人抓狂的底层异常。换成 DSPy Compiler 编译后的版本不仅错误率压到 0.3%更关键的是——当某天客户突然要求把“风控阈值”从 50 万改成 30 万时我只需要改一行代码里的阈值常量重新 run 一次compile()整个 Pipeline 就自动适配了新规则下的最优 prompt 模板和 retrieval 策略连模型微调都不用动。这种“意图驱动、自动适配”的能力才是它被称为“Self-Improving AI Systems”的底层底气。它解决的从来不是“怎么让模型多说一句话”而是“怎么让整个 AI 工作流在无人值守的情况下越跑越准、越跑越省、越跑越稳”。2. Pipeline 架构设计为什么不能只是把几个 Module 串起来2.1 传统链式调用的三大硬伤很多刚接触 DSPy 的人第一反应是写个 Python 函数里面依次调用retriever(...),reasoner(...),formatter(...)。这看似合理但埋下了三个深坑我在三个不同客户的项目里都亲手踩过状态丢失与隐式耦合retriever输出一个字典{text: ..., score: 0.92}reasoner却硬编码去取result[text]。一旦retriever因升级返回了{content: ..., relevance: 0.92}整个链就断了报错位置还在reasoner里调试时得倒着查两层。这不是模块化这是“胶水式耦合”。错误传播不可控retriever在 5% 的 case 里返回空列表[]reasoner拿到空输入要么抛出IndexError要么胡乱生成一个答案。你没法在reasoner里优雅地处理“上游没给料”这个业务语义只能加一堆if not result:判断代码迅速变得臃肿且语义模糊。无法全局优化你发现reasoner的准确率卡在 78%想提升。但它是独立模块你只能单独给它喂更多训练数据、调它的 prompt。可问题可能出在retriever返回的文本质量不高或者formatter的输出格式让reasoner的上下文理解变差。单点优化就像给一辆四个轮子气压都不一样的车只拼命踩油门。提示DSPy Pipeline 的核心设计哲学是把“数据流”和“控制流”彻底分离。Pipeline 只定义“数据该往哪走”不定义“数据该怎么处理”而每个 Module 只专注“我拿到数据后怎么处理”不关心“数据从哪来、到哪去”。这种契约式接口是后续所有自动优化的前提。2.2 DSPy Pipeline 的三层结构解析一个健壮的 DSPy Pipeline 不是扁平的函数链而是有清晰分层的“操作系统”第一层Signature签名层—— 定义契约这是 Pipeline 的“宪法”。它用dspy.Signature明确声明输入是什么input: str输出是什么output: str中间状态有哪些字段context: List[str],risk_score: float每个字段的语义约束dspy.input_field(descriptionThe raw contract text)这份签名就是所有下游 Module 必须遵守的 ABI应用二进制接口。retriever必须产出context字段reasoner必须消费context并产出risk_scoreformatter必须消费risk_score。Compiler 后续的所有优化都是在这个强契约基础上进行的。第二层Module Graph模块图—— 描述拓扑这是 Pipeline 的“电路图”。它用dspy.Pipeline类或dspy.dsp.Program构建一个有向无环图DAGclass ContractReviewPipeline(dspy.Pipeline): def __init__(self): super().__init__() self.retriever dspy.Retrieve(k3) self.reasoner RiskAnalyzer() self.formatter RiskReporter() def forward(self, input): # 声明数据流向input - retriever - context context self.retriever(input).passages # context - reasoner - risk_score risk_score self.reasoner(contextcontext).risk_score # risk_score input - formatter - output output self.formatter(risk_scorerisk_score, inputinput).report return dspy.Prediction(outputoutput, contextcontext, risk_scorerisk_score)关键在于forward()方法里每一行赋值都是一条显式的数据边。Compiler 能看到context是retriever的输出也是reasoner的输入能看到risk_score是reasoner的输出也是formatter的输入。这种显式依赖让 Compiler 能做“跨模块联合优化”。第三层Execution Engine执行引擎—— 实现调度这是 Pipeline 的“调度中心”。它不直接调用Module.forward()而是通过dspy.LM和dspy.Retrieve的统一抽象将forward()中的每一步翻译成对底层 LLM 或检索服务的 API 调用并自动处理缓存相同input的retriever结果会被缓存避免重复调用。重试若某次 LLM 调用超时或返回格式错误引擎会按预设策略重试如换 prompt 模板、降采样温度。日志与追踪为每一步生成唯一 trace_id方便在 Prometheus 里看retriever_latency_p95和reasoner_accuracy的关联性。这三层结构共同构成了一个“可编程、可观测、可优化”的 AI 工作流底座。它让复杂任务的开发从“写一堆胶水代码”变成了“定义契约、画张流程图、交给 Compiler 去跑”。3. Compiler 的工作原理它到底在“编译”什么3.1 编译目标从“能跑”到“跑得又快又好”很多人误以为 DSPy Compiler 就是把你的 Python 代码转成字节码。完全不是。它的编译对象是Prompt Engineering Space提示工程空间和Model Configuration Space模型配置空间的联合搜索。它要找的不是一个“正确”的答案而是一组能让整个 Pipeline 在验证集上综合得分最高的参数组合。这个“综合得分”由你定义的metric决定。它可以是纯准确率exact_match(prediction.output, gold_answer)成本敏感0.7 * accuracy - 0.3 * (num_tokens_used / 1000)鼓励用更少 token 达到同等效果鲁棒性加权accuracy * (1 - std_dev_of_latency)既准又稳Compiler 的终极目标就是最大化这个 metric。它不关心你的forward()函数怎么写只关心给定我的验证数据什么样的 prompt 模板、什么样的 retrieval 策略、什么样的模型参数temperature, max_tokens能让 pipeline 的整体 metric 最高3.2 编译四阶段一个真实案例的全程跟踪我以一个真实的“医疗问答 Pipeline”为例带你走一遍 Compiler 的完整生命周期。这个 Pipeline 有三步SymptomExtractor→DiseaseClassifier→TreatmentSuggester。验证集有 200 个医生标注的 QA 对。阶段一Initialization初始化—— 建立搜索起点Compiler 不会从零开始瞎猜。它首先基于你的Signature和Module定义生成一组合理的初始候选方案SymptomExtractor的初始 prompt 模板请从以下患者描述中提取所有明确提到的症状名词用逗号分隔。患者描述{input}DiseaseClassifier的初始 prompt 模板根据以下症状列表判断最可能的三种疾病按可能性从高到低排序。症状{context}TreatmentSuggester的初始 prompt 模板针对疾病 {disease}给出三条最权威、最常用的治疗建议每条不超过 15 字。检索策略默认BM25k5LLM 参数temperature0.3, max_tokens256这组初始方案就是搜索的“原点”。Compiler 会先用它在验证集上跑一遍得到 baseline metric 62.4%。阶段二Exploration探索—— 大胆试错Compiler 开始在 prompt space 里“撒网”。它不是随机改而是有策略地变异Prompt 模板变异对SymptomExtractor它可能生成请严格按以下格式输出[症状1, 症状2, ...]。不要解释不要添加任何其他字符。患者描述{input}请识别患者描述中的医学症状实体。症状是指可观察、可测量的身体异常表现如发烧、皮疹不包括病因如病毒感染或感受如很疼。输出格式JSON {\symptoms\: [\...\, \...\]}Retrieval 策略变异尝试dense_vector_search替代BM25或调整k3/k7。LLM 参数变异尝试temperature0.1更确定或temperature0.7更多样。每次变异Compiler 都会在验证集的一个 mini-batch比如 20 个样本上快速评估。它记录下每次变异带来的 metric delta。如果某个新 prompt 让SymptomExtractor的抽取 F1 提升了 5%但DiseaseClassifier的准确率掉了 3%Compiler 会计算这个“跨模块影响”并决定是否保留这个变异。阶段三Exploitation深耕—— 精细调优当 Exploration 找到几个 promising 的方向比如“强制 JSON 输出”对SymptomExtractor效果显著Compiler 就进入 Exploitation 阶段对这些方向做精细化搜索对 JSON 模板它会尝试不同的 schema{symptoms: [...]}vs{entities: {symptoms: [...]}}对 temperature它会在[0.05, 0.15, 0.25]上做网格搜索。对 retrieval它会测试dense_vector_search在不同 embedding modelall-MiniLM-L6-v2vse5-small上的表现。这个阶段Compiler 会使用更小的 learning rate更密集的采样目标是找到局部最优解。阶段四Compilation Export编译与导出—— 交付可运行产物当 Exploration 和 Exploitation 收敛比如连续 5 次迭代 metric 提升 0.1%Compiler 就停止搜索。它会固化最优配置将最终选定的 prompt 模板、retrieval 策略、LLM 参数全部写入 Pipeline 的__dict__。生成可复现报告输出一个compilation_report.json里面详细记录{ final_metric: 78.9, improvement_over_baseline: 16.5, best_prompt_for_SymptomExtractor: 请严格按以下格式输出[症状1, 症状2, ...]。不要解释..., best_retrieval_strategy: dense_vector_search, best_embedding_model: e5-small, best_temperature: 0.15, search_steps: 142 }导出轻量级 Pipeline生成一个optimized_pipeline.py文件里面只有最精简的、已固化配置的 Pipeline 类不包含任何 Compiler 逻辑可以直接扔进生产环境的 Flask API 里。这个过程本质上是一场全自动的、端到端的 Prompt Engineering Model Selection Hyperparameter Tuning。你付出的只是写好Signature、Module和metricCompiler 付出的是数小时的 GPU 时间和一套成熟的搜索算法。最终交付的是一个“开箱即用、效果拔群”的 AI 流水线。4. 实操从零构建并编译一个电商客服 Pipeline4.1 明确业务需求与定义 Signature我们的客户是一家卖高端耳机的电商。客服痛点是用户问“我的耳机左耳没声音了怎么办”客服要先查是否在保修期再查是否是常见故障如耳塞堵塞最后给对应解决方案。人工处理平均耗时 4 分钟。我们目标是构建一个 Pipeline能在 2 秒内给出准确、可执行的回复。第一步定义Signature这是契约的基石import dspy class WarrantyCheck(dspy.Signature): Check if the users device is still under warranty based on purchase date. purchase_date dspy.InputField(descThe date when the user purchased the device, e.g., 2023-05-12) current_date dspy.InputField(descTodays date, e.g., 2024-08-28) warranty_period_months dspy.InputField(descWarranty duration in months, e.g., 24) is_under_warranty dspy.OutputField(descTrue if under warranty, False otherwise) class FaultDiagnosis(dspy.Signature): Diagnose the most likely cause of the reported audio issue. user_description dspy.InputField(descUsers description of the problem, e.g., left ear no sound) known_issues dspy.InputField(descList of common issues for this model, e.g., [earbud blockage, bluetooth pairing error]) most_likely_cause dspy.OutputField(descThe single most likely cause from the known_issues list) class SolutionGenerator(dspy.Signature): Generate a clear, step-by-step solution for the diagnosed fault. fault dspy.InputField(descThe diagnosed fault, e.g., earbud blockage) solution_steps dspy.OutputField(descA numbered list of 2-3 actionable steps, e.g., 1. Use a soft brush to clean the earbud mesh. 2. Try resetting the headphones.)注意这里的关键设计purchase_date和current_date是字符串不是datetime对象。因为 LLM 处理字符串更稳WarrantyCheck模块内部可以安全地用datetime.strptime()解析而不会让 Compiler 在搜索 prompt 时被类型问题干扰。known_issues是InputField而不是硬编码在 prompt 里。这给了 Compiler 自由度它可以在编译时根据验证集里高频出现的问题动态调整known_issues的内容让FaultDiagnosis更聚焦。4.2 构建 Pipeline Graph 与 Module 实现接下来我们实现具体的 Module 和 Pipelineclass EcommerceSupportPipeline(dspy.Pipeline): def __init__(self, lmNone, retrieveNone): super().__init__() self.lm lm or dspy.OpenAI(modelgpt-4-turbo) self.retrieve retrieve or dspy.Retrieve(k3) # 初始化各模块注意此时它们还没有被编译只是占位符 self.warranty_checker dspy.ChainOfThought(WarrantyCheck) self.fault_diagnoser dspy.ChainOfThought(FaultDiagnosis) self.solution_generator dspy.ChainOfThought(SolutionGenerator) def forward(self, user_query: str, purchase_date: str): # Step 1: Retrieve relevant knowledge (common issues, warranty policy) # 这里我们用 user_query 作为检索关键词去查知识库 retrieved self.retrieve(user_query) # retrieved.passages 是一个字符串列表我们把它拼成一个 context context \n.join(retrieved.passages) # Step 2: Check warranty using current date import datetime current_date datetime.date.today().isoformat() warranty_result self.warranty_checker( purchase_datepurchase_date, current_datecurrent_date, warranty_period_months24 ) # Step 3: Diagnose fault # 我们把 retrieved.passages 当作 known_issues 的来源 diagnosis_result self.fault_diagnoser( user_descriptionuser_query, known_issuesretrieved.passages ) # Step 4: Generate solution solution_result self.solution_generator( faultdiagnosis_result.most_likely_cause ) # 返回所有中间状态便于调试和监控 return dspy.Prediction( outputsolution_result.solution_steps, is_under_warrantywarranty_result.is_under_warranty, most_likely_causediagnosis_result.most_likely_cause, contextcontext ) # 创建未编译的 Pipeline 实例 uncompiled_pipeline EcommerceSupportPipeline()这个forward()方法就是我们前面说的“电路图”。它清晰地画出了user_query→retrieve→context→fault_diagnoser→most_likely_cause→solution_generator→output的路径。Compiler 会逐行解析这个图。4.3 设计 Metric 与准备验证集Metric 是 Compiler 的“方向盘”。我们不能只看最终output是否匹配因为一个完美的solution_steps如果建立在错误的is_under_warranty判断上就是灾难性的。所以我们设计一个分层加权 metricdef ecommerce_metric(gold, pred, traceNone): gold: dict with keys output, is_under_warranty, most_likely_cause pred: same structure from dspy.Prediction score 0.0 # 1. Warranty check must be 100% correct (business critical) if pred.is_under_warranty gold[is_under_warranty]: score 0.4 # 2. Fault diagnosis F1 score (using set intersection) pred_set set(pred.most_likely_cause.lower().split()) gold_set set(gold[most_likely_cause].lower().split()) if pred_set and gold_set: precision len(pred_set gold_set) / len(pred_set) recall len(pred_set gold_set) / len(gold_set) f1 2 * (precision * recall) / (precision recall 1e-8) score 0.4 * f1 # 3. Solution output must contain at least one key action verb (clean, reset, charge, etc.) key_verbs [clean, reset, charge, pair, update, check] if any(verb in pred.output.lower() for verb in key_verbs): score 0.2 return score # 准备一个小型但高质量的验证集 (20 samples) validation_set [ { user_query: My left earbud isnt playing any sound., purchase_date: 2023-06-15, gold: { is_under_warranty: True, most_likely_cause: earbud blockage, output: 1. Use a soft brush to clean the earbud mesh. 2. Try resetting the headphones. } }, # ... 19 more samples ]这个 metric 强制 Compiler 优先保证warranty_check的绝对正确因为这是法律和财务红线。fault_diagnosis其次solution_output是锦上添花。Compiler 会严格遵循这个权重去搜索。4.4 执行 Compilation参数详解与实操技巧现在到了最关键的compile()步骤。这不是一个黑盒命令每个参数都值得深究from dspy.teleprompt import BootstrapFewShot # 创建 Compiler 实例 teleprompter BootstrapFewShot( metricecommerce_metric, max_bootstrapped_demos8, # 每个 Module 最多用 8 个示例做 few-shot max_labeled_demos4, # 每个 Module 最多用 4 个人工标注的示例 teacher_settings{temperature: 0.1}, # 教师模型用于生成示例用低温度更确定 ) # 执行编译这一步会启动搜索 compiled_pipeline teleprompter.compile( uncompiled_pipeline, trainsetvalidation_set, # 注意这里传 validation_set不是 train_set valsetvalidation_set[:5], # 专门留 5 个样本做验证防止过拟合 requires_permission_to_runTrue, # 重要确保你有权限运行本地或远程 display_progressTrue, # 实时打印搜索进度 num_trials100 # 最多尝试 100 种配置组合 )关键参数解读与我的实操心得trainsetvalidation_set这是 DSPy 的一个反直觉设计。BootstrapFewShot的“训练”本质是让 Compiler 基于验证集自动生成高质量的 few-shot 示例demos然后用这些示例去微调 prompt。所以它需要一个“有代表性的数据集”来学习这个数据集就是你的验证集。别被名字骗了。valsetvalidation_set[:5]必须预留一小部分样本作为搜索过程中的“验证者”。Compiler 每生成一个新配置都会在这 5 个样本上快速跑一遍看 metric 是否提升。没有它Compiler 就是闭着眼睛瞎猜。num_trials100这是搜索预算。100 次听起来不多但对于一个三步 Pipeline每次 trial 都要跑 20 个样本就是 2000 次 LLM 调用。我建议首次编译从num_trials30开始快速验证流程是否通再逐步加到 100。teacher_settings{temperature: 0.1}这是“教师模型”的设置。Compiler 会用这个模型通常是更强的 GPT-4来生成 few-shot 示例。低 temperature 让它输出更稳定、更标准的示例避免引入噪声。实操中必踩的坑与避坑指南注意编译前务必检查你的validation_set里gold字段的结构是否和Prediction的返回结构完全一致。我曾因为gold[output]是字符串而pred.output是dspy.Prediction对象导致ecommerce_metric里pred.output.lower()报错编译直接中断。解决方案在 metric 函数开头加if hasattr(pred, output): pred_output pred.output else: pred_output 。提示如果你的 Pipeline 里有dspy.Retrieve确保你的retrieve服务无论是本地 ChromaDB 还是远程 ElasticSearch在编译期间是可访问且响应稳定的。Compiler 会在搜索过程中疯狂调用它。我建议在编译前先用retrieve(test query)手动测试 10 次确认 P95 延迟 500ms。实操心得编译不是“一键生成神迹”而是“人机协同的精细调教”。第一次编译完成后不要急着上线。打开生成的compilation_report.json重点看best_prompt_for_*。你会发现 Compiler 生成的 prompt往往比你手写的更啰嗦、更强调格式约束。比如它可能把SolutionGenerator的 prompt 写成“你是一个专业的耳机技术支持工程师。请严格按以下 JSON 格式输出{steps: [step1, step2]}。不要输出任何其他文字。用户问题{user_query}。诊断结果{fault}。” 这种“过度工程化”的 prompt恰恰是它对抗 LLM 随机性的有效手段。接受它而不是试图“美化”它。5. 常见问题与排查技巧实录那些编译失败时的深夜救星5.1 “Compilation failed with RuntimeError: No valid demos found”这是新手遇到的第一个拦路虎。错误信息很模糊但原因非常集中你的validation_set里没有任何一个样本能让初始的、未编译的 Pipeline 产生一个“看起来还行”的输出。排查思路手动运行 baseline在compile()之前先用uncompiled_pipeline跑一个validation_set[0]sample validation_set[0] pred uncompiled_pipeline( user_querysample[user_query], purchase_datesample[purchase_date] ) print(Baseline Prediction:, pred)如果pred.output是空字符串、是乱码、或者根本没包含is_under_warranty字段说明你的初始 Pipeline 就有问题。根因分析Prompt 太弱WarrantyCheck的初始 prompt 可能太简单Is it under warranty?LLM 直接回答Yes而不是布尔值True。解决方案在Signature的OutputField里加更强的desc比如descAnswer ONLY with the word True or False, nothing else.。Retrieval 失败retrieve(user_query)返回空列表导致FaultDiagnosis拿不到known_issues。解决方案检查你的知识库是否真的包含了user_query的相关文档或者在forward()里加 fallbackknown_issues retrieved.passages or [earbud blockage, bluetooth pairing error, battery drain]。终极急救如果时间紧可以跳过BootstrapFewShot用更简单的MIPROMeta-Interpretive Program OptimizationCompiler它对初始 demo 质量要求更低from dspy.teleprompt import MIPRO teleprompter MIPRO(metricecommerce_metric, num_candidates10) compiled_pipeline teleprompter.compile(uncompiled_pipeline, trainsetvalidation_set)5.2 “Compilation hangs forever at trial #X”编译卡死99% 的原因是某次 trial 的 LLM 调用超时或返回了格式错误而 Pipeline 没有设置超时和重试机制。排查与解决检查 LLM 设置确保你在创建Pipeline时传入的lm对象设置了max_retries和timeoutlm dspy.OpenAI( modelgpt-4-turbo, max_retries3, # 失败后最多重试 3 次 timeout30 # 单次调用最长等待 30 秒 )检查 Retrieval 设置同理dspy.Retrieve也要设置超时retrieve dspy.Retrieve( k3, timeout10 )启用详细日志在编译前加import logging; logging.basicConfig(levellogging.DEBUG)这样卡死时你能看到最后一条 log 是在调用哪个模块从而精准定位。5.3 “Compiled pipeline works on validation set, but fails miserably on real user queries”这是最令人沮丧的情况花了 3 小时编译指标从 62% 涨到 79%结果一上线用户问“我的耳机连不上手机”就返回了“请清洁耳塞”。典型的分布偏移Distribution Shift。应对策略扩大验证集多样性最初的validation_set很可能只覆盖了“左耳没声音”、“右耳没声音”这类规范问题。立刻收集线上真实 bad case加入验证集。我通常会建一个real_world_failures.jsonl文件每天同步。引入对抗性测试在验证集中主动加入一些“边界 case”模糊描述It sounds weird sometimes.错别字my lef earbud is brocken多问题混合Left ear no sound AND battery drains fast无购买日期How do I fix left ear no sound?这时purchase_date是NonePipeline 必须优雅处理Pipeline 内置 fallback 逻辑在forward()里对关键步骤加try/except并提供降级方案try: warranty_result self.warranty_checker(...) except Exception as e: # fallback: assume under warranty for better UX warranty_result dspy.Prediction(is_under_warrantyTrue)5.4 “Compilation report shows high metric, but the generated prompts are unreadable”你打开compilation_report.json发现best_prompt_for_FaultDiagnosis是一长串包含 12 个示例、嵌套了 3 层 JSON 的怪物。这很正常而且往往是好事。为什么它要这么复杂因为 Compiler 发现FaultDiagnosis这个模块对 prompt 的微小变化极其敏感。一个词的差别listvsenumerate就能让 F1 掉 5%。为了对抗这种敏感性它不得不堆砌大量示例和冗余约束来“锚定”LLM 的行为。你应该怎么做不要手改删掉一个示例或者把JSON改成plain text大概率会让 metric 断崖下跌。Compiler 的搜索是全局的你改一个地方可能破坏了它精心构建的平衡。拥抱它然后封装它把这个“怪物 prompt” 当作一个黑盒组件。你的工作是确保FaultDiagnosis模块的InputField和OutputField定义足够清晰让这个黑盒能稳定地吐出你需要的字段。这才是 DSPy 的哲学关注接口而非实现。6. 生产部署与持续演进让 Pipeline 活下去6.1 从 Jupyter Notebook 到生产 API 的三步打包编译完成的compiled_pipeline是一个 Python 对象不能直接扔进 Docker。你需要把它变成一个可部署的服务。Step 1: 序列化与固化import pickle # 将编译好的 pipeline 保存为 .pkl 文件 with open(ecommerce_pipeline_v1.pkl, wb) as f: pickle.dump(compiled_pipeline, f) # 在生产环境加载 with open(ecommerce_pipeline_v1.pkl, rb) as f: pipeline pickle.load(f)注意pickle有安全风险仅限可信环境。生产环境推荐用dspy.save()和dspy.load()它们会序列化成更安全的 JSON 格式。Step 2: 构建 FastAPI 服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel import dspy app FastAPI() class SupportRequest(BaseModel): user_query: str purchase_date: str app.post(/support) def get_support_response(request: SupportRequest): try: # 加载已编译的 pipeline pipeline dspy.load(ecommerce_pipeline_v1.pkl) # 执行预测 result pipeline( user_queryrequest.user_query, purchase_daterequest.purchase_date ) return { solution: result.output, warranty_status: result.is_under_warranty, diagnosis: result.most_likely_cause, latency_ms: result._latency * 1000 # DSPy 自动记录 latency } except Exception as e: raise HTTPException(status_code500, detailf