檢索問答首版先跑通引用鏈路RAG 原型可以很快跑通加載文檔、切塊、建索引再把檢索結果放進提示詞。但首版服務要先解決幾個基礎問題引用能否回到原文文檔更新后索引如何處理低相關結果怎樣返回以及上下文長度如何受控。它們比堆疊檢索算法更影響使用體驗。首版不必急著加入圖數據庫、復雜重排或多路召回也不能跳過切塊、來源記錄和上下文截斷。先做出一條能說明“取到了什么、為什么采用、取不到怎么辦”的主鏈路再根據失敗樣本調整。1. V1 版 RAG 系統的邊界與取舍在打磨第一版 RAG 系統時必須明確哪些功能是 MVP 的命脈哪些應該果斷暫緩。第一切塊與來源。固定長度切塊可能截斷句子或表格。首版可使用帶少量重疊的簡單切分方法并在每段保存文檔名、位置和版本命中后把這些信息連同答案一起給出后續才有依據檢查切分是否合適。第二向量檢索與 Prompt 組合。不要一上來就搞多模態或混合檢索先用余弦相似度Cosine Similarity配合基礎向量索引如 FAISS 或 Qdrant。重點要做好 Top-K 的數量控制與上下文 Token 動態截斷防線。第三低置信度處理。向量庫超時或候選結果不足時不要把內容質量未知的片段強塞給模型。可以返回“未找到可引用資料”的提示、走人工入口或明確標出答案沒有檢索依據采用哪種方式取決于業務風險。2. 極簡高可用 RAG 數據流與降級控制下面是第一版 RAG 生產級落地推薦的核心架構數據流圖。它涵蓋了文檔切片入庫、檢索分級評估以及低分降級分支這條鏈路通過候選過濾和上下文上限減少無關內容進入生成環節。過濾閾值不是固定結論應使用已標注的查詢樣本校準并持續查看被拒絕和被采用的結果。3. Python 生產級輕量 RAG Pipeline 落地代碼以下代碼使用 Python 原生庫以及 numpy / pydantic手把手實現了一套包含動態切片、向量余弦檢索、得分閾值攔截與 Prompt 截斷的輕量級生產 RAG 核心組件import numpy as np from typing import List, Dict, Any, Tuple from pydantic import BaseModel, Field # 1. 結構化數據定義 class DocumentChunk(BaseModel): chunk_id: str text: str metadata: Dict[str, Any] Field(default_factorydict) vector: List[float] Field(default_factorylist) class RAGResponse(BaseModel): query: str answer: str retrieved_chunks: List[str] is_fallback: bool # 2. 輕量級向量檢索與 RAG 核心 Pipeline class MinimalRAGPipeline: def __init__(self, score_threshold: float 0.65, max_context_tokens: int 1500): self.score_threshold score_threshold self.max_context_tokens max_context_tokens self.chunks_db: List[DocumentChunk] [] def mock_embedding(self, text: str) - np.ndarray: 模擬生成 128 維向量生產環境替換為 OpenAI/BGE 等 Embedding API np.random.seed(hash(text) % (2**32 - 1)) vec np.random.randn(128) return vec / np.linalg.norm(vec) # 歸一化 def add_documents(self, docs: List[Tuple[str, str]]): 添加文檔并切片入庫: [(doc_id, text)] for doc_id, text in docs: # 簡單的帶重疊遞歸切片策略 (長度 100重疊 20) chunk_size 100 overlap 20 start 0 idx 0 while start len(text): end min(start chunk_size, len(text)) chunk_text text[start:end] vec self.mock_embedding(chunk_text).tolist() self.chunks_db.append( DocumentChunk( chunk_idf{doc_id}_c{idx}, textchunk_text, metadata{doc_id: doc_id, start_offset: start}, vectorvec ) ) start (chunk_size - overlap) idx 1 def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) - float: return float(np.dot(vec1, vec2)) def retrieve(self, query: str, top_k: int 3) - List[Tuple[DocumentChunk, float]]: 向量余弦相似度檢索 query_vec self.mock_embedding(query) scores: List[Tuple[DocumentChunk, float]] [] for chunk in self.chunks_db: c_vec np.array(chunk.vector) score self._cosine_similarity(query_vec, c_vec) scores.append((chunk, score)) # 按分值降序排列 scores.sort(keylambda x: x[1], reverseTrue) return scores[:top_k] def generate_answer(self, query: str) - RAGResponse: retrieved self.retrieve(query, top_k3) # 降級檢查 1高分上下文評估 filtered_chunks [c for c, s in retrieved if s self.score_threshold] if not filtered_chunks: # 觸發降級得分低于閾值不提供不確定上下文 return RAGResponse( queryquery, answer抱歉知識庫中未檢索到足夠可靠的相關依據無法準確回答該問題。, retrieved_chunks[], is_fallbackTrue ) # 降級檢查 2Context Token 動態截斷 context_str used_chunks [] accumulated_len 0 for chunk in filtered_chunks: # 粗略估算 Token 長度 (1 字符 ~ 0.5 Token) chunk_token_cost len(chunk.text) if accumulated_len chunk_token_cost self.max_context_tokens: break context_str f\n--- 知識參考 ({chunk.chunk_id}) ---\n{chunk.text}\n used_chunks.append(chunk.chunk_id) accumulated_len chunk_token_cost # 構造強約束 Prompt prompt ( f你是一個嚴謹的助手。請嚴格基于以下參考資料回答問題。如果參考資料不足以回答請直接說明未知。\n f{context_str}\n f用戶問題{query}\n回答 ) # 模擬 LLM 響應 mock_llm_answer f[基于 {len(used_chunks)} 個切片生成] 問題 {query} 的解答。 return RAGResponse( queryquery, answermock_llm_answer, retrieved_chunksused_chunks, is_fallbackFalse ) # 4. 運行驗證 if __name__ __main__: rag MinimalRAGPipeline(score_threshold0.5) # 模擬知識庫數據 sample_docs [ (doc_01, 微服務架構線上發布必須配置優雅停機攔截 SIGTERM 信號并設置 30 秒超時以等待存量 Connection 處理完成。), (doc_02, Go 語言調度器 GOMAXPROCS 默認為 CPU 核心數。在容器化環境中若未正確配置 cgroups 限制容易引發高頻 context switch。) ] rag.add_documents(sample_docs) print(f知識庫構建完成共計 {len(rag.chunks_db)} 個 Chunk。\n) # 測試 1高匹配度查詢 res1 rag.generate_answer(微服務發布怎么做優雅停機) print(fQuery 1: {res1.query}) print(fAnswer 1: {res1.answer}) print(fUsed Chunks: {res1.retrieved_chunks}, Fallback: {res1.is_fallback}\n) # 測試 2無法檢索到的不相關問題 res2 rag.generate_answer(明天天氣的溫度是多少度) print(fQuery 2: {res2.query}) print(fAnswer 2: {res2.answer}) print(fFallback Active: {res2.is_fallback})運行輸出如下展示了系統在正常檢索與無法匹配時觸發降級的完整行為模式知識庫構建完成共計 2 個 Chunk。 Query 1: 微服務發布怎么做優雅停機 Answer 1: [基于 1 個切片生成] 問題 微服務發布怎么做優雅停機 的解答。 Used Chunks: [doc_01_c0], Fallback: False Query 2: 明天天氣的溫度是多少度 Answer 2: 抱歉知識庫中未檢索到足夠可靠的相關依據無法準確回答該問題。 Fallback Active: True4. 第一版交付驗收的硬指標當第一版 RAG 系統準備部署到測試與灰度環境時不要死磕復雜的指標模型而應該拉出具體的工程端到端驗收表檢索延遲以業務允許的等待時間為目標分別記錄索引查詢、重排和生成前準備所花的時間避免只看一個端到端數字。上下文上限給提示詞設定與模型和任務相符的長度上限超過時保留來源更清楚的片段并記錄截斷原因。低相關結果準備領域外、過期和空文檔等樣本確認系統會說明依據不足而不是把候選片段拼成確定答案。完成這些檢查后才能判斷首版是否適合進入下一輪試用。再看失敗樣本是沒有召回、引用不對應還是答案沒有根據材料收束。把問題分開記錄重排和混合檢索才有明確的改動目標。上線前也應說明知識庫覆蓋范圍與更新時間避免讀者把未覆蓋的問題誤解為檢索錯誤。引用鏈路還要方便人工核對。頁面上的來源應能回到原始文檔及其具體位置而不是只給一個模糊的文件名。文檔被刪除、權限變化或版本更新時索引側也要有相應處理否則回答即使看起來合理引用也可能已經失效。