基于 LangChain 的 RAG 与对话记忆实现

前言

在大模型应用中，RAG（检索增强生成）已成为企业级知识库和智能文档问答的标配。它能让大模型读取私有文件（txt、docx、pdf），并基于文件内容精准回答问题。

对话记忆则能让 AI 记住历史对话，实现连贯的多轮问答。

本文搭建一套完整智能问答系统，支持本地私有文档读取、RAG 知识库检索及对话历史记忆。

核心功能

读取本地文档（TXT、DOCX 等）
RAG 检索：从文档中精准找答案
对话记忆：记住之前说过的话
多轮问答：上下文连贯
兼容在线大模型（通义千问/Qwen）

# ===================== 配置 ===================== MODEL = "qwen3-max#######" BASE_URL = "https://aiapi.#######.com/qwen/v1" API_KEY = "##########" DOCX_FILE = "test.txt" # 你的文档（我是小明，今年 25 岁，喜欢打篮球和编程。我最喜欢的编程语言是 Python。） # ==================================================== from langchain_openai import ChatOpenAI from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings import FakeEmbeddings from langchain_community.vectorstores import DocArrayInMemorySearch from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.chains import create_history_aware_retriever, create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain import warnings from langchain_community.document_loaders import TextLoader from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory warnings.filterwarnings('ignore') # ===================== 加载文档 ===================== print("正在加载文档...") loader = TextLoader(DOCX_FILE, encoding="utf-8") documents = loader.load() # 文档切分 text_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100) texts = text_splitter.split_documents(documents) # 构建轻量级向量库 embeddings = FakeEmbeddings(size=100) db = DocArrayInMemorySearch.from_documents(texts, embeddings) retriever = db.as_retriever(search_kwargs={"k": 2}) # ===================== 模型初始化 ===================== llm = ChatOpenAI( model=MODEL, base_url=BASE_URL, api_key=API_KEY, temperature=0.1, ) # ===================== 对话记忆模板 ===================== # 历史对话理解 prompt1 = ChatPromptTemplate.from_messages([ ("system", "根据对话历史理解用户问题，不要忽略用户刚说过的内容"), MessagesPlaceholder("chat_history"), ("human", "{input}"), ]) retriever_chain = create_history_aware_retriever(llm, retriever, prompt1) # 优先遵守用户指令 + 参考文档 prompt2 = ChatPromptTemplate.from_messages([ ("system", "请遵守用户最新指令！如果用户修改信息，优先使用最新内容，文档仅做参考。文档内容：\n\n{context}"), MessagesPlaceholder("chat_history"), ("human", "{input}"), ]) doc_chain = create_stuff_documents_chain(llm, prompt2) rag_chain = create_retrieval_chain(retriever_chain, doc_chain) # ===================== 记忆存储 ===================== store = {} def get_history(session_id): if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # 最终带记忆的 RAG 链 chain = RunnableWithMessageHistory( rag_chain, get_history, input_messages_key="input", history_messages_key="chat_history", output_messages_key="answer", ) # ===================== 启动对话 ===================== print("\n✅ 带记忆的文档问答已启动！输入 exit 退出") print("=" * 60) while True: q = input("\n你：") if q.lower() in ["exit", "q"]: print("AI：结束！") break res = chain.invoke({"input": q}, config={"configurable": {"session_id": "test1"}}) print("\nAI：", res["answer"])

基于 LangChain 的 RAG 与对话记忆实现