概述

在构建生产级 AI Agent 的过程中,基础链式调用往往难以满足复杂业务需求。本文深入探讨 LangChain 中链的高级应用,涵盖函数集成、动态配置、记忆管理、智能路由与容错机制。所有示例均基于模拟模型,无需 API Key,开箱即用。

纲要

  • 链中的函数集成
    • @chain 装饰器:将普通函数转换为 Runnable
    • RunnableLambda:在链中嵌入 lambda 函数
    • 自定义流式函数:基于生成器实现流式输出处理
  • 值透传:RunnablePassthrough 的使用场景
  • 运行时动态配置
    • 动态调整模型参数:configurable_fields 详解
    • 动态切换提示词模板
  • 链的记忆机制
    • 短时记忆:InMemoryChatMessageHistoryRunnableWithMessageHistory
    • 长期记忆:基于 Redis 实现持久化
  • 智能路由链:基于 LLM 分类的请求分发
  • 容错与回退:.with_fallbacks() 实现高可用
  • 完整 Demo:整合全部高级特性的可运行代码

链中的函数集成

在实际业务中,我们经常需要在 LCEL 管道中嵌入自定义逻辑。LangChain 提供了多种将 Python 函数集成到链中的方式。

使用 @chain 装饰器快速生成链

@chain 装饰器可将任意 Python 函数转换为 Runnable 对象,使其能够无缝融入 LCEL 管道。被装饰的函数可以直接调用 .invoke().stream() 等方法。

from langchain_core.runnables import chain
from langchain_community.chat_models.fake import FakeListChatModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = FakeListChatModel(responses=["This is a joke about dogs."])
prompt = ChatPromptTemplate.from_template("Tell a joke about {topic}")

@chain
def custom_chain(topic: str) -> str:
    # 函数内部自由组合 LCEL 组件
    chain = prompt | model | StrOutputParser()
    return chain.invoke({"topic": topic})

# custom_chain 现在是一个 Runnable 对象
print(custom_chain.invoke("dogs"))

使用 RunnableLambda 嵌入 lambda 函数

对于简单的逻辑,RunnableLambda 可将 lambda 函数直接嵌入链中。

from langchain_core.runnables import RunnableLambda

# 在链末端添加一个计算长度的步骤
length_func = RunnableLambda(lambda x: len(x))
chain = (
    ChatPromptTemplate.from_template("Just say: {text}")
    | model
    | StrOutputParser()
    | length_func
)
print(chain.invoke({"text": "Hello"}))  # 输出数字 5

自定义支持流式输出的函数

若链的末端需要处理流式输出并保持流式特性,必须使用生成器(yield),而非 returnreturn 会等待全部数据收集完毕后才返回,会阻塞流式传输。

from typing import Iterator

def stream_splitter(input_stream: Iterator[str]) -> Iterator[str]:
    buffer = ""
    for chunk in input_stream:
        buffer += chunk
        while "," in buffer:
            idx = buffer.index(",")
            yield buffer[:idx+1]
            buffer = buffer[idx+1:]
    if buffer:
        yield buffer

# 模拟流式输入
mock_stream = iter(["Cat,", "Dog", ",Bird"])
for item in stream_splitter(mock_stream):
    print(item)
# 输出:
# Cat,
# Dog,
# Bird

将该生成器函数封装为 RunnableLambda 后,即可在链中使用,并兼容流式调用。

值透传:RunnablePassthrough

RunnablePassthrough 是一个特殊组件,它不修改输入,直接将输入原封不动地传递给下游。它最典型的应用场景是与 RunnableParallel 配合,在并行分支中保留原始输入。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

parallel = RunnableParallel(
    unchanged=RunnablePassthrough(),
    doubled=RunnableLambda(lambda x: x * 2)
)
print(parallel.invoke(5))  # 输出: {'unchanged': 5, 'doubled': 10}

在更复杂的链中,RunnablePassthrough 常用于将原始用户输入传递给多个处理分支,最终将各分支结果合并输出。

运行时动态配置

LangChain 提供了 configurable_fields 机制,允许在运行时动态调整链的配置参数,如模型温度、提示词模板等。

动态调节模型参数

通过 .configurable_fields() 为模型标记可配置字段,然后在调用时通过 .with_config() 覆盖参数值。

from langchain_openai import ChatOpenAI  # 真实场景使用
from langchain_community.chat_models.fake import FakeListChatModel

# 使用 FakeListChatModel 模拟,temperature 参数在此模拟器中无实际效果
# 此处仅展示配置接口的使用方式
model = FakeListChatModel(responses=["42"])
configurable_model = model.configurable_fields(
    temperature={"default": 0.7}  # 声明 temperature 为可配置字段
)
# 运行时覆盖 temperature 值
result = configurable_model.with_config(
    configurable={"temperature": 0.9}
).invoke("")
print(result.content)

注意FakeListChatModel 并不真正支持 temperature 参数,上述示例仅演示 API 用法。在生产环境中,此机制对 ChatOpenAIChatAnthropic 等真实模型有效。

动态切换提示词模板

利用 configurable_fields 也可在运行时切换不同的提示词模板。

from langchain_core.prompts import PromptTemplate

prompt_a = PromptTemplate.from_template("Hello {name}")
prompt_b = PromptTemplate.from_template("Hi {name} from template B")

config_prompt = prompt_a.configurable_fields(
    template={"default": prompt_a.template}  # 将 template 标记为可配置
)
# 实际切换需要传入完整的 PromptTemplate 对象,此处为概念演示
new_prompt = config_prompt.with_config(
    configurable={"prompt": prompt_b}
)
print(new_prompt.invoke({"name": "Alice"}).text)

链的记忆机制

记忆(Memory)是构建对话式 AI Agent 的核心能力。LangChain 通过 RunnableWithMessageHistory 将历史消息管理无缝集成到链中。

短时记忆:InMemoryChatMessageHistory

InMemoryChatMessageHistory 将对话历史存储在内存中,适用于单次会话场景。通过 RunnableWithMessageHistory 包装链,即可自动携带历史消息进行多轮对话。

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate

# 会话存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("placeholder", "{history}"),
    ("human", "{input}")
])

model = FakeListChatModel(responses=["I remember you said: 'Hello'."])
chain = prompt | model | StrOutputParser()

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 第一轮对话
response1 = chain_with_history.invoke(
    {"input": "Hello, my name is Alice"},
    config={"configurable": {"session_id": "user123"}}
)
print(response1)  # I remember you said: 'Hello'.

# 第二轮对话,自动携带历史上下文
response2 = chain_with_history.invoke(
    {"input": "What is my name?"},
    config={"configurable": {"session_id": "user123"}}
)
print(response2)  # 模型可基于历史回答 "Alice"

长期记忆:使用 Redis 持久化

当需要跨会话、跨重启持久保存历史记录时,可使用 RedisChatMessageHistory。它基于 Redis 存储,确保数据持久化。

# 需要先安装 redis 和 langchain-community
# pip install redis langchain-community

from langchain_community.chat_message_histories import RedisChatMessageHistory

# 确保本地 Redis 服务已启动(默认端口 6379)
history = RedisChatMessageHistory(
    session_id="user_001",
    url="redis://localhost:6379"
)
history.clear()  # 清空旧数据
history.add_user_message("Hi")
history.add_ai_message("Hello! How can I help?")
print(history.messages)  # 重启后仍可获取

RedisChatMessageHistory 的用法与 InMemoryChatMessageHistory 完全一致,可直接替换,实现零成本升级到长期记忆。

智能路由链

路由链(Routing Chain)根据用户输入的内容动态选择不同的处理链路。其核心是通过 LLM 对输入进行分类,然后使用 RunnableLambdaRunnableBranch 将请求分发到对应的专业链。

from langchain_core.runnables import RunnableLambda

# 1. 分类链:判断问题类型
classify_prompt = ChatPromptTemplate.from_template(
    "Classify the following question into 'math' or 'general'. Only return one word.\nQuestion: {question}"
)
classify_model = FakeListChatModel(responses=["math"])  # 模拟分类结果
classify_chain = classify_prompt | classify_model | StrOutputParser()

# 2. 专业处理链
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()
)

# 3. 路由函数:根据分类结果选择链
def route(info: dict):
    if "math" in info["topic"]:
        return math_chain
    else:
        return general_chain

# 4. 构建路由链
routing_chain = (
    {"topic": classify_chain, "question": lambda x: x["question"]}
    | RunnableLambda(route)
)

print(routing_chain.invoke({"question": "What is 2+2?"}))
# 输出: 2 + 2 = 4

实际生产环境中,分类模型可用更精细的 Prompt 和真实 LLM 替代,以实现更准确的分类。

math

general

用户输入

分类链 LLM 分类

分类结果

数学专业链

通用问答链

输出结果

容错与回退

在生产环境中,LLM API 可能因速率限制、网络波动、服务不可用等原因失败。LangChain 提供 .with_fallbacks() 方法,允许为链设置备用模型,当主模型失败时自动切换。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

primary_model = FakeListChatModel(responses=["Primary response"])
# 注意:FakeListChatModel 不会抛出异常,此处仅演示 API 用法
# 真实场景中,若 primary_model 因 API 限流抛出异常,会自动切换到 fallback
fallback_model = FakeListChatModel(responses=["Fallback response"])

chain = (
    ChatPromptTemplate.from_template("Say something")
    | primary_model
    | StrOutputParser()
)

# 设置回退链
chain_with_fallback = chain.with_fallbacks([fallback_model])

print(chain_with_fallback.invoke({}))
# 若 primary 失败,输出: Fallback response

.with_fallbacks() 可接受多个备用模型,按顺序尝试,直到成功为止。该机制极大提升了系统的健壮性。

API 速览

本节梳理本文涉及的核心 API,供快速查阅。

@chain 装饰器

属性 说明
所属库 langchain_core.runnables
用法 将普通函数转换为 Runnable 对象
返回值 Runnable 对象,支持 .invoke().stream() 等方法

RunnableLambda

属性 说明
所属库 langchain_core.runnables
构造方法 RunnableLambda(func)
参数 func 可调用对象(函数或 lambda
返回值 Runnable 对象

RunnablePassthrough

属性 说明
所属库 langchain_core.runnables
构造方法 RunnablePassthrough()
行为 透传输入,不做任何修改

RunnableWithMessageHistory

属性 说明
所属库 langchain_core.runnables.history
构造参数 runnableget_session_historyinput_messages_keyhistory_messages_key
返回值 支持多轮对话的 Runnable 对象

RedisChatMessageHistory

属性 说明
所属库 langchain_community.chat_message_histories
构造参数 session_idurl(Redis 连接地址)
核心方法 add_user_message()add_ai_message()messages(属性)

完整 Demo 示例

以下代码整合了本文介绍的全部高级技巧,均使用模拟模型,无需任何 API Key 即可运行。

运行说明

  1. 确保 Python 环境已安装依赖:
    pip install langchain langchain-core langchain-community redis
    
  2. 如需测试 Redis 长期记忆,请确保本地 Redis 服务已启动(redis-server)。
  3. 直接运行 Python 文件即可。

代码示例

# complete_demo.py
from langchain_core.runnables import (
    chain, RunnableLambda, RunnablePassthrough,
    RunnableParallel
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_community.chat_models.fake import FakeListChatModel
from typing import Iterator

# ---------- 1. @chain 装饰器 ----------
model = FakeListChatModel(responses=["A joke about cats."])

@chain
def quick_chain(topic: str) -> str:
    prompt = ChatPromptTemplate.from_template("Tell a joke about {topic}")
    return (prompt | model | StrOutputParser()).invoke({"topic": topic})

print("@chain 输出:", quick_chain.invoke("cats"))

# ---------- 2. 流式处理函数 ----------
def aggregate_commas(stream: Iterator[str]) -> Iterator[str]:
    buf = ""
    for chunk in stream:
        buf += chunk
        while "," in buf:
            idx = buf.index(",")
            yield buf[:idx+1]
            buf = buf[idx+1:]
    if buf:
        yield buf

mock_stream = iter(["A,B", ",C"])
print("流式分割:", list(aggregate_commas(mock_stream)))

# ---------- 3. RunnablePassthrough ----------
res = RunnableParallel(
    orig=RunnablePassthrough(),
    mod=RunnableLambda(lambda x: x * 2)
).invoke(10)
print("Passthrough 示例:", res)

# ---------- 4. 动态配置(模拟) ----------
base_model = FakeListChatModel(responses=["42"])
# 实际场景:真实模型可配置 temperature
print("动态配置示例(模拟):", base_model.invoke("").content)

# ---------- 5. 记忆示例 ----------
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("placeholder", "{history}"),
    ("human", "{input}")
])
memory_model = FakeListChatModel(responses=["Yes, your name is Alice."])
memory_chain = prompt | memory_model | StrOutputParser()

with_memory = RunnableWithMessageHistory(
    memory_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="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_model = FakeListChatModel(responses=["math", "general"])
classify_prompt = ChatPromptTemplate.from_template(
    "Classify the question into 'math' or 'general'. Only one word.\nQuestion: {question}"
)
classify_chain = classify_prompt | classify_model | StrOutputParser()

math_chain = (
    ChatPromptTemplate.from_template("Math answer: {question}")
    | FakeListChatModel(responses=["2+2=4"])
    | StrOutputParser()
)
general_chain = (
    ChatPromptTemplate.from_template("General answer: {question}")
    | FakeListChatModel(responses=["General reply"])
    | StrOutputParser()
)

def route(info):
    return math_chain if "math" in info["topic"] else general_chain

routing_chain = (
    {"topic": classify_chain, "question": lambda x: x["question"]}
    | RunnableLambda(route)
)
print("路由结果:", routing_chain.invoke({"question": "What is 2+2?"}))

# ---------- 7. 回退机制 ----------
primary = FakeListChatModel(responses=["Primary"])
fallback = FakeListChatModel(responses=["Fallback"])
fallback_chain = (
    ChatPromptTemplate.from_template("say")
    | primary
    | StrOutputParser()
).with_fallbacks([fallback])
print("回退输出:", fallback_chain.invoke({}))

技术点总结

此 Demo 演示了以下核心技术:

  • 使用 @chain 将普通函数转为 Runnable
  • 使用生成器函数处理流式输出
  • 使用 RunnablePassthrough 实现值透传
  • 使用 RunnableWithMessageHistoryInMemoryChatMessageHistory 实现短时记忆
  • 使用 LLM 分类实现智能路由
  • 使用 .with_fallbacks() 实现容错回退

参考文档

官方文档

参考链接

总结

本文系统梳理了 LangChain 链的高级应用技巧。函数集成方面,通过 @chainRunnableLambda 可灵活嵌入自定义逻辑,流式函数需基于生成器实现。

动态配置允许在运行时调整模型参数和提示词模板,提升了链的灵活性。记忆机制通过 RunnableWithMessageHistory 统一管理,InMemoryChatMessageHistory 适用于短时会话,RedisChatMessageHistory 支持跨会话持久化。

智能路由链结合 LLM 分类实现请求分发,而 .with_fallbacks() 则提供了关键的容错能力,保障系统高可用。这些技术共同构成了构建生产级 AI Agent 的基础设施。

更多推荐