RAG 系統實作完全指南：打造企業級智能問答系統
RAG (Retrieval-Augmented Generation) 是當前最熱門的 LLM 應用架構，能夠讓大型語言模型回答專業領域問題，並大幅降低幻覺(Hallucination)問題。本文將詳細介紹如何從零開始建立一套生產級的 RAG 系統。
RAG 系統架構概覽
核心組件
┌─────────────────────────────────────────────────────────┐
│ RAG 系統架構 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 文件處理 │ ───→ │ 向量化 │ ───→ │ 向量資料庫│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ ↑ │
│ │ │ │
│ ↓ │ │
│ ┌──────────┐ ┌─────────┐ │
│ │ 使用者 │ ─── 查詢 ───→ 向量檢索 ──→│ LLM │ │
│ │ 查詢 │ ↓ │ 生成 │ │
│ └──────────┘ ┌─────────┐ └─────────┘ │
│ │ Re-rank │ │ │
│ └─────────┘ │ │
│ │ │ │
│ └────→ 整合 ←───────┘ │
│ ↓ │
│ ┌──────────┐ │
│ │ 回答 │ │
│ └──────────┘ │
└─────────────────────────────────────────────────────────┘
技術選型建議
| 組件 | 推薦方案 | 適用場景 |
|---|---|---|
| 向量資料庫 | Pinecone | 雲端服務，易於擴展 |
| Qdrant | 自建部署，隱私優先 | |
| Milvus | 大規模生產環境 | |
| 嵌入模型 | OpenAI text-embedding-3-large | 高品質英文 |
| bge-large-zh-v1.5 | 中文優化 | |
| Voyage AI | 領域專用 | |
| LLM | GPT-4 Turbo | 通用高品質 |
| Claude 3 Opus | 長文本處理 | |
| Mixtral 8x7B | 自建部署 | |
| 框架 | LangChain | 快速原型開發 |
| LlamaIndex | 文件索引優化 | |
| Haystack | 生產級部署 |
步驟 1：文件處理與分塊策略
智能文件分塊
文件分塊是 RAG 系統的基礎，直接影響檢索品質。
基本分塊策略：
from langchain.text_splitter import RecursiveCharacterTextSplitter
def smart_text_splitter(documents, chunk_size=1000, chunk_overlap=200):
"""
智能文字分塊器
Parameters:
- chunk_size: 每塊文字大小 (token 數量)
- chunk_overlap: 重疊部分，保持上下文連貫性
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=[
"\n\n", # 段落分隔
"\n", # 行分隔
"。", # 中文句號
".", # 英文句號
"!",
"?",
";",
":",
" ", # 空格
"", # 字符
]
)
chunks = text_splitter.split_documents(documents)
return chunks
# 使用範例
from langchain.document_loaders import PyPDFLoader, TextLoader
# 載入文件
pdf_loader = PyPDFLoader("company_handbook.pdf")
documents = pdf_loader.load()
# 分塊處理
chunks = smart_text_splitter(documents, chunk_size=800, chunk_overlap=150)
print(f"Total documents: {len(documents)}")
print(f"Total chunks: {len(chunks)}")
進階：語意分塊
傳統字符分塊可能破壞語意完整性，使用語意分塊可以改善：
from langchain.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
def semantic_chunking(documents, breakpoint_threshold_type="percentile"):
"""
基於語意相似度的智能分塊
breakpoint_threshold_type:
- percentile: 百分位數閾值
- standard_deviation: 標準差閾值
- interquartile: 四分位數閾值
"""
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
text_splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type=breakpoint_threshold_type,
breakpoint_threshold_amount=75 # 75th percentile
)
chunks = text_splitter.create_documents([doc.page_content for doc in documents])
return chunks
# 比較兩種分塊方式
traditional_chunks = smart_text_splitter(documents)
semantic_chunks = semantic_chunking(documents)
print(f"Traditional chunks: {len(traditional_chunks)}")
print(f"Semantic chunks: {len(semantic_chunks)}")
文件元數據增強
添加豐富的元數據可以改善檢索精度：
def enhance_chunks_metadata(chunks, source_info):
"""
為文件塊添加結構化元數據
"""
enhanced_chunks = []
for i, chunk in enumerate(chunks):
# 基本元數據
chunk.metadata.update({
'chunk_id': i,
'source': source_info['filename'],
'document_type': source_info['type'],
'created_date': source_info['created_date'],
'department': source_info.get('department', 'general'),
'security_level': source_info.get('security_level', 'public'),
})
# 內容分析元數據
chunk.metadata.update({
'word_count': len(chunk.page_content.split()),
'has_code': '```' in chunk.page_content,
'has_table': '|' in chunk.page_content,
'language': detect_language(chunk.page_content),
})
# 語意標籤 (可選)
chunk.metadata['tags'] = extract_key_topics(chunk.page_content)
enhanced_chunks.append(chunk)
return enhanced_chunks
def extract_key_topics(text, max_topics=5):
"""
使用 LLM 提取關鍵主題
"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{
"role": "user",
"content": f"請從以下文字提取 {max_topics} 個關鍵主題，用逗號分隔:\n\n{text}"
}],
temperature=0.3,
)
topics = response.choices[0].message.content.strip().split(',')
return [topic.strip() for topic in topics]
步驟 2：向量嵌入與索引建立
選擇合適的嵌入模型
中文嵌入模型比較：
from sentence_transformers import SentenceTransformer
from openai import OpenAI
import numpy as np
class EmbeddingComparison:
def __init__(self):
# 本地模型
self.bge_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
self.m3e_model = SentenceTransformer('moka-ai/m3e-base')
# API 模型
self.openai_client = OpenAI()
def get_embeddings(self, text, model='bge'):
"""獲取文本嵌入向量"""
if model == 'bge':
return self.bge_model.encode(text, normalize_embeddings=True)
elif model == 'm3e':
return self.m3e_model.encode(text, normalize_embeddings=True)
elif model == 'openai':
response = self.openai_client.embeddings.create(
model="text-embedding-3-large",
input=text
)
return np.array(response.data[0].embedding)
def compare_models(self, queries, docs):
"""比較不同模型的檢索效果"""
results = {}
for model_name in ['bge', 'm3e', 'openai']:
# 嵌入文檔
doc_embeddings = [self.get_embeddings(doc, model_name) for doc in docs]
# 嵌入查詢
query_embedding = self.get_embeddings(queries[0], model_name)
# 計算相似度
similarities = [
np.dot(query_embedding, doc_emb)
for doc_emb in doc_embeddings
]
# 排序
ranked_indices = np.argsort(similarities)[::-1]
results[model_name] = {
'top_3_indices': ranked_indices[:3].tolist(),
'top_3_scores': [similarities[i] for i in ranked_indices[:3]]
}
return results
# 使用範例
comparator = EmbeddingComparison()
queries = ["如何申請員工旅遊補助？"]
docs = [
"員工福利包含旅遊補助，每年最高 5000 元...",
"請假流程：先填寫請假單，再由主管核准...",
"旅遊補助申請需檢附發票，於活動後 30 天內提交...",
]
results = comparator.compare_models(queries, docs)
for model, result in results.items():
print(f"\n{model} 模型結果:")
print(f" Top 3 索引: {result['top_3_indices']}")
print(f" Top 3 分數: {result['top_3_scores']}")
建立向量索引 - Qdrant 實作
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer
import uuid
class QdrantRAGIndex:
def __init__(self, collection_name="company_docs"):
# 初始化 Qdrant 客戶端
self.client = QdrantClient(url="http://localhost:6333")
self.collection_name = collection_name
# 嵌入模型
self.embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')
self.embedding_dim = self.embedder.get_sentence_embedding_dimension()
# 建立 collection
self._create_collection()
def _create_collection(self):
"""建立或重建 collection"""
try:
self.client.recreate_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=self.embedding_dim,
distance=Distance.COSINE # 使用餘弦相似度
)
)
print(f"Collection '{self.collection_name}' created successfully")
except Exception as e:
print(f"Error creating collection: {e}")
def index_documents(self, documents, batch_size=100):
"""
批次索引文件
documents: List of (text, metadata) tuples
"""
points = []
for i, (text, metadata) in enumerate(documents):
# 生成向量
vector = self.embedder.encode(text, normalize_embeddings=True).tolist()
# 建立 point
point = PointStruct(
id=str(uuid.uuid4()),
vector=vector,
payload={
'text': text,
'metadata': metadata,
'chunk_index': i
}
)
points.append(point)
# 批次上傳
if len(points) >= batch_size:
self.client.upsert(
collection_name=self.collection_name,
points=points
)
print(f"Indexed {i+1} documents...")
points = []
# 上傳剩餘文件
if points:
self.client.upsert(
collection_name=self.collection_name,
points=points
)
print(f"Total {i+1} documents indexed successfully")
def search(self, query, top_k=5, score_threshold=0.7, filters=None):
"""
搜尋相關文件
query: 查詢文本
top_k: 返回結果數量
score_threshold: 最低相似度閾值
filters: 元數據過濾條件
"""
# 生成查詢向量
query_vector = self.embedder.encode(query, normalize_embeddings=True).tolist()
# 執行搜尋
search_result = self.client.search(
collection_name=self.collection_name,
query_vector=query_vector,
limit=top_k,
score_threshold=score_threshold,
query_filter=filters # 可選的元數據過濾
)
return search_result
# 使用範例
rag_index = QdrantRAGIndex(collection_name="company_knowledge_base")
# 準備文件
documents = [
("旅遊補助申請流程：需檢附發票正本，於活動後 30 天內提交...",
{"category": "福利", "department": "HR"}),
("請假規定：病假需檢附醫生證明，事假需提前 3 天申請...",
{"category": "人事", "department": "HR"}),
# ... 更多文件
]
# 索引文件
rag_index.index_documents(documents)
# 搜尋
results = rag_index.search(
query="如何申請旅遊補助？",
top_k=3,
score_threshold=0.75
)
for result in results:
print(f"Score: {result.score:.4f}")
print(f"Text: {result.payload['text'][:100]}...")
print(f"Metadata: {result.payload['metadata']}\n")
步驟 3：檢索優化策略
Hybrid Search (混合檢索)
結合向量檢索與關鍵字檢索，提升召回率：
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
def __init__(self, vector_index, documents):
self.vector_index = vector_index
self.documents = documents
# 建立 BM25 索引
tokenized_docs = [doc.split() for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
def retrieve(self, query, top_k=10, alpha=0.5):
"""
混合檢索
alpha: 向量檢索權重 (0-1)
1-alpha: BM25 權重
"""
# 1. 向量檢索
vector_results = self.vector_index.search(query, top_k=top_k*2)
vector_scores = {
result.payload['chunk_index']: result.score
for result in vector_results
}
# 2. BM25 關鍵字檢索
tokenized_query = query.split()
bm25_scores = self.bm25.get_scores(tokenized_query)
# 正規化 BM25 分數到 [0, 1]
max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
normalized_bm25 = bm25_scores / max_bm25
# 3. 混合分數計算
hybrid_scores = {}
all_indices = set(vector_scores.keys()) | set(range(len(bm25_scores)))
for idx in all_indices:
vec_score = vector_scores.get(idx, 0)
bm25_score = normalized_bm25[idx] if idx < len(normalized_bm25) else 0
# 加權組合
hybrid_scores[idx] = alpha * vec_score + (1 - alpha) * bm25_score
# 4. 排序並返回 top-k
sorted_indices = sorted(
hybrid_scores.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return [
{
'index': idx,
'text': self.documents[idx],
'score': score
}
for idx, score in sorted_indices
]
# 使用範例
hybrid_retriever = HybridRetriever(rag_index, [doc[0] for doc in documents])
results = hybrid_retriever.retrieve(
query="旅遊補助需要什麼文件？",
top_k=5,
alpha=0.7 # 70% 向量檢索，30% BM25
)
Re-ranking (重新排序)
使用交叉編碼器對初步檢索結果重新排序：
from sentence_transformers import CrossEncoder
class ReRanker:
def __init__(self, model_name='BAAI/bge-reranker-large'):
self.model = CrossEncoder(model_name)
def rerank(self, query, documents, top_k=5):
"""
重新排序檢索結果
Returns: List of (document, score) sorted by relevance
"""
# 建立 query-document pairs
pairs = [[query, doc] for doc in documents]
# 計算相關性分數
scores = self.model.predict(pairs)
# 排序
ranked_results = sorted(
zip(documents, scores),
key=lambda x: x[1],
reverse=True
)
return ranked_results[:top_k]
# 整合到 RAG pipeline
class RAGPipelineWithRerank:
def __init__(self, retriever, reranker, llm):
self.retriever = retriever
self.reranker = reranker
self.llm = llm
def query(self, question, retrieve_k=20, final_k=5):
# 1. 初步檢索 (召回更多候選)
candidates = self.retriever.retrieve(question, top_k=retrieve_k)
candidate_texts = [c['text'] for c in candidates]
# 2. Re-ranking (精準排序)
reranked = self.reranker.rerank(question, candidate_texts, top_k=final_k)
# 3. 構建 prompt
context = "\n\n".join([doc for doc, score in reranked])
prompt = f"""基於以下資訊回答問題：
{context}
問題：{question}
請提供詳細且準確的回答。如果資訊不足以回答問題，請明確說明。"""
# 4. 生成答案
response = self.llm.generate(prompt)
return {
'answer': response,
'sources': reranked,
'context': context
}
步驟 4：LLM 整合與 Prompt 工程
Prompt 模板設計
from langchain.prompts import PromptTemplate
# 基礎 RAG prompt
basic_rag_template = """你是一個專業的企業知識庫助手。請根據提供的上下文資訊回答用戶問題。
上下文資訊：
{context}
用戶問題：{question}
回答要求：
1. 基於提供的上下文資訊回答
2. 如果上下文中沒有相關資訊，請明確說明
3. 提供具體的參考來源
4. 回答要清晰、專業、易懂
回答："""
basic_prompt = PromptTemplate(
template=basic_rag_template,
input_variables=["context", "question"]
)
# 進階 RAG prompt (帶思考鏈)
advanced_rag_template = """你是一個專業的企業知識庫助手。請仔細分析提供的上下文資訊，並回答用戶問題。
上下文資訊：
{context}
用戶問題：{question}
請按照以下步驟回答：
1. **理解問題**：首先分析用戶問題的核心意圖
2. **資訊檢索**：從上下文中找出相關資訊
- 列出所有相關的段落
- 標註資訊來源
3. **推理分析**：
- 整合多個資訊來源
- 分析資訊的相關性和可靠性
- 識別可能存在的矛盾或不一致
4. **生成答案**：
- 提供清晰、完整的回答
- 引用具體的來源
- 如果資訊不足，明確指出缺失的部分
5. **建議行動**：如適用，提供下一步建議
回答格式：
【問題分析】
<你的問題分析>
【相關資訊】
<從上下文提取的關鍵資訊>
【答案】
<詳細回答>
【參考來源】
<引用的段落編號和摘要>
【建議】（如適用）
<下一步建議>
"""
advanced_prompt = PromptTemplate(
template=advanced_rag_template,
input_variables=["context", "question"]
)
完整 RAG Chain 實作
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
class ProductionRAG System:
def __init__(
self,
vector_store,
llm_model="gpt-4-turbo-preview",
temperature=0.1,
streaming=True
):
# LLM 配置
callbacks = [StreamingStdOutCallbackHandler()] if streaming else []
self.llm = ChatOpenAI(
model=llm_model,
temperature=temperature,
callbacks=callbacks
)
# Retriever 配置
self.retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": 10, # 初步檢索數量
"score_threshold": 0.7 # 最低相似度
}
)
# Re-ranker
self.reranker = ReRanker()
# 構建 chain
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # 或 "map_reduce", "refine"
retriever=self.retriever,
return_source_documents=True,
chain_type_kwargs={
"prompt": advanced_prompt
}
)
def query(self, question, rerank=True):
"""
執行 RAG 查詢
Returns:
answer: 生成的答案
sources: 參考來源文檔
metadata: 額外的元數據（分數、時間等）
"""
import time
start_time = time.time()
# 1. 檢索相關文檔
docs = self.retriever.get_relevant_documents(question)
# 2. Re-ranking (可選)
if rerank and len(docs) > 3:
doc_texts = [doc.page_content for doc in docs]
reranked = self.reranker.rerank(question, doc_texts, top_k=5)
docs = [doc for doc, score in reranked]
# 3. 生成答案
result = self.qa_chain.invoke({"query": question})
# 4. 整理輸出
elapsed_time = time.time() - start_time
return {
'answer': result['result'],
'sources': result['source_documents'],
'metadata': {
'elapsed_time': elapsed_time,
'num_sources': len(result['source_documents']),
'model': self.llm.model_name
}
}
# 使用範例
rag_system = ProductionRAGSystem(
vector_store=qdrant_vector_store,
llm_model="gpt-4-turbo-preview",
temperature=0.1,
streaming=True
)
# 查詢
result = rag_system.query("員工旅遊補助的申請流程是什麼？")
print(f"\n答案：\n{result['answer']}\n")
print(f"參考來源數量：{result['metadata']['num_sources']}")
print(f"查詢時間：{result['metadata']['elapsed_time']:.2f} 秒")
步驟 5：評估與優化
RAG 系統評估指標
from ragas import evaluate
from ragas.metrics import (
answer_relevancy,
faithfulness,
context_recall,
context_precision
)
class RAGEvaluator:
def __init__(self, rag_system):
self.rag_system = rag_system
self.metrics = [
answer_relevancy,
faithfulness,
context_recall,
context_precision
]
def create_evaluation_dataset(self, qa_pairs):
"""
準備評估數據集
qa_pairs: List of {
'question': str,
'ground_truth': str, # 標準答案
'context': str # 黃金上下文（可選）
}
"""
dataset = {
'question': [],
'answer': [],
'contexts': [],
'ground_truths': []
}
for qa in qa_pairs:
# 執行 RAG 查詢
result = self.rag_system.query(qa['question'])
dataset['question'].append(qa['question'])
dataset['answer'].append(result['answer'])
dataset['contexts'].append([
doc.page_content for doc in result['sources']
])
dataset['ground_truths'].append([qa['ground_truth']])
return dataset
def evaluate(self, dataset):
"""執行評估"""
from datasets import Dataset
eval_dataset = Dataset.from_dict(dataset)
# 執行評估
result = evaluate(
dataset=eval_dataset,
metrics=self.metrics
)
return result
# 使用範例
evaluator = RAGEvaluator(rag_system)
# 準備測試數據
test_qa_pairs = [
{
'question': '如何申請旅遊補助？',
'ground_truth': '需要在旅遊後 30 天內提交發票正本和申請表單...'
},
# ... 更多測試案例
]
# 建立評估數據集
eval_dataset = evaluator.create_evaluation_dataset(test_qa_pairs)
# 執行評估
evaluation_results = evaluator.evaluate(eval_dataset)
print("評估結果：")
print(f"Answer Relevancy: {evaluation_results['answer_relevancy']:.4f}")
print(f"Faithfulness: {evaluation_results['faithfulness']:.4f}")
print(f"Context Recall: {evaluation_results['context_recall']:.4f}")
print(f"Context Precision: {evaluation_results['context_precision']:.4f}")
A/B 測試框架
import random
from collections import defaultdict
class RAGABTester:
def __init__(self, system_a, system_b):
self.system_a = system_a
self.system_b = system_b
self.results = defaultdict(list)
def run_ab_test(self, queries, user_feedback_func=None):
"""
執行 A/B 測試
user_feedback_func: Optional function to collect user ratings
"""
for query in queries:
# 隨機分配到 A 或 B
system = random.choice(['A', 'B'])
if system == 'A':
result = self.system_a.query(query)
else:
result = self.system_b.query(query)
# 記錄結果
self.results[system].append({
'query': query,
'answer': result['answer'],
'elapsed_time': result['metadata']['elapsed_time'],
'num_sources': result['metadata']['num_sources']
})
# 收集用戶反饋（可選）
if user_feedback_func:
rating = user_feedback_func(query, result['answer'])
self.results[system][-1]['user_rating'] = rating
return self.analyze_results()
def analyze_results(self):
"""分析 A/B 測試結果"""
analysis = {}
for system in ['A', 'B']:
results = self.results[system]
analysis[system] = {
'avg_response_time': sum(r['elapsed_time'] for r in results) / len(results),
'avg_num_sources': sum(r['num_sources'] for r in results) / len(results),
'total_queries': len(results)
}
# 如果有用戶評分
if 'user_rating' in results[0]:
analysis[system]['avg_rating'] = sum(r['user_rating'] for r in results) / len(results)
return analysis
生產環境部署
FastAPI 服務包裝
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
app = FastAPI(title="RAG API Service", version="1.0.0")
# 全局 RAG 系統實例
rag_system = None
class Query(BaseModel):
question: str
top_k: Optional[int] = 5
rerank: Optional[bool] = True
class Answer(BaseModel):
answer: str
sources: List[dict]
metadata: dict
@app.on_event("startup")
async def startup_event():
"""啟動時初始化 RAG 系統"""
global rag_system
rag_system = ProductionRAGSystem(
vector_store=initialize_vector_store(),
llm_model="gpt-4-turbo-preview"
)
print("RAG System initialized successfully")
@app.post("/query", response_model=Answer)
async def query_endpoint(query: Query):
"""RAG 查詢端點"""
try:
result = rag_system.query(
question=query.question,
rerank=query.rerank
)
return Answer(
answer=result['answer'],
sources=[
{
'content': doc.page_content,
'metadata': doc.metadata
}
for doc in result['sources'][:query.top_k]
],
metadata=result['metadata']
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""健康檢查端點"""
return {"status": "healthy", "version": "1.0.0"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Docker 容器化
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安裝依賴
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 複製應用程式
COPY . .
# 下載嵌入模型（如果使用本地模型）
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-large-zh-v1.5')"
# 暴露端口
EXPOSE 8000
# 啟動應用
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
qdrant:
image: qdrant/qdrant:latest
ports:
- "6333:6333"
volumes:
- qdrant_storage:/qdrant/storage
rag-api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- QDRANT_URL=http://qdrant:6333
depends_on:
- qdrant
volumes:
- ./data:/app/data
volumes:
qdrant_storage:
實際應用案例
案例 1：企業內部知識庫
場景：某科技公司建立員工手冊、技術文檔問答系統
實作重點：
- 文檔分層索引（部門、類別）
- 權限控制與資訊安全
- 多語言支援（中英文）
- 歷史對話追蹤
案例 2：客戶服務智能助手
場景：電商平台客服機器人
實作重點：
- 產品資訊實時更新
- 訂單狀態查詢整合
- 情感分析與人工轉接
- 多輪對話管理
案例 3：法律文件分析系統
場景：法律事務所判例搜尋與分析
實作重點：
- 長文本處理（判決書）
- 引用網絡建立
- 時間序列分析
- 專業術語識別
常見問題與解決方案
Q1: 如何處理超長文檔？
方案：Hierarchical Retrieval
# 先檢索章節，再檢索段落
def hierarchical_retrieval(query, document_chunks):
# 第一層：檢索相關章節
chapter_results = retrieve_chapters(query, top_k=3)
# 第二層：在相關章節中檢索段落
paragraph_results = []
for chapter in chapter_results:
paragraphs = retrieve_paragraphs(query, chapter, top_k=5)
paragraph_results.extend(paragraphs)
return paragraph_results
Q2: 如何減少幻覺(Hallucination)？
方案：
- 強化 prompt 約束
- 加入引用驗證
- 使用 self-consistency 檢查
- 降低 temperature 參數
Q3: 如何提升中文處理效果？
方案：
- 使用中文專用嵌入模型 (bge-large-zh)
- 中文分詞優化
- 繁簡轉換處理
- 領域詞彙增強
總結
建立生產級 RAG 系統的關鍵要素：
- 文檔處理 - 智能分塊與元數據增強
- 嵌入模型 - 選擇合適的向量化方案
- 檢索策略 - Hybrid Search + Re-ranking
- Prompt 工程 - 結構化的指令設計
- 持續優化 - 評估、測試、迭代
在 BASHCAT，我們擁有豐富的 RAG 系統建置經驗，已成功為多家企業打造客製化的智能問答系統。如果您正在考慮導入 RAG 技術，歡迎與我們聯繫討論您的需求。