Langchain构建RAG对话应用

本文：关注检索器与上下文的子链、父链；即检索器也需要上下文内容。

RAG是一种增强LLM知识的方法，通过引入额外的数据来实现。

实现思路：加载—》分割—》存储—》检索—》生成。

初始化

import os
import bs4
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.chains.retrieval import create_retrieval_chain
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_openai import ChatOpenAI, OpenAIEmbeddingsos.environ['http_proxy'] = '127.0.0.1:7890'
os.environ['https_proxy'] = '127.0.0.1:7890'os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "LangchainDemo"
os.environ["LANGCHAIN_API_KEY"] = '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'
# os.environ["TAVILY_API_KEY"] = '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'# 聊天机器人案例
# 创建模型
model = ChatOpenAI(model='gpt-4-turbo')

加载数据

通过angchain_community..document_loaders import WebBaseLoader加载博客内容数据

返回一个Document列表，每个Document包含web元数据（metadata）、内容（page_content）；

## angchain_community.包含大量工具。

#WebBaseLoader 相当于一个爬虫，可以爬取多个网页。

# Beautiful Soup的SoupStrainer用于解析HTML文档时仅提取特定（class_）的部分
# 1、加载数据: 一篇博客内容数据
loader = WebBaseLoader(web_paths=['https://lilianweng.github.io/posts/2023-06-23-agent/'],bs_kwargs=dict(parse_only=bs4.SoupStrainer(class_=('post-header', 'post-title', 'post-content')))
)docs = loader.load()# print(len(docs))
# print(docs)

切割

可以看到，上述Document的内容非常大；因此要进行切割。

# from langchain_text_splitters import RecursiveCharacterTextSplitter

## chunk_size：分割块大小；chunk_overlap：允许重复字符（保证语句完整性）

# splitter包含多种切割，还有split_text：切割string、bytes
# 2、大文本的切割
# text = "hello world, how about you? thanks, I am fine.  the machine learning class. So what I wanna do today is just spend a little time going over the logistics of the class, and then we'll start to talk a bit about machine learning"
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)splits = splitter.split_documents(docs)
test示例

#可以看到第一块，和第二块末尾都有how。第一块 how因补充chunk_size；第二块how因语序完整性及chunk_overlap而存在。

存储与检索

# 根据切割结果，创建向量数据库

# 2、存储
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())

# 检索器

# 3、检索器
retriever = vectorstore.as_retriever()

创建Prompt和chain

# 创PromptTemplate：系统定位、聊天历史、用户输入。

# 创 chain：create_stuff_documents_chain(model, prompt) ：创建多文本的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{context}
"""
prompt = ChatPromptTemplate.from_messages(  # 提问和回答的 历史记录  模板[("system", system_prompt),MessagesPlaceholder("chat_history"),  #("human", "{input}"),]
)chain1 = create_stuff_documents_chain(model, prompt)# 下述测试，因为没有 chat_history,因此会报错。
### 将检索器和已有chain链接
# chain2 = create_retrieval_chain(retriever, chain1)# resp = chain2.invoke({'input': "What is Task Decomposition?"})
### resp包含多个key，只取answer。
# print(resp['answer'])

子链-历史记录

'''注意：
一般情况下，我们构建的链（chain）直接使用输入问答记录来关联上下文。

但在此案例中，查询检索器也需要对话上下文才能被理解。

#解决办法：
添加一个子链(chain)，它采用最新用户问题和聊天历史，并在它引用历史信息中的任何信息时重新表述问题。这可以被简单地认为是构建一个新的“历史感知”检索器。
这个子链的目的：让检索过程融入了对话的上下文。

#eg：HumanMessage：它的方法有哪些；这里的‘ 它 ’即需要结合上下文，才能理解。”

# 子链提示词模板：重新定义子链AI的定位（根据用户新问题和上下文，分析并返回最新真实问题）。

##create_history_aware_retriever（模型、检索器、子链Prompt）：用于创建一种能够感知对话历史的检索器（Retriever）。它的核心作用是让检索过程动态结合之前的对话上下文，从而使当前查询的检索结果更精准、更相关。

##create_retrieval_chain(history_chain, chain1)：整合俩个链。
# 创建一个子链
# 子链的提示模板
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."""retriever_history_temp = ChatPromptTemplate.from_messages([('system', contextualize_q_system_prompt),MessagesPlaceholder('chat_history'),("human", "{input}"),]
)# 创建一个子链
history_chain = create_history_aware_retriever(model, retriever, retriever_history_temp)# 保持问答的历史记录
store = {}def get_session_history(session_id: str):if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]# 创建父链chain: 把前两个链整合
chain = create_retrieval_chain(history_chain, chain1)# 创建携带history的Runnable对象
result_chain = RunnableWithMessageHistory(chain,get_session_history,input_messages_key='input',history_messages_key='chat_history',output_messages_key='answer'
)
1

多轮会话

# 第一轮对话
resp1 = result_chain.invoke({'input': 'What is Task Decomposition?'},config={'configurable': {'session_id': 'zs123456'}}
)print(resp1['answer'])# 第二轮对话
resp2 = result_chain.invoke({'input': 'What are common ways of doing it?'},config={'configurable': {'session_id': 'ls123456'}}
)print(resp2['answer'])