LangChain中create_history_aware_retriever与RunnableWithMessageHistory协同及Stream报错咨询

• create_history_aware_retriever：核心作用是将用户当前问题与历史对话结合，生成适配检索器的精准查询语句，再调用检索器获取相关上下文文档，为后续QA环节提供针对性参考信息。
• RunnableWithMessageHistory：负责会话历史的全生命周期管理，根据session_id从存储中拉取对应聊天记录，将历史消息注入到目标Runnable（如rag_chain）的输入中；在Runnable执行完成后，自动将当前输入与输出追加到会话历史，保证后续对话能复用历史上下文。
• 协同流程：
  1. 用户发起请求，RunnableWithMessageHistory按session_id获取对应聊天历史。
  2. 将历史消息与当前输入一同传入rag_chain，rag_chain先调用history_aware_retriever生成检索查询。
  3. history_aware_retriever调用检索器拿到相关文档，返回给QA链。
  4. QA链结合文档与历史生成回答，RunnableWithMessageHistory更新会话历史。

错误原因 #

该问题源于LangChain框架中RunnableWithMessageHistory的流式回调逻辑缺陷：

• invoke模式下，链执行完成后返回包含answer键的完整结果，回调可正常读取该字段。
• stream模式下，链分块返回结果，单个chunk可能不包含answer键，但内部的RootListenersTracer回调仍尝试读取该字段，从而触发KeyError。

解决方法 #

方法1：适配流式输出结构 #

修改rag_chain，确保每个流式chunk都被包装为包含answer键的字典，让回调能正确识别：

from langchain_core.runnables import RunnableLambda

def wrap_stream_chunk(chunk):
    return {"answer": chunk}

# 给rag_chain添加输出包装
wrapped_rag_chain = rag_chain | RunnableLambda(wrap_stream_chunk)

# 重新创建带历史的会话链
conversational_rag_chain = RunnableWithMessageHistory(
    wrapped_rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

方法2：禁用回调追踪 #

调用stream时传入空的回调列表，跳过触发错误的回调逻辑：

for chunk in conversational_rag_chain.stream(
    {"input": question}, 
    config={"configurable": {"session_id": "abc123"}, "callbacks": []}
):
    print(chunk)

方法3：升级LangChain版本 #

该缺陷已在LangChain后续版本中修复，直接升级到最新稳定版即可解决：

pip install --upgrade langchain langchain-core

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