摘要
在简单的RAG应用中，该RAG应用并不具备联系历史消息的能力，因此，本文将在上文的基础上实现带有历史消息的RAG应用，所用组件以上文一致。带有历史消息的RAG与上文中构建的RAG存在不同。
在上文中，在获取到用户query后，检索器根据用户的query获取到上下文context，并将query和context填充到Prompt中，给到大模型给出回答。
在带有chat_history的RAG中，在获取到用户查询query和用户历史消息chat_history后，会根据两者使用大模型生成一个新的search_query（检索查询），根据search_query查询上下文context，之后根据用户query和context填充Prompt，最后生成模型回答。
对比发现，主要区别在于多了一个chat_history、一个search_query生成的过程。具体如下。本文还将补充内置链相关内容。
RAG组件初始化
from langchain_openai import ChatOpenAIfrom langchain_community.embeddings import ZhipuAIEmbeddingsfrom langchain_chroma import Chromafrom dotenv import load_dotenvimport osload_dotenv()api_key = os.getenv("api_key")base_url = os.getenv("base_url")zhipu_api_key = os.getenv("ZHIPU_API_KEY")zhipu_model_name = os.getenv("ZHIPU_EMBEDDING_MODEL")model = ChatOpenAI(model = "deepseek-chat",api_key=api_key, base_url=base_url)embeddings = ZhipuAIEmbeddings(api_key=zhipu_api_key, model=zhipu_model_name)vector_store = Chroma(embedding_function=embeddings,persist_directory="./chroma_langchain_rag")retriever = vector_store.as_retriever()基于内置链构建带有历史消息的RAG应用
基于内置链实现简单的RAG应用
代码如下：
from langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParsersystem = ( "你是一个基于深度学习的数据融合方法研究者，你能根据问题和上下文，给出相关的研究成果。" "根据以下相关内容给出问题的答案。" "如果你不知道，你就说不知道。" "保持对话的连贯性和简洁。" "同时要注意，你不能提供任何与问题无关的信息，并需要反思你所提供的信息适合符合逻辑、是否合理。" "\n\n" "{context}")prompt = ChatPromptTemplate.from_messages([ ("system",system), ("human","{input}"),])# 将prompt和llm结合起来，使用create_stuff_documents_chain构建文档填充链question_answer_chain = create_stuff_documents_chain( llm, prompt)# 将检索器与文档填充链结合起来，构建检索链rag_chain_with_inner_chain = create_retrieval_chain( retriever, question_answer_chain)"""rag_chain = ( {"context": retriever | format_doc, "input": RunnablePassthrough()} | prompt | model | StrOutputParser())"""代码解释：
内置链（Built-in chain）是LangChain对LCEL的封装，实现一些特定功能。
create_stuff_documents_chain是将上一步骤查询出的Document列表进行格式化，并将格式化后的内容填充到prompt的context中，最后通过llm输出。
create_stuff_documents_chain源码如下：
def create_stuff_documents_chain( llm: LanguageModelLike, #大语言模型 prompt: BasePromptTemplate, # prompt *, output_parser: Optional[BaseOutputParser] = None, # 默认为StrOutputParser() document_prompt: Optional[BasePromptTemplate] = None, document_separator: str = DEFAULT_DOCUMENT_SEPARATOR, # 默认为'/n/n' document_variable_name: str = DOCUMENTS_KEY, # 默认为'context') -> Runnable[Dict[str, Any], Any]: _validate_prompt(prompt, document_variable_name) ## 验证context是否在prompt中 _document_prompt = document_prompt or DEFAULT_DOCUMENT_PROMPT #默认为{page_content} _output_parser = output_parser or StrOutputParser() # 格式化文档 def format_docs(inputs: dict) -> str: return document_separator.join( format_document(doc, _document_prompt) for doc in inputs[document_variable_name] ) return ( RunnablePassthrough.assign(**{document_variable_name: format_docs}).with_config( run_name="format_inputs" ) | prompt | llm | _output_parser ).with_config(run_name="stuff_documents_chain")create_retrieval_chain是将接收用户查询，然后传递给检索器以获取相关文档。随后，将这些文档（以及原始输入）传递给question_answer_chain以生成响应。
create_retrieval_chain源码如下：
def create_retrieval_chain( retriever: Union[BaseRetriever, Runnable[dict, RetrieverOutput]], #可以是一个检索器或一个输出的检索结果 combine_docs_chain: Runnable[Dict[str, Any], str],# 可执行的链) -> Runnable: if not isinstance(retriever, BaseRetriever): retrieval_docs: Runnable[dict, RetrieverOutput] = retriever else: retrieval_docs = (lambda x: x["input"]) | retriever # 如果是可执行的检索器，则将输入中的'input'给到检索器进行检索。 retrieval_chain = ( RunnablePassthrough.assign( context=retrieval_docs.with_config(run_name="retrieve_documents"), ).assign(answer=combine_docs_chain) ).with_config(run_name="retrieval_chain") return retrieval_chain基于内置链实现历史消息的RAG应用
源码如下：
from langchain.chains.combine_documents import create_stuff_documents_chainfrom langchain.chains.retrieval import create_retrieval_chainfrom langchain.chains.history_aware_retriever import create_history_aware_retrieverfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.prompts import MessagesPlaceholderfrom langchain_core.chat_history import BaseChatMessageHistoryfrom langchain_core.runnables import RunnableWithMessageHistory, RunnablePassthroughfrom langchain_community.chat_message_histories import ChatMessageHistory# 创建上下文链的prompt，用于将当前消息基于历史消息进行重构contextualize_system_prompt = ("给定一个聊天历史和最新的用户问题，该最新的用户问题可能需要引用聊天历史中的上下文，制定一个独立的问题，可以在没有聊天历史的情况下理解。不要回答任何问题，只要在需要时重新表述最新问题，否则就原样返回。")contextualize_prompt = ChatPromptTemplate( [ ("system", contextualize_system_prompt), MessagesPlaceholder("chat_history"), ("human", "{input}"), ])# 构建问答链的prompt，用于回答问题question_answer_system_prompt = ("你是一个基于深度学习的数据融合方法研究者，你能根据问题和上下文，给出相关的研究成果。" "根据以下相关内容给出问题的答案。" "如果你不知道，你就说不知道。" "保持对话的连贯性和简洁。" "同时要注意，你不能提供任何与问题无关的信息，并需要反思你所提供的信息适合符合逻辑、是否合理。" "\n\n" "{context}")question_answer_prompt = ChatPromptTemplate( [ ("system", question_answer_system_prompt), MessagesPlaceholder("chat_history"), ("human", "{input}"), ])# 创建上下文链contextualize_chain = create_history_aware_retriever(model, retriever, contextualize_prompt) # 创建上下文链# 创建文本融合链（问答链）combine_docs_chain = create_stuff_documents_chain(model, question_answer_prompt) # 创建文本融合链# 创建RAG链rag_chain = create_retrieval_chain(contextualize_chain, combine_docs_chain) # 创建rag链"""def format_doc(docs): return "\n\n".join([f"{doc.page_content}" for doc in docs])from langchain_core.output_parsers import StrOutputParserrag_chain_with_lcel = ( {"context":contextualize_prompt | model | StrOutputParser() | retriever | format_doc, "input":RunnablePassthrough()} | question_answer_prompt | model | StrOutputParser())"""# 实现历史消息自动管理store = {} # 创建一个空字典，用于存储历史消息def get_session_history(session_id:str) -> BaseChatMessageHistory: # 创建一个函数，用于获取历史消息 if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id]conversational_rag_chain = RunnableWithMessageHistory( rag_chain, get_session_history, input_messages_key="input", history_messages_key="chat_history", output_messages_key="answer")conversational_rag_chain.invoke( {"input": "什么是数据融合"}, config={ "configurable": {"session_id": "abc123"} }, )["answer"]代码解释：
contextualize_prompt在将query重构至search_query时，需要调用大模型，该Prompt用于此次调用。
question_answer_prompt在根据context、chat_history、input生成回答时，调用该Prompt。
create_history_aware_retriever内置链，该链接将收集对话历史记录，然后将其用于生成传递给底层检索器的搜索查询。其源码如下：
def create_history_aware_retriever( llm: LanguageModelLike, retriever: RetrieverLike, prompt: BasePromptTemplate,) -> RetrieverOutputLike: if "input" not in prompt.input_variables: #chain调用必须使用‘input’ raise ValueError( "Expected `input` to be a prompt variable, " f"but got {prompt.input_variables}" ) retrieve_documents: RetrieverOutputLike = RunnableBranch( ( # Both empty string and empty list evaluate to False lambda x: not x.get("chat_history", False), # If no chat history, then we just pass input to retriever (lambda x: x["input"]) | retriever, ), # If chat history, then we pass inputs to LLM chain, then to retriever prompt | llm | StrOutputParser() | retriever, # 默认执行链 ).with_config(run_name="chat_retriever_chain") return retrieve_documentsRunnableWithMessageHistory BaseChatMessageHistory：BaseChatMessageHistory保存聊天历史的对象，能够被RunnableWithMessageHistory调用，根据session_id区分；get_session_history方法用于获取历史消息，被RunnableWithMessageHistory包装起来。
基于Agent实现历史消息管理的RAG应用
将查询上下文包装成为一个工具，供Agent调用，再一次对话中，Agent可能多次调用。此种方式时LangChain较为推荐的。
代码如下：
from langgraph.checkpoint.memory import MemorySaverfrom langgraph.prebuilt import create_react_agentfrom langchain.tools.retriever import create_retriever_toolfrom langchain_core.messages import HumanMessagememory_saver = MemorySaver()# 创建工具retriever_tool = create_retriever_tool( retriever, "retriever", "Retrieve relevant documents for a given question")tools = [retriever_tool]# 创建Agentagent = create_react_agent( model, tools, checkpointer=memory_saver)config = {"configurable": {"thread_id": "abc123"}}for s in agent.stream( {"messages": [HumanMessage(content="我叫AfroNick")]}, config=config): print(s) print("----")目前，未涉及LangGraph学习，待后续补充。
原文地址：https://www.cnblogs.com/AfroNicky/p/18909198
