1. 项目概述为什么需要封装AI执行链如果你已经跟着这个系列走过了前面的十四天那么恭喜你你已经掌握了从环境搭建、模型调用到Prompt工程、记忆管理等一系列AI应用开发的核心技能。但不知道你有没有发现一个问题当我们把一个个独立的组件比如模型、提示词模板、记忆、工具拼凑成一个能完成复杂任务的AI应用时代码往往会变得冗长、耦合度高而且难以维护和复用。今天我们就来解决这个痛点——使用LangChain来封装AI执行链。简单来说AI执行链就是将多个步骤如数据检索、模型推理、结果后处理串联起来形成一个自动化工作流的“管道”。想象一下你要开发一个智能客服机器人它需要1. 理解用户问题2. 从知识库中检索相关信息3. 结合检索到的信息生成回答4. 记录对话历史。如果每一步都手动写代码去连接不仅容易出错而且当你想调整流程比如在生成前加一个安全检查时改动会非常麻烦。而LangChain的“链”抽象正是为了解决这种编排问题而生的。在当前的AI应用开发热潮中能否高效、优雅地构建和编排复杂的工作流已经成为区分“玩具项目”和“生产级应用”的关键。无论是构建RAG问答系统、多智能体协作平台还是自动化数据分析流水线其底层核心都是一个或多个精心设计的执行链。因此掌握链的封装意味着你从“会调用API”迈向了“能设计AI系统架构”的新阶段。接下来的内容我将带你从零开始理解链的核心思想并手把手教你封装几种最常用、最实用的链。2. 核心概念深入理解LangChain中的“链”在动手之前我们必须把几个核心概念掰扯清楚。很多人刚接触LangChain时会被Chain、Runnable、LCEL这些术语搞晕。别担心我会用最直白的方式给你讲明白。2.1 链的本质可组合的Runnable在LangChain中一个Chain本质上是一个实现了Runnable协议的对象。什么是Runnable你可以把它理解为一个“可运行单元”它接受一个输入字典经过内部处理输出一个结果字典。这个处理过程可以非常简单比如只是一个提示词模板也可以非常复杂包含条件判断、循环、并行调用等。链的强大之处在于可组合性。任何Runnable包括模型、提示词模板、工具、甚至另一个链都可以像乐高积木一样通过管道操作符|连接起来形成一个新的、更复杂的Runnable。这就是LangChain Expression Language (LCEL) 的核心思想。例如chain prompt | model | output_parser这行代码就定义了一个链用户输入先经过prompt模板格式化然后送给model推理最后用output_parser解析模型输出。整个chain本身也是一个Runnable可以被继续组合或调用。2.2 链与智能体Agent的区别这是另一个常见的困惑点。简单来说链Chain是确定性的工作流。给定相同的输入它会执行一系列预定义好的步骤产生可预测的输出。比如一个翻译链总是“接收文本-格式化提示词-调用模型-解析结果”。智能体Agent是具备决策能力的工作流。它内部包含一个“大脑”通常是LLM根据当前状态和目标动态决定下一步该使用哪个工具或执行哪个子任务。它的执行路径是不确定的。你可以把链看作是智能体的“子程序”或“基础能力模块”。一个复杂的智能体内部可能会调用多个不同的链来完成子任务。今天我们先聚焦在确定性链的封装上这是构建更高级应用包括智能体的基石。2.3 为什么需要自定义封装LangChain提供了很多预置链如LLMChainRetrievalQA那为什么我们还要自己封装呢原因有三业务逻辑定制预置链是通用设计而你的业务逻辑是独特的。封装允许你将业务规则如数据校验、特定格式转换固化到链中。提升可维护性将一段复杂的处理逻辑封装成一个有明确名称和接口的链就像在代码中定义了一个函数大大提升了代码的可读性和可维护性。便于测试与复用封装好的链可以独立进行单元测试。一旦验证通过就可以像标准库一样在不同的项目中复用极大提升开发效率。注意虽然LCEL用|符号连接非常简洁但在封装复杂链时我强烈建议使用Runnable类的子类如RunnableSequence,RunnableParallel或自定义类来显式定义。这能让链的结构在代码中更清晰尤其是在需要添加条件逻辑或复杂分支时。3. 实战封装一构建一个基础的问答链让我们从一个最简单的例子开始封装一个问答链。它的功能是接收一个用户问题调用大模型返回一个答案。虽然简单但这是理解封装流程的绝佳起点。3.1 步骤拆解与依赖准备这个链只需要三个组件提示词模板PromptTemplate将用户问题嵌入到预设的指令模板中。大语言模型LLM执行推理生成回答。输出解析器OutputParser将模型的原始文本输出解析成我们需要的格式这里我们暂时用字符串。首先确保你已经安装了必要的库并准备好了模型API密钥以OpenAI为例pip install langchain langchain-openai python-dotenv在你的项目根目录创建.env文件存放你的API密钥OPENAI_API_KEY你的密钥3.2 代码实现与逐行解析接下来我们使用LCEL来定义并封装这个链。# basic_qa_chain.py import os from dotenv import load_dotenv from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnableSequence # 1. 加载环境变量 load_dotenv() # 2. 定义组件 # 提示词模板我们设计一个简单的系统指令让模型扮演助手。 prompt_template PromptTemplate.from_template( “”” 你是一个乐于助人的AI助手。请用中文回答用户的问题。 问题{question} 回答 “”” ) # 大语言模型使用ChatOpenAI指定模型和温度。 # 温度temperature控制创造性0.0更确定1.0更多变。对于问答通常设低一点。 llm ChatOpenAI( model“gpt-3.5-turbo”, temperature0.1, api_keyos.getenv(“OPENAI_API_KEY”) ) # 输出解析器将模型的AIMessage对象转换为纯字符串。 output_parser StrOutputParser() # 3. 使用LCEL组装链 # 这是最简洁的方式使用管道操作符 | simple_qa_chain prompt_template | llm | output_parser # 4. 调用链 if __name__ “__main__”: question “LangChain是什么” # 调用invoke方法执行链 answer simple_qa_chain.invoke({“question”: question}) print(f“问题{question}”) print(f“回答{answer}”)代码解析与实操要点PromptTemplate.from_template这是创建模板的便捷方法。模板中的{question}是一个变量会在链执行时被替换。ChatOpenAI参数temperature0.1使得回答更聚焦、更确定适合事实性问答。如果你希望回答更有创意可以调高。StrOutputParser这是最常用的解析器之一。因为ChatOpenAI返回的是AIMessage对象我们需要用这个解析器提取其中的文本内容。invoke方法这是执行链的标准方法传入一个字典字典的键必须与提示词模板中的变量名匹配。3.3 进阶将链封装为可配置的类上面的方式很简洁但如果我们想给这个链增加更多功能比如日志记录、输入验证或者想更方便地配置它最好将其封装成一个类。# configurable_qa_chain.py import os from typing import Dict, Any from dotenv import load_dotenv from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnableSequence class ConfigurableQAChain: “”“一个可配置的问答链”“” def __init__(self, model_name: str “gpt-3.5-turbo”, temperature: float 0.1): “”“ 初始化链。 Args: model_name: 使用的模型名称。 temperature: 模型温度参数。 “”“ load_dotenv() # 确保环境变量已加载 self.model_name model_name self.temperature temperature # 初始化核心组件 self.prompt self._create_prompt() self.llm self._create_llm() self.output_parser StrOutputParser() # 组装链 self.chain self.prompt | self.llm | self.output_parser def _create_prompt(self) - PromptTemplate: “”“创建提示词模板。这里可以扩展为从文件加载等。”“” template “”” 你是一个专业的AI助手。请根据你的知识清晰、准确地回答以下问题。 如果问题涉及你不确定的信息请如实说明。 用户问题{question} 请开始你的回答 “”” return PromptTemplate.from_template(template) def _create_llm(self) - ChatOpenAI: “”“创建LLM实例。这里可以方便地切换不同的模型提供商。”“” return ChatOpenAI( modelself.model_name, temperatureself.temperature, api_keyos.getenv(“OPENAI_API_KEY”), # 可以添加其他参数如请求超时时间 request_timeout60 ) def invoke(self, question: str) - str: “”“调用链回答问题。”“” # 这里可以添加前置处理如输入清洗、日志记录 print(f“[QA Chain] 正在处理问题: {question}”) try: result self.chain.invoke({“question”: question}) print(f“[QA Chain] 处理完成。”) return result except Exception as e: print(f“[QA Chain] 处理出错: {e}”) return f“抱歉处理问题时出现错误{e}” def get_chain_config(self) - Dict[str, Any]: “”“获取当前链的配置信息便于调试。”“” return { “chain_type”: “ConfigurableQAChain”, “model”: self.model_name, “temperature”: self.temperature } # 使用示例 if __name__ “__main__”: # 创建链实例可以轻松修改配置 qa_chain ConfigurableQAChain(model_name“gpt-4”, temperature0.2) print(“链配置”, qa_chain.get_chain_config()) answer qa_chain.invoke(“如何学习AI应用开发”) print(“回答”, answer)封装带来的好处配置集中管理模型、温度等参数在初始化时设定修改一处即可影响整个链。易于扩展可以在invoke方法前后轻松添加日志、监控、异常处理等逻辑。更好的抽象对外部调用者而言他只需要知道ConfigurableQAChain这个类有一个invoke方法可以回答问题无需关心内部是如何组装的。便于测试你可以对这个类进行单元测试模拟llm的返回验证整个处理流程。实操心得在项目初期用LCEL的管道语法快速原型验证。当逻辑稳定、需要投入生产时强烈建议将其封装成类。这看似多写了一些代码但长期来看对于代码的维护、团队协作和功能扩展有巨大的好处。4. 实战封装二构建带检索的增强生成RAG链单纯的问答链只能依赖模型自身的知识。要让它能回答特定领域如你的公司文档、个人知识库的问题就需要引入检索增强生成RAG。这是一个更复杂但也更实用的链。4.1 RAG链的工作流程一个典型的RAG链包含以下核心步骤检索Retrieve根据用户问题从向量数据库中检索出最相关的文档片段。增强Augment将检索到的文档片段与原始问题组合构建一个包含上下文信息的增强提示词。生成Generate将增强后的提示词发送给大模型生成最终答案。我们的目标就是将这三个步骤封装成一个流畅的链。4.2 准备工作创建向量存储假设我们已经有一批文本数据比如Markdown格式的文档我们需要先将其转换为向量并存储起来。这里我们使用ChromaDB轻量级、内存友好和OpenAI的嵌入模型。# prepare_vector_store.py import os from dotenv import load_dotenv from langchain_community.document_loaders import TextLoader, DirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma load_dotenv() def create_knowledge_base(data_dir: str “./data”, persist_dir: str “./chroma_db”): “”“从文档目录创建向量知识库”“” # 1. 加载文档这里假设data_dir下都是.txt文件 loader DirectoryLoader(data_dir, glob“**/*.txt”, loader_clsTextLoader) documents loader.load() print(f“已加载 {len(documents)} 个文档。”) # 2. 分割文本 # 大模型有上下文长度限制所以需要把长文档切分成小块。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块大约500字符 chunk_overlap50 # 块之间重叠50字符保持语义连贯 ) splits text_splitter.split_documents(documents) print(f“文档被分割成 {len(splits)} 个文本块。”) # 3. 创建向量存储 embeddings OpenAIEmbeddings(model“text-embedding-3-small”) vectorstore Chroma.from_documents( documentssplits, embeddingembeddings, persist_directorypersist_dir # 指定持久化目录 ) vectorstore.persist() # 保存到磁盘 print(f“向量数据库已创建并保存至 {persist_dir}”) return vectorstore if __name__ “__main__”: # 假设你的文档放在 ./data 目录下 create_knowledge_base(data_dir“./data”)关键参数解析chunk_size这是最重要的参数之一。太小会丢失上下文太大会超出模型上下文窗口。一般根据你使用的模型和文档特点调整。500-1000是常见起点。chunk_overlap重叠部分可以防止一个完整的句子或概念被硬生生切断有助于提升检索质量。persist_directory指定后Chroma会将索引持久化到磁盘下次启动无需重新生成节省时间和API费用。4.3 封装RAG链现在我们有了向量存储可以来封装RAG链了。# rag_chain.py import os from typing import List from dotenv import load_dotenv from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough, RunnableParallel from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings load_dotenv() class RAGChain: “”“一个完整的RAG问答链”“” def __init__(self, persist_dir: str “./chroma_db”, model_name: str “gpt-3.5-turbo”): “”“ 初始化RAG链。 Args: persist_dir: 向量数据库持久化目录。 model_name: 生成答案使用的LLM模型。 “”“ self.persist_dir persist_dir self.model_name model_name # 加载已有的向量数据库 self.embeddings OpenAIEmbeddings() self.vectorstore Chroma( persist_directorypersist_dir, embedding_functionself.embeddings ) # 将向量数据库转换为检索器Retriever # search_kwargs 可以控制返回的相关文档数量 self.retriever self.vectorstore.as_retriever(search_kwargs{“k”: 4}) # 初始化LLM self.llm ChatOpenAI(modelmodel_name, temperature0.1) # 定义提示词模板 # 注意这里我们使用了更复杂的模板明确指令模型基于上下文回答。 self.prompt_template ChatPromptTemplate.from_messages([ (“system”, “””你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。 如果你在上下文中找不到答案或者上下文信息不足以回答问题请直接说“根据提供的资料我无法回答这个问题。”不要编造信息。 上下文信息 {context} 问题{question} 请基于上下文给出答案”“”), ]) # 使用LCEL组装链 self._build_chain() def _build_chain(self): “”“组装RAG链的核心逻辑”“” # 步骤1定义处理函数 def format_docs(docs: List) - str: “”“将检索到的文档列表格式化为一个字符串。”“” return “\n\n”.join([doc.page_content for doc in docs]) # 步骤2使用RunnableParallel并行处理虽然这里检索和问题传递是串行的但此结构易于扩展 # RunnablePassthrough() 用于传递用户的原始问题。 setup RunnableParallel( contextself.retriever | format_docs, # 先检索再格式化 questionRunnablePassthrough() # 直接传递问题 ) # 步骤3组装完整链 # setup的输出是一个包含“context”和“question”键的字典正好匹配prompt模板的输入。 self.chain ( setup | self.prompt_template | self.llm | StrOutputParser() ) def invoke(self, question: str) - str: “”“执行RAG问答”“” print(f“[RAG Chain] 检索并回答: {question}”) try: return self.chain.invoke(question) except Exception as e: print(f“[RAG Chain] 错误: {e}”) return “系统处理问题时发生错误。” def similarity_search(self, query: str, k: int 3): “”“辅助方法直接查看检索到的原始文档用于调试。”“” docs self.vectorstore.similarity_search(query, kk) for i, doc in enumerate(docs): print(f“--- 相关文档 {i1} ---”) print(doc.page_content[:200] “...”) # 打印前200字符 print() return docs # 使用示例 if __name__ “__main__”: rag RAGChain(persist_dir“./chroma_db”) # 测试1正常问答 question “LangChain中的Chain是什么” answer rag.invoke(question) print(f“问题{question}”) print(f“答案{answer}”) print(“-” * 50) # 测试2查看检索结果调试用 print(“检索到的相关文档片段”) rag.similarity_search(question, k2)链组装逻辑深度解析RunnableParallel这是一个关键组件。它允许我们并行执行多个分支。在这里我们定义了两个分支一个分支通过retriever获取context另一个分支通过RunnablePassthrough()直接传递question。这两个分支的结果会自动合并成一个字典{“context”: “…”, “question”: “…”}作为下一阶段的输入。self.retriever | format_docs这是一个子链。self.retriever检索返回一个文档列表然后通过format_docs函数将其格式化为一个字符串。LCEL的优雅之处在于任何函数只要签名匹配都可以通过管道接入。链的最终形态setup | self.prompt_template | self.llm | StrOutputParser()。这清晰地定义了数据流先准备上下文和问题然后填充提示词接着发送给LLM最后解析输出。注意事项提示词模板中的system消息至关重要。它明确限制了模型只能基于提供的context回答这是减少模型“幻觉”即编造信息的关键。在实际应用中你可能需要根据业务场景反复打磨这个提示词。5. 实战封装三构建带条件判断的路由链现实中的任务并非总是线性的。有时我们需要根据用户输入的内容决定走哪条处理路径。这就是路由链Router Chain的用武之地。例如用户输入可能是一个需要翻译的句子也可能是一个需要总结的文档我们需要先判断意图再分发给不同的子链处理。5.1 设计思路与架构我们将构建一个简单的路由链它能区分两种请求翻译请求如果用户输入包含“翻译”或“translate”关键词则调用翻译链。总结请求如果用户输入包含“总结”或“summarize”关键词则调用总结链。默认处理如果都不匹配则调用通用的问答链。我们将使用RunnableBranch来实现条件路由它是LangChain中用于构建条件逻辑的强大工具。5.2 实现各功能子链首先我们实现三个子链翻译链、总结链和通用问答链。为了简化我们使用同一个LLM但提示词不同。# router_chain_components.py from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnableBranch # 初始化共享的LLM llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.1) # 1. 翻译链 translation_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个专业的翻译家。将用户输入的内容准确、流畅地翻译成中文。”), (“user”, “{input}”) ]) translation_chain translation_prompt | llm | StrOutputParser() # 2. 总结链 summarization_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个文本总结助手。请用简洁的语言概括以下文本的核心内容不超过100字。”), (“user”, “{input}”) ]) summarization_chain summarization_prompt | llm | StrOutputParser() # 3. 通用问答链复用之前的基础链 qa_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个AI助手。请回答用户的问题。”), (“user”, “{input}”) ]) qa_chain qa_prompt | llm | StrOutputParser()5.3 实现路由逻辑与主链接下来我们实现核心的路由判断函数并用RunnableBranch将各个子链组合起来。# router_chain_main.py from langchain.schema.runnable import RunnableBranch, RunnableLambda def route_function(input_data: dict) - str: “”“ 根据输入内容决定路由到哪个链。 返回子链对应的标识符。 “”“ user_input input_data.get(“input”, “”).lower() # 简单的关键词匹配路由逻辑 # 在实际项目中这里可以用更复杂的分类模型如另一个LLM来实现 if any(keyword in user_input for keyword in [“翻译”, “translate”]): return “translation” elif any(keyword in user_input for keyword in [“总结”, “summarize”, “概括”]): return “summarization” else: return “qa” # 使用RunnableLambda将判断函数包装成Runnable router RunnableLambda(route_function) # 定义分支每个分支是一个(condition, runnable)对 branch RunnableBranch( (lambda x: x “translation”, translation_chain), (lambda x: x “summarization”, summarization_chain), qa_chain # 默认分支 ) # 组装主链先路由判断再执行分支 # 这里有个技巧router的输出是字符串标识但branch的每个条件函数接收的是这个标识。 # 我们需要让router的输出直接作为branch的输入。 main_chain router | branch # 为了方便调用我们封装一个函数将用户输入包装成字典 def process_input(user_input: str) - str: return main_chain.invoke({“input”: user_input}) # 测试 if __name__ “__main__”: test_cases [ “请将‘Hello, world!’翻译成中文。”, “总结一下《红楼梦》的主要情节。”, “太阳为什么从东边升起”, “translate ‘good morning’ to Chinese.” ] for test in test_cases: print(f“输入{test}”) result process_input(test) print(f“输出{result}”) print(“-” * 30)路由逻辑的扩展性 上面的route_function使用了简单的关键词匹配这在生产环境中可能不够健壮。更高级的做法是使用一个分类链Classification Chain。这个分类链本身是一个小型的LLM调用它分析用户输入并输出一个预定义的类别标签如“translation” “summarization” “qa”。这样路由的准确率会高得多。你可以尝试将router替换成这样一个分类链# 进阶使用LLM进行分类路由示例 from langchain.prompts import PromptTemplate classification_prompt PromptTemplate.from_template(“”” 请判断用户意图属于以下哪一类 - translation: 用户要求进行语言翻译。 - summarization: 用户要求总结文本。 - qa: 其他一般性问题。 只输出类别名称不要输出其他任何内容。 用户输入{input} 意图类别“””) classification_chain classification_prompt | llm | StrOutputParser() # 然后将 main_chain 改为classification_chain | branch实操心得路由链是构建复杂AI应用如智能客服、工作流引擎的核心模式。在设计时要确保各个子链的输入输出接口保持一致例如都接受一个包含input键的字典这样它们才能被RunnableBranch无缝调度。同时路由逻辑本身也可以很复杂甚至是一个小型的决策树或多级路由这完全取决于你的业务需求。6. 链的调试、监控与最佳实践封装好链只是第一步让链在生产环境中稳定、可靠地运行更为关键。这部分分享一些我踩过坑后总结的调试、监控和最佳实践。6.1 利用LangSmith进行可视化调试与追踪LangChain官方提供了强大的可视化调试平台LangSmith。它能记录链的每一次调用展示完整的执行流程、每一步的输入输出、耗时和Token使用情况是开发和调试的利器。基础配置import os os.environ[“LANGCHAIN_TRACING_V2”] “true” os.environ[“LANGCHAIN_ENDPOINT”] “https://api.smith.langchain.com” os.environ[“LANGCHAIN_API_KEY”] “你的-langchain-api-key” # 在LangSmith设置中获取 os.environ[“LANGCHAIN_PROJECT”] “My_AI_Project” # 设置项目名配置好后你正常调用链所有执行信息都会自动同步到LangSmith的网页端。你可以清晰地看到链的调用树。每个节点如PromptTemplate LLM Retriever的输入输出。任何过程中发生的错误。每次调用的延迟和成本如果配置了。这对于理解复杂链的执行过程、定位性能瓶颈和排查错误不可或缺。6.2 为链添加日志与自定义监控除了LangSmith在链的关键节点添加自定义日志也很有帮助。# 在自定义链类的invoke方法或关键函数中添加日志 import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class MonitoredQAChain(ConfigurableQAChain): def invoke(self, question: str) - str: logger.info(f“收到问题: {question}”) start_time time.time() # ... 执行链的核心逻辑 ... end_time time.time() logger.info(f“问题处理完毕耗时 {end_time - start_time:.2f} 秒”) return result你还可以记录更细粒度的信息如检索到的文档数量、模型生成的长度等方便后续分析和优化。6.3 性能优化与缓存策略链的调用可能很慢尤其是涉及LLM和向量检索时。以下是一些优化思路LLM调用缓存对于相同或相似的输入直接返回缓存结果节省成本和时间。LangChain内置了InMemoryCache、SQLiteCache等。from langchain.globals import set_llm_cache from langchain.cache import SQLiteCache set_llm_cache(SQLiteCache(database_path“.langchain.db”))向量检索优化索引选择对于大规模数据考虑使用FAISS、Pinecone等高性能向量数据库。检索参数调整search_kwargs中的k返回数量和score_threshold相似度阈值在精度和速度间取得平衡。异步调用如果你的应用需要高并发使用链的ainvoke或abatch异步方法可以显著提升吞吐量。# 异步调用示例 import asyncio async def process_questions(questions): tasks [chain.ainvoke({“question”: q}) for q in questions] results await asyncio.gather(*tasks) return results6.4 错误处理与链的健壮性链中的任何一个环节都可能出错网络超时、API限额、模型生成不符合格式等。健壮的链必须有完善的错误处理。class RobustChain: def invoke_safely(self, input_data: dict, max_retries: int 2) - dict: “”“带重试和降级处理的链调用”“” last_exception None for attempt in range(max_retries 1): try: return self.chain.invoke(input_data) except Exception as e: last_exception e logger.warning(f“链调用第{attempt1}次失败: {e}”) if attempt max_retries: time.sleep(1 * (attempt 1)) # 指数退避 else: # 所有重试都失败执行降级策略 logger.error(f“链调用最终失败启用降级策略。”) return self._fallback_response(input_data) def _fallback_response(self, input_data: dict) - dict: “”“降级策略返回一个友好的默认响应。”“” return {“output”: “系统暂时无法处理您的请求请稍后再试。”}在封装链时考虑加入重试逻辑、超时设置以及最终的降级方案能极大提升用户体验和系统可用性。7. 常见问题与排查技巧实录在实际开发和部署链的过程中你会遇到各种各样的问题。这里我记录了一些最常见的问题和解决方法希望能帮你少走弯路。7.1 链执行报错输入/输出结构不匹配问题现象调用chain.invoke()时出现类似ValueError: Missing some input keys: [‘context’]的错误。根本原因链中某个组件的输入期望与上游组件的输出不匹配。这是使用LCEL组装链时最容易出错的地方。排查步骤逐段测试不要一次性组装整个长链。先测试第一部分如prompt | llm确保它能正常工作并输出你期望的格式。打印中间结果使用RunnableLambda在链中插入打印语句。from langchain.schema.runnable import RunnableLambda def debug_print(x): print(f“DEBUG: 中间结果类型{type(x)}, 值{x}”) return x # 将debug_print插入到链中怀疑有问题的地方 debug_chain prompt | debug_print | llm | output_parser检查提示词变量确保你的PromptTemplate中定义的变量名如{context},{question}与上游传递来的字典键名完全一致包括大小写。7.2 检索链效果不佳返回不相关文档问题现象RAG链给出的答案与问题无关或者“幻觉”严重。排查与优化检查检索结果使用我们上面封装的similarity_search方法直接查看针对某个问题向量库返回了哪些文档。如果文档完全不相关问题出在检索环节。优化文本分割这是影响检索质量的最关键因素之一。调整chunk_size如果文档块太大可能包含多个不相关主题导致检索精度下降。尝试减小chunk_size如从1000降到500。调整chunk_overlap适当增加重叠可以防止拆分切断重要信息。尝试不同的分割器RecursiveCharacterTextSplitter是通用选择。对于代码可以用LanguageTextSplitter对于Markdown可以用MarkdownHeaderTextSplitter按标题分割效果更好。优化嵌入模型不同的嵌入模型对语义的理解能力不同。可以尝试换用其他模型如text-embedding-3-large或者开源模型通过HuggingFaceEmbeddings加载。优化提示词在系统指令中反复强调“仅根据上下文回答”并设计更严格的格式要求。有时可以要求模型在答案中引用来源文档的编号。7.3 链的执行速度太慢问题现象用户请求响应时间过长。性能瓶颈定位使用LangSmith这是最直观的方法。查看Trace详情找出耗时最长的环节。通常是LLM调用或向量检索。分段计时在代码中手动添加时间戳记录每个主要步骤的耗时。针对性优化LLM慢考虑使用更快的模型如gpt-3.5-turbo比gpt-4快、降低max_tokens、启用流式响应streamingTrue让用户感知更快。检索慢检查向量数据库的索引类型、是否在内存中。对于大规模数据确保使用了高效的索引如HNSW。考虑对检索结果进行缓存。网络延迟确保你的服务部署在离模型API服务器较近的区域。7.4 如何处理流式输出很多场景下我们希望LLM的生成结果能够逐字逐句地返回给前端而不是等待全部生成完毕。LangChain对流式输出有很好的支持。# 流式调用示例 from langchain.schema.output_parser import StrOutputParser from langchain.schema.runnable import RunnablePassthrough simple_chain prompt | llm # 使用stream方法 for chunk in simple_chain.stream({“question”: “你好”}): # chunk可能是AIMessageChunk等类型需要从中提取内容 if hasattr(chunk, ‘content’): print(chunk.content, end“”, flushTrue) # 逐块打印在封装链时你可以提供一个stream方法内部调用底层链的stream方法并将处理好的文本块通过生成器yield返回这样就能轻松实现与前端如WebSocket的流式对接。封装AI执行链是将零散的AI能力模块化、工程化的关键一步。它让我们的代码从“脚本”走向了“系统”。今天介绍的基础问答链、RAG链和路由链是三种最核心的模式掌握了它们你就能应对绝大多数AI应用编排的需求。记住封装的核心思想是高内聚、低耦合一个链最好只做好一件事复杂的流程通过链的组合来实现。多利用LangSmith进行调试多思考异常处理和性能优化你的AI应用就会越来越稳健、高效。
