檢索問答首版先跑通引用鏈路RAG 原型可以很快跑通加載文檔、切塊、建索引再把檢索結(jié)果放進(jìn)提示詞。但首版服務(wù)要先解決幾個(gè)基礎(chǔ)問題引用能否回到原文文檔更新后索引如何處理低相關(guān)結(jié)果怎樣返回以及上下文長度如何受控。它們比堆疊檢索算法更影響使用體驗(yàn)。首版不必急著加入圖數(shù)據(jù)庫、復(fù)雜重排或多路召回也不能跳過切塊、來源記錄和上下文截?cái)唷Ｏ茸龀鲆粭l能說明“取到了什么、為什么采用、取不到怎么辦”的主鏈路再根據(jù)失敗樣本調(diào)整。1. V1 版 RAG 系統(tǒng)的邊界與取舍在打磨第一版 RAG 系統(tǒng)時(shí)必須明確哪些功能是 MVP 的命脈哪些應(yīng)該果斷暫緩。第一切塊與來源。固定長度切塊可能截?cái)嗑渥踊虮砀?。首版可使用帶少量重疊的簡(jiǎn)單切分方法并在每段保存文檔名、位置和版本命中后把這些信息連同答案一起給出后續(xù)才有依據(jù)檢查切分是否合適。第二向量檢索與 Prompt 組合。不要一上來就搞多模態(tài)或混合檢索先用余弦相似度Cosine Similarity配合基礎(chǔ)向量索引如 FAISS 或 Qdrant。重點(diǎn)要做好 Top-K 的數(shù)量控制與上下文 Token 動(dòng)態(tài)截?cái)喾谰€。第三低置信度處理。向量庫超時(shí)或候選結(jié)果不足時(shí)不要把內(nèi)容質(zhì)量未知的片段強(qiáng)塞給模型?？梢苑祷亍拔凑业娇梢觅Y料”的提示、走人工入口或明確標(biāo)出答案沒有檢索依據(jù)采用哪種方式取決于業(yè)務(wù)風(fēng)險(xiǎn)。2. 極簡(jiǎn)高可用 RAG 數(shù)據(jù)流與降級(jí)控制下面是第一版 RAG 生產(chǎn)級(jí)落地推薦的核心架構(gòu)數(shù)據(jù)流圖。它涵蓋了文檔切片入庫、檢索分級(jí)評(píng)估以及低分降級(jí)分支這條鏈路通過候選過濾和上下文上限減少無關(guān)內(nèi)容進(jìn)入生成環(huán)節(jié)。過濾閾值不是固定結(jié)論應(yīng)使用已標(biāo)注的查詢樣本校準(zhǔn)并持續(xù)查看被拒絕和被采用的結(jié)果。3. Python 生產(chǎn)級(jí)輕量 RAG Pipeline 落地代碼以下代碼使用 Python 原生庫以及 numpy / pydantic手把手實(shí)現(xiàn)了一套包含動(dòng)態(tài)切片、向量余弦檢索、得分閾值攔截與 Prompt 截?cái)嗟妮p量級(jí)生產(chǎn) RAG 核心組件import numpy as np from typing import List, Dict, Any, Tuple from pydantic import BaseModel, Field # 1. 結(jié)構(gòu)化數(shù)據(jù)定義 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. 輕量級(jí)向量檢索與 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 維向量生產(chǎn)環(huán)境替換為 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: # 簡(jiǎn)單的帶重疊遞歸切片策略 (長度 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) # 降級(jí)檢查 1高分上下文評(píng)估 filtered_chunks [c for c, s in retrieved if s self.score_threshold] if not filtered_chunks: # 觸發(fā)降級(jí)得分低于閾值不提供不確定上下文 return RAGResponse( queryquery, answer抱歉知識(shí)庫中未檢索到足夠可靠的相關(guān)依據(jù)無法準(zhǔn)確回答該問題。, retrieved_chunks[], is_fallbackTrue ) # 降級(jí)檢查 2Context Token 動(dòng)態(tài)截?cái)?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--- 知識(shí)參考 ({chunk.chunk_id}) ---\n{chunk.text}\n used_chunks.append(chunk.chunk_id) accumulated_len chunk_token_cost # 構(gòu)造強(qiáng)約束 Prompt prompt ( f你是一個(gè)嚴(yán)謹(jǐn)?shù)闹?。?qǐng)嚴(yán)格基于以下參考資料回答問題。如果參考資料不足以回答請(qǐng)直接說明未知。\n f{context_str}\n f用戶問題{query}\n回答 ) # 模擬 LLM 響應(yīng) mock_llm_answer f[基于 {len(used_chunks)} 個(gè)切片生成] 問題 {query} 的解答。 return RAGResponse( queryquery, answermock_llm_answer, retrieved_chunksused_chunks, is_fallbackFalse ) # 4. 運(yùn)行驗(yàn)證 if __name__ __main__: rag MinimalRAGPipeline(score_threshold0.5) # 模擬知識(shí)庫數(shù)據(jù) sample_docs [ (doc_01, 微服務(wù)架構(gòu)線上發(fā)布必須配置優(yōu)雅停機(jī)攔截 SIGTERM 信號(hào)并設(shè)置 30 秒超時(shí)以等待存量 Connection 處理完成。), (doc_02, Go 語言調(diào)度器 GOMAXPROCS 默認(rèn)為 CPU 核心數(shù)。在容器化環(huán)境中若未正確配置 cgroups 限制容易引發(fā)高頻 context switch。) ] rag.add_documents(sample_docs) print(f知識(shí)庫構(gòu)建完成共計(jì) {len(rag.chunks_db)} 個(gè) Chunk。\n) # 測(cè)試 1高匹配度查詢 res1 rag.generate_answer(微服務(wù)發(fā)布怎么做優(yōu)雅停機(jī)) print(fQuery 1: {res1.query}) print(fAnswer 1: {res1.answer}) print(fUsed Chunks: {res1.retrieved_chunks}, Fallback: {res1.is_fallback}\n) # 測(cè)試 2無法檢索到的不相關(guān)問題 res2 rag.generate_answer(明天天氣的溫度是多少度) print(fQuery 2: {res2.query}) print(fAnswer 2: {res2.answer}) print(fFallback Active: {res2.is_fallback})運(yùn)行輸出如下展示了系統(tǒng)在正常檢索與無法匹配時(shí)觸發(fā)降級(jí)的完整行為模式知識(shí)庫構(gòu)建完成共計(jì) 2 個(gè) Chunk。 Query 1: 微服務(wù)發(fā)布怎么做優(yōu)雅停機(jī) Answer 1: [基于 1 個(gè)切片生成] 問題 微服務(wù)發(fā)布怎么做優(yōu)雅停機(jī) 的解答。 Used Chunks: [doc_01_c0], Fallback: False Query 2: 明天天氣的溫度是多少度 Answer 2: 抱歉知識(shí)庫中未檢索到足夠可靠的相關(guān)依據(jù)無法準(zhǔn)確回答該問題。 Fallback Active: True4. 第一版交付驗(yàn)收的硬指標(biāo)當(dāng)?shù)谝话?RAG 系統(tǒng)準(zhǔn)備部署到測(cè)試與灰度環(huán)境時(shí)不要死磕復(fù)雜的指標(biāo)模型而應(yīng)該拉出具體的工程端到端驗(yàn)收表檢索延遲以業(yè)務(wù)允許的等待時(shí)間為目標(biāo)分別記錄索引查詢、重排和生成前準(zhǔn)備所花的時(shí)間避免只看一個(gè)端到端數(shù)字。上下文上限給提示詞設(shè)定與模型和任務(wù)相符的長度上限超過時(shí)保留來源更清楚的片段并記錄截?cái)嘣?。低相關(guān)結(jié)果準(zhǔn)備領(lǐng)域外、過期和空文檔等樣本確認(rèn)系統(tǒng)會(huì)說明依據(jù)不足而不是把候選片段拼成確定答案。完成這些檢查后才能判斷首版是否適合進(jìn)入下一輪試用。再看失敗樣本是沒有召回、引用不對(duì)應(yīng)還是答案沒有根據(jù)材料收束。把問題分開記錄重排和混合檢索才有明確的改動(dòng)目標(biāo)。上線前也應(yīng)說明知識(shí)庫覆蓋范圍與更新時(shí)間避免讀者把未覆蓋的問題誤解為檢索錯(cuò)誤。引用鏈路還要方便人工核對(duì)。頁面上的來源應(yīng)能回到原始文檔及其具體位置而不是只給一個(gè)模糊的文件名。文檔被刪除、權(quán)限變化或版本更新時(shí)索引側(cè)也要有相應(yīng)處理否則回答即使看起來合理引用也可能已經(jīng)失效。