LangChain中RunnableLambda分支对话历史重复记录问题求助

问题根源分析 #

外层full_chain和分支内的RunnableWithMessageHistory重复实例化，导致LangSmith中对话历史被两次加载/记录；移除外层后，分支组件无法获取conversation_id、user_id等必要会话参数，因此报错。核心解决思路是只保留一个层级的RunnableWithMessageHistory，并确保会话参数能完整传递到该组件。

方案1：仅在分支层级使用RunnableWithMessageHistory，手动传递会话参数 #

直接移除外层的RunnableWithMessageHistory，通过输入参数将user_id、conversation_id传递给分支内的对话历史组件，确保仅一次加载/记录对话历史。

示例代码：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI

# 1. 定义对话历史获取函数（替换为你的实际存储逻辑，比如Redis/数据库）
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
    return InMemoryChatMessageHistory(session_id=f"{user_id}_{conversation_id}")

# 2. 定义分支内的生成链（包含对话历史）
llm = ChatOpenAI(model="gpt-3.5-turbo")
branch_llm_chain = llm | (lambda x: x.content)

branch_chain = RunnableWithMessageHistory(
    branch_llm_chain,
    get_session_history=get_session_history,
    input_messages_key="user_input",
    history_messages_key="chat_history",
    # 指定从输入中提取会话参数的key
    user_id_key="user_id",
    conversation_id_key="conversation_id"
)

# 3. 定义分支路由逻辑
def route_to_branch(inputs):
    # 这里替换为你的实际分支判断逻辑（比如根据输入类型选择不同分支）
    return branch_chain.invoke(inputs)

# 4. 外层链：仅负责传递参数，不嵌套对话历史组件
full_chain = RunnablePassthrough() | RunnableLambda(route_to_branch)

# 调用示例
full_chain.invoke({
    "user_id": "user_123",
    "conversation_id": "conv_456",
    "user_input": "今天天气怎么样？"
})

方案2：外层保留逻辑但剥离对话历史，通过参数透传传递会话配置 #

如果外层需要处理输入预处理、权限校验等逻辑，可将外层的RunnableWithMessageHistory替换为普通逻辑链，仅在分支内处理对话历史，通过RunnablePassthrough确保会话参数完整传递。

示例代码：

# 外层预处理链（比如输入清洗、权限校验）
preprocess_chain = RunnableLambda(lambda inputs: {
    **inputs,
    "user_input": inputs["raw_input"].strip()  # 示例预处理逻辑
})

# 组合外层与分支链
full_chain = preprocess_chain | RunnableLambda(route_to_branch)

# 调用方式与方案1一致
full_chain.invoke({
    "user_id": "user_123",
    "conversation_id": "conv_456",
    "raw_input": "   今天天气怎么样？   "
})

方案3：禁用其中一个实例的LangSmith跟踪（临时 workaround） #

如果因业务需求必须保留双层RunnableWithMessageHistory，可通过禁用其中一个实例的跟踪，避免重复记录。但此方法会丢失对应层级的跟踪数据，仅作为临时方案。

示例代码：

# 禁用分支内对话历史组件的LangSmith跟踪
branch_chain = RunnableWithMessageHistory(
    branch_llm_chain,
    get_session_history=get_session_history,
    input_messages_key="user_input",
    history_messages_key="chat_history",
    user_id_key="user_id",
    conversation_id_key="conversation_id",
    config={"callbacks": []}  # 清空回调，禁用跟踪
)

关键注意事项 #

• 确保所有调用full_chain的场景，都传入user_id和conversation_id参数；
• 对话历史获取函数get_session_history需根据实际存储实现，避免内存存储（InMemoryChatMessageHistory）仅用于测试；
• 优先选择方案1/2，从架构层面避免重复实例化，而非依赖跟踪禁用的临时方案。

内容的提问来源于stack exchange，提问作者Keita Kiuchi