你可以這樣理解：切分就像把一長串香腸切成小段，每段長度 chunk_size，但是切的時候刀口稍微退回去一點（chunk_overlap），這樣相鄰兩段會有一部分重合，保證接頭處的信息不丟失。

05 第三步：向量嵌入——給文字“貼坐標”

計算機不認識文字，但認識數字。嵌入模型的作用就是把一段文字轉換成一串數字（向量），這串數字代表了文字的“語義”。

想象一下：你把所有文檔塊放在一個巨大的多維空間里，意思相近的塊會靠得很近。比如“蘋果手機”和“iPhone”這兩個詞的向量距離很近；而“蘋果手機”和“香蕉”的距離很遠。

5.1 選擇一個好用的中文嵌入模型

目前國產的 BAAI（北京智源研究院）推出的 BGE 系列模型，對中文的支持非常好，而且是開源的。

向量維度：可以理解為坐標的“維度”。三維空間能區(qū)分前后左右上下，維度越高，能表達的語義細節(jié)越豐富，但計算量也越大。

5.2 用代碼生成向量

# pip install sentence-transformers langchain_huggingface
from langchain_huggingface import HuggingFaceEmbeddings
 
# 加載模型（第一次運行會自動下載，大概 400MB）
# 如果你已經下載到本地，可以用 os.path.expanduser("~/models/bge-base-zh-v1.5")
embed_model = HuggingFaceEmbeddings(
? ? model_name="BAAI/bge-base-zh-v1.5" ? # 也可以換成其他模型名
)
 
# 單個文本嵌入（用于用戶的問題）
question = "年假有多少天？"
question_vector = embed_model.embed_query(question)
print(f"問題向量長度：{len(question_vector)}") ?# 輸出 768
print(f"向量前5個值：{question_vector[:5]}") ? ?# 一堆小數
 
# 批量文本嵌入（用于文檔塊）
doc_texts = ["員工每年享有5天帶薪年假", "產假為98天", "婚假3天"]
doc_vectors = embed_model.embed_documents(doc_texts)
print(f"文檔塊數量：{len(doc_vectors)}，每個向量長度：{len(doc_vectors[0])}")

注意：embed_query 和 embed_documents 雖然底層都是生成向量，但有些模型會對查詢和文檔做不同的處理（比如添加特殊前綴），所以不要混用。用戶的問題永遠用 embed_query，文檔塊永遠用 embed_documents。

06 第四步：向量數據庫——給向量們安個家

生成了向量之后，需要把它們存起來，并且能夠快速查找。這就是向量 數據庫 干的事。

你可以把向量數據庫想象成一個超級智能的“圖書館”：你給它一段文字，它立刻告訴你圖書館里哪幾本書的內容和這段文字最像。

6.1 選擇哪個向量數據庫？

• Chroma

  ：輕量級，純 Python，幾行代碼就能跑起來，適合學習和原型驗證。

• FAISS

  ：Facebook 出品，檢索速度極快，但不提供持久化存儲（重啟就丟），適合研究用途。

• Milvus

  ：功能最強大，支持分布式、標量過濾、高并發(fā)，適合生產環(huán)境。

我們這里用?Milvus Lite?版本，它可以直接在本地文件上運行，不需要搭建服務器，非常適合學習。

6.2 安裝和初始化

pip install pymilvus[milvus-lite]

from pymilvus import MilvusClient
 
# 創(chuàng)建一個客戶端，數據會保存在當前目錄下的 milvus_demo.db 文件里
client = MilvusClient(uri="./milvus_demo.db")

6.3 設計數據表結構（Schema）

在 Milvus 里，表叫 Collection。我們需要告訴它每個字段叫什么、什么類型。

from pymilvus import DataType
 
# 定義表結構
def create_schema():
? ? schema = MilvusClient.create_schema(
? ? ? ? auto_id=True, ? ? ? ? ? # 自動生成主鍵ID，不需要自己提供
? ? ? ? enable_dynamic_field=True ?# 允許動態(tài)添加未定義的字段
? ? )
? ? # 添加字段：id（主鍵，整數類型）
? ? schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
? ? # 添加字段：vector（向量，768維，浮點數）
? ? schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=768)
? ? # 添加字段：text（原文內容，最長1024個字符）
? ? schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1024)
? ? # 添加字段：metadata（元數據，JSON格式，可以放任意結構）
? ? schema.add_field(field_name="metadata", datatype=DataType.JSON)
? ? return schema
 
# 創(chuàng)建索引（索引能加速檢索）
def create_index():
? ? index_params = MilvusClient.prepare_index_params()
? ? index_params.add_index(
? ? ? ? field_name="vector", ? ? ? ?# 對向量字段建立索引
? ? ? ? index_type="AUTOINDEX", ? ? # 自動選擇最合適的索引類型
? ? ? ? metric_type="L2" ? ? ? ? ? ?# 相似度計算方式：L2=歐氏距離（值越小越相似）
? ? )
? ? return index_params

metric_type 的三種選擇：

L2

（歐氏距離）：兩點之間的直線距離，越小越相似。最直觀。

IP

（內積）：向量點積，越大越相似（需要先對向量做歸一化）。

COSINE

（余弦相似度）：關注方向而非長度，1表示完全相同。文本相似度通常用這個。

為了簡單，我們先用 L2。

6.4 創(chuàng)建表并插入數據

# 如果表已經存在，先刪除（方便重復運行）
collection_name = "demo_collection"
if client.has_collection(collection_name):
? ? client.drop_collection(collection_name)
 
# 創(chuàng)建表
client.create_collection(
? ? collection_name=collection_name,
? ? schema=create_schema(),
? ? index_params=create_index()
)
 
# 假設我們已經有了切分好的 chunks 和對應的向量
# chunks: List[Document]，每個 Document 有 page_content 和 metadata
# embeddings: List[List[float]]，每個向量與 chunks 一一對應
 
# 把數據組裝成 Milvus 需要的格式
data_to_insert = []
for chunk, vector in zip(chunks, embeddings):
? ? data_to_insert.append({
? ? ? ? "vector": vector, ? ? ? ? ? ? ? ? ? ? # 向量
? ? ? ? "text": chunk.page_content, ? ? ? ? ? # 原文
? ? ? ? "metadata": chunk.metadata ? ? ? ? ? ?# 元數據（如文件名、頁碼等）
? ? })
 
# 插入數據
insert_result = client.insert(
? ? collection_name=collection_name,
? ? data=data_to_insert
)
print(f"插入了 {insert_result['insert_count']} 條數據")

6.5 檢索：根據問題找最相似的塊

當用戶問一個問題時，我們先把問題轉成向量，然后去向量數據庫里找最相似的 k 個文檔塊。

def search_similar(query_text, embed_model, client, top_k=3):
? ? # 1. 把問題轉成向量
? ? query_vector = embed_model.embed_query(query_text)
 
? ? # 2. 在數據庫里搜索
? ? results = client.search(
? ? ? ? collection_name="demo_collection",
? ? ? ? data=[query_vector], ? ? ? ? ? ? ?# 要搜索的向量
? ? ? ? anns_field="vector", ? ? ? ? ? ? ?# 在哪個向量字段里搜索
? ? ? ? search_params={"metric_type": "L2"}, ?# 使用歐氏距離
? ? ? ? output_fields=["text", "metadata"], ? # 要返回哪些字段
? ? ? ? limit=top_k ? ? ? ? ? ? ? ? ? ? ? # 返回最相似的前 top_k 個
? ? )
 
? ? # results 的結構：[[{id, distance, entity}, ...]]
? ? # 提取出原文
? ? contexts = [hit["entity"]["text"] for hit in results[0]]
? ? return contexts
 
# 測試
query = "公司年假怎么計算？"
contexts = search_similar(query, embed_model, client)
for i, ctx in enumerate(contexts):
? ? print(f"結果{i+1}:\n{ctx}\n")

6.6 關于“近似最近鄰”（ ANN ）的通俗解釋

你可能會好奇：數據庫里可能有幾十萬條向量，怎么能在幾毫秒內找到最相似的？

如果老老實實把所有向量都算一遍（這叫 KNN，精確最近鄰），數據量一大就慢死了。

聰明的方法是近似最近鄰（ANN）：事先給向量建一個“索引”，就像書的目錄。查找時，不需要翻遍整本書，而是先看目錄找到大概的章節(jié)，再去那幾頁里仔細找。

Milvus 默認用的 HNSW 算法，你可以這樣理解：

• 想象你要在一個大城市里找一個人。如果你從大街上一個人一個人地問，累死。

• 更好的辦法：先找這個人可能在哪個區(qū)（上層索引），再去那個區(qū)里找街道（中層），最后到街道上找門牌號（底層）。

• HNSW 就是建立了一個多層地圖，上層“跳”得快，下層找得準。

這個索引只需要建一次（插入數據時），之后每次檢索都享受加速。

07 第五步：生成——讓大模型看著參考資料寫答案

最后一步，把檢索到的相關文檔塊，連同用戶的問題，一起發(fā)給大模型，讓它生成答案。

7.1 構建提示詞（Prompt）

提示詞就是你對大模型說的話。一個好的提示詞要包含三部分：

1. 角色設定

   ：你是一個專業(yè)的HR助手...

2. 參考資料

   ：以下是從員工手冊中檢索到的內容...

3. 用戶問題

   ：請根據參考資料回答...

from langchain_core.prompts import ChatPromptTemplate
 
template = ChatPromptTemplate.from_messages([
? ? ("system",?
? ? ?"你是一個專業(yè)的HR助手。請根據下面的【參考資料】回答用戶的問題。"
? ? ?"如果參考資料里沒有相關信息，請如實說‘資料中沒有提到’，不要編造答案。"
? ? ?"\n\n【參考資料】\n{context}"
? ? ),
? ? ("human", "{question}")
])

7.2 調用大模型

你可以使用 OpenAI、國產模型（如智譜、通義千問），或者本地部署的模型。這里以 OpenAI 兼容接口為例：

# pip install langchain-openai
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(
? ? model="gpt-3.5-turbo", ? # 或 "gpt-4"
? ? temperature=0, ? ? ? ? ? # 溫度越低，回答越確定（不隨意發(fā)揮）
? ? api_key="你的API密鑰", ? ?# 也可以從環(huán)境變量讀取
? ? base_url="https://api.openai.com/v1" ?# 如果是其他廠商，換成對應地址
)

7.3 把檢索和生成串起來

LangChain 提供了 RunnablePassthrough 和 RunnableLambda 來構建處理鏈。

from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
 
# 定義一個函數：檢索上下文
def retrieve_context(question):
? ? return search_similar(question, embed_model, client, top_k=3)
 
# 構建鏈
rag_chain = (
? ? {
? ? ? ? "question": RunnablePassthrough(), ? ? ? ? ? # 原樣傳遞問題
? ? ? ? "context": RunnableLambda(retrieve_context) ?# 調用檢索函數
? ? }
? ? | template ? ? ? ? ? ?# 填入提示詞模板
? ? | llm ? ? ? ? ? ? ? ? # 調用大模型
? ? | StrOutputParser() ? # 提取輸出文本
)
 
# 執(zhí)行
question = "新員工的年假是怎么計算的？"
answer = rag_chain.invoke(question)
print(answer)

7.4 流式輸出 （像 ChatGPT 那樣一個字一個字出來）

如果你想實現打字機效果，可以用 stream 方法：

for chunk in rag_chain.stream("試用期員工有年假嗎？"):
? ? print(chunk, end="", flush=True)

08 總結：RAG 的優(yōu)缺點和適用場景

優(yōu)點

1. 知識實時更新

   ：只需要往數據庫里加新文檔，不需要重新訓練模型。

2. 可溯源

   ：答案有據可查，可以告訴用戶“這是根據第X頁的文檔得出的”。

3. 降低幻覺

   ：大模型有參考資料，胡說八道的概率大大降低。

4. 保護隱私

   ：私有數據留存在你自己的數據庫里，不用交給模型廠商。

缺點

1. 響應延遲

   ：每次問答都要做檢索+生成兩步，比直接調用大模型慢一些。

2. 消耗更多 Token

   ：把檢索到的上下文也塞進提示詞里，Token 用量會增加。

3. 檢索效果決定上限

   ：如果沒檢索到相關的資料，大模型再強也沒用。

什么時候該用 RAG？

??適合 RAG：

• 知識頻繁更新（公司制度、產品文檔、新聞資訊）

• 領域知識不在大模型訓練數據中（企業(yè)內部資料、私有數據庫）

• 需要答案有據可查（法律、醫(yī)療、金融）

• 不想承擔微調模型的高昂成本

??不適合 RAG：

• 任務本身不需要外部知識（比如寫一首詩、做數學計算）

• 實時性要求極高（毫秒級響應）

• 文檔量極小，且基本不變（那直接提示詞里寫死就行了）

最后的話

RAG 并不是什么高深莫測的技術，它只是巧妙地利用了“檢索”這個傳統(tǒng)技術，給大模型裝上了一雙可以翻書的“手”。很多人一開始覺得大模型很神奇，但當你理解了它的局限，再配合 RAG，你會發(fā)現：一個可控、可靠、可更新的 AI 系統(tǒng)，并不需要幾十億的投入，只需要你按照這篇文章的步驟，一步步搭建起來。

如果你的團隊也想落地一個智能問答機器人，建議從最簡單的 Chroma + 本地小模型開始，跑通整個流程。等有了實際需求和數據量，再平滑遷移到 Milvus + 商業(yè)大模型。一步一個腳印，你會發(fā)現 RAG 遠比想象中簡單。