本文介紹

從本節開始，有了上一節的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_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
vectorstore = Chroma(collection_name="ai_learning",embedding_function=OpenAIEmbeddings(),persist_directory="vectordb"
)
vectorstore.add_documents(splits)

這里的 TextLoader 屬于 DocumentLoader。在 LangChain 中，有一個很重要的概念叫文檔（Document），它包括文檔的內容（page_content）以及相關的元數據（metadata）。所有原始信息都是文檔，索引信息的第一步就是把這些文檔加載進來，這就是 DocumentLoader 的作用。

除了這里用到的 TextLoader，LangChain 社區里已經實現了大量的 DocumentLoader，比如，從數據庫里加載數據的 SQLDatabaseLoader，從亞馬遜 S3 加載文件的 S3FileLoader。基本上，大部分我們需要的文檔加載器都可以找到直接的實現。

拆分加載進來的文檔是 TextSplitter 的主要職責。

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)

雖然都是文本，但怎樣拆分還是有講究的，拆分源代碼和拆分普通文本，處理方法就是不一樣的。LangChain 社區里同樣實現了大量的 TextSplitter，我們可以根據自己的業務特點進行選擇。我們這里使用了 RecursiveCharacterTextSplitter，它會根據常見的分隔符（比如換行符）遞歸地分割文檔，直到把每個塊拆分成適當的大小。

將向量存放到向量數據庫中

做好基礎的準備之后，就要把拆分的文檔存放到向量數據庫里了：

vectorstore = Chroma(collection_name="ai_learning",embedding_function=OpenAIEmbeddings(),persist_directory="vectordb"
)
vectorstore.add_documents(splits)

LangChain 支持了很多的向量數據庫，它們都有一個統一的接口：VectorStore，在這個接口中包含了向量數據庫的統一操作，比如添加、查詢之類的。這個接口屏蔽了向量數據庫的差異，在向量數據庫并不為所有程序員熟知的情況下，給嘗試不同的向量數據庫留下了空間。各個具體實現負責實現這些接口，我們這里采用的實現是 Chroma。

在 Chroma 初始化的過程中，我們指定了 Embedding 函數，它負責把文本變成向量。這里我們采用了 OpenAI 的 Embeddings 實現，你完全可以根據自己的需要選擇相應的實現，LangChain 社區同樣提供了大量的實現，比如，你可以指定 Hugging Face 這個模型社區中的特定模型來做 Embedding。

到這里，我們就完成了索引的過程，看上去還是比較簡單的。為了驗證我們索引的結果，我們可以調用 similarity_search 檢索向量數據庫的數據：

vectorstore = Chroma(collection_name="ai_learning",embedding_function=OpenAIEmbeddings(),persist_directory="vectordb"
)
documents = vectorstore.similarity_search("專欄的作者是誰？")
print(documents)

這里用的 similarity_search 表示的是根據相似度進行搜索，還可以使用 max_marginal_relevance_search，它會采用 MMR（Maximal 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_name="ai_learning",embedding_function=OpenAIEmbeddings(),persist_directory="vectordb"
)retriever = vectorstore.as_retriever(search_type="similarity")# 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 = "user"elif isinstance(msg, AIMessage):role = "assistant"elif isinstance(msg, ToolMessage):role = "tool"elif isinstance(msg, SystemMessage):role = "system"else:raise ValueError(f"Unsupported 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_tokens=4096,strategy="last",token_counter=tiktoken_counter,include_system=True,
)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 don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.Context: {context}""",),MessagesPlaceholder(variable_name="history"),("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(context=context)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`，提取 `.messages`if 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_history=get_session_history,input_messages_key="question",history_messages_key="history",
)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},  # 直接傳入字符串config=config)# print(prompt)for chunk in stream:print(chunk.content, end='', flush=True)print()

示例代碼運行如下：

為了進行檢索，我們需要指定數據源，這里就是我們的向量數據庫，其中存放著我們前面已經索引過的數據：

vectorstore = Chroma(collection_name="ai_learning",embedding_function=OpenAIEmbeddings(),persist_directory="vectordb"
)retriever = vectorstore.as_retriever(search_type="similarity")

為什么不直接使用向量數據庫呢？因為 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 don't know the answer, just say that you don't 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 don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.Context: {context}""",),MessagesPlaceholder(variable_name="history"),("human", "{question}"),]
)

構成一條鏈

接下來，就是把各個組件組裝到一起，構成一條完整的鏈：

context = itemgetter("question") | retriever | format_docs
first_step = RunnablePassthrough.assign(context=context)
chain = first_step | prompt | trimmer | modelwith_message_history = RunnableWithMessageHistory(chain,get_session_history=get_session_history,input_messages_key="question",history_messages_key="history",
)

在這段代碼里，我們首先構建了一個 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(context=context)

RunnablePassthrough.assign 這個函數就是在不改變鏈當前狀態值的前提下，添加新的狀態值。前面我們說了，這里賦給 context 變量的值是一個鏈，我們可以把它理解成一個函數，它會在運行期執行，其參數就是我們當前的狀態值。現在你可以理解 itemgetter(“question”) 的參數是從哪來的了。這個函數的返回值會用來在當前的狀態里添加一個叫 context 的變量，以便在后續使用。