大模型RAG+AI Agent企業級落地實戰：檢索增強生成架構與生產部署全指南
人工智能
摘要
- 掌握RAG系統的核心架構與文件處理管線，理解從原始文件到高品質檢索結果的完整鏈路
- 深入混合檢索（向量+關鍵詞+知識圖譜）與重排序技術，實現企業級95%+的檢索準確率
- RAG+AI Agent深度融合實戰：工具增強檢索、多輪對話記憶、企業知識庫權限控制與生產部署
目錄
- 一、RAG系統架構與核心流程
- 二、文件處理與分塊策略
- 三、Embedding模型選型與最佳化
- 四、混合檢索與重排序
- 五、RAG+AI Agent深度融合
- 六、企業知識庫權限與安全
- 七、生產級部署與效能最佳化
- 八、總結與展望
一、RAG系統架構與核心流程
1.1 RAG的核心價值與限制
檢索增強生成（Retrieval-Augmented Generation, RAG）是2024-2026年大模型應用最廣泛的技術範式。其核心價值在於：無需微調模型，透過檢索外部知識庫來增強大模型的回答品質與事實準確性。
然而，生產級RAG系統面臨的核心挑戰遠不止「檢索+拼接」那麼簡單：
- 檢索品質瓶頸：向量檢索的語意相似不等於答案相關，Top-K結果可能包含大量雜訊
- 上下文視窗浪費：無關的檢索結果佔用有限的上下文視窗，降低模型推理品質
- 多跳推理缺失：複雜問題需要多步檢索和推理，單次檢索無法滿足
- 即時性要求：企業知識庫頻繁更新，索引需要即時同步
1.2 生產級RAG架構
一個生產級RAG系統的架構遠比簡單的「查詢→檢索→生成」複雜：
┌──────────────────────────────────────────────────┐
│ Query Understanding │
│ 意圖識別 · 查詢改寫 · 實體抽取 · 多跳分解 │
├──────────────────────────────────────────────────┤
│ Hybrid Retrieval │
│ 向量檢索 · BM25關鍵詞 · 知識圖譜 · SQL查詢 │
├──────────────────────────────────────────────────┤
│ Reranking & Fusion │
│ 交叉編碼器重排 · 互資訊最大化 · 結果去重 │
├──────────────────────────────────────────────────┤
│ Context Assembly │
│ 相關性過濾 · 上下文壓縮 · 結構化組織 │
├──────────────────────────────────────────────────┤
│ Generation & Verification │
│ Chain-of-Thought · 事實校驗 · 幻覺偵測 │
└──────────────────────────────────────────────────┘
Query Understanding層負責理解使用者查詢的真實意圖，包括查詢改寫、實體抽取和多跳問題分解。Hybrid Retrieval層使用多種檢索策略的融合。Reranking層對初步檢索結果進行精細排序。Context Assembly層組裝最佳上下文。Generation層生成最終答案並進行事實校驗。
1.3 RAG vs 微調 vs 預訓練
| 維度 | RAG | 微調 | 預訓練 |
|---|---|---|---|
| 知識更新 | 即時 | 需重新訓練 | 需重新訓練 |
| 成本 | 低 | 中 | 極高 |
| 事實準確性 | 高 | 中 | 低 |
| 領域適應性 | 強 | 強 | 最強 |
| 部署複雜度 | 中 | 低 | 高 |
| 幻覺控制 | 好 | 一般 | 差 |
企業級場景中，RAG+微調的組合是最佳實踐：RAG保證事實準確性，微調最佳化模型對特定領域的理解和輸出風格。
二、文件處理與分塊策略
2.1 文件解析管線
企業知識庫的文件格式多樣（PDF、Word、Excel、PPT、HTML、Markdown），需要統一的解析管線：
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import hashlib
@dataclass
class ParsedDocument:
doc_id: str
title: str
content: str
metadata: dict
source_path: str
checksum: str
page_count: int
language: str
class DocumentParser:
def __init__(self, ocr_enabled: bool = True, table_enabled: bool = True):
self.ocr_enabled = ocr_enabled
self.table_enabled = table_enabled
def parse(self, file_path: str) -> ParsedDocument:
path = Path(file_path)
suffix = path.suffix.lower()
match suffix:
case ".pdf":
return self._parse_pdf(file_path)
case ".docx" | ".doc":
return self._parse_docx(file_path)
case ".xlsx" | ".xls":
return self._parse_excel(file_path)
case ".pptx":
return self._parse_pptx(file_path)
case ".md":
return self._parse_markdown(file_path)
case ".html" | ".htm":
return self._parse_html(file_path)
case _:
return self._parse_plain_text(file_path)
def _parse_pdf(self, file_path: str) -> ParsedDocument:
import fitz
doc = fitz.open(file_path)
content_parts = []
page_count = len(doc)
for page_num in range(page_count):
page = doc[page_num]
text = page.get_text()
if text.strip():
content_parts.append(text)
if self.table_enabled:
tables = page.find_tables()
for table in tables:
table_text = table.to_pandas().to_markdown()
content_parts.append(f"\n[第 {page_num + 1} 頁的表格]\n{table_text}")
if self.ocr_enabled:
images = page.get_images()
for img_idx, img in enumerate(images):
xref = img[0]
base_image = doc.extract_image(xref)
if base_image:
image_bytes = base_image["image"]
ocr_text = self._ocr_image(image_bytes)
if ocr_text:
content_parts.append(
f"\n[第 {page_num + 1} 頁的第 {img_idx + 1} 張圖片]\n{ocr_text}"
)
content = "\n\n".join(content_parts)
checksum = hashlib.sha256(content.encode()).hexdigest()[:16]
return ParsedDocument(
doc_id=f"doc_{checksum}",
title=Path(file_path).stem,
content=content,
metadata={"format": "pdf", "page_count": page_count},
source_path=file_path,
checksum=checksum,
page_count=page_count,
language=self._detect_language(content),
)
def _ocr_image(self, image_bytes: bytes) -> Optional[str]:
try:
import pytesseract
from PIL import Image
import io
image = Image.open(io.BytesIO(image_bytes))
return pytesseract.image_to_string(image, lang='chi_sim+eng')
except Exception:
return None
def _detect_language(self, text: str) -> str:
sample = text[:500]
chinese_chars = sum(1 for c in sample if '\u4e00' <= c <= '\u9fff')
if chinese_chars / max(len(sample), 1) > 0.3:
return "zh-TW"
return "en"
2.2 分塊策略深度對比
文件分塊是RAG系統中最關鍵的環節之一，分塊策略直接影響檢索品質：
固定長度分塊（Fixed-Size Chunking）：按固定Token數切分，實作簡單但可能切斷語意完整性。
遞迴字元分塊（Recursive Character Chunking）：按段落→句子→字元的優先順序遞迴切分，保持語意完整性。LangChain的RecursiveCharacterTextSplitter即採用此策略。
語意分塊（Semantic Chunking）：使用Embedding模型計算相鄰句子的語意相似度，在語意斷點處切分。品質最高但運算成本大。
文件結構分塊（Structure-Aware Chunking）：利用文件的標題層級、章節結構進行分塊，保持文件的邏輯結構。
from dataclasses import dataclass
from typing import Callable
import numpy as np
@dataclass
class Chunk:
chunk_id: str
content: str
metadata: dict
start_index: int
end_index: int
token_count: int
parent_doc_id: str
class SemanticChunker:
def __init__(
self,
embedding_fn: Callable[[str], list[float]],
similarity_threshold: float = 0.5,
min_chunk_size: int = 100,
max_chunk_size: int = 1000,
):
self.embedding_fn = embedding_fn
self.similarity_threshold = similarity_threshold
self.min_chunk_size = min_chunk_size
self.max_chunk_size = max_chunk_size
def chunk(self, text: str, doc_id: str) -> list[Chunk]:
sentences = self._split_sentences(text)
if len(sentences) <= 1:
return [Chunk(
chunk_id=f"{doc_id}_0",
content=text,
metadata={"chunk_type": "semantic"},
start_index=0,
end_index=len(text),
token_count=len(text) // 4,
parent_doc_id=doc_id,
)]
embeddings = [self.embedding_fn(s) for s in sentences]
similarities = [
self._cosine_similarity(embeddings[i], embeddings[i + 1])
for i in range(len(embeddings) - 1)
]
breakpoints = []
for i, sim in enumerate(similarities):
if sim < self.similarity_threshold:
breakpoints.append(i + 1)
chunks = []
current_start = 0
chunk_idx = 0
for bp in breakpoints + [len(sentences)]:
chunk_text = " ".join(sentences[current_start:bp])
token_count = len(chunk_text) // 4
if token_count >= self.min_chunk_size:
chunks.append(Chunk(
chunk_id=f"{doc_id}_{chunk_idx}",
content=chunk_text,
metadata={
"chunk_type": "semantic",
"sentence_count": bp - current_start,
},
start_index=current_start,
end_index=bp,
token_count=token_count,
parent_doc_id=doc_id,
))
chunk_idx += 1
elif chunks:
chunks[-1].content += " " + chunk_text
chunks[-1].end_index = bp
chunks[-1].token_count += token_count
current_start = bp
for chunk in chunks:
if chunk.token_count > self.max_chunk_size:
self._split_oversized(chunk, doc_id, chunks)
return chunks
def _split_sentences(self, text: str) -> list[str]:
import re
sentences = re.split(r'(?<=[。！？.!?])\s*', text)
return [s.strip() for s in sentences if s.strip()]
def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
a_arr = np.array(a)
b_arr = np.array(b)
return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr) + 1e-8))
def _split_oversized(self, chunk: Chunk, doc_id: str, chunks: list[Chunk]) -> None:
pass
2.3 元資料增強分塊
為每個分塊添加豐富的元資料，支援檢索時的精細過濾：
@dataclass
class EnhancedChunk(Chunk):
heading_path: list[str]
keywords: list[str]
entities: list[dict]
summary: str
access_level: int
department: str
doc_type: str
created_at: str
updated_at: str
class MetadataEnricher:
def __init__(self, llm_client):
self.llm = llm_client
async def enrich(self, chunk: Chunk, doc: ParsedDocument) -> EnhancedChunk:
keywords = await self._extract_keywords(chunk.content)
entities = await self._extract_entities(chunk.content)
summary = await self._generate_summary(chunk.content)
return EnhancedChunk(
**chunk.__dict__,
heading_path=self._extract_heading_path(doc, chunk),
keywords=keywords,
entities=entities,
summary=summary,
access_level=doc.metadata.get("access_level", 0),
department=doc.metadata.get("department", ""),
doc_type=doc.metadata.get("format", ""),
created_at=doc.metadata.get("created_at", ""),
updated_at=doc.metadata.get("updated_at", ""),
)
async def _extract_keywords(self, text: str) -> list[str]:
prompt = f"從以下文字中提取5-10個關鍵術語，以JSON陣列格式回傳：\n\n{text[:1000]}"
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return []
async def _extract_entities(self, text: str) -> list[dict]:
prompt = f"""從以下文字中提取命名實體，回傳JSON陣列，每個實體包含name、type、value欄位：
實體類型包括：PERSON, ORGANIZATION, PRODUCT, DATE, LOCATION, TECHNOLOGY
文字：
{text[:2000]}"""
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return []
async def _generate_summary(self, text: str) -> str:
prompt = f"用一句話總結以下文字的核心內容：\n\n{text[:500]}"
return await self.llm.generate(prompt)
def _extract_heading_path(self, doc: ParsedDocument, chunk: Chunk) -> list[str]:
return doc.metadata.get("heading_path", [])
三、Embedding模型選型與最佳化
3.1 主流Embedding模型對比
| 模型 | 維度 | MTEB得分 | 中文能力 | 推理速度 | 授權條款 |
|---|---|---|---|---|---|
| BGE-M3 | 1024 | 73.5 | 優秀 | 中 | MIT |
| GTE-Qwen2-7B | 3584 | 76.2 | 優秀 | 慢 | Apache 2.0 |
| text-embedding-3-large | 3072 | 74.5 | 良好 | 快 | 商業 |
| Jina-Embeddings-v3 | 1024 | 72.8 | 良好 | 中 | CC-BY-4.0 |
| BCE-Embedding | 768 | 71.2 | 優秀 | 快 | MIT |
選型建議：
- 中文場景首選BGE-M3，性價比最高
- 追求極致效果選GTE-Qwen2-7B，但推理成本高
- 需要多語言支援選Jina-Embeddings-v3
- 使用OpenAI API的場景選text-embedding-3-large
3.2 Embedding服務部署
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
from sentence_transformers import SentenceTransformer
app = FastAPI()
class EmbedRequest(BaseModel):
texts: list[str]
normalize: bool = True
class EmbedResponse(BaseModel):
embeddings: list[list[float]]
model: str
dimension: int
model = SentenceTransformer("BAAI/bge-m3")
@app.post("/embed", response_model=EmbedResponse)
async def embed(request: EmbedRequest):
embeddings = model.encode(
request.texts,
normalize_embeddings=request.normalize,
show_progress_bar=False,
)
return EmbedResponse(
embeddings=embeddings.tolist(),
model="bge-m3",
dimension=embeddings.shape[1],
)
@app.post("/embed/batch", response_model=EmbedResponse)
async def embed_batch(request: EmbedRequest):
batch_size = 64
all_embeddings = []
for i in range(0, len(request.texts), batch_size):
batch = request.texts[i:i + batch_size]
batch_embeddings = model.encode(
batch,
normalize_embeddings=request.normalize,
batch_size=len(batch),
)
all_embeddings.append(batch_embeddings)
embeddings = np.vstack(all_embeddings)
return EmbedResponse(
embeddings=embeddings.tolist(),
model="bge-m3",
dimension=embeddings.shape[1],
)
3.3 查詢側Embedding最佳化
查詢側的Embedding最佳化是提升檢索品質的關鍵技巧：
查詢擴展（Query Expansion）：使用LLM將使用者查詢擴展為多個相關查詢，增加檢索覆蓋面。
假設性文件嵌入（HyDE）：先讓LLM生成一個假設性答案，再用假設答案的Embedding去檢索，比直接用查詢檢索效果更好。
指令前綴（Instruction Prefix）：在查詢前添加任務指令，如「為以下查詢檢索相關文件：」，對齊訓練時的輸入格式。
class QueryOptimizer:
def __init__(self, llm_client, embedding_fn):
self.llm = llm_client
self.embedding_fn = embedding_fn
async def expand_query(self, query: str, num_expansions: int = 3) -> list[str]:
prompt = f"""將以下查詢改寫為{num_expansions}個不同角度的等價查詢，以JSON陣列格式回傳。
原始查詢：{query}
要求：
1. 保持原始查詢的核心意圖
2. 使用不同的表述方式和關鍵詞
3. 涵蓋不同的專業術語和通俗說法"""
response = await self.llm.generate(prompt)
import json
try:
expansions = json.loads(response)
return [query] + expansions[:num_expansions]
except:
return [query]
async def hyde_embed(self, query: str) -> list[float]:
prompt = f"""請針對以下問題，寫一段詳細的回答（即使你不確定答案，也請給出合理的假設性回答）：
問題：{query}"""
hypothetical_answer = await self.llm.generate(prompt)
return self.embedding_fn(hypothetical_answer)
def instruction_embed(self, query: str, task: str = "search") -> list[float]:
prefixes = {
"search": "為以下查詢檢索相關文件：",
"similarity": "尋找與以下內容相似的文件：",
"classification": "對以下內容進行分類：",
}
prefix = prefixes.get(task, "")
return self.embedding_fn(f"{prefix}{query}")
四、混合檢索與重排序
4.1 混合檢索架構
單一向量檢索無法涵蓋所有場景。關鍵詞檢索（BM25）擅長精確比對（產品型號、專有名詞），向量檢索擅長語意比對（概念相似），知識圖譜檢索擅長關係推理。混合檢索是生產級RAG的必選項。
from dataclasses import dataclass
from typing import Optional
@dataclass
class RetrievalResult:
chunk_id: str
content: str
score: float
source: str
metadata: dict
class HybridRetriever:
def __init__(
self,
vector_store,
bm25_store,
kg_store=None,
vector_weight: float = 0.6,
bm25_weight: float = 0.3,
kg_weight: float = 0.1,
):
self.vector_store = vector_store
self.bm25_store = bm25_store
self.kg_store = kg_store
self.vector_weight = vector_weight
self.bm25_weight = bm25_weight
self.kg_weight = kg_weight
async def retrieve(
self,
query: str,
query_embedding: list[float],
top_k: int = 20,
filters: Optional[dict] = None,
) -> list[RetrievalResult]:
vector_results = await self.vector_store.search(
query_embedding, top_k=top_k * 2, filters=filters
)
bm25_results = await self.bm25_store.search(
query, top_k=top_k * 2, filters=filters
)
kg_results = []
if self.kg_store:
kg_results = await self.kg_store.search(
query, top_k=top_k
)
merged = self._reciprocal_rank_fusion(
vector_results, bm25_results, kg_results
)
return merged[:top_k]
def _reciprocal_rank_fusion(
self,
vector_results: list[RetrievalResult],
bm25_results: list[RetrievalResult],
kg_results: list[RetrievalResult],
k: int = 60,
) -> list[RetrievalResult]:
scores: dict[str, float] = {}
result_map: dict[str, RetrievalResult] = {}
for rank, result in enumerate(vector_results):
rrf_score = self.vector_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
result_map[result.chunk_id] = result
for rank, result in enumerate(bm25_results):
rrf_score = self.bm25_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
if result.chunk_id not in result_map:
result_map[result.chunk_id] = result
for rank, result in enumerate(kg_results):
rrf_score = self.kg_weight / (k + rank + 1)
scores[result.chunk_id] = scores.get(result.chunk_id, 0.0) + rrf_score
if result.chunk_id not in result_map:
result_map[result.chunk_id] = result
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [
RetrievalResult(
chunk_id=cid,
content=result_map[cid].content,
score=scores[cid],
source=result_map[cid].source,
metadata=result_map[cid].metadata,
)
for cid in sorted_ids
]
4.2 交叉編碼器重排序
初步檢索結果使用雙編碼器（Bi-Encoder）計算，速度快但精度有限。交叉編碼器（Cross-Encoder）將查詢和文件一起輸入模型，精度更高但速度慢，適合對Top-K結果進行精排：
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name, max_length=512)
def rerank(
self,
query: str,
results: list[RetrievalResult],
top_k: int = 10,
) -> list[RetrievalResult]:
pairs = [(query, result.content) for result in results]
scores = self.model.predict(pairs)
scored_results = list(zip(results, scores))
scored_results.sort(key=lambda x: x[1], reverse=True)
return [
RetrievalResult(
chunk_id=result.chunk_id,
content=result.content,
score=float(score),
source=result.source,
metadata={**result.metadata, "rerank_score": float(score)},
)
for result, score in scored_results[:top_k]
]
4.3 相關性過濾
重排序後仍需過濾低相關性結果，避免雜訊進入上下文：
class RelevanceFilter:
def __init__(self, min_score: float = 0.3, max_chunks: int = 8):
self.min_score = min_score
self.max_chunks = max_chunks
def filter(self, results: list[RetrievalResult]) -> list[RetrievalResult]:
filtered = [r for r in results if r.score >= self.min_score]
return filtered[:self.max_chunks]
def adaptive_filter(
self,
results: list[RetrievalResult],
max_context_tokens: int = 4000,
) -> list[RetrievalResult]:
selected = []
total_tokens = 0
for result in results:
if result.score < self.min_score:
continue
chunk_tokens = len(result.content) // 4
if total_tokens + chunk_tokens > max_context_tokens:
break
selected.append(result)
total_tokens += chunk_tokens
return selected
五、RAG+AI Agent深度融合
5.1 工具增強檢索Agent
將RAG檢索能力封裝為AI Agent可呼叫的工具，實作更智慧的檢索策略：
from typing import Annotated
class RAGAgentTools:
def __init__(self, retriever: HybridRetriever, reranker: Reranker):
self.retriever = retriever
self.reranker = reranker
def search_knowledge_base(
self,
query: Annotated[str, "知識庫搜尋查詢"],
top_k: Annotated[int, "回傳結果數量"] = 5,
filters: Annotated[dict | None, "元資料過濾條件"] = None,
) -> str:
"""搜尋企業知識庫中的相關文件。"""
query_embedding = self.get_embedding(query)
results = self.retriever.retrieve(query, query_embedding, top_k=top_k * 2, filters=filters)
reranked = self.reranker.rerank(query, results, top_k=top_k)
if not reranked:
return "未找到相關文件。"
formatted = []
for i, result in enumerate(reranked):
formatted.append(
f"[文件 {i + 1}] (分數: {result.score:.3f})\n"
f"來源: {result.metadata.get('source', '未知')}\n"
f"內容: {result.content}\n"
)
return "\n---\n".join(formatted)
def search_by_entity(
self,
entity_name: Annotated[str, "要搜尋的實體名稱"],
entity_type: Annotated[str, "實體類型：PERSON, ORGANIZATION, PRODUCT 等"] = "",
) -> str:
"""搜尋提及特定實體的文件。"""
filters = {"entities": {"name": entity_name}}
if entity_type:
filters["entities"]["type"] = entity_type
return self.search_knowledge_base(entity_name, filters=filters)
def compare_documents(
self,
topic: Annotated[str, "要跨文件比較的主題"],
doc_ids: Annotated[list[str], "要比較的文件ID"] = None,
) -> str:
"""跨多份文件比較某主題的資訊。"""
query_embedding = self.get_embedding(topic)
filters = {"parent_doc_id": {"$in": doc_ids}} if doc_ids else None
results = self.retriever.retrieve(topic, query_embedding, top_k=20, filters=filters)
reranked = self.reranker.rerank(topic, results, top_k=10)
grouped = {}
for result in reranked:
doc_id = result.metadata.get("parent_doc_id", "unknown")
if doc_id not in grouped:
grouped[doc_id] = []
grouped[doc_id].append(result)
output = []
for doc_id, chunks in grouped.items():
output.append(f"文件: {doc_id}")
for chunk in chunks:
output.append(f" - {chunk.content[:200]}...")
return "\n\n".join(output)
5.2 多輪對話RAG
企業級RAG系統需要支援多輪對話，維護對話上下文和檢索歷史：
class ConversationalRAG:
def __init__(self, llm_client, retriever, reranker):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
async def chat(
self,
query: str,
conversation_history: list[dict],
max_context_tokens: int = 4000,
) -> dict:
rewritten_query = await self._rewrite_query(query, conversation_history)
query_embedding = self.get_embedding(rewritten_query)
results = await self.retriever.retrieve(
rewritten_query, query_embedding, top_k=20
)
reranked = self.reranker.rerank(rewritten_query, results, top_k=10)
context = self._assemble_context(reranked, max_context_tokens)
prompt = self._build_prompt(query, context, conversation_history)
answer = await self.llm.generate(prompt)
return {
"answer": answer,
"sources": [
{
"chunk_id": r.chunk_id,
"content": r.content[:200],
"score": r.score,
"source": r.metadata.get("source", ""),
}
for r in reranked[:5]
],
"rewritten_query": rewritten_query,
}
async def _rewrite_query(self, query: str, history: list[dict]) -> str:
if not history:
return query
history_text = "\n".join([
f"{'使用者' if h['role'] == 'user' else '助手'}: {h['content']}"
for h in history[-6:]
])
prompt = f"""基於以下對話歷史，將使用者的最新問題改寫為獨立的、完整的檢索查詢。
只回傳改寫後的查詢，不要解釋。
對話歷史：
{history_text}
最新問題：{query}
改寫後的查詢："""
return await self.llm.generate(prompt)
def _assemble_context(self, results: list[RetrievalResult], max_tokens: int) -> str:
parts = []
total = 0
for result in results:
tokens = len(result.content) // 4
if total + tokens > max_tokens:
break
parts.append(f"[來源: {result.metadata.get('source', '未知')}]\n{result.content}")
total += tokens
return "\n\n---\n\n".join(parts)
def _build_prompt(self, query: str, context: str, history: list[dict]) -> str:
return f"""你是一個專業的企業知識庫助手。請基於以下檢索到的文件內容回答使用者的問題。
要求：
1. 只基於提供的文件內容回答，不要編造資訊
2. 如果文件中沒有相關資訊，明確告知使用者
3. 引用資訊時標註來源文件
4. 使用清晰的結構化格式
檢索到的文件：
{context}
使用者問題：{query}
回答："""
5.3 多跳推理RAG
複雜問題需要多步檢索和推理，單次RAG無法滿足。多跳RAG透過Agent的規劃能力，將複雜問題分解為多步檢索鏈：
class MultiHopRAG:
def __init__(self, llm_client, retriever, reranker, max_hops: int = 3):
self.llm = llm_client
self.retriever = retriever
self.reranker = reranker
self.max_hops = max_hops
async def answer(self, query: str) -> dict:
hop_results = []
current_query = query
all_contexts = []
for hop in range(self.max_hops):
query_embedding = self.get_embedding(current_query)
results = await self.retriever.retrieve(
current_query, query_embedding, top_k=10
)
reranked = self.reranker.rerank(current_query, results, top_k=5)
all_contexts.extend(reranked)
hop_results.append({
"hop": hop + 1,
"query": current_query,
"results_count": len(reranked),
})
next_action = await self._decide_next_hop(query, all_contexts, hop)
if next_action["action"] == "answer":
break
elif next_action["action"] == "search":
current_query = next_action["query"]
context = self._assemble_context(all_contexts)
answer = await self._generate_answer(query, context)
return {
"answer": answer,
"hops": hop_results,
"total_contexts": len(all_contexts),
}
async def _decide_next_hop(self, original_query: str, contexts: list, hop: int) -> dict:
if hop >= self.max_hops - 1:
return {"action": "answer"}
context_summary = "\n".join([
f"- {c.content[:200]}" for c in contexts[-5:]
])
prompt = f"""基於原始問題和已檢索到的資訊，判斷是否需要繼續檢索。
原始問題：{original_query}
已檢索資訊：
{context_summary}
請判斷：
1. 如果已有足夠資訊回答問題，回傳 {{"action": "answer"}}
2. 如果需要更多資訊，回傳 {{"action": "search", "query": "下一步檢索查詢"}}
以JSON格式回傳："""
response = await self.llm.generate(prompt)
import json
try:
return json.loads(response)
except:
return {"action": "answer"}
六、企業知識庫權限與安全
6.1 文件級權限控制
企業知識庫的文件通常有嚴格的存取權限控制，RAG檢索必須遵守权限規則：
class PermissionAwareRetriever:
def __init__(self, base_retriever, permission_service):
self.base_retriever = base_retriever
self.permission_service = permission_service
async def retrieve(
self,
query: str,
query_embedding: list[float],
user_id: str,
top_k: int = 20,
) -> list[RetrievalResult]:
user_permissions = await self.permission_service.get_user_permissions(user_id)
accessible_departments = user_permissions.get("departments", [])
access_level = user_permissions.get("access_level", 0)
filters = {
"$or": [
{"department": {"$in": accessible_departments}},
{"access_level": {"$lte": access_level}},
]
}
results = await self.base_retriever.retrieve(
query, query_embedding, top_k=top_k * 2, filters=filters
)
verified_results = []
for result in results:
if await self._verify_access(result, user_permissions):
verified_results.append(result)
return verified_results[:top_k]
async def _verify_access(self, result: RetrievalResult, permissions: dict) -> bool:
doc_department = result.metadata.get("department", "")
doc_access_level = result.metadata.get("access_level", 99)
if doc_access_level <= permissions.get("access_level", 0):
return True
if doc_department in permissions.get("departments", []):
return True
return False
6.2 資料去識別化
檢索結果在送入大模型前，需要自動去識別化敏感資訊：
import re
class DataMasker:
PATTERNS = {
"phone": (r'1[3-9]\d{9}', lambda m: m.group()[:3] + "****" + m.group()[-4:]),
"email": (r'[\w.-]+@[\w.-]+\.\w+', lambda m: m.group()[0] + "***@" + m.group().split("@")[1]),
"id_card": (r'\d{17}[\dXx]', lambda m: m.group()[:6] + "********" + m.group()[-4:]),
"bank_card": (r'\d{16,19}', lambda m: m.group()[:4] + "****" + m.group()[-4:]),
}
def mask(self, text: str, enabled_types: list[str] | None = None) -> str:
types = enabled_types or list(self.PATTERNS.keys())
for type_name in types:
if type_name in self.PATTERNS:
pattern, replacer = self.PATTERNS[type_name]
text = re.sub(pattern, replacer, text)
return text
6.3 稽核日誌
所有RAG查詢和檢索操作需要記錄稽核日誌：
from datetime import datetime
class RAGAuditLogger:
def __init__(self, log_store):
self.log_store = log_store
async def log_query(
self,
user_id: str,
query: str,
rewritten_query: str | None,
results_count: int,
answer_preview: str,
latency_ms: int,
):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"user_id": user_id,
"original_query": query,
"rewritten_query": rewritten_query,
"results_count": results_count,
"answer_preview": answer_preview[:200],
"latency_ms": latency_ms,
}
await self.log_store.insert(entry)
async def log_access_denied(self, user_id: str, query: str, denied_doc_ids: list[str]):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": "access_denied",
"user_id": user_id,
"query": query,
"denied_documents": denied_doc_ids,
}
await self.log_store.insert(entry)
七、生產級部署與效能最佳化
7.1 快取策略
RAG系統的多層快取策略：
import hashlib
from functools import lru_cache
class RAGCache:
def __init__(self, redis_client, ttl: int = 3600):
self.redis = redis_client
self.ttl = ttl
def _cache_key(self, query: str, user_id: str, filters: dict | None = None) -> str:
raw = f"{query}:{user_id}:{filters}"
return f"rag:cache:{hashlib.md5(raw.encode()).hexdigest()}"
async def get(self, query: str, user_id: str, filters: dict | None = None) -> dict | None:
key = self._cache_key(query, user_id, filters)
cached = await self.redis.get(key)
if cached:
import json
return json.loads(cached)
return None
async def set(self, query: str, user_id: str, result: dict, filters: dict | None = None):
key = self._cache_key(query, user_id, filters)
import json
await self.redis.setex(key, self.ttl, json.dumps(result, ensure_ascii=False))
async def invalidate_doc(self, doc_id: str):
pattern = f"rag:cache:*"
async for key in self.redis.scan_iter(pattern):
cached = await self.redis.get(key)
if cached and doc_id in cached.decode():
await self.redis.delete(key)
7.2 非同步索引更新
企業知識庫的文件頻繁更新，需要非同步索引更新機制：
import asyncio
from typing import Callable
class AsyncIndexUpdater:
def __init__(
self,
embedding_fn: Callable,
vector_store,
chunker,
batch_size: int = 100,
flush_interval: int = 30,
):
self.embedding_fn = embedding_fn
self.vector_store = vector_store
self.chunker = chunker
self.batch_size = batch_size
self.flush_interval = flush_interval
self.pending_updates: list[dict] = []
self._running = False
async def start(self):
self._running = True
asyncio.create_task(self._flush_loop())
async def stop(self):
self._running = False
if self.pending_updates:
await self._flush()
async def add_document(self, doc: ParsedDocument):
self.pending_updates.append({
"action": "add",
"doc_id": doc.doc_id,
"content": doc.content,
"metadata": doc.metadata,
})
if len(self.pending_updates) >= self.batch_size:
await self._flush()
async def delete_document(self, doc_id: str):
self.pending_updates.append({
"action": "delete",
"doc_id": doc_id,
})
async def _flush_loop(self):
while self._running:
await asyncio.sleep(self.flush_interval)
if self.pending_updates:
await self._flush()
async def _flush(self):
updates = self.pending_updates[:]
self.pending_updates.clear()
to_add = [u for u in updates if u["action"] == "add"]
to_delete = [u for u in updates if u["action"] == "delete"]
if to_add:
all_chunks = []
for update in to_add:
chunks = self.chunker.chunk(update["content"], update["doc_id"])
all_chunks.extend(chunks)
texts = [c.content for c in all_chunks]
embeddings = [self.embedding_fn(t) for t in texts]
await self.vector_store.upsert(
ids=[c.chunk_id for c in all_chunks],
embeddings=embeddings,
metadatas=[c.metadata for c in all_chunks],
documents=texts,
)
if to_delete:
doc_ids = [u["doc_id"] for u in to_delete]
await self.vector_store.delete_by_doc_ids(doc_ids)
7.3 效能指標與SLA
| 指標 | SLA目標 | 監控方式 |
|---|---|---|
| 端到端延遲P95 | < 3s | Prometheus histogram |
| 檢索延遲P95 | < 500ms | 自訂指標 |
| 檢索準確率 | > 95% | 人工抽檢+自動評估 |
| 索引更新延遲 | < 60s | 文件寫入→可檢索時間差 |
| 系統可用性 | > 99.9% | 健康檢查+告警 |
| 並發QPS | > 100 | 壓測驗證 |
八、總結與展望
大模型RAG+AI Agent的企業級落地是2026年AI應用最核心的技術方向之一。本文從RAG架構、文件處理、Embedding最佳化、混合檢索、Agent融合、權限安全和生產部署七個維度，系統性地闡述了企業級RAG系統的建構方法。
關鍵要點回顧：
- 混合檢索：向量+BM25+知識圖譜的RRF融合，是生產級RAG檢索的標配
- 重排序：交叉編碼器精排+相關性過濾，將檢索準確率從70%提升到95%+
- 查詢最佳化：查詢擴展+HyDE+指令前綴，顯著提升檢索召回率
- Agent融合：RAG能力封裝為Agent工具，支援多輪對話和多跳推理
- 安全合規：文件級權限+資料去識別化+稽核日誌，是企業級部署的必要條件
未來，RAG技術將向更智慧的方向演進：自適應檢索策略（根據查詢類型自動選擇檢索方式）、持續學習（從使用者回饋中最佳化檢索品質）、多模態RAG（支援影像、表格、程式碼等非文字內容的檢索與生成）。AI Agent與RAG的深度融合將使知識庫從被動檢索工具，進化為主動知識助手。
相關閱讀
權威參考
本站提供瀏覽器本地工具，免註冊即可試用 →
#RAG#AI Agent#大模型#向量检索#企业级落地#2026