1. 项目概述为什么我们需要“写明白”LangChain的链与历史如果你最近在折腾大语言模型应用开发大概率绕不开LangChain这个名字。它像是一个乐高积木箱提供了各种标准化的组件让你能把大模型、外部工具、数据源和业务逻辑拼装成一个能跑起来的智能应用。但很多开发者包括我自己在初学阶段都有过类似的困惑代码跑起来了结果也出来了但中间到底发生了什么为什么这次调用成功了上次却报了个莫名其妙的错那个叫“链”的东西数据到底是怎么一步步流过去的这就是标题里“写明白”三个字的由来。它指的不仅仅是把代码写对更是要把LangChain内部那种“黑盒”般的执行过程变得透明、可追溯、可调试。核心痛点集中在两块链式调用和历史记录。链式调用是LangChain的骨架它定义了任务执行的流程和顺序而历史记录则是这个流程的“行车记录仪”记录了每一次交互的输入、输出、中间状态乃至错误信息。不理解链你就无法设计出高效、可靠的流程不掌握历史你就无法在出问题时进行有效的诊断和优化。网络上关于“接受家庭邀请失败。根据您的购买历史记录您似乎与此steam家庭的其他成员不在同”这类错误提示的讨论虽然场景不同但内核是相通的——系统基于历史记录做出了一个判断。在LangChain构建的AI应用中历史记录同样至关重要它决定了智能体Agent的上下文理解、链Chain的下一步决策以及整个应用的可观测性。把这两样东西“写明白”意味着从“能用”到“好用且可控”的关键一跃。这篇文章我就结合自己趟过的坑来拆解如何清晰地构建链并有效地记录和利用每一次调用的历史。2. 核心概念拆解链、历史与可观测性在动手写代码之前我们必须把几个核心概念及其关系理清楚。很多教程直接上代码但概念模糊会导致后续的设计举步维艰。2.1 链式调用不止是顺序执行LangChain中的“链”其核心思想是将多个对大语言模型的调用、工具的使用或数据处理步骤组合成一个更高阶的工作流。最简单的链是LLMChain它基本上就是“提示词模板 LLM调用”。但LangChain的强大之处在于链的组装。1. 链的类型与选择顺序链 (SequentialChain)这是最直观的链前一个步骤的输出作为后一个步骤的输入。适合有明确前后依赖关系的线性任务比如“总结文本 - 提取关键词 - 生成标题”。转换链 (TransformChain)用于对数据进行纯Python函数的转换不调用LLM比如清洗格式、提取特定字段。它常作为预处理或后处理环节嵌入到更大的链中。路由链 (RouterChain)这是一个高级概念根据输入内容动态决定下一步调用哪个子链。这构成了智能体Agent决策逻辑的基础。当你看到网络热词里提到的LangGraph它本质上就是为管理更复杂、有状态、可能循环或并行的路由工作流而设计的框架。LangChain提供了基础的链式组装而LangGraph更擅长描述带有循环和状态的工作流图。自定义链通过继承Chain基类你可以完全控制输入/输出的格式、内部步骤的执行逻辑和中间结果的保存方式。这是实现复杂、定制化业务逻辑的终极手段。选择逻辑如果你的流程是线性的用SequentialChain如果需要复杂的条件分支和循环考虑LangGraph或自定义链如果只是简单的提示词LLMLLMChain就够了。2. 数据流与invoke/stream调用一个链主要使用invoke或stream方法。invoke是同步调用一次性返回所有结果stream则返回一个生成器用于流式输出尤其适合需要实时看到LLM生成过程的场景比如聊天。网络热词中提到的“流式输出吞掉reasoning-content字段”就是在此场景下可能遇到的序列化或框架兼容性问题。注意在组装链时务必确认每一步的输入/输出键名。一个常见的坑是上一步的输出字典键名与下一步输入期望的键名不匹配导致链执行失败。清晰的命名和文档是避免此问题的关键。2.2 历史记录上下文、记忆与调试的基石历史记录在LangChain生态中有多层含义对应不同的使用场景1. 对话记忆 (Memory)这是最常说的“历史记录”用于让LLM记住当前会话中之前的对话内容。ConversationBufferMemory会把所有历史对话都存进去简单但可能导致上下文过长ConversationSummaryMemory则会自动对历史进行摘要以节省TokenConversationBufferWindowMemory只保留最近K轮对话。选择依据完全取决于你的应用场景和对上下文长度的要求。2. 调用追踪与日志 (Callbacks,Tracing)这是实现“写明白”的技术核心。它记录的不是对话内容而是链的每一次内部调用Invocation的详细信息。Callbacks(回调)你可以在调用链时传入一个回调处理器如StdOutCallbackHandler它会在控制台实时打印出链的每一步执行信息包括调用了哪个组件、输入是什么、输出是什么。这对于本地调试极其有用。LangSmith(官方追踪平台)这是生产级应用的首选。通过配置一个环境变量LangChain会自动将每次链执行的详细轨迹包括耗时、输入输出、中间步骤、Token消耗、成本发送到LangSmith平台。你可以可视化地查看整个链的“执行树”精准定位性能瓶颈或逻辑错误。网络热词中提到的“langchain 打印invoke发送的内容”用回调或LangSmith可以完美解决。3. 应用状态历史对于像智能体这样的应用历史还包括其调用过的工具、工具返回的结果、以及自身内部状态的变化。这部分历史对于实现复杂的、多步骤的推理任务至关重要。把历史“写明白”意味着你需要根据目的选择合适的工具调试用回调或LangSmith维持对话上下文用Memory分析智能体行为则需要结合工具调用记录。2.3 可观测性将链与历史关联起来可观测性不是某个具体功能而是一种设计目标。它的实现依赖于我们主动地在链的执行过程中注入和暴露关键信息。一个具备良好可观测性的链应该能轻松回答以下问题用户的原始输入是什么链的每个步骤分别收到了什么输入产出了什么输出调用LLM的实际提示词Prompt是什么每一步耗时多少消耗了多少Token如果出错了错误发生在哪个具体环节当时的上下文数据是什么通过结合结构清晰的链设计和详尽的历史记录我们就能构建出可观测的系统。例如在自定义链的_call方法中除了执行业务逻辑你还应该有意识地将中间变量通过run_manager的on_text等方法输出到回调或者确保它们被包含在返回的字典中以便被LangSmith捕获。3. 实战构建一个可观测的问答链理论说再多不如动手。我们来实现一个经典的RAG检索增强生成问答链并确保它的每一步都是清晰可见的。这个链的流程是用户提问 - 将问题转换为向量进行检索 - 从知识库获取相关文档 - 组合文档和问题生成最终答案。3.1 环境准备与链设计首先假设我们已经有一个可用的向量数据库如Chroma和嵌入模型。我们设计一个简单的RAGChain。# 基础组件导入 from langchain.chains import LLMChain, RetrievalQA from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory from langchain.callbacks import StdOutCallbackHandler from langchain_community.vectorstores import Chroma from langchain_openai import ChatOpenAI, OpenAIEmbeddings # 1. 初始化核心组件 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) embeddings OpenAIEmbeddings() vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索3条最相关文档 # 2. 设计提示词模板 prompt_template 你是一个专业的助手请根据以下上下文来回答问题。如果上下文不包含答案请直接说“根据提供的资料我无法回答这个问题”不要编造信息。 上下文 {context} 问题{question} 请给出答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 3. 使用标准RetrievalQA链它本身就是一个封装好的链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最简单的文档整合方式 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 关键返回源文档方便追溯 )这个qa_chain已经可以工作了但它的内部过程对我们来说是黑盒。我们接下来让它“透明”起来。3.2 注入回调实时查看执行过程我们创建一个自定义的回调处理器来打印更详细的信息。同时使用StdOutCallbackHandler作为基础。from langchain.callbacks.base import BaseCallbackHandler from typing import Any, Dict, List import json class DetailedLoggingCallbackHandler(BaseCallbackHandler): 一个详细记录链执行步骤的回调处理器 def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs) - None: chain_name serialized.get(name, serialized.get(id, [unknown])[-1]) print(f\n [链开始] {chain_name}) print(f 输入: {json.dumps(inputs, indent2, ensure_asciiFalse)}) def on_chain_end(self, outputs: Dict[str, Any], **kwargs) - None: print(f\n✅ [链结束]) # 注意避免打印可能很长的完整文档内容 filtered_outputs {k: (v[:200] ... if isinstance(v, str) and len(v) 200 else v) for k, v in outputs.items()} print(f 输出: {json.dumps(filtered_outputs, indent2, ensure_asciiFalse)}) def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) - None: print(f\n [LLM调用开始]) # 打印前500个字符的提示词用于检查 preview prompts[0][:500] print(f 提示词预览: {preview}...) def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) - None: tool_name serialized.get(name, unknown) print(f\n️ [工具调用开始] {tool_name}) print(f 工具输入: {input_str}) # 使用回调 callbacks [StdOutCallbackHandler(), DetailedLoggingCallbackHandler()] result qa_chain.invoke({query: LangChain中链的主要类型有哪些}, config{callbacks: callbacks}) print(\n最终答案:, result[result])运行这段代码你会在控制台看到类似下面的输出清晰地展示了链的激活、检索工具的调用、LLM的提示词生成以及最终结果的返回。 [链开始] RetrievalQA 输入: { query: LangChain中链的主要类型有哪些 } ️ [工具调用开始] Retriever 工具输入: LangChain中链的主要类型有哪些 [LLM调用开始] 提示词预览: 你是一个专业的助手请根据以下上下文来回答问题... ... ✅ [链结束] 输出: { result: LangChain中主要的链类型包括..., source_documents: [...] }3.3 集成LangSmith生产级追踪对于线上应用控制台日志不够用。我们需要LangSmith。注册并获取API密钥在LangSmith官网注册创建一个项目获得LANGSMITH_API_KEY。配置环境变量export LANGCHAIN_TRACING_V2true export LANGCHAIN_ENDPOINThttps://api.smith.langchain.com export LANGCHAIN_API_KEY你的-api-key export LANGCHAIN_PROJECT你的项目名 # 可选默认是default运行你的链无需修改代码只要环境变量设置正确LangChain会自动将追踪数据发送到LangSmith。在LangSmith界面查看登录LangSmith进入对应项目你会看到每一次调用的记录。点击进入可以看到一个完整的执行轨迹树精确展示了从RetrievalQA链开始到Retriever工具调用再到LLMChain生成答案的全过程每一步的输入输出、耗时、Token数一目了然。这是“写明白”的终极形态。3.4 封装自定义链以获得完全控制标准链有时无法满足复杂需求。我们封装一个自定义链将检索和生成步骤更显式地分开便于插入自定义逻辑和记录。from langchain.chains.base import Chain from typing import Dict, List, Optional, Any class ObservableRAGChain(Chain): 一个可观测的自定义RAG链 retriever: Any # 检索器 llm_chain: LLMChain # 用于生成答案的LLM链 memory: Optional[ConversationBufferMemory] None # 可选记忆 property def input_keys(self) - List[str]: return [question] property def output_keys(self) - List[str]: return [answer, source_documents, intermediate_steps] def _call(self, inputs: Dict[str, Any], run_manager None) - Dict[str, Any]: 核心执行逻辑 question inputs[question] intermediate_steps [] # 用于记录中间步骤 # 步骤1检索 if run_manager: run_manager.on_text(f开始检索与问题相关文档...\n, colorgreen) docs self.retriever.get_relevant_documents(question) intermediate_steps.append({step: retrieval, query: question, doc_count: len(docs)}) if run_manager: run_manager.on_text(f检索到 {len(docs)} 条相关文档。\n, colorgreen) # 步骤2组合上下文 context \n\n.join([doc.page_content for doc in docs]) intermediate_steps.append({step: context_assembly, context_preview: context[:500]}) # 步骤3调用LLM生成答案 llm_inputs {context: context, question: question} if run_manager: run_manager.on_text(f调用LLM生成最终答案...\n, coloryellow) answer_result self.llm_chain.invoke(llm_inputs, config{callbacks: run_manager.get_child() if run_manager else None}) answer answer_result[text] intermediate_steps.append({step: generation, llm_inputs_keys: list(llm_inputs.keys())}) # 步骤4更新记忆如果存在 if self.memory: self.memory.save_context({input: question}, {output: answer}) # 返回结果包含中间步骤 return { answer: answer, source_documents: docs, intermediate_steps: intermediate_steps # 关键暴露内部状态 } # 使用自定义链 llm_chain LLMChain(llmllm, promptPROMPT) custom_rag_chain ObservableRAGChain( retrieverretriever, llm_chainllm_chain, memoryConversationBufferMemory() ) result custom_rag_chain.invoke( {question: 解释一下链式调用和智能体的区别}, config{callbacks: [DetailedLoggingCallbackHandler()]} ) print(答案, result[answer]) print(\n--- 中间步骤详情 ---) for step in result[intermediate_steps]: print(step)这个自定义链的_call方法清晰地定义了三个步骤并通过intermediate_steps列表和run_manager回调主动暴露了所有关键中间状态。无论你是查看控制台日志还是通过LangSmith追踪都能对数据流了如指掌。4. 常见问题与排查技巧实录在实际使用中你会遇到各种问题。下面是我总结的一些典型场景和解决方法。4.1 链执行报错“Missing required input keys”问题描述调用链时提示缺少某个输入键。排查思路检查链的input_keys每个链都有定义的输入键。使用chain.input_keys查看。检查链的输入字典确保你传入的invoke或run的字典包含了所有必需的键。检查顺序链的串联在SequentialChain中确保前一个链的output_keys与后一个链的input_keys完全匹配。一个常见的错误是命名不一致。实操技巧在组装复杂链时为每个子链的输入输出键使用有明确意义的常量字符串避免硬编码的魔法字符串。例如SUMMARY_KEY “summary_text” KEYWORDS_KEY “extracted_keywords” # 然后在链定义和串联时都使用这些常量4.2 历史记录Memory混乱或超出上下文问题描述对话进行几轮后LLM的回答开始偏离主题或忘记之前的内容或者提示词因过长而报错。排查思路确认Memory类型你用的是ConversationBufferMemory吗它会无限制增长。对于长对话应换用ConversationSummaryMemory或ConversationBufferWindowMemory。检查Memory的存储键确保保存上下文save_context和加载上下文load_memory_variables时使用的输入/输出键名与链的期望一致。手动查看Memory内容在调试时直接打印memory.buffer或memory.load_memory_variables({})来确认里面存储了什么。实操技巧在开发阶段可以写一个简单的路由当用户输入“/debug_memory”时将当前记忆的内容安全地注意过滤隐私返回给用户便于诊断。4.3 LangSmith看不到追踪数据或数据不全问题描述配置了环境变量但LangSmith控制台没有收到数据或者数据缺少中间步骤。排查思路验证环境变量确保LANGCHAIN_TRACING_V2true注意是V2并且API密钥和终端点正确。检查网络和代理确保运行环境能访问https://api.smith.langchain.com。检查链是否支持追踪标准LangChain组件都支持。但如果你用了大量自定义Python函数且未通过run_manager传递回调这些内部步骤可能不会被记录。使用traceable装饰器对于你想追踪的自定义函数可以用langchain_core.tracers.traceable装饰器来手动标记。实操技巧在代码开头添加一段检查快速确认配置是否生效import os if os.getenv(“LANGCHAIN_TRACING_V2”): print(“LangSmith 追踪已启用”) else: print(“警告: LangSmith 追踪未启用”)4.4 流式输出Streaming时信息丢失问题描述使用stream时某些中间信息如网络热词中提到的reasoning-content在最终输出中看不到。排查思路理解流式输出对象stream返回的是异步生成器产出的是AIMessageChunk等增量对象。最终完整的消息属性可能分布在多个chunk中。检查回调处理在流式模式下确保你的回调处理器正确处理了on_llm_new_token和on_llm_end等事件以捕获完整信息。聚合Chunks如果需要获取完整的响应内容包括reasoning等字段你需要手动聚合所有chunks。对于OpenAI的某些模型reasoning_content可能是一个独立的输出流需要特殊处理。实操技巧如果不必须流式输出在调试阶段先用invoke获取完整响应确认数据结构。然后再处理流式逻辑。对于复杂响应参考官方模型文档了解其流式输出的具体格式。4.5 智能体Agent决策过程不透明问题描述智能体直接给出了最终答案但不知道它为什么选择调用某个工具思考过程是什么。排查思路启用详细输出在初始化智能体时设置verboseTrue这会在控制台打印出智能体的“思考”Thought过程。使用LangSmith这是最佳实践。LangSmith可以完整记录智能体的每一次思考、每一次工具调用和观察。解析返回的中间步骤大多数智能体执行方法如agent.invoke的返回结果中都包含一个intermediate_steps键里面按顺序记录了Thought, Action, Observation的元组列表。这是分析其推理路径的原始数据。实操技巧编写一个后处理函数将intermediate_steps格式化成更易读的字符串方便日志记录或展示给用户在安全的前提下这极大地增强了智能体行为的可信度。把链和历史“写明白”本质上是一种工程素养的体现。它要求我们在追求功能实现的同时必须考虑系统的可调试性、可维护性和可观测性。从清晰地设计数据流到有策略地记录关键状态再到利用好LangSmith这样的专业工具每一步都是在为应用的稳定性和开发效率加码。当你能够清晰地回答“我的AI应用为什么输出了这个答案”时你就真正掌握了LangChain也就能更自信地构建更复杂的智能系统。
LangChain链式调用与历史记录：构建可观测AI应用的核心实践
1. 项目概述为什么我们需要“写明白”LangChain的链与历史如果你最近在折腾大语言模型应用开发大概率绕不开LangChain这个名字。它像是一个乐高积木箱提供了各种标准化的组件让你能把大模型、外部工具、数据源和业务逻辑拼装成一个能跑起来的智能应用。但很多开发者包括我自己在初学阶段都有过类似的困惑代码跑起来了结果也出来了但中间到底发生了什么为什么这次调用成功了上次却报了个莫名其妙的错那个叫“链”的东西数据到底是怎么一步步流过去的这就是标题里“写明白”三个字的由来。它指的不仅仅是把代码写对更是要把LangChain内部那种“黑盒”般的执行过程变得透明、可追溯、可调试。核心痛点集中在两块链式调用和历史记录。链式调用是LangChain的骨架它定义了任务执行的流程和顺序而历史记录则是这个流程的“行车记录仪”记录了每一次交互的输入、输出、中间状态乃至错误信息。不理解链你就无法设计出高效、可靠的流程不掌握历史你就无法在出问题时进行有效的诊断和优化。网络上关于“接受家庭邀请失败。根据您的购买历史记录您似乎与此steam家庭的其他成员不在同”这类错误提示的讨论虽然场景不同但内核是相通的——系统基于历史记录做出了一个判断。在LangChain构建的AI应用中历史记录同样至关重要它决定了智能体Agent的上下文理解、链Chain的下一步决策以及整个应用的可观测性。把这两样东西“写明白”意味着从“能用”到“好用且可控”的关键一跃。这篇文章我就结合自己趟过的坑来拆解如何清晰地构建链并有效地记录和利用每一次调用的历史。2. 核心概念拆解链、历史与可观测性在动手写代码之前我们必须把几个核心概念及其关系理清楚。很多教程直接上代码但概念模糊会导致后续的设计举步维艰。2.1 链式调用不止是顺序执行LangChain中的“链”其核心思想是将多个对大语言模型的调用、工具的使用或数据处理步骤组合成一个更高阶的工作流。最简单的链是LLMChain它基本上就是“提示词模板 LLM调用”。但LangChain的强大之处在于链的组装。1. 链的类型与选择顺序链 (SequentialChain)这是最直观的链前一个步骤的输出作为后一个步骤的输入。适合有明确前后依赖关系的线性任务比如“总结文本 - 提取关键词 - 生成标题”。转换链 (TransformChain)用于对数据进行纯Python函数的转换不调用LLM比如清洗格式、提取特定字段。它常作为预处理或后处理环节嵌入到更大的链中。路由链 (RouterChain)这是一个高级概念根据输入内容动态决定下一步调用哪个子链。这构成了智能体Agent决策逻辑的基础。当你看到网络热词里提到的LangGraph它本质上就是为管理更复杂、有状态、可能循环或并行的路由工作流而设计的框架。LangChain提供了基础的链式组装而LangGraph更擅长描述带有循环和状态的工作流图。自定义链通过继承Chain基类你可以完全控制输入/输出的格式、内部步骤的执行逻辑和中间结果的保存方式。这是实现复杂、定制化业务逻辑的终极手段。选择逻辑如果你的流程是线性的用SequentialChain如果需要复杂的条件分支和循环考虑LangGraph或自定义链如果只是简单的提示词LLMLLMChain就够了。2. 数据流与invoke/stream调用一个链主要使用invoke或stream方法。invoke是同步调用一次性返回所有结果stream则返回一个生成器用于流式输出尤其适合需要实时看到LLM生成过程的场景比如聊天。网络热词中提到的“流式输出吞掉reasoning-content字段”就是在此场景下可能遇到的序列化或框架兼容性问题。注意在组装链时务必确认每一步的输入/输出键名。一个常见的坑是上一步的输出字典键名与下一步输入期望的键名不匹配导致链执行失败。清晰的命名和文档是避免此问题的关键。2.2 历史记录上下文、记忆与调试的基石历史记录在LangChain生态中有多层含义对应不同的使用场景1. 对话记忆 (Memory)这是最常说的“历史记录”用于让LLM记住当前会话中之前的对话内容。ConversationBufferMemory会把所有历史对话都存进去简单但可能导致上下文过长ConversationSummaryMemory则会自动对历史进行摘要以节省TokenConversationBufferWindowMemory只保留最近K轮对话。选择依据完全取决于你的应用场景和对上下文长度的要求。2. 调用追踪与日志 (Callbacks,Tracing)这是实现“写明白”的技术核心。它记录的不是对话内容而是链的每一次内部调用Invocation的详细信息。Callbacks(回调)你可以在调用链时传入一个回调处理器如StdOutCallbackHandler它会在控制台实时打印出链的每一步执行信息包括调用了哪个组件、输入是什么、输出是什么。这对于本地调试极其有用。LangSmith(官方追踪平台)这是生产级应用的首选。通过配置一个环境变量LangChain会自动将每次链执行的详细轨迹包括耗时、输入输出、中间步骤、Token消耗、成本发送到LangSmith平台。你可以可视化地查看整个链的“执行树”精准定位性能瓶颈或逻辑错误。网络热词中提到的“langchain 打印invoke发送的内容”用回调或LangSmith可以完美解决。3. 应用状态历史对于像智能体这样的应用历史还包括其调用过的工具、工具返回的结果、以及自身内部状态的变化。这部分历史对于实现复杂的、多步骤的推理任务至关重要。把历史“写明白”意味着你需要根据目的选择合适的工具调试用回调或LangSmith维持对话上下文用Memory分析智能体行为则需要结合工具调用记录。2.3 可观测性将链与历史关联起来可观测性不是某个具体功能而是一种设计目标。它的实现依赖于我们主动地在链的执行过程中注入和暴露关键信息。一个具备良好可观测性的链应该能轻松回答以下问题用户的原始输入是什么链的每个步骤分别收到了什么输入产出了什么输出调用LLM的实际提示词Prompt是什么每一步耗时多少消耗了多少Token如果出错了错误发生在哪个具体环节当时的上下文数据是什么通过结合结构清晰的链设计和详尽的历史记录我们就能构建出可观测的系统。例如在自定义链的_call方法中除了执行业务逻辑你还应该有意识地将中间变量通过run_manager的on_text等方法输出到回调或者确保它们被包含在返回的字典中以便被LangSmith捕获。3. 实战构建一个可观测的问答链理论说再多不如动手。我们来实现一个经典的RAG检索增强生成问答链并确保它的每一步都是清晰可见的。这个链的流程是用户提问 - 将问题转换为向量进行检索 - 从知识库获取相关文档 - 组合文档和问题生成最终答案。3.1 环境准备与链设计首先假设我们已经有一个可用的向量数据库如Chroma和嵌入模型。我们设计一个简单的RAGChain。# 基础组件导入 from langchain.chains import LLMChain, RetrievalQA from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory from langchain.callbacks import StdOutCallbackHandler from langchain_community.vectorstores import Chroma from langchain_openai import ChatOpenAI, OpenAIEmbeddings # 1. 初始化核心组件 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) embeddings OpenAIEmbeddings() vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索3条最相关文档 # 2. 设计提示词模板 prompt_template 你是一个专业的助手请根据以下上下文来回答问题。如果上下文不包含答案请直接说“根据提供的资料我无法回答这个问题”不要编造信息。 上下文 {context} 问题{question} 请给出答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 3. 使用标准RetrievalQA链它本身就是一个封装好的链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最简单的文档整合方式 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 关键返回源文档方便追溯 )这个qa_chain已经可以工作了但它的内部过程对我们来说是黑盒。我们接下来让它“透明”起来。3.2 注入回调实时查看执行过程我们创建一个自定义的回调处理器来打印更详细的信息。同时使用StdOutCallbackHandler作为基础。from langchain.callbacks.base import BaseCallbackHandler from typing import Any, Dict, List import json class DetailedLoggingCallbackHandler(BaseCallbackHandler): 一个详细记录链执行步骤的回调处理器 def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs) - None: chain_name serialized.get(name, serialized.get(id, [unknown])[-1]) print(f\n [链开始] {chain_name}) print(f 输入: {json.dumps(inputs, indent2, ensure_asciiFalse)}) def on_chain_end(self, outputs: Dict[str, Any], **kwargs) - None: print(f\n✅ [链结束]) # 注意避免打印可能很长的完整文档内容 filtered_outputs {k: (v[:200] ... if isinstance(v, str) and len(v) 200 else v) for k, v in outputs.items()} print(f 输出: {json.dumps(filtered_outputs, indent2, ensure_asciiFalse)}) def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) - None: print(f\n [LLM调用开始]) # 打印前500个字符的提示词用于检查 preview prompts[0][:500] print(f 提示词预览: {preview}...) def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) - None: tool_name serialized.get(name, unknown) print(f\n️ [工具调用开始] {tool_name}) print(f 工具输入: {input_str}) # 使用回调 callbacks [StdOutCallbackHandler(), DetailedLoggingCallbackHandler()] result qa_chain.invoke({query: LangChain中链的主要类型有哪些}, config{callbacks: callbacks}) print(\n最终答案:, result[result])运行这段代码你会在控制台看到类似下面的输出清晰地展示了链的激活、检索工具的调用、LLM的提示词生成以及最终结果的返回。 [链开始] RetrievalQA 输入: { query: LangChain中链的主要类型有哪些 } ️ [工具调用开始] Retriever 工具输入: LangChain中链的主要类型有哪些 [LLM调用开始] 提示词预览: 你是一个专业的助手请根据以下上下文来回答问题... ... ✅ [链结束] 输出: { result: LangChain中主要的链类型包括..., source_documents: [...] }3.3 集成LangSmith生产级追踪对于线上应用控制台日志不够用。我们需要LangSmith。注册并获取API密钥在LangSmith官网注册创建一个项目获得LANGSMITH_API_KEY。配置环境变量export LANGCHAIN_TRACING_V2true export LANGCHAIN_ENDPOINThttps://api.smith.langchain.com export LANGCHAIN_API_KEY你的-api-key export LANGCHAIN_PROJECT你的项目名 # 可选默认是default运行你的链无需修改代码只要环境变量设置正确LangChain会自动将追踪数据发送到LangSmith。在LangSmith界面查看登录LangSmith进入对应项目你会看到每一次调用的记录。点击进入可以看到一个完整的执行轨迹树精确展示了从RetrievalQA链开始到Retriever工具调用再到LLMChain生成答案的全过程每一步的输入输出、耗时、Token数一目了然。这是“写明白”的终极形态。3.4 封装自定义链以获得完全控制标准链有时无法满足复杂需求。我们封装一个自定义链将检索和生成步骤更显式地分开便于插入自定义逻辑和记录。from langchain.chains.base import Chain from typing import Dict, List, Optional, Any class ObservableRAGChain(Chain): 一个可观测的自定义RAG链 retriever: Any # 检索器 llm_chain: LLMChain # 用于生成答案的LLM链 memory: Optional[ConversationBufferMemory] None # 可选记忆 property def input_keys(self) - List[str]: return [question] property def output_keys(self) - List[str]: return [answer, source_documents, intermediate_steps] def _call(self, inputs: Dict[str, Any], run_manager None) - Dict[str, Any]: 核心执行逻辑 question inputs[question] intermediate_steps [] # 用于记录中间步骤 # 步骤1检索 if run_manager: run_manager.on_text(f开始检索与问题相关文档...\n, colorgreen) docs self.retriever.get_relevant_documents(question) intermediate_steps.append({step: retrieval, query: question, doc_count: len(docs)}) if run_manager: run_manager.on_text(f检索到 {len(docs)} 条相关文档。\n, colorgreen) # 步骤2组合上下文 context \n\n.join([doc.page_content for doc in docs]) intermediate_steps.append({step: context_assembly, context_preview: context[:500]}) # 步骤3调用LLM生成答案 llm_inputs {context: context, question: question} if run_manager: run_manager.on_text(f调用LLM生成最终答案...\n, coloryellow) answer_result self.llm_chain.invoke(llm_inputs, config{callbacks: run_manager.get_child() if run_manager else None}) answer answer_result[text] intermediate_steps.append({step: generation, llm_inputs_keys: list(llm_inputs.keys())}) # 步骤4更新记忆如果存在 if self.memory: self.memory.save_context({input: question}, {output: answer}) # 返回结果包含中间步骤 return { answer: answer, source_documents: docs, intermediate_steps: intermediate_steps # 关键暴露内部状态 } # 使用自定义链 llm_chain LLMChain(llmllm, promptPROMPT) custom_rag_chain ObservableRAGChain( retrieverretriever, llm_chainllm_chain, memoryConversationBufferMemory() ) result custom_rag_chain.invoke( {question: 解释一下链式调用和智能体的区别}, config{callbacks: [DetailedLoggingCallbackHandler()]} ) print(答案, result[answer]) print(\n--- 中间步骤详情 ---) for step in result[intermediate_steps]: print(step)这个自定义链的_call方法清晰地定义了三个步骤并通过intermediate_steps列表和run_manager回调主动暴露了所有关键中间状态。无论你是查看控制台日志还是通过LangSmith追踪都能对数据流了如指掌。4. 常见问题与排查技巧实录在实际使用中你会遇到各种问题。下面是我总结的一些典型场景和解决方法。4.1 链执行报错“Missing required input keys”问题描述调用链时提示缺少某个输入键。排查思路检查链的input_keys每个链都有定义的输入键。使用chain.input_keys查看。检查链的输入字典确保你传入的invoke或run的字典包含了所有必需的键。检查顺序链的串联在SequentialChain中确保前一个链的output_keys与后一个链的input_keys完全匹配。一个常见的错误是命名不一致。实操技巧在组装复杂链时为每个子链的输入输出键使用有明确意义的常量字符串避免硬编码的魔法字符串。例如SUMMARY_KEY “summary_text” KEYWORDS_KEY “extracted_keywords” # 然后在链定义和串联时都使用这些常量4.2 历史记录Memory混乱或超出上下文问题描述对话进行几轮后LLM的回答开始偏离主题或忘记之前的内容或者提示词因过长而报错。排查思路确认Memory类型你用的是ConversationBufferMemory吗它会无限制增长。对于长对话应换用ConversationSummaryMemory或ConversationBufferWindowMemory。检查Memory的存储键确保保存上下文save_context和加载上下文load_memory_variables时使用的输入/输出键名与链的期望一致。手动查看Memory内容在调试时直接打印memory.buffer或memory.load_memory_variables({})来确认里面存储了什么。实操技巧在开发阶段可以写一个简单的路由当用户输入“/debug_memory”时将当前记忆的内容安全地注意过滤隐私返回给用户便于诊断。4.3 LangSmith看不到追踪数据或数据不全问题描述配置了环境变量但LangSmith控制台没有收到数据或者数据缺少中间步骤。排查思路验证环境变量确保LANGCHAIN_TRACING_V2true注意是V2并且API密钥和终端点正确。检查网络和代理确保运行环境能访问https://api.smith.langchain.com。检查链是否支持追踪标准LangChain组件都支持。但如果你用了大量自定义Python函数且未通过run_manager传递回调这些内部步骤可能不会被记录。使用traceable装饰器对于你想追踪的自定义函数可以用langchain_core.tracers.traceable装饰器来手动标记。实操技巧在代码开头添加一段检查快速确认配置是否生效import os if os.getenv(“LANGCHAIN_TRACING_V2”): print(“LangSmith 追踪已启用”) else: print(“警告: LangSmith 追踪未启用”)4.4 流式输出Streaming时信息丢失问题描述使用stream时某些中间信息如网络热词中提到的reasoning-content在最终输出中看不到。排查思路理解流式输出对象stream返回的是异步生成器产出的是AIMessageChunk等增量对象。最终完整的消息属性可能分布在多个chunk中。检查回调处理在流式模式下确保你的回调处理器正确处理了on_llm_new_token和on_llm_end等事件以捕获完整信息。聚合Chunks如果需要获取完整的响应内容包括reasoning等字段你需要手动聚合所有chunks。对于OpenAI的某些模型reasoning_content可能是一个独立的输出流需要特殊处理。实操技巧如果不必须流式输出在调试阶段先用invoke获取完整响应确认数据结构。然后再处理流式逻辑。对于复杂响应参考官方模型文档了解其流式输出的具体格式。4.5 智能体Agent决策过程不透明问题描述智能体直接给出了最终答案但不知道它为什么选择调用某个工具思考过程是什么。排查思路启用详细输出在初始化智能体时设置verboseTrue这会在控制台打印出智能体的“思考”Thought过程。使用LangSmith这是最佳实践。LangSmith可以完整记录智能体的每一次思考、每一次工具调用和观察。解析返回的中间步骤大多数智能体执行方法如agent.invoke的返回结果中都包含一个intermediate_steps键里面按顺序记录了Thought, Action, Observation的元组列表。这是分析其推理路径的原始数据。实操技巧编写一个后处理函数将intermediate_steps格式化成更易读的字符串方便日志记录或展示给用户在安全的前提下这极大地增强了智能体行为的可信度。把链和历史“写明白”本质上是一种工程素养的体现。它要求我们在追求功能实现的同时必须考虑系统的可调试性、可维护性和可观测性。从清晰地设计数据流到有策略地记录关键状态再到利用好LangSmith这样的专业工具每一步都是在为应用的稳定性和开发效率加码。当你能够清晰地回答“我的AI应用为什么输出了这个答案”时你就真正掌握了LangChain也就能更自信地构建更复杂的智能系统。