实现了一个带上下文记忆的 RAG（检索增强生成）问答系统，核心能力是：
1.从指定网页加载 Agent 相关知识并构建向量数据库；
2.基于用户问题从向量库检索相关上下文；
3.结合聊天历史理解用户问题（比如处理指代 / 省略）；
4.生成简洁、基于上下文的回答，并保留会话历史。
from langchain_chroma import Chroma from langchain_classic.chains.combine_documents import create_stuff_documents_chain from langchain_classic.chains.history_aware_retriever import create_history_aware_retriever from langchain_classic.chains.retrieval import create_retrieval_chain from langchain_community.document_loaders import WebBaseLoader from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.embeddings import Embeddings from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables import RunnableWithMessageHistory from langchain_text_splitters import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer import bs4 from langchain_demo.my_llm import llm #1.构建向量数据库 class CustomQwen3Embeddings(Embeddings): ''' 定义一个Qwen3的Embedding和lang'chain整合的类 ''' def __init__(self,model_name): self.qwen3_embedding = SentenceTransformer(model_name) #输入的问题向量化 def embed_query(self, text: str) -> list[float]: return self.embed_documents([text])[0] #文本内容向量化 def embed_documents(self, texts: list[str]) -> list[list[float]]: return self.qwen3_embedding.encode(texts) qwen3=CustomQwen3Embeddings('Qwen/Qwen3-Embedding-0.6B') vector_store=Chroma( collection_name='t_agent_blog', embedding_function=qwen3, persist_directory='../chroma.db' ) #2.把网络的关于Agent的博客数据写入向量数据库 def create_dense_db(): loader=WebBaseLoader( web_path=('https://lilianweng.github.io/posts/2023-06-23-agent/',), bs_kwargs=dict( parse_only=bs4.SoupStrainer(class_=("post-content","post-title","post-header")) ) ) docs_list=loader.load() #切割 #初始化文本切割器，设置块大小1000，重叠200 text_splitter=RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200) #分割文档 splits=text_splitter.split_documents(docs_list) print('doc的数量为：',len(splits)) ids=['id'+str(i+1) for i in range(len(splits))] vector_store.add_documents(documents=splits,ids=ids) #create_dense_db()#首次运行需打开 #3.问题上下文化 #系统提示词 contextualize_q_system_prompt=( "给定聊天历史和最新的用户问题(可能引用聊天历史中的上下文)，" "将起重新表述为一个独立的问题(不需要聊天历史也能理解)。" "不要回答问题，只需在需要时重新表述问题，否则保持原样。" ) contextualize_q_prompt=ChatPromptTemplate.from_messages( [ ("system",contextualize_q_system_prompt),#系统角色提示 MessagesPlaceholder("chat_history"),#聊天历史占位符 ("human","{input}"),#用户输入占位符 ] ) #创建一个向量数据库的检索器 retriever=vector_store.as_retriever() #创建一个上下文感知的检索器 history_aware_retriever=create_history_aware_retriever( llm,retriever,contextualize_q_prompt ) #4.RAG的代码 #系统提示词：定义助手的行为和回答规范 system_prompt=( "你是一个问答任务助手。" "使用以下检索到的上下文来回答问题。" "如果不知道答案，就说你不知道。" "回答最多三句话，保持简洁。" "\n\n" "{context}"#从向量数据库中检索处的doc ) #创建问答提示模板 qa_prompt=ChatPromptTemplate.from_messages( [ ("system",system_prompt), MessagesPlaceholder("chat_history"), ("human","{input}"), ] ) question_chain=create_stuff_documents_chain(llm,qa_prompt) RAG_chain=create_retrieval_chain(history_aware_retriever,question_chain) #5.保存历史消息 store={}#用来保存历史消息，key:会话id session_id def get_session_history(session_id:str): """从内存中的历史消息列表中，返回当前会话的所有历史消息""" if session_id not in store: store[session_id] = InMemoryChatMessageHistory() 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',#输出消息的键 ) resp1=conversational_rag_chain.invoke( {"input":"What is Task Decomposition?"}, config={ "configurable":{"session_id":"abc123"} } ) print(resp1['answer']) resp2=conversational_rag_chain.invoke( {"input":"What are common ways of doing it"}, config={ "configurable":{"session_id":"abc123"} } ) print(resp2['answer'])输出：
