1. 从“静态”到“动态”为什么运行时数据注入是LangChain的灵魂如果你用过LangChain大概率是从它的链Chain或者代理Agent开始的。你写好了提示词模板PromptTemplate定义好了工具Tool然后满怀期待地运行起来。但很快你就会遇到一个非常现实的问题我的链或代理怎么才能知道“现在”发生了什么比如用户刚刚上传了一个PDF对话历史里提到了某个关键信息或者系统刚刚从数据库里查询到了一组实时数据。你不可能把这些动态变化的信息在程序启动时就一股脑全塞进一个固定的提示词模板里。这就是“运行时数据注入”要解决的核心痛点。在LangChain的语境里Context和Runtime这两个词几乎可以看作是“动态智能”的代名词。Context上下文指的是在链或代理执行过程中那些动态存在、随时可能被访问和修改的数据。Runtime运行时则描述了这些数据在程序实际运行时的生命周期、传递路径和作用范围。很多人把LangChain用成了“高级的字符串拼接工具”就是因为只用了静态的PromptTemplate而忽略了运行时数据的动态注入。结果就是构建的应用僵硬、死板无法根据实时情况做出灵活响应。一个真正智能的、能处理复杂工作流的应用其核心能力恰恰体现在对运行时数据的精细化管理上。这不仅仅是技术实现更是一种设计思维的转变——从编写固定的执行脚本转变为设计一个能够感知环境、动态调整的数据流系统。2. 理解LangChain的运行时数据流Runnable与RunnableConfig的协奏要掌握数据注入必须先理解LangChain的执行模型。从v0.1.0开始LangChain全面转向了基于Runnable协议的架构。几乎一切可执行对象——链、模型、提示词、工具、输出解析器——都是Runnable。2.1Runnable一切执行的基石Runnable定义了一个标准的接口它接收一个输入字典并返回一个输出字典。这个简单的抽象是数据流动的基础。from langchain_core.runnables import RunnableLambda # 一个最简单的Runnable将输入翻倍 double RunnableLambda(lambda x: x * 2) print(double.invoke(5)) # 输出: 10但Runnable的强大之处在于它的组合性。你可以通过|操作符类似于Unix管道将多个Runnable连接起来形成复杂的数据处理流水线。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义提示词模板 prompt ChatPromptTemplate.from_template(请将以下数字翻倍{number}) # 定义模型 model ChatOpenAI(modelgpt-3.5-turbo) # 组合成链 chain prompt | model # 调用链时number就是我们需要注入的运行时数据 result chain.invoke({number: 5})在这个链被invoke或batch调用时字典{number: 5}就是我们在运行时注入的数据。它流经prompt被填充到模板中然后传递给model。2.2RunnableConfig运行时的控制面板如果Runnable是发动机那么RunnableConfig就是驾驶舱里的控制面板。它是一个字典包含了影响Runnable执行行为的各种配置。数据注入的许多高级技巧都依赖于对RunnableConfig的运用。RunnableConfig中几个关键的、与数据注入相关的字段包括callbacks: 用于记录日志、追踪执行过程。tags: 给当前执行打标签用于分类或过滤。metadata: 附加的元数据字典可以存放任意信息。configurable: 这是一个特殊字段是运行时数据注入的核心载体。你可以通过它来动态覆盖链中某些组件的配置。当你调用chain.invoke(input, configconfig)时这个config对象会随着执行流向下传递链中的每一个Runnable都可以访问到它。注意config的传递是自动的、隐式的。你不需要手动将它从一个组件传递到另一个组件LangChain框架会负责这件事。这保证了运行时上下文在复杂链中也能无损传递。3. 核心注入模式一通过configurable_fields动态覆盖配置这是最直接、最强大的运行时配置注入方式。它允许你在不修改链定义代码的情况下在调用时动态改变链中特定组件的属性。想象一个场景你有一个客服聊天链核心是一个LLM。在白天你希望使用快速但能力稍弱的模型如gpt-3.5-turbo来应对大量咨询在深夜流量低时你希望切换到能力更强但更慢的模型如gpt-4来处理复杂问题。如果为此写两个不同的链就太笨重了。configurable_fields完美解决了这个问题。3.1 定义可配置的链首先在定义链时使用configurable_fields方法将需要动态改变的组件“标记”出来。from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import ConfigurableField from langchain_openai import ChatOpenAI # 1. 定义一个基础模型并使其可配置 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7).configurable_fields( modelConfigurableField( idmodel_name, nameLLM Model Name, descriptionThe name of the LLM model to use, ), temperatureConfigurableField( idtemperature, nameLLM Temperature, descriptionThe temperature for the LLM, ) ) # 2. 定义提示词 prompt ChatPromptTemplate.from_template(你是一个专业的客服。用户说{query}) # 3. 组合成链 chain prompt | llm现在llm的model和temperature字段都变成了可配置项。id这里是model_name和temperature是我们后续在运行时引用它们的钥匙。3.2 在运行时注入配置在调用链时通过config参数的configurable字段来注入新的值。# 场景A白天使用快速模型 daytime_config { configurable: { model_name: gpt-3.5-turbo, # 覆盖默认的model temperature: 0.7 } } result_day chain.invoke({query: 我的订单什么时候发货}, configdaytime_config) # 场景B深夜使用更强模型处理复杂问题 night_config { configurable: { model_name: gpt-4, # 运行时动态切换为gpt-4 temperature: 0.3 # 同时降低temperature让回答更确定 } } result_night chain.invoke({query: 请根据我的历史购买记录和当前库存为我推荐三款可能感兴趣的新品并说明理由。}, confignight_config)通过这种方式同一个链在不同的运行时配置下表现出了完全不同的行为。这实现了业务逻辑链的结构与运行时策略模型参数的彻底解耦。3.3 实战心得不仅仅是模型configurable_fields的应用远不止于LLM模型。你可以用它来动态切换数据库连接根据用户区域切换不同的数据库实例。检索器Retriever在普通搜索和向量化语义搜索之间动态选择。提示词模板根据用户语言动态切换不同语种的模板。API密钥在多个服务商之间做负载均衡或故障转移。我个人的经验是在项目初期就思考哪些组件可能在运行时需要变化并提前将它们configurable_fields化。这能为后续的灵活扩展省去大量重构代码的麻烦。一个常见的“坑”是忘记给ConfigurableField设置清晰的description导致后期自己或团队成员看不懂这个配置项是干什么用的。好的描述是一种文档。4. 核心注入模式二利用Runnable.bind动态绑定工具与停止词Runnable.bind()方法允许你在运行时为某个Runnable特别是LLM动态地“绑定”一些信息这些信息会成为该次调用上下文的一部分。这在实现动态工具调用和动态停止词场景中极为有用。4.1 动态绑定工具让Agent能力“随用随取”传统的LangChain Agent在初始化时就需要确定好所有可用的工具。但在很多场景下工具集是动态的。例如一个数据分析Agent只有当用户上传了Excel文件后“读取Excel”这个工具才应该被启用或者一个系统管理Agent只有拥有管理员权限的会话才能使用“重启服务”这样的高危工具。from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain.tools import Tool from langchain_core.prompts import ChatPromptTemplate # 定义一些工具 def search_order(order_id: str) - str: return f订单 {order_id} 的状态是已发货。 tool_search Tool(nameSearchOrder, funcsearch_order, description根据订单ID查询订单状态) def cancel_order(order_id: str) - str: return f订单 {order_id} 已取消。 tool_cancel Tool(nameCancelOrder, funccancel_order, description根据订单ID取消订单) # 基础LLM llm ChatOpenAI(modelgpt-3.5-turbo) # 基础提示词 prompt ChatPromptTemplate.from_messages([ (system, 你是一个订单助手。请根据用户需求使用工具。注意只有VIP用户才能取消订单。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # **关键步骤在运行时动态绑定工具** def get_agent_executor(is_vip: bool): # 所有用户都有的工具 available_tools [tool_search] # 只有VIP用户才增加取消工具 if is_vip: available_tools.append(tool_cancel) # 动态创建Agent agent create_tool_calling_agent(llmllm, toolsavailable_tools, promptprompt) executor AgentExecutor(agentagent, toolsavailable_tools, verboseTrue) return executor # 模拟普通用户请求 print( 普通用户 ) executor_normal get_agent_executor(is_vipFalse) result executor_normal.invoke({input: 帮我取消订单12345。}) # LLM只会看到SearchOrder工具因此它可能会回答“我没有取消订单的权限”。 # 模拟VIP用户请求 print(\n VIP用户 ) executor_vip get_agent_executor(is_vipTrue) result executor_vip.invoke({input: 帮我取消订单12345。}) # LLM看到了SearchOrder和CancelOrder工具因此它会尝试调用CancelOrder。虽然上面的例子是通过重新创建AgentExecutor来实现的但更优雅的方式是利用bind的思想或者结合configurable_fields来动态构建工具列表。核心逻辑是在调用链的瞬间根据运行时上下文如用户身份决定给LLM绑定哪些工具。4.2 动态绑定停止词控制生成边界停止词Stop Sequences用于告诉LLM在生成到特定字符序列时停止。动态绑定停止词可以精确控制生成内容的格式和长度。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo) # 场景生成一个列表。我们希望LLM在生成完第三项后立刻停止。 dynamic_stop [\n4., 第四项] # 告诉模型看到“\n4.”或“第四项”就停 # 使用bind动态注入停止词 bounded_llm llm.bind(stopdynamic_stop) prompt 请列出水果的三个好处 response bounded_llm.invoke(prompt) print(response.content) # 输出可能为“1. 富含维生素和矿物质... 2. 提供膳食纤维... 3. 有助于补充水分...” # 因为模型在即将生成“4.”时被强制停止了。这个技巧在需要严格输出格式如生成特定数量的JSON对象、固定行数的诗歌时非常有效。它比在提示词里写“请只列出三点”要可靠得多因为后者模型可能不遵守而stop参数是模型API层面的强制约束。5. 核心注入模式三通过RunnablePassthrough传递与加工上下文RunnablePassthrough是一个看似简单却极其强大的组件。它的核心作用是让数据“流过”而不做改变或者对数据进行简单的复制、赋值操作。它是构建复杂、分支数据流的粘合剂。5.1 基本用法传递输入最直接的用法是作为数据管道中的一个“透明”节点确保上游的输出能原封不动地传递到下游的某个输入槽位。from langchain_core.runnables import RunnablePassthrough, RunnableParallel # 假设我们有一个链需要同时使用原始问题和问题的一个变体 chain RunnableParallel({ original_question: RunnablePassthrough(), # 直接传递整个输入 rephrased_question: some_rephrasing_chain, # 另一个链处理输入生成改写后的问题 }) # 当invoke时original_question键的值就是原始的输入字典。5.2 高级用法分配与合并上下文RunnablePassthrough.assign(**kwargs)方法允许你基于当前上下文一个字典计算新的值并将其添加到上下文中。这是实现上下文增量构建的关键。假设我们有一个RAG检索增强生成流程先根据问题检索文档然后结合检索结果和原始问题来生成答案。但我们需要在最终生成前对检索到的文档做一些后处理如去重、摘要。from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 模拟一个检索器 def retriever(query: str): # 模拟返回相关文档 return [f文档A关于{query}, f文档B也提到了{query}] # 模拟一个文档后处理函数 def summarize_docs(docs): return .join(docs) 。以上是摘要 # 1. 定义子链检索并处理 retrieve_and_process RunnableParallel({ query: RunnablePassthrough(), # 传递原始问题 docs: (lambda x: retriever(x[query])) | summarize_docs, # 检索并摘要 }) # 此时retrieve_and_process的输出是 {query: “原始问题” “docs”: “摘要后的文档”} # 2. 构建最终生成链 template 基于以下背景知识回答问题。 背景{docs} 问题{query} 答案 prompt ChatPromptTemplate.from_template(template) llm ChatOpenAI(modelgpt-3.5-turbo) # 使用RunnablePassthrough.assign将上一步的输出作为prompt的输入 full_chain retrieve_and_process | RunnablePassthrough.assign( answerprompt | llm ) # 最终输出会是{query: ..., docs: ..., answer: LLM生成的答案} result full_chain.invoke({query: LangChain是什么}) print(result[answer])在这个例子中RunnablePassthrough.assign扮演了“上下文组装工”的角色。它接收retrieve_and_process输出的字典包含query和docs然后运行prompt | llm这个子链该子链会使用字典中的query和docs来填充模板最后将子链的结果以answer为键合并回原有的字典中。这样最终输出的上下文就包含了从原始问题到最终答案的全链路信息。实操心得RunnablePassthrough.assign是构建清晰数据流水线的利器。它迫使你思考每一步输入和输出的数据结构让数据流变得显式化和可调试。当链变得复杂时我习惯为每一步的输入输出字典设计好Schema哪怕只是心理上的这能极大减少因为键名错误导致的bug。6. 实战构建一个上下文感知的对话链让我们综合运用以上技巧构建一个更真实的例子一个支持多轮对话、能根据对话历史动态调整检索策略的智能助手。需求维护完整的对话历史。根据最新问题和历史动态决定是否需要检索知识库例如如果是问候语就不检索。如果需要检索则根据整个对话上下文而不仅仅是最后一个问题来优化检索查询词。将检索到的文档、对话历史、当前问题一起交给LLM生成回答。from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_openai import ChatOpenAI from langchain_community.vectorstores import Chroma # 假设使用Chroma向量库 from langchain_openai import OpenAIEmbeddings # ---------- 1. 初始化组件 ---------- llm ChatOpenAI(modelgpt-3.5-turbo) embeddings OpenAIEmbeddings() # 假设已有向量库 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # ---------- 2. 定义决策与优化链 ---------- def should_retrieve(chat_history, current_question): 决策函数是否需要检索 # 简单规则如果问题很短或像是问候则不检索 no_retrieve_keywords [你好, 嗨, 早上好, 在吗] if len(current_question.strip()) 5 or any(kw in current_question for kw in no_retrieve_keywords): return False return True def build_enhanced_query(chat_history, current_question): 优化查询词利用历史对话丰富当前问题 # 将最近几轮历史对话的“用户问题”部分拼接起来作为上下文 recent_user_turns [] for msg in chat_history[-4:]: # 取最近4轮 if isinstance(msg, HumanMessage): recent_user_turns.append(msg.content) context .join(recent_user_turns[-2:]) # 取最近2个用户问题作为上下文 enhanced_query f{context} {current_question}.strip() return enhanced_query if enhanced_query else current_question # ---------- 3. 构建主链 ---------- # 链的输入预期是一个字典{chat_history: List[BaseMessage], input: 用户当前问题} # 步骤A决策与查询优化 decision_and_query_chain RunnableLambda( lambda x: { need_retrieve: should_retrieve(x[chat_history], x[input]), query_for_retrieve: build_enhanced_query(x[chat_history], x[input]) if should_retrieve(x[chat_history], x[input]) else , original_input: x[input], chat_history: x[chat_history] } ) # 步骤B条件检索 def conditional_retrieve(info): 根据need_retrieve标志决定是否检索 if info[need_retrieve]: docs retriever.invoke(info[query_for_retrieve]) return {retrieved_docs: docs, **info} # 将检索结果合并进上下文 else: return {retrieved_docs: [], **info} retrieval_chain RunnableLambda(conditional_retrieve) # 步骤C准备LLM的输入消息 def format_messages(info): 组装最终的对话消息列表 messages [] # 系统消息可动态配置 system_msg SystemMessage(content你是一个有帮助的助手请根据对话历史和提供的资料回答问题。) messages.append(system_msg) # 历史消息 messages.extend(info[chat_history]) # 当前人类消息附加上下文 human_content info[original_input] if info[retrieved_docs]: docs_text \n.join([doc.page_content for doc in info[retrieved_docs]]) human_content f参考信息\n{docs_text}\n\n基于以上参考信息请回答{info[original_input]} messages.append(HumanMessage(contenthuman_content)) return {messages: messages} formatting_chain RunnableLambda(format_messages) # 步骤D调用LLM并解析 llm_chain RunnableLambda(lambda x: llm.invoke(x[messages])) # 步骤E更新对话历史用于下一轮 def update_history(info, ai_message): 将本轮问答加入历史 new_history info[chat_history] [ HumanMessage(contentinfo[original_input]), AIMessage(contentai_message.content) ] # 可选限制历史长度防止token超限 max_length 10 if len(new_history) max_length * 2: # 每轮包含一问一答 new_history new_history[-max_length*2:] return {answer: ai_message.content, chat_history: new_history} update_chain RunnableLambda(lambda x: update_history(x[0], x[1])) # ---------- 4. 组装完整链 ---------- full_conversation_chain ( decision_and_query_chain | retrieval_chain | formatting_chain | RunnableParallel({ # 并行将格式化后的消息传给LLM同时保留原始上下文 formatted: RunnablePassthrough(), llm_input: RunnablePassthrough() }) | RunnableLambda(lambda x: (x[llm_input], llm_chain.invoke(x[formatted]))) | update_chain ) # ---------- 5. 运行测试 ---------- chat_history [] user_inputs [你好, LangChain是什么, 它和LangGraph有什么区别] for inp in user_inputs: print(f\n用户: {inp}) result full_conversation_chain.invoke({ chat_history: chat_history, input: inp }) print(f助手: {result[answer]}) chat_history result[chat_history] # 更新历史用于下一轮这个链展示了运行时数据注入的多个层面动态决策should_retrieve函数基于当前对话上下文决定工作流分支。上下文感知的检索build_enhanced_query利用历史优化当前查询使检索更精准。条件执行conditional_retrieve根据运行时标志决定是否执行检索操作。上下文组装format_messages将原始问题、检索结果、对话历史动态组装成LLM所需的消息格式。状态维护update_history在链的末尾更新对话历史这个新的历史状态就是下一轮对话的运行时上下文。整个应用的状态对话历史在每一轮调用间流动、更新驱动着应用的行为不断演进。这才是基于LangChain构建复杂、有状态应用的真正模样。7. 避坑指南运行时数据注入的常见陷阱与最佳实践在实践中我踩过不少坑也总结出一些让运行时数据注入更稳健、更高效的方法。7.1 陷阱一上下文字典的键冲突与污染当使用RunnableParallel或多次RunnablePassthrough.assign时很容易发生键名冲突。下游的Runnable可能意外使用了上游某个临时键导致结果错误。最佳实践命名空间化为不同阶段或模块的数据加上前缀。例如检索前的原始查询叫raw_query优化后的查询叫enhanced_query检索结果叫retrieved_docs_v1。及时清理对于只在中间步骤需要的临时变量可以在下一步用RunnableLambda将其从字典中弹出pop避免传递到不必要的下游。使用Pydantic模型对于复杂的上下文定义一个PydanticBaseModel来明确数据结构。虽然Runnable主要处理字典但你可以用RunnableLambda进行验证和转换这能在开发早期发现类型和结构错误。7.2 陷阱二配置Config的传递与覆盖链RunnableConfig的传递是自动的但理解其覆盖规则很重要。当你嵌套调用多个链时内层链的config会继承外层链的config但内层链自己的config定义具有更高优先级。最佳实践显式传递关键配置对于像callbacks用于追踪这类需要贯穿整个执行过程的配置确保在顶层调用invoke时传入。谨慎使用全局配置避免在链内部写死某些配置如model_name尽量通过configurable_fields使其可从外部配置。这样测试和部署会更灵活。利用config传递业务参数除了LangChain内置字段你可以将一些业务相关的全局参数如user_id,session_id放在config.metadata中这样链中的任何组件都能访问到无需通过复杂的函数参数传递。7.3 陷阱三状态管理在并发环境下的问题上面的对话链例子在单线程、顺序请求下工作良好。但在多线程或异步Web服务器如FastAPI中直接修改和传递一个全局的或共享的chat_history列表会导致数据竞争和混乱。最佳实践无状态设计链本身应该是无状态的。所有状态如对话历史都应作为输入的一部分从外部传入并作为输出的一部分返回。由调用者如Web服务器的路由处理函数来管理状态的存储如在数据库或Redis中和传递。使用RunnableWithMessageHistory对于简单的对话历史管理LangChain提供了RunnableWithMessageHistory这个包装器。它抽象了历史记录的获取和保存接口你只需要实现一个BaseChatMessageHistory的存储后端如RedisChatMessageHistory它就能帮你自动管理历史注入。这是生产环境更推荐的做法。为状态设计唯一键无论是自己管理还是用现成组件确保每个会话或对话都有唯一的标识符如session_id并用这个键去存储和获取对应的状态。7.4 陷阱四过度设计导致的复杂性运行时数据注入能力强大但滥用会导致链的逻辑极其复杂、难以调试。一个由几十个RunnableLambda和RunnableParallel组成的链其数据流会像一团乱麻。最佳实践模块化将功能独立的子链封装成函数或单独的Runnable。例如把“查询优化与检索”封装成一个retrieval_subchain把“响应生成与格式化”封装成generation_subchain。然后在顶层用清晰的管道|或并行RunnableParallel将它们组合起来。可视化与日志充分利用LangChain的callbacks如LangChainTracer来追踪链的执行过程。看到每个步骤的输入输出是调试复杂数据流的不二法门。从简单开始不要一开始就追求完美的动态化。先实现一个能工作的静态链然后识别出哪些部分真正需要根据运行时数据变化再逐步引入configurable_fields或条件逻辑。运行时数据注入是LangChain从“玩具”走向“生产级工具”的关键阶梯。它要求开发者从编写线性脚本的思维升级到设计动态数据流系统的思维。一开始可能会觉得绕但一旦掌握你将能构建出真正灵活、强大、可适应复杂业务场景的AI应用。记住最好的设计往往是让链的每个部分都只关注一件事并通过清晰的上下文数据流将它们连接起来。
