LangChain实现RAG历史会话：构建上下文感知AI问答系统

1. 项目概述：基于LangChain的RAG应用开发

在当今AI技术快速发展的背景下，检索增强生成（Retrieval-Augmented Generation，简称RAG）已成为构建智能问答系统的核心技术之一。RAG通过结合信息检索和大型语言模型的生成能力，能够提供更准确、更有上下文的回答。而LangChain作为当前最流行的AI应用开发框架之一，为RAG系统的实现提供了强大的工具链支持。

这个项目专注于在RAG应用中添加历史会话消息功能，这是构建真正对话式AI系统的关键一步。想象一下，当你与ChatGPT进行多轮对话时，它能记住之前的交流内容，这种"记忆"能力正是通过chat_history机制实现的。在商业客服、教育辅导、技术支持等场景中，这种上下文感知能力尤为重要。

2. 核心组件解析

2.1 LangChain框架基础

LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套标准化的接口和组件，使得开发者能够轻松构建复杂的AI应用链。在RAG场景中，LangChain主要处理以下几个核心环节：

• 文档加载与处理：从各种来源（网页、PDF、数据库等）加载文档
• 文本分割：将大文档切分为适合处理的片段
• 向量化存储：将文本转换为向量并存入向量数据库
• 检索与生成：根据查询检索相关内容并生成回答

2.2 RAG架构详解

RAG系统通常由三个主要部分组成：

1. 检索器(Retriever)：负责从知识库中查找与问题相关的文档片段
2. 生成器(Generator)：基于检索到的内容和问题生成回答
3. 对话管理器：处理多轮对话的上下文和历史

传统RAG系统的一个主要局限是它们通常将每个查询视为独立的，缺乏对话上下文的理解能力。这正是本项目要解决的核心问题。

3. 历史会话消息的实现

3.1 会话感知检索器

LangChain提供了create_history_aware_retriever构造函数来创建能够理解对话上下文的检索器。这个检索器会接收两个输入：

• 用户当前的问题(input)
• 之前的对话历史(chat_history)

其工作原理是将对话历史与当前问题结合，重新构造一个独立的、包含完整上下文的问题。例如：

原始对话历史： 用户：什么是任务分解？ AI：任务分解是将复杂任务拆分为更小步骤的技术...

用户新问题：常见的方法有哪些？

重构后的问题： "任务分解的常见方法有哪些？（上下文：任务分解是将复杂任务拆分为更小步骤的技术）"

这种重构使得检索器能够找到更相关的文档片段。

3.2 实现步骤详解

3.2.1 构建基础检索器

首先需要建立一个标准的RAG检索系统：

from langchain_chroma import Chroma from langchain_community.document_loaders import WebBaseLoader from langchain_openai import OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter # 加载文档 loader = WebBaseLoader( web_paths=("https://example.com/your-knowledge-source",), bs_kwargs=dict( parse_only=bs4.SoupStrainer( class_=("content-class", "title-class") ) ), ) docs = loader.load() # 分割文档 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) splits = text_splitter.split_documents(docs) # 创建向量存储 vectorstore = Chroma.from_documents( documents=splits, embedding=OpenAIEmbeddings() ) retriever = vectorstore.as_retriever()

3.2.2 添加历史感知能力

接下来，我们创建能够理解对话历史的检索器：

from langchain.chains import create_history_aware_retriever from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder contextualize_q_system_prompt = ( "Given a chat history and the latest user question " "which might reference context in the chat history, " "formulate a standalone question which can be understood " "without the chat history. Do NOT answer the question, " "just reformulate it if needed and otherwise return it as is." ) contextualize_q_prompt = ChatPromptTemplate.from_messages( [ ("system", contextualize_q_system_prompt), MessagesPlaceholder("chat_history"), ("human", "{input}"), ] ) history_aware_retriever = create_history_aware_retriever( llm, retriever, contextualize_q_prompt )

3.2.3 构建完整问答链

将检索器与生成器结合，创建完整的问答链：

from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain system_prompt = ( "You are an assistant for question-answering tasks. " "Use the following pieces of retrieved context to answer " "the question. If you don't know the answer, say that you " "don't know. Use three sentences maximum and keep the " "answer concise.\n\n" "{context}" ) qa_prompt = ChatPromptTemplate.from_messages( [ ("system", system_prompt), MessagesPlaceholder("chat_history"), ("human", "{input}"), ] ) question_answer_chain = create_stuff_documents_chain(llm, qa_prompt) rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

4. 对话状态管理

4.1 会话历史存储

为了实现真正的多轮对话，我们需要存储和管理对话历史。LangChain提供了ChatMessageHistory类来帮助管理：

from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.chat_history import BaseChatMessageHistory store = {} def get_session_history(session_id: str) -> BaseChatMessageHistory: if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id]

4.2 集成历史管理的对话链

使用RunnableWithMessageHistory将历史管理集成到问答链中：

from langchain_core.runnables.history import RunnableWithMessageHistory conversational_rag_chain = RunnableWithMessageHistory( rag_chain, get_session_history, input_messages_key="input", history_messages_key="chat_history", output_messages_key="answer", )

4.3 使用示例

现在可以这样使用对话系统：

# 第一轮对话 result = conversational_rag_chain.invoke( {"input": "什么是任务分解？"}, config={"configurable": {"session_id": "user123"}}, ) print(result["answer"]) # 第二轮对话（会记住之前的上下文） result = conversational_rag_chain.invoke( {"input": "有哪些常见方法？"}, config={"configurable": {"session_id": "user123"}}, ) print(result["answer"])

5. 高级功能与优化

5.1 使用代理(Agent)模式

对于更复杂的场景，可以使用LangChain的代理模式，让LLM自主决定何时使用检索器：

from langchain.tools.retriever import create_retriever_tool from langgraph.prebuilt import create_react_agent # 将检索器转换为工具 tool = create_retriever_tool( retriever, "knowledge_retriever", "Searches and returns information from knowledge base.", ) tools = [tool] # 创建代理 agent_executor = create_react_agent(llm, tools) # 使用代理进行对话 for s in agent_executor.stream( {"messages": [HumanMessage(content="什么是任务分解？")]}, config={"configurable": {"thread_id": "user123"}}, ): print(s)

5.2 持久化存储方案

在生产环境中，应该使用更可靠的存储方案替代内存存储：

# 使用Redis存储对话历史 from langchain_community.chat_message_histories import RedisChatMessageHistory def get_redis_history(session_id: str) -> BaseChatMessageHistory: return RedisChatMessageHistory( session_id=session_id, url="redis://localhost:6379/0", )

5.3 性能优化技巧

1. 分块策略优化：根据文档类型调整chunk_size和chunk_overlap
2. 检索优化：使用混合检索(结合关键词和向量检索)
3. 缓存机制：对常见问题答案进行缓存
4. 异步处理：对耗时操作使用异步实现

6. 常见问题与解决方案

6.1 上下文窗口限制

问题：长对话历史可能超出模型的上下文窗口限制。

解决方案：

• 实现对话历史摘要功能
• 只保留最近N轮对话
• 使用向量检索从历史中选择最相关的部分

6.2 信息不一致

问题：模型可能生成与检索内容不一致的回答。

解决方案：

• 在提示中强制要求引用来源
• 实现一致性检查机制
• 使用RAG-Fusion等技术改进检索

6.3 会话隔离

问题：不同用户的会话需要隔离。

解决方案：

• 确保每个用户/会话有唯一的session_id
• 使用数据库或缓存系统存储历史
• 实现会话过期机制

7. 实际应用案例

7.1 技术文档助手

为开发团队构建的技术文档问答系统，能够理解开发者的问题上下文，比如：

用户：如何配置数据库连接？ （系统回答...） 用户：连接池参数有哪些？ （系统能理解这是前一个问题的延伸）

7.2 教育辅导系统

智能辅导系统可以记住学生的知识水平和使用习惯，提供个性化的学习建议：

学生：请解释一下牛顿第一定律 （系统回答...） 学生：能举个例子吗？ （系统能提供与力学相关的例子）

7.3 商业客服机器人

电商客服机器人能够记住用户的订单信息和之前的咨询内容：

用户：我上周买的手机什么时候到货？ （系统查询订单...） 用户：能改送到公司地址吗？ （系统理解这是关于同一订单的请求）

8. 部署与扩展

8.1 部署方案

1. 本地开发：使用FastAPI或Flask构建简单的API服务
2. 云服务部署：AWS Lambda、Google Cloud Functions等无服务方案
3. 容器化：使用Docker打包，部署到Kubernetes集群

8.2 监控与日志

• 使用LangSmith跟踪链的执行情况
• 记录用户查询和系统响应用于后续分析
• 实现反馈机制收集用户满意度

8.3 扩展方向

1. 多模态支持：处理图像、表格等非文本内容
2. 个性化适配：基于用户画像调整回答风格
3. 实时学习：允许用户纠正错误并更新知识库

9. 性能评估与调优

9.1 评估指标

1. 检索相关性：命中文档与问题的匹配程度
2. 回答质量：准确性、完整性和流畅性
3. 响应时间：从提问到获得回答的延迟
4. 上下文保持：在多轮对话中保持主题一致性的能力

9.2 A/B测试策略

• 对比有无历史会话功能的用户体验
• 测试不同检索策略的效果
• 评估不同LLM模型的性能差异

9.3 持续改进流程

1. 收集真实用户交互数据
2. 识别常见问题和失败模式
3. 调整提示词和检索策略
4. 部署更新并监控改进效果

10. 安全与隐私考虑

10.1 数据安全

• 对话历史加密存储
• 实现数据访问控制
• 定期清理过期数据

10.2 隐私保护

• 匿名化处理用户数据
• 提供数据删除接口
• 遵守相关数据保护法规

10.3 内容过滤

• 实现输入输出过滤机制
• 检测并阻止不当内容
• 记录审核日志

在实际开发中，我发现历史会话管理是RAG系统中最容易被低估的复杂部分。一个常见的陷阱是过度依赖对话历史，导致系统在长对话中性能下降。最佳实践是动态调整历史上下文的长度和相关性，而不是简单地保留所有历史消息。另一个实用技巧是在检索前对历史消息进行轻量级摘要，既能保留关键信息，又能节省上下文窗口的空间。