本文详解如何在 LangChain 构建的 RAG 系统中，无缝融合对话历史（chat history）与检索源（sources），通过 Runnable 组合、自定义 Prompt 模板和上下文注入实现可追溯、有记忆的问答 API。
本文详解如何在 langchain 构建的 rag 系统中，无缝融合对话历史（chat history）与检索源（sources），通过 `runnable` 组合、自定义 prompt 模板和上下文注入实现可追溯、有记忆的问答 api。
在 LangChain 的现代 RAG 实践中，单纯使用 ConversationRetrievalChain 已不被推荐（v0.1+ 中已弃用），官方更鼓励使用函数式、可组合的 Runnable 接口。而用户常遇到的核心矛盾是：既要保留多轮对话状态（如上一轮问题/回答），又要显式返回支撑答案的原始文档来源（source metadata）——这两者在默认链路中天然存在张力：retriever 仅作用于当前问题，而 chat_history 需参与 prompt 构建，但又不能干扰检索逻辑。
解决方案的关键在于解耦“检索”、“历史注入”与“生成”三个阶段，并通过 RunnableParallel + assign 显式控制数据流。以下是经过生产验证的结构化实现：
✅ 正确的数据流设计（推荐模式）
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, SystemMessage
# 1. 定义带历史占位符的 Prompt 模板（关键！）
SYSTEM_TEMPLATE = "你是一个专业的产品顾问。请基于以下上下文回答问题，并在结尾注明所参考的产品信息。"
HUMAN_TEMPLATE = "问题：{question}\n历史对话：{chat_history}"
chat_prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_TEMPLATE),
MessagesPlaceholder(variable_name="chat_history"), # ← 支持动态插入历史
("human", HUMAN_TEMPLATE),
])
# 2. 构建带历史感知的 RAG 链
def format_docs(docs):
return "\n\n".join([
f"- {doc.metadata.get('name', '未知')}, "
f"{doc.metadata.get('brand', 'N/A')}, "
f"{doc.metadata.get('country', 'N/A')}"
for doc in docs
])
rag_chain = (
# 并行执行：检索 + 透传问题 & 历史
RunnableParallel({
"context": retriever, # 当前问题触发的检索结果
"question": RunnablePassthrough(),
"chat_history": RunnablePassthrough(), # 历史作为独立输入传入
})
# 注入格式化后的 context 到 prompt
.assign(context=lambda x: format_docs(x["context"]))
# 组装完整 prompt 并调用模型
.assign(answer=(
chat_prompt
| model
| StrOutputParser()
))
# 最终输出 answer + 原始 context（含 source metadata）
.pick(["answer", "context"])
)
✅ FastAPI 接口集成（支持状态管理）
from typing import List, Dict, Any
from langchain_core.messages import get_buffer_string
# 模拟内存存储（生产环境应替换为 Redis / DB）
chat_histories: Dict[str, List[Dict]] = {}
@app.post("/invoke")
def invoke_rag(
question: Question,
session_id: str = "default" # 用于区分不同用户会话
):
# 获取并格式化历史消息（转为字符串或 Message 对象）
history = chat_histories.get(session_id, [])
# 转为 LangChain 兼容的 Message 列表（如需结构化处理）
# 或直接转为字符串：get_buffer_string(history)
# 执行链（传入 question + history）
result = rag_chain.invoke({
"question": question.question,
"chat_history": history
})
# 更新历史（追加用户问题 + AI 回答）
if session_id not in chat_histories:
chat_histories[session_id] = []
chat_histories[session_id].extend([
{"role": "user", "content": question.question},
{"role": "assistant", "content": result["answer"]}
])
return {
"answer": result["answer"],
"sources": [
{
"product_name": doc.metadata.get("name"),
"brand": doc.metadata.get("brand"),
"country": doc.metadata.get("country"),
"page_content_preview": doc.page_content[:100] + "..."
}
for doc in result["context"]
]
}
⚠️ 注意事项与最佳实践
- 不要滥用 ConversationRetrievalChain：它内部硬编码了 prompt 和 memory 逻辑，难以定制 source 格式与历史处理方式，且与新版 Runnable 生态不兼容。
- MessagesPlaceholder 是历史注入的核心：必须在 ChatPromptTemplate 中显式声明，否则 chat_history 不会被渲染进 prompt。
- Source 返回需保留原始 Document 对象：retriever 输出的 Document 包含完整 metadata（如 source, uri, page_number），切勿仅返回 page_content 字符串而丢失溯源能力。
- 流式响应兼容性：若需 stream()，请改用 RunnableLambda 封装分块逻辑，并确保 pick() 前不触发最终解析（避免阻塞）。
- 安全提示：FAISS.load_local(..., allow_dangerous_deserialization=True) 存在反序列化风险，生产环境务必改用 Chroma、PGVector 等安全向量库。
该方案已在多个 FastAPI + LangChain RAG 项目中稳定运行，兼顾可维护性、可调试性与可扩展性。核心思想是：让每个 Runnable 只做一件事，用 assign 显式传递中间态，用 pick 精确控制输出契约——这正是 LangChain v0.1+ 函数式范式的精髓所在。
