DSPy和LangChain的无缝集成

书接上文《DSPy Visualizer：可视化Prompt优化过程》，我们通过可视化的方式观察了DSPy如何一步步优化Prompt。今天这篇文章，想换个角度——深入聊聊DSPy和LangChain这对好搭档到底怎么无缝集成的。主要从三个方面展开：第一，DSPy VS LangChain，到底谁更擅长什么；第二，LangChain和DSPy具体怎么结合；第三，用一个实实在在的例子，演示如何用DSPy优化LCEL。

1. 概述

随着大型语言模型（LLMs）和向量存储的能力越来越强，一批新框架也应运而生，它们的目标都是让AI应用的开发变得更简单——从检索增强生成（RAG）到复杂的对话机器人，再到那些需要多步推理才能跑起来的应用。这些框架里，最出名的恐怕要数LangChain了。它由Harrison Chase在2022年10月发布，随后迅速火起来，GitHub上几百位开发者贡献代码，文档、数据源、API的支持面铺得特别广。配合Qdrant这类向量存储，再加上把多个LLM串联起来的能力，开发者确实能搭出复杂的AI应用，不用从头造轮子。 不过话说回来，哪怕LangChain已经解锁了这么多能力，开发者还是得自己掌握提示工程的那套手艺，才能写出好Prompt。而且，要优化这些Prompt、构建多阶段推理的AI，在现有框架里仍然是个难题。实际上，当你开始搭建生产级的AI应用时，很快就会发现单次LLM调用根本不够用——你得做一个工作流，让模型跟外部工具（比如浏览器）互动，抽出文档里的相关片段，再汇总到一个多阶段推理管道里。这里面涉及的中间输出处理、提示调整，全凭手动提示工程的话，效率很快就跟不上了。 2023年10月，斯坦福NLP的研究团队放出了DSPy这个库，它的核心思路是：完全自动化LLM的提示和权重优化过程，换句话说，把人工提示工程给省了。DSPy最厉害的地方在于，它会自动调整LLM提示，尤其是当你的应用里需要多次调用LLM时，这种自动优化的威力就完全体现出来了。那么问题来了：在构建基于LLM和向量存储的AI应用时，到底该选LangChain还是DSPy？下面我们就从它们各自的能力、适用场景出发，做个深入对比。

2. DSPy VS LangChain

两个框架都是搭建AI应用的好手，都吃LLM和向量搜索这套，但侧重点和做法很不一样。先看一个对比表格：

特点	LangChain	DSPy
核心关注点	提供大量构建模块，简化使用LLMs与用户指定数据源结合的应用程序开发。	自动化和模块化LLM交互，消除手动提示工程，提高系统可靠性。
方法	利用模块化组件和可以使用LangChain表达语言（LCEL）链接在一起的链。	通过编程而非提示来简化LLM交互，并自动优化提示和权重。
复杂管道	通过LCEL创建链，支持异步执行和与各种数据源及API的集成。	使用模块和优化器简化多阶段推理管道，并通过减少手动干预确保可扩展性。
优化	依赖用户的提示工程和多个LLM调用的链式操作。	内置优化器自动调整提示和权重，提高LLM管道的效率和效果。
社区与支持	拥有庞大的开源社区，文档丰富，示例众多。	新兴框架，社区支持不断增长，带来LLM提示的新范式。

**LangChain 的优势**： - **数据源和API**：各种各样的数据源和API都能接入，集成起来毫无压力，特别适合要跟不同数据打交道的AI应用。 - **模块化组件**：用LCEL这套声明式语法，把模块拼在一起，工作流的构建和管理都变得很清爽。 - **文档和示例**：作为老牌框架，文档又厚又全，示例成千上万，新手也能很快上手。 **LangChain 的劣势**： - **复杂推理任务**：一旦涉及多步推理，就得靠手动提示工程来调，费时费力还容易出错。 - **可扩展性问题**：多个LLM调用的工作流，要管理和扩展起来，挑战不小。 - **提示工程门槛**：开发者必须对提示工程有深入理解，才能构建高质量的多步推理应用。

DSPy

**DSPy 的优势**： - **自动优化**：提示生成和优化完全自动化，人工干预大幅减少，用LLM更省心，还能轻松搭建可扩展的工作流。 - **内置优化器**：像BootstrapFewShot、MIPRO这些优化器，能自动微调提示并适配特定数据集。 - **简化复杂性**：用通用模块+优化器，把提示设计的复杂细节藏起来，构建多步推理应用变得很轻松。 - **多LLM支持**：支持多种LLM，同一个程序里灵活切换。 **DSPy 的劣势**： - **社区较小**：相比于LangChain，资源、示例和社区支持都还有限。 - **文档偏少**：教程和指南虽然有，但跟LangChain一比，厚度差不少，刚开始接触时会有点吃力。 - **模式和模块的束缚**：刚上手时可能会觉得框架给你的模式有点僵，不那么自由。

3. LangChain和DSPy的结合

下面我们从两个方向来看看怎么把这两个框架融合起来。首先是**把LangChain集成到DSPy里**，让LCEL也能享受DSPy的优化；其次是**把DSPy集成到LangChain里**，这条路目前还处于探索阶段，LangChain官方暂时不支持，但我们可以给出一些思路。

3.1 将LangChain集成到DSPy

为了让LCEL能用上DSPy的优化，DSPy官方提供了两个核心类：`LangChainModule` 和 `LangChainPredict`。它们分别继承自 `dspy.Module` 和 `dspy.Predict`，同时实现了必要的方法，让LangChain的组件能无缝接入DSPy的优化和评估流程。 **LangChainModule**：在DSPy里，预测、优化、验证的主体就是Module。LangChainModule能把完整的LCEL封装起来，这样你依然可以用DSPy原生的 `Optimizer.compile()` 和 `Evaluate.evaluate()` 接口，传入LCEL后，里面的提示词和大模型组件就能被优化了。 ```python class LangChainModule(dspy.Module): def __init__(self, lcel): super().__init__() modules = [] for name, node in lcel.get_graph().nodes.items(): if isinstance(node.data, LangChainPredict): modules.append(node.data) self.modules = modules self.chain = lcel def forward(self, **kwargs): output_keys = ['output', self.modules[-1].output_field_key] output = self.chain.invoke(dict(**kwargs)) try: output = output.content except Exception: pass return dspy.Prediction({k: output for k in output_keys}) def invoke(self, d, *args, **kwargs): return self.forward(**d).output ``` 从代码可以看出，`LangChainModule` 实现了 `dspy.Module` 必需的 `forward` 方法，核心逻辑就是执行 `self.chain.invoke`，也就是跑LCEL，最后把结果包装成DSPy标准的 `dspy.Prediction` 对象返回。 **LangChainPredict**：这个类覆盖了 `Predict` 的功能，同时又带有LCEL的特性，所以它能跟其他 `langchain.Runnable` 子类一起串到LCEL里。 ```python class LangChainPredict(Predict, Runnable): class Config: extra = Extra.allow def __init__(self, prompt, llm, **config): Runnable.__init__(self) Parameter.__init__(self) self.langchain_llm = ShallowCopyOnly(llm) try: langchain_template = '\n'.join([msg.prompt.template for msg in prompt.messages]) except AttributeError: langchain_template = prompt.template self.stage = random.randbytes(8).hex() self.signature, self.output_field_key = self._build_signature(langchain_template) self.config = config self.reset() def reset(self): ... def dump_state(self): ... def load_state(self, state): ... def __call__(self, *arg, **kwargs): if len(arg) > 0: kwargs = {**arg[0], **kwargs} return self.forward(**kwargs) def _build_signature(self, template): gpt4T = dspy.OpenAI(model='gpt-4-1106-preview', max_tokens=4000, model_type='chat') with dspy.context(lm=gpt4T): parts = dspy.Predict(Template2Signature)(template=template) inputs = {k.strip(): OldInputField() for k in parts.input_keys.split(',')} outputs = {k.strip(): OldOutputField() for k in parts.output_key.split(',')} for k, v in inputs.items(): v.finalize(k, infer_prefix(k)) for k, v in outputs.items(): output_field_key = k v.finalize(k, infer_prefix(k)) return dsp.Template(parts.essential_instructions, **inputs, **outputs), output_field_key def forward(self, **kwargs): signature = kwargs.pop("signature", self.signature) demos = kwargs.pop("demos", self.demos) config = dict(**self.config, **kwargs.pop("config", {})) prompt = signature(dsp.Example(demos=demos, **kwargs)) output = self.langchain_llm.invoke(prompt, **config) try: content = output.content except AttributeError: content = output pred = Prediction.from_completions([{self.output_field_key: content}], signature=signature) dspy.settings.langchain_history.append((prompt, pred)) if dsp.settings.trace is not None: trace = dsp.settings.trace trace.append((self, {**kwargs}, pred)) return output def invoke(self, d, *args, **kwargs): return self.forward(**d) ``` `LangChainPredict` 初始化时接收LangChain的模型和提示词，然后根据提示词构建 `dspy.Signature`。预测时调用 `forward`，用 `self.signature` 把输入转成字符串，再传给 `self.langchain_llm.invoke` 进行推理——说白了，实际推理还是LangChain在干活，只是结果被包装成 `Prediction` 并存到DSPy的上下文里。另外，它还实现了 `invoke` 方法，所以可以跟其他LCEL组件串起来用。

3.2. DSPy集成到LangChain

上面聊的是用 LangChainPredict 和 LangChainModule 让 DSPy 去优化 LCEL 里的 prompt 和 LLM。那优化好了之后，这些优化过的提示词和大模型，怎么再回到LangChain的LCEL里用呢？这里给出两种思路。 **1）直接把优化后的LangChainPredict实例串到LCEL里** 因为LangChainPredict本身继承了 `Runnable`，天然就能在LangChain里使用。所以只需要把训练好的 `LangChainPredict` 实例直接加入chain即可： ```python trained_langchain_predict = LangChainPredict(prompt, llm) # 此处省略优化步骤，得到优化后的 trained_langchain_predict ... chain = trained_langchain_predict | StrOutputParser() ``` **2）提取优化后的提示词，手动加到LangChain的Prompt里** 目前业内还没看到这方面的公开探索，这里给出一个思路：从2.1的源码分析可知，`LangChainPredict` 类包含 `signature` 属性，调用 `forward` 时会把 `signature` 转成字符串。我们可以仿照这个逻辑，先获取 `signature`，再把 `demos` 传进去，得到字符串形式的提示词。 ```python # 获取signature signature = trained_langchain_predict.signature # 获取demos demos = trained_langchain_predict.demos # 得到提示词字符串 prompt_content = signature(demos) # 转换为提示词类 prompt = PromptTemplate.from_template(prompt_content) # 创建大模型 llm = OllamaLLM(model="llama3", stream=False) # 连接成 LCEL chain = prompt | llm ```

4. 实践示例：使用DSPy优化LCEL

这一节我们走一遍完整的流程：先写一个LCEL，用 LangChainPredict 和 LangChainModule 封装好，然后用 DSPy 的 Optimizer 和 Evaluate 进行优化和验证。

4.1 准备工作

确保安装了以下Python包：

• langchain
• langchain_ollama / langchain_openai
• dspy
• langchain_core
• langchain_community

4.2 示例介绍

**1）模块初始化** 导入相关模块，初始化LangChain的大模型、检索器和提示词。这里用Ollama提供的llama3模型和维基百科检索器。 ```python import dspy from langchain.cache import SQLiteCache from langchain.globals import set_llm_cache from langchain_ollama import OllamaLLM from langchain_community.retrievers import WikipediaRetriever set_llm_cache(SQLiteCache(database_path="cache.db")) llm = OllamaLLM(model="llama3", stream=False) retriever = WikipediaRetriever(load_max_docs=1) prompt = PromptTemplate.from_template( "Given {context}, answer the question `{question}` as a tweet." ) def retrieve(inputs): return [doc.page_content[:1024] for doc in retriever.get_relevant_documents(query=inputs["question"])] def retrieve_eval(inputs): return [{"text": doc.page_content[:1024]} for doc in retriever.get_relevant_documents(query=inputs["question"])] question = "where was MS Dhoni born?" ``` **2）创建LangChainModule实例** 先把llm和prompt封装成LangChainPredict实例，再把检索器、LangChainPredict、输出解析器串成LCEL，最后用LangChainModule包起来。 ```python from dspy.predict.langchain import LangChainModule, LangChainPredict zeroshot_chain = LangChainModule( RunnablePassthrough.assign(context=retrieve) | LangChainPredict(prompt, llm) | StrOutputParser() ) ``` **3）构建数据集** 引入HotPotQA数据集，拆分为训练集、验证集和测试集。 ```python from dspy.primitives.example import Example from datasets import load_dataset dataset = load_dataset('hotpot_qa', 'fullwiki') trainset = [ Example(dataset['train'][i]).without("id", "type", "level", "supporting_facts", "context").with_inputs("question") for i in range(0, 50) ] valset = [ Example(dataset['validation'][i]).without("id", "type", "level", "supporting_facts", "context").with_inputs("question") for i in range(0, 10) ] testset = [ Example(dataset['validation'][i]).without("id", "type", "level", "supporting_facts", "context").with_inputs("question") for i in range(10, 20) ] ``` **4）创建Metrics** 这里我们用LLM来评判回答质量，从`correct`、`engaging`、`faithful`三个维度打分。 ```python class Assess(dspy.Signature): """Assess the quality of a tweet along the specified dimension.""" context = dspy.InputField(desc="ignore if N/A") assessed_text = dspy.InputField() assessment_question = dspy.InputField() assessment_answer = dspy.OutputField(desc="Yes or No") optimiser_model = dspy.OpenAI(model="gpt-4-turbo", max_tokens=1000, model_type="chat") METRIC = None def metric(gold, pred, trace=None): question, answer, tweet = gold.question, gold.answer, pred.output context = retrieve_eval({'question': question}) engaging = "Does the assessed text make for a self-contained, engaging tweet?" faithful = "Is the assessed text grounded in the context? Say no if it includes significant facts not in the context." correct = f"The text above is should answer `{question}`. The gold answer is `{answer}`. Does the assessed text above contain the gold answer?" with dspy.context(lm=optimiser_model): faithful = dspy.Predict(Assess)( context=context, assessed_text=tweet, assessment_question=faithful ) correct = dspy.Predict(Assess)( context="N/A", assessed_text=tweet, assessment_question=correct ) engaging = dspy.Predict(Assess)( context="N/A", assessed_text=tweet, assessment_question=engaging ) correct, engaging, faithful = [ m.assessment_answer.split()[0].lower() == "yes" for m in [correct, engaging, faithful] ] score = (correct + engaging + faithful) if correct and (len(tweet) <= 280) else 0 if METRIC is not None: if METRIC == "correct": return correct if METRIC == "engaging": return engaging if METRIC == "faithful": return faithful if trace is not None: return score >= 3 return score / 3.0 ``` **5）优化程序** 跟优化普通dspy.Module一样，直接对LangChainModule实例调用compile。 ```python from dspy.teleprompt import BootstrapFewShotWithRandomSearch optimizer = BootstrapFewShotWithRandomSearch( metric=metric, max_bootstrapped_demos=3, num_candidate_programs=3 ) optimized_chain = optimizer.compile(zeroshot_chain, trainset=trainset, valset=valset) ``` **6）验证结果** ```python from dspy.evaluate.evaluate import Evaluate evaluate = Evaluate( metric=metric, devset=testset, num_threads=8, display_progress=True, display_table=5 ) evaluate(optimized_chain) # A verage Metric: 3.0 / 10 (30.0): 100%|██████████| 10/10 [00:22<00:00, 2.22s/it] ``` 到这里，我们就完成了把LCEL集成到DSPy中并利用其自动优化能力的完整流程。整个示例说明了一件事：LangChain的生态优势加上DSPy的自动优化能力，组合起来确实能帮我们省掉不少手工调Prompt的苦差事。