视频网站采集规则,泰安市住房和城乡建设局网站,如意宝魔方建站,网站友情链接要加什么用文章目录 本文介绍向量和向量数据库向量向量数据库 索引开始动手实现rag加载文档数据并建立索引将向量存放到向量数据库中检索生成构成一条链 本文介绍
从本节开始#xff0c;有了上一节的langchain基础学习#xff0c;接下来使用langchain实现一个rag应用#xff0c;并稍微… 文章目录 本文介绍向量和向量数据库向量向量数据库 索引开始动手实现rag加载文档数据并建立索引将向量存放到向量数据库中检索生成构成一条链 本文介绍
从本节开始有了上一节的langchain基础学习接下来使用langchain实现一个rag应用并稍微深入的讲解一下流程
向量和向量数据库
向量
在rag中不得不提向量和向量数据库在ai检索中多数采用特征在文本中多数采用向量来表示文本。采用向量后“语义”的匹配程度就转换成了向量之间的相似程度。计算向量相似度的算法有很多比如余弦相似度、内积、欧氏距离等等。
有了向量当用户提出问题时处理过程就变成了将问题转换为向量然后计算向量之间的距离找到与问题向量最接近的文档向量从而实现“语义”的匹配。
OpenAI 提供了一个专门负责将文本转换成向量的 API——Embeddings。我们可以根据需要选择自己部署模型或是选择别人提供的服务。不同的 Embedding 模型之间的差异主要取决于训练样本比如有的模型会在中文处理上表现得比较好。
向量数据库
在 RAG 系统中我们要把数据存放到哪里呢我们需要一个数据库只不过我们需要的既不是 Oracle、MySQL 这样的关系数据库也不是 MongoDB、Redis 这样的 NoSQL 数据库。因为我们后续处理的都是向量所以我们需要的是向量数据库。
向量数据库与传统数据库有很大的差别在使用方式上传统数据库搜索信息倾向于精确匹配而向量数据库的匹配则是语义上的接近。
在实现上二者也存在不小的差别比如由于向量本身通常维度会很多如果按照传统数据库的方式直接进行存储将会带来很多问题。向量数据库需要把向量数据作为一个完整的单元处理底层存储结构也需要根据这个特点进行规划。另外向量数据格式也相对单一每个维度的数据往往都是固定的数据格式浮点数、二进制整数等。
索引
下面是一个常见的索引过程 在这个过程里面我们会先对信息源做一次信息提取。信息源可能是各种文档比如 Word 文档、PDF 文件Web 页面甚至是一些图片。从这些信息源中我们把内容提取出来也就是其中的文本。
接下来我们会把这些文本进行拆分将其拆分成更小的文本块。之所以要拆分主要是原始的文本可能会比较大这并不利于检索还有一点重要原因是我们前面说过要把检索到的信息拼装到提示词里过大的文本可能会造成提示词超过模型有限的上下文窗口。
再来就是把文本块转换成向量也就是得到 Embedding 的过程。前面我们说过这个过程往往是通过训练好的模型来完成的。到这里我们就把信息转换成了向量。最后一步就是把得到的向量存储到向量数据库中供后续的检索使用。
至此我们对常见的 RAG 流程已经有了基本了解。但实际上RAG 领域正处于一个快速发展的过程中有很多相关技术也在不断地涌现
虽然采用向量搜索对于语义理解很有帮助但一些人名、缩写、特定 ID 之类的信息却是传统搜索的强项有人提出混合搜索的概念将二者结合起来通过各种搜索方式我们会得到很多的候选内容但到底哪个与我们的问题更相关有人引入了重排序Rerank模型以此决定候选内容与查询问题的相关程度除了在已有方向的努力甚至还有人提出了 RAG 的新方向。我们前面讨论的流程前提条件是把原始信息转换成了向量但这本质上还是基于文本的更适合回答一些事实性问题。它无法理解更复杂的关系比如我的朋友里谁在 AI 领域里工作。所以有人提出了基于知识图谱的 RAG知识图谱是一种结构化的语义知识库特别适合找出信息之间的关联。
开始动手实现rag
加载文档数据并建立索引
建索引使用TextLoader从txt中加载数据
from langchain_community.document_loaders import TextLoaderloader TextLoader(introduction.txt)
docs loader.load()text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200)
splits text_splitter.split_documents(docs)
vectorstore Chroma(collection_nameai_learning,embedding_functionOpenAIEmbeddings(),persist_directoryvectordb
)
vectorstore.add_documents(splits)这里的 TextLoader 属于 DocumentLoader。在 LangChain 中有一个很重要的概念叫文档Document它包括文档的内容page_content以及相关的元数据metadata。所有原始信息都是文档索引信息的第一步就是把这些文档加载进来这就是 DocumentLoader 的作用。
除了这里用到的 TextLoaderLangChain 社区里已经实现了大量的 DocumentLoader比如从数据库里加载数据的 SQLDatabaseLoader从亚马逊 S3 加载文件的 S3FileLoader。基本上大部分我们需要的文档加载器都可以找到直接的实现。
拆分加载进来的文档是 TextSplitter 的主要职责。
text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200)
splits text_splitter.split_documents(docs)虽然都是文本但怎样拆分还是有讲究的拆分源代码和拆分普通文本处理方法就是不一样的。LangChain 社区里同样实现了大量的 TextSplitter我们可以根据自己的业务特点进行选择。我们这里使用了 RecursiveCharacterTextSplitter它会根据常见的分隔符比如换行符递归地分割文档直到把每个块拆分成适当的大小。
将向量存放到向量数据库中
做好基础的准备之后就要把拆分的文档存放到向量数据库里了
vectorstore Chroma(collection_nameai_learning,embedding_functionOpenAIEmbeddings(),persist_directoryvectordb
)
vectorstore.add_documents(splits)LangChain 支持了很多的向量数据库它们都有一个统一的接口VectorStore在这个接口中包含了向量数据库的统一操作比如添加、查询之类的。这个接口屏蔽了向量数据库的差异在向量数据库并不为所有程序员熟知的情况下给尝试不同的向量数据库留下了空间。各个具体实现负责实现这些接口我们这里采用的实现是 Chroma。
在 Chroma 初始化的过程中我们指定了 Embedding 函数它负责把文本变成向量。这里我们采用了 OpenAI 的 Embeddings 实现你完全可以根据自己的需要选择相应的实现LangChain 社区同样提供了大量的实现比如你可以指定 Hugging Face 这个模型社区中的特定模型来做 Embedding。
到这里我们就完成了索引的过程看上去还是比较简单的。为了验证我们索引的结果我们可以调用 similarity_search 检索向量数据库的数据
vectorstore Chroma(collection_nameai_learning,embedding_functionOpenAIEmbeddings(),persist_directoryvectordb
)
documents vectorstore.similarity_search(专栏的作者是谁)
print(documents)这里用的 similarity_search 表示的是根据相似度进行搜索还可以使用 max_marginal_relevance_search它会采用 MMRMaximal Marginal Relevance最大边际相关性算法。这个算法可以在保持结果相关性的同时尽量选择与已选结果不相似的内容以增加结果的多样性。
检索生成
现在我们已经为我们 RAG 应用准备好了数据。接下来就该正式地构建我们的 RAG 应用了。我在之前的聊天机器上做了一些修改让它能够支持 RAG代码如下
from operator import itemgetter
from typing import List
import tiktoken
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage, SystemMessage, trim_messages
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import OpenAIEmbeddings
from langchain_openai.chat_models import ChatOpenAI
from langchain_chroma import Chromavectorstore Chroma(collection_nameai_learning,embedding_functionOpenAIEmbeddings(),persist_directoryvectordb
)retriever vectorstore.as_retriever(search_typesimilarity)# Token 计算
def str_token_counter(text: str) - int:enc tiktoken.get_encoding(o200k_base)return len(enc.encode(text))def tiktoken_counter(messages: List[BaseMessage]) - int:num_tokens 3tokens_per_message 3tokens_per_name 1for msg in messages:if isinstance(msg, HumanMessage):role userelif isinstance(msg, AIMessage):role assistantelif isinstance(msg, ToolMessage):role toolelif isinstance(msg, SystemMessage):role systemelse:raise ValueError(fUnsupported messages type {msg.__class__})# 确保 msg.content 不是 Nonecontent msg.content or num_tokens (tokens_per_message str_token_counter(role) str_token_counter(content))if msg.name:num_tokens tokens_per_name str_token_counter(msg.name)return num_tokens# 限制 token 长度
trimmer trim_messages(max_tokens4096,strategylast,token_countertiktoken_counter,include_systemTrue,
)store {}def get_session_history(session_id: str) - BaseChatMessageHistory:if session_id not in store:store[session_id] InMemoryChatMessageHistory()return store[session_id]model ChatOpenAI()prompt ChatPromptTemplate.from_messages([(system,You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you dont know the answer, just say that you dont know. Use three sentences maximum and keep the answer concise.Context: {context},),MessagesPlaceholder(variable_namehistory),(human, {question}),]
)def format_docs(docs):return \n\n.join(doc.page_content for doc in docs)context itemgetter(question) | retriever | format_docs
# first_step的内容有可能是这样的
# {
# question: 什么是LangChain,
# context: LangChain 是一个用于构建基于 LLM 的应用的框架。\n它支持多种数据检索方式。
# }
first_step RunnablePassthrough.assign(contextcontext)from langchain_core.prompt_values import ChatPromptValue
def format_prompt_output(input):确保 prompt 生成的内容是 List[BaseMessage]# print(\n[DEBUG] Prompt 输出类型:, type(input))# print([DEBUG] Prompt 输出内容:, input)# 如果 input 是 ChatPromptValue提取 .messagesif isinstance(input, ChatPromptValue):return input.messages # 直接返回 List[BaseMessage]# 如果 input 已经是 List[BaseMessage]直接返回if isinstance(input, list) and all(isinstance(msg, BaseMessage) for msg in input):return input# 其他情况报错raise TypeError(f❌ format_prompt_output 传入了不正确的格式: {type(input)})chain first_step | prompt | format_prompt_output | trimmer | modelwith_message_history RunnableWithMessageHistory(chain,get_session_historyget_session_history,input_messages_keyquestion,history_messages_keyhistory,
)config {configurable: {session_id: dreamhead}}while True:user_input input(You: )if user_input.lower() exit:breakif user_input.strip() :continue# 修正 stream 传参stream with_message_history.stream({question: user_input}, # 直接传入字符串configconfig)# print(prompt)for chunk in stream:print(chunk.content, end, flushTrue)print()示例代码运行如下 为了进行检索我们需要指定数据源这里就是我们的向量数据库其中存放着我们前面已经索引过的数据
vectorstore Chroma(collection_nameai_learning,embedding_functionOpenAIEmbeddings(),persist_directoryvectordb
)retriever vectorstore.as_retriever(search_typesimilarity)为什么不直接使用向量数据库呢因为 Retriever 并不只有向量数据库一种实现比如WikipediaRetriever 可以从 Wikipedia 上进行搜索。所以一个 Retriever 接口就把具体的实现隔离开来。
回到向量数据库上当我们调用 as_retriever 创建 Retriever 时还传入了搜索类型search_type这里的搜索类型和前面讲到向量数据库的检索方式是一致的这里我们传入的是 similarity当然也可以传入 mmr。
文档检索出来并不能直接就和我们的问题拼装到一起。这时就轮到提示词登场了。下面是我们在代码里用到的提示词
You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you dont know the answer, just say that you dont know. Use three sentences maximum and keep the answer concise.
Context: {context}在这段提示词里我们告诉大模型根据提供的上下文回答问题不知道就说不知道。这是一个提示词模板在提示词的最后是我们给出的上下文Context。这里上下文是根据问题检索出来的内容。
有了这个提示词再加上聊天历史和我们的问题就构成了一个完整的提示词模板
prompt ChatPromptTemplate.from_messages([(system,You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you dont know the answer, just say that you dont know. Use three sentences maximum and keep the answer concise.Context: {context},),MessagesPlaceholder(variable_namehistory),(human, {question}),]
)构成一条链
接下来就是把各个组件组装到一起构成一条完整的链
context itemgetter(question) | retriever | format_docs
first_step RunnablePassthrough.assign(contextcontext)
chain first_step | prompt | trimmer | modelwith_message_history RunnableWithMessageHistory(chain,get_session_historyget_session_history,input_messages_keyquestion,history_messages_keyhistory,
)在这段代码里我们首先构建了一个 context 变量它也一条链。第一步是从传入参数中获取到 question 属性也就是我们的问题然后把它传给 retriever。retriever 会根据问题去做检索对应到我们这里的实现就是到向量数据库中检索检索的结果是一个文档列表。
文档是 LangChain 应用内部的表示要传给大模型我们需要把它转成文本这就是 format_docs 做的事情它主要是把文档内容取出来拼接到一起
def format_docs(docs):return \n\n.join(doc.page_content for doc in docs)这里补充几句实现细节。在 LangChain 代码里 | 运算符被用作不同组件之间的连接其实现的关键就是大部分组件都实现了 Runnable 接口在这个接口里实现了 or 和 ror。or 表示这个对象出现在| 左边时的处理相应的 ror 表示这个对象出现在右边时的处理。 Python 在处理 a | b 这个表达式时它会先尝试找 a 的 or如果找不到它会尝试找 b 的 ror。所以在 context 的处理中 来自标准库的 itemgetter 虽然没有实现__or__但 retriever 因为实现了 Runnable 接口所以它也实现了 ror。所以这段代码才能组装出我们所需的链。 有了 context 变量我们可以用它构建了另一个变量 first_step
first_step RunnablePassthrough.assign(contextcontext)RunnablePassthrough.assign 这个函数就是在不改变链当前状态值的前提下添加新的状态值。前面我们说了这里赋给 context 变量的值是一个链我们可以把它理解成一个函数它会在运行期执行其参数就是我们当前的状态值。现在你可以理解 itemgetter(“question”) 的参数是从哪来的了。这个函数的返回值会用来在当前的状态里添加一个叫 context 的变量以便在后续使用。
