内容纲要在链中使用函数chain装饰器将普通函数转换为RunnableRunnableLambda与lambda函数在链中的集成自定义支持流式输出的函数yield实现生成器值的透传RunnablePassthrough运行时动态配置动态调整模型温度configurable_fields动态切换提示词模板为链增加记忆能力短时记忆InMemoryHistory实现多轮对话长期记忆使用 Redis 持久化聊天记录自定义路由链基于 LLM 分类的智能分发回退机制with_fallbacks在主模型失败时切换备用模型完整可运行代码使用模拟模型无需 API Key引言在掌握了 LCELLangChain Expression Language的基本链式调用后面对复杂的业务场景往往需要在链中嵌入自定义函数、添加记忆、实现动态路由以及错误容错。本文将深入讲解这些高级技巧从函数集成、记忆机制到路由与回退每个知识点都配有可运行的代码示例帮助你将链的开发能力提升到生产级水平。在链中使用函数使用chain装饰器快速生成链通过chain装饰器可以将任意 Python 函数转换为Runnable对象无缝融入 LCEL 管道。fromlangchain_core.runnablesimportchainfromlangchain_community.chat_models.fakeimportFakeListChatModelfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelFakeListChatModel(responses[This is a joke about dogs.])promptChatPromptTemplate.from_template(Tell a joke about {topic})chaindefcustom_chain(topic:str)-str:# 函数内部可以自由组合组件chainprompt|model|StrOutputParser()returnchain.invoke({topic:topic})# 现在 custom_chain 就是一个 Runnableprint(custom_chain.invoke(dogs))使用RunnableLambda嵌入 lambda 函数fromlangchain_core.runnablesimportRunnableLambda# 定义一个计算长度的函数length_funcRunnableLambda(lambdax:len(x))chain(ChatPromptTemplate.from_template(Just say: {text})|model|StrOutputParser()|length_func)print(chain.invoke({text:Hello}))# 输出数字自定义支持流式输出的函数若要在链的末端添加一个处理逻辑并保持流式输出必须用yield实现生成器避免使用return会阻塞直到全部完成。fromtypingimportIteratordefstream_splitter(input_stream:Iterator[str])-Iterator[str]:bufferforchunkininput_stream:bufferchunkwhile,inbuffer:idxbuffer.index(,)yieldbuffer[:idx1]bufferbuffer[idx1:]ifbuffer:yieldbuffer# 模拟流式输入mock_streamiter([Cat,,Dog,,Bird])foriteminstream_splitter(mock_stream):print(item)实际在链中使用时可将该生成器函数封装为RunnableLambda并正确设置afunc等以兼容流式调用。值的透传RunnablePassthrough当需要将原始输入原封不动地传递给下游时使用RunnablePassthrough常见于与并行分支配合。fromlangchain_core.runnablesimportRunnableParallel,RunnablePassthrough parallelRunnableParallel(unchangedRunnablePassthrough(),doubledRunnableLambda(lambdax:x*2))print(parallel.invoke(5))# {unchanged: 5, doubled: 10}运行时动态配置利用configurable_fields可在运行时调整模型的参数或切换提示词。动态调节温度fromlangchain_openaiimportChatOpenAI# 假设使用真实模型此处用 FakeListChatModel 模拟modelFakeListChatModel(responses[42])# 为模型添加可配置字段config_modelmodel.configurable_fields(temperaturelambda:None# 实际可设 temperature)# 运行时覆盖resultconfig_model.with_config(configurable{temperature:0.9}).invoke()print(result.content)动态切换提示词fromlangchain_core.promptsimportPromptTemplate prompt_aPromptTemplate.from_template(Hello {name})prompt_bPromptTemplate.from_template(Hi {name} from template B)config_promptprompt_a.configurable_fields()new_promptconfig_prompt.with_config(configurable{prompt:prompt_b})print(new_prompt.invoke({name:Alice}).text)为链增加记忆能力短时记忆InMemoryHistory使用InMemoryHistory存储会话中的对话历史配合RunnableWithMessageHistory实现多轮对话。fromlangchain_core.chat_historyimportInMemoryChatMessageHistoryfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.messagesimportHumanMessage store{}defget_session_history(session_id:str):ifsession_idnotinstore:store[session_id]InMemoryChatMessageHistory()returnstore[session_id]promptChatPromptTemplate.from_messages([(system,You are a helpful assistant.),(placeholder,{history}),(human,{input})])modelFakeListChatModel(responses[I remember you said: Hello.])chainprompt|model|StrOutputParser()chain_with_historyRunnableWithMessageHistory(chain,get_session_history,input_messages_keyinput,history_messages_keyhistory,)# 第一轮对话response1chain_with_history.invoke({input:Hello, my name is Alice},config{configurable:{session_id:user123}})print(response1)# 第二轮对话自动携带历史response2chain_with_history.invoke({input:What is my name?},config{configurable:{session_id:user123}})print(response2)长期记忆使用 Redis 持久化安装redis和langchain-community后可用RedisChatMessageHistory持久保存历史记录重启后依然存在。# 请确保本地 Redis 服务已启动fromlangchain_community.chat_message_historiesimportRedisChatMessageHistory historyRedisChatMessageHistory(session_iduser_001,urlredis://localhost:6379)history.clear()history.add_user_message(Hi)history.add_ai_message(Hello! How can I help?)print(history.messages)# 重启后仍可获取自定义路由链借助 LLM 对用户输入进行分类然后根据分类结果路由到不同的专业链。fromlangchain_core.runnablesimportRunnableBranch# 分类链返回 math 或 generalclassify_promptChatPromptTemplate.from_template(Classify the following question into math or general. Only return one word.\nQuestion: {question})classify_chainclassify_prompt|FakeListChatModel(responses[math])|StrOutputParser()# 专业链math_chain(ChatPromptTemplate.from_template(Answer the math question: {question})|FakeListChatModel(responses[2 2 4])|StrOutputParser())general_chain(ChatPromptTemplate.from_template(Answer the general question: {question})|FakeListChatModel(responses[This is a general answer.])|StrOutputParser())# 路由函数defroute(info):ifmathininfo[topic]:returnmath_chainelse:returngeneral_chain full_chain(classify_chain|(lambdatopic:{topic:topic,question:lambdad:d[question]})# 简化处理|RunnableLambda(route))# 实际调用print(full_chain.invoke({question:What is 22?}))回退机制使用.with_fallbacks()设置备用模型当主模型因速率限制等原因失败时自动切换到备选模型。primary_modelFakeListChatModel(responses[Primary response])# 模拟主模型失败可以让它抛出异常这里用模拟方式跳过fallback_modelFakeListChatModel(responses[Fallback response])# 构建链设置回退实际主模型失败时会自动切换chain(ChatPromptTemplate.from_template(Say something)|primary_model|StrOutputParser())chain_with_fallbackchain.with_fallbacks([fallback_model])# 此处因 FakeListChatModel 不会抛异常真实场景如 API 限制会触发print(chain_with_fallback.invoke({}))完整可运行代码整合示例以下代码整合了上述关键技巧全部使用模拟模型无需任何 API Key 即可运行。# 安装依赖pip install langchain langchain-core langchain-communityfromlangchain_core.runnablesimport(chain,RunnableLambda,RunnablePassthrough,RunnableParallel,RunnableBranch)fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistoryfromlangchain_core.messagesimportHumanMessage,AIMessagefromlangchain_community.chat_models.fakeimportFakeListChatModelfromtypingimportIterator# ---------- 1. chain 装饰器 ----------modelFakeListChatModel(responses[A joke about cats.])chaindefquick_chain(topic:str)-str:promptChatPromptTemplate.from_template(Tell a joke about {topic})return(prompt|model|StrOutputParser()).invoke({topic:topic})print(chain 输出:,quick_chain.invoke(cats))# ---------- 2. 流式处理函数 ----------defaggregate_commas(stream:Iterator[str])-Iterator[str]:bufforchunkinstream:bufchunkwhile,inbuf:idxbuf.index(,)yieldbuf[:idx1]bufbuf[idx1:]ifbuf:yieldbuf mock_streamiter([A,B,,C])print(流式分割:,list(aggregate_commas(mock_stream)))# ---------- 3. RunnablePassthrough ----------resRunnableParallel(origRunnablePassthrough(),modRunnableLambda(lambdax:x*2)).invoke(10)print(Passthrough:,res)# ---------- 4. 动态配置模拟 ----------base_modelFakeListChatModel(responses[42])configurable_modelbase_model.configurable_fields()modified_modelconfigurable_model.with_config(configurable{temperature:0.9})print(动态温度:,modified_model.invoke().content)# ---------- 5. 记忆示例 ----------store{}defget_session_history(session_id:str):ifsession_idnotinstore:store[session_id]InMemoryChatMessageHistory()returnstore[session_id]promptChatPromptTemplate.from_messages([(system,You are a helpful assistant.),(placeholder,{history}),(human,{input})])memory_modelFakeListChatModel(responses[Yes, your name is Alice.])memory_chainprompt|memory_model|StrOutputParser()with_memoryRunnableWithMessageHistory(memory_chain,get_session_history,input,history)print(第一轮:,with_memory.invoke({input:My name is Alice},config{configurable:{session_id:1}}))print(第二轮:,with_memory.invoke({input:What is my name?},config{configurable:{session_id:1}}))# ---------- 6. 路由链 ----------classify_responses[math,general]classify_modelFakeListChatModel(responsesclassify_responses)classify_promptChatPromptTemplate.from_template(Classify the question into math or general. Only one word.\nQuestion: {question})classify_chainclassify_prompt|classify_model|StrOutputParser()math_chain(ChatPromptTemplate.from_template(Math answer: {question})|FakeListChatModel(responses[224])|StrOutputParser())general_chain(ChatPromptTemplate.from_template(General answer: {question})|FakeListChatModel(responses[General reply])|StrOutputParser())defroute(info):returnmath_chainifmathininfo[topic]elsegeneral_chain# 由于分类模型返回次序问题这里简化演示直接使用分类结果routing_chain({topic:classify_chain,question:lambdax:x[question]}|RunnableLambda(route))print(路由结果:,routing_chain.invoke({question:What is 22?}))# ---------- 7. 回退机制 ----------primaryFakeListChatModel(responses[Primary])fallbackFakeListChatModel(responses[Fallback])fallback_chain(ChatPromptTemplate.from_template(say)|primary|StrOutputParser()).with_fallbacks([fallback])print(回退输出:,fallback_chain.invoke({}))总结通过本文的实战示例我们掌握了在 LangChain 链中嵌入函数、配置动态参数、赋予链记忆能力、构建智能路由以及设置容错回退等高级技巧。这些能力是构建健壮、灵活且具备生产级品质的 AI Agent 应用的核心基础。