構建可用于生產環境的 RAG 智能體：開發者完整指南
你有沒有納悶過，為什么你的 RAG 系統總是返回一堆無關的結果，或者漏掉顯而易見的答案？你不是一個人！很多開發者一開始用 vector search，然后一臉懵地發現，他們的“智能”AI 連一個簡單的產品代碼都找不到。
如果你試過搭建一個 Retrieval-Augmented Generation (RAG) 系統，可能也撞過我一樣的南墻。你的 chatbot 有時候回答得很棒，但有時候完全答非所問，返回一些概念上相似但實際沒用的信息。
問題出在哪兒？大部分 RAG 教程只關注那些酷炫的東西——embeddings、vector databases 和 LLMs，卻忽略了一個樸實無華的真相：搜索其實很難，而 relevance 才是王道。
今天，我要帶你一步步打造一個在生產環境中真正好用的 RAG 智能體。我們會聊聊不同的搜索策略，什么時候用哪種，以及怎么組合它們來達到最高準確率。
RAG 現實檢查：為什么單靠 Vector Search 不夠
先講個可能你也覺得眼熟的故事。
我第一次建 RAG 系統時，用的是標準套路：把文檔切塊，用 OpenAI 生成 embeddings，存到 vector database，檢索 top-k 相似塊。看起來優雅又現代。
然后我問它：“P0420 錯誤代碼的解決辦法是什么？”
系統回了三段關于排氣系統和排放的文字——技術上有點相關，但完全沒提到 P0420。而我需要的那個具體公告，埋在第47個 chunk 里。
這就是 vector search 的陷阱。語義相似不等于 relevance。
完整的搜索策略工具箱
我學到的經驗是：不同的問題需要不同的搜索策略。來逐一拆解什么時候用哪種方法。
1. Keyword Search：精準的冠軍
是什么：傳統的文本匹配，用像 BM25 這樣的算法。想象一下2005年的 Google 搜索。
什么時候用：
- 搜索具體術語、代碼或 ID
- 法律文檔這種需要精確措辭的場景
- 用戶知道具體術語的時候
用例：比如“查找安全公告 SB-2024-003”
# 簡單關鍵詞搜索實現
def keyword_search(query, documents):
query_terms = query.lower().split()
scored_docs = []
for doc in documents:
score = sum(1 for term in query_terms if term in doc.lower())
if score > 0:
scored_docs.append((doc, score))
return sorted(scored_docs, key=lambda x: x[1], reverse=True)注意：它很“死腦筋”。你搜“car repair”，它不會找到“automobile maintenance”。
2. Vector Search：語義偵探
是什么：把文本轉成高維向量，捕捉語義。相似的概念會在向量空間里聚在一起。
什么時候用：
- 開放式問題
- 用戶的措辭和文檔不完全一致時
- 知識庫內容風格多變
用例：比如“怎么修車上怪響？” → 能找到關于引擎故障排查的文檔
import chromadb
from chromadb.utils import embedding_functions
# Vector search 設置
client = chromadb.Client()
embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
collection = client.create_collection(
name="knowledge_base",
embedding_functinotallow=embedding_fn
)
defvector_search(query, top_k=5):
results = collection.query(
query_texts=[query],
n_results=top_k
)
return results["documents"][0]注意：有時候會返回“相關”但沒用的結果。recall 高，precision 低。
3. Hybrid Search：兩全其美
是什么：結合 keyword 和 vector search，然后合并結果。
什么時候用：
- 生產系統，不能漏掉任何關鍵信息
- 查詢類型混合（有的很具體，有的很開放）
- 準確性比簡單性更重要
秘訣：用 Reciprocal Rank Fusion (RRF) 合并排名：
def reciprocal_rank_fusion(keyword_results, vector_results, k=60):
"""用 RRF 算法合并兩個排名列表"""
scores = {}
# 給關鍵詞結果打分
for rank, doc inenumerate(keyword_results, 1):
scores[doc] = scores.get(doc, 0) + 1 / (k + rank)
# 給向量結果打分
for rank, doc inenumerate(vector_results, 1):
scores[doc] = scores.get(doc, 0) + 1 / (k + rank)
# 按綜合得分排序
returnsorted(scores.items(), key=lambda x: x[1], reverse=True)4. Database Search：事實核查員
是什么：對結構化數據進行 SQL 查詢。
什么時候用：
- 當前價格、庫存、用戶數據
- 需要精確、實時的數字
- 結合文本搜索給完整答案
5. Graph Search：關系專家
是什么：查詢知識圖譜，找相關聯的信息。
什么時候用：
- “誰是 Alice 的上司的匯報對象？”
- 復雜多跳推理
- 關系比內容更重要
這里有個決策流程圖幫你選：
Pure Vector Search 的隱藏問題
在建智能體之前，先看看會出什么問題。我遇到過的十大失敗模式：
打造生產級 RAG 智能體
現在來實戰。我會教你怎么用 ChromaDB 做 vector search，加上 reranking，用《回到未來》電影劇本作為知識庫。
步驟 1：設置知識庫
import chromadb
import openai
import tiktoken
from pathlib import Path
from typing importList
# 智能切塊函數
defsmart_chunk(text: str, max_tokens: int = 200) -> List[str]:
"""按語義切分文本，限制 token 數"""
tokenizer = tiktoken.get_encoding("cl100k_base")
words = text.split()
chunks = []
current_chunk = []
for word in words:
# 檢查加這個詞會不會超 token 限制
test_chunk = " ".join(current_chunk + [word])
iflen(tokenizer.encode(test_chunk)) > max_tokens:
if current_chunk: # 不加空塊
chunks.append(" ".join(current_chunk))
current_chunk = [word]
else:
current_chunk.append(word)
# 加最后一塊
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
# 加載并處理劇本
script_text = Path("back_to_the_future.txt").read_text(encoding="utf-8")
chunks = smart_chunk(script_text, max_tokens=200)
print(f"從劇本創建了 {len(chunks)} 個塊")步驟 2：創建向量存儲
from chromadb.utils import embedding_functions
import uuid
# 初始化帶持久化的 ChromaDB
client = chromadb.Client(chromadb.config.Settings(
persist_directory="./movie_knowledge_base"
))
# 設置 embedding 函數
embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key=openai.api_key,
model_name="text-embedding-3-small"
)
# 創建或獲取集合
collection_name = "bttf_script"
try:
collection = client.get_collection(collection_name)
print("找到已有集合")
except:
collection = client.create_collection(
name=collection_name,
embedding_functinotallow=embedding_fn
)
print("創建新集合")
# 如果集合為空，添加文檔
if collection.count() == 0:
print("正在添加文檔到集合...")
collection.add(
ids=[str(uuid.uuid4()) for _ in chunks],
documents=chunks,
metadatas=[{"chunk_index": i} for i inrange(len(chunks))]
)
client.persist()
print("文檔已添加并持久化")步驟 3：實現帶 Reranking 的智能搜索
def smart_search(query: str, top_k: int = 5) -> str:
"""
多階段檢索：
1. 用 vector search 廣撒網
2. 按關鍵詞重疊 rerank
3. 返回最佳結果
"""
# 第一階段：vector search 提高 recall
initial_results = collection.query(
query_texts=[query],
n_results=min(top_k * 3, 15) # 獲取更多候選
)
docs = initial_results["documents"][0]
ifnot docs:
return"未找到相關信息。"
# 第二階段：按關鍵詞重疊 rerank
query_words = set(query.lower().split())
scored_docs = []
for doc in docs:
doc_words = set(doc.lower().split())
keyword_score = len(query_words.intersection(doc_words))
scored_docs.append((doc, keyword_score))
# 按關鍵詞重疊排序，保留最佳結果
reranked = sorted(scored_docs, key=lambda x: x[1], reverse=True)
best_docs = [doc for doc, score in reranked[:top_k]]
return"\n\n---\n\n".join(best_docs)步驟 4：用 OpenAI Agents SDK 打造 RAG 智能體
這里是魔法發生的地方。OpenAI Agents SDK 讓構建會用工具的智能體變得超簡單。下面是一個完整的工作示例：
import uuid
from pathlib import Path
import chromadb
import tiktoken
from agents import Agent, Runner, function_tool
from dotenv import load_dotenv
# 加載環境變量
load_dotenv()
# 加載并切分劇本
script_text = Path("back_to_the_future.txt").read_text(encoding="utf-8")
defsimple_chunk(text, max_tokens=200):
tokenizer = tiktoken.get_encoding("cl100k_base")
words, chunk, chunks = text.split(), [], []
for w in words:
iflen(tokenizer.encode(" ".join(chunk + [w]))) > max_tokens:
chunks.append(" ".join(chunk))
chunk = [w]
else:
chunk.append(w)
if chunk:
chunks.append(" ".join(chunk))
return chunks
docs = simple_chunk(script_text, max_tokens=200)
# 設置帶持久化的 ChromaDB
client = chromadb.PersistentClient(path="./chroma_script_store")
collection_name = "bttf_script"
# 獲取或創建集合
try:
collection = client.get_collection(collection_name)
except Exception:
collection = client.create_collection(name=collection_name)
# 如果集合為空，添加文檔
if collection.count() == 0:
collection.add(
ids=[str(uuid.uuid4()) for _ in docs],
documents=docs
)
# 定義搜索工具，帶正確的裝飾器
@function_tool
defsearch_script(query: str, top_k: int = 3) -> str:
"""搜索《回到未來》劇本中的相關片段"""
res = collection.query(query_texts=[query], n_results=top_k)
if res and"documents"in res and res["documents"] and res["documents"][0]:
return"\n\n".join(res["documents"][0])
return"未找到相關文檔。"
# 創建智能體
agent = Agent(
name="Script Agent",
instructinotallow=(
"你回答關于《回到未來》電影的問題。\n"
"需要時調用 `search_script` 工具獲取片段，"
"然后引用或改述它們來回答。"
),
tools=[search_script],
)
# 測試智能體
query = "Doc 在哪里讓 Marty 見他，幾點鐘？"
result = Runner.run_sync(agent, query)
print("\n--- 答案 ---\n", result.final_output)
query = "凌晨 1:15 發生了什么？"
result = Runner.run_sync(agent, query)
print("\n--- 答案 ---\n", result.final_output)優雅之處：
- @function_tool 裝飾器自動把你的 Python 函數變成智能體能用的工具
- 智能體指令告訴 LLM 什么時候、怎么用搜索工具
- Runner.run_sync() 管理整個對話流程——智能體決定什么時候搜索，處理結果，生成最終答案
- 持久化存儲讓你不用每次重啟都重新嵌入文檔
秘訣：讓它達到生產級
以下是區分業余項目和生產系統的關鍵：
1. 智能切塊策略
別光按 token 數切分，還要考慮：
- 句子邊界
- 段落分隔
- 主題轉換
- 重疊塊保留上下文
2. 多階段檢索
# 生產級檢索管道
def production_search(query: str):
# 第一階段：快速檢索（廣撒網）
candidates = vector_search(query, k=20)
# 第二階段：關鍵詞加權
keyword_boosted = boost_keyword_matches(candidates, query)
# 第三階段：交叉編碼器 rerank（如果預算夠）
final_results = cross_encoder_rerank(keyword_boosted, query, k=5)
return final_results3. 評估與監控
跟蹤這些指標：
- Hit Rate：檢索到相關文檔的問題百分比
- Answer Quality：人工評分或用 LLM 做裁判
- Latency：端到端響應時間
- Cost：嵌入和生成成本
4. 錯誤處理
def robust_search(query: str):
try:
return smart_search(query)
except Exception as e:
# 回退到簡單搜索
logging.error(f"智能搜索失敗: {e}")
return simple_keyword_search(query)什么時候用什么：你的決策矩陣
這是我的速查表，幫你選對方法：
從 Vector Search 開始，如果：
- 內容多變，偏自然語言
- 用戶提開放式問題
- 想要開箱即用的好結果
加 Keyword Search，如果：
- 用戶搜具體術語、代碼、名字
- 有結構化或一致的術語
- 精準度比召回率更重要
用 Hybrid Search，如果：
- 建生產系統
- 不能漏掉重要結果
- 有工程帶寬
考慮 Graph/SQL，如果：
- 需要關系查詢
- 有結構化數據
- 實時準確性關鍵
總結
打造一個牛逼的 RAG 系統，不是用最新的 embedding 模型或最炫的 vector database，而是要懂你的用戶、你的數據，選對每種場景的檢索策略。
從簡單的 vector search 開始，衡量關鍵指標，逐步增加復雜度。最重要的是，永遠用真實用戶查詢測試——別光用演示里完美的例子。
你的 RAG 翻車故事是啥？在下面留言吧！我想聽聽你的經歷和解決辦法。
想深入了解？我正在寫一個完整的 RAG 實現指南，包含生產級示例。關注我獲取更新，告訴我你想讓我下次講啥具體話題。
本文轉載自?????PyTorch研習社?????，作者：AI研究生
