# 基于RAG的中土世界知識問答系統(tǒng)構(gòu)建實踐
## 一、項目概述與核心架構(gòu)設(shè)計
### 項目背景與目標(biāo)
在《魔戒》、《霍比特人》等作品構(gòu)建的中土世界，存在著龐大而復(fù)雜的知識體系。傳統(tǒng)的問答系統(tǒng)難以準(zhǔn)確理解這個奇幻世界的獨特術(shù)語、地理關(guān)系和歷史脈絡(luò)。本文將探討如何利用檢索增強生成技術(shù)，構(gòu)建一個專門針對中土世界知識的智能問答系統(tǒng)。
### 系統(tǒng)架構(gòu)概覽
```
中土世界知識問答系統(tǒng)架構(gòu)：
├── 知識庫構(gòu)建層
│? ├── 數(shù)據(jù)采集與預(yù)處理
│? ├── 文本分割與向量化
│? └── 向量數(shù)據(jù)庫存儲
├── 檢索增強層
│? ├── 語義檢索模塊
│? ├── 重排序模塊
│? └── 上下文構(gòu)建模塊
└── 生成推理層
? ? ├── LLM集成接口
? ? ├── 提示詞工程
? ? └── 響應(yīng)生成與驗證
```
### 技術(shù)棧選型
```python
# requirements.txt 核心依賴
langchain==0.1.0
chromadb==0.4.22
sentence-transformers==2.2.2
faiss-cpu==1.7.4
openai==1.3.0? # 或其他本地LLM
pydantic==2.5.0
fastapi==0.104.0
```
## 二、中土世界知識庫構(gòu)建
### 數(shù)據(jù)源收集與處理
中土世界知識數(shù)據(jù)主要來源于多個權(quán)威渠道：
```python
import json
from typing import List, Dict
from dataclasses import dataclass
@dataclass
class MiddleEarthDocument:
? ? """中土世界文檔數(shù)據(jù)結(jié)構(gòu)"""
? ? title: str
? ? content: str
? ? category: str? # 如: 種族、地理、歷史、人物
? ? source: str
? ? metadata: Dict
class KnowledgeCollector:
? ? def __init__(self):
? ? ? ? self.documents: List[MiddleEarthDocument] = []
? ? def load_tolkien_texts(self, file_path: str):
? ? ? ? """加載托爾金原著文本"""
? ? ? ? with open(file_path, 'r', encoding='utf-8') as f:
? ? ? ? ? ? texts = json.load(f)
? ? ? ? for text in texts:
? ? ? ? ? ? doc = MiddleEarthDocument(
? ? ? ? ? ? ? ? title=text.get('title', ''),
? ? ? ? ? ? ? ? content=text['content'],
? ? ? ? ? ? ? ? category=text.get('category', 'literature'),
? ? ? ? ? ? ? ? source='tolkien_original',
? ? ? ? ? ? ? ? metadata={
? ? ? ? ? ? ? ? ? ? 'book': text.get('book', ''),
? ? ? ? ? ? ? ? ? ? 'chapter': text.get('chapter', ''),
? ? ? ? ? ? ? ? ? ? 'era': text.get('era', 'Third Age')
? ? ? ? ? ? ? ? }
? ? ? ? ? ? )
? ? ? ? ? ? self.documents.append(doc)
? ? def load_wiki_data(self, wiki_dump_path: str):
? ? ? ? """加載中土世界維基數(shù)據(jù)"""
? ? ? ? # 解析維基格式，提取結(jié)構(gòu)化信息
? ? ? ? pass
```
### 文本分割與處理策略
考慮到中土世界文本的特殊性，需要定制化的分割策略：
```python
from langchain.text_splitter import RecursiveCharacterTextSplitter
class MiddleEarthTextSplitter:
? ? def __init__(self):|6H.E8P.HK|BV.R6T.HK|LZ.P8H.HK
? ? ? ? # 中土世界特有的分隔符
? ? ? ? self.separators = [
? ? ? ? ? ? "\n## ",? # 章節(jié)標(biāo)題
? ? ? ? ? ? "\n### ",
? ? ? ? ? ? "\n\n",
? ? ? ? ? ? "。",
? ? ? ? ? ? ".",
? ? ? ? ? ? " ",
? ? ? ? ? ? "",
? ? ? ? ]
? ? ? ? self.text_splitter = RecursiveCharacterTextSplitter(
? ? ? ? ? ? chunk_size=500,
? ? ? ? ? ? chunk_overlap=50,
? ? ? ? ? ? separators=self.separators,
? ? ? ? ? ? length_function=len,
? ? ? ? )
? ? def split_document(self, document: MiddleEarthDocument) -> List[str]:
? ? ? ? """分割中土世界文檔"""
? ? ? ? chunks = self.text_splitter.split_text(document.content)
? ? ? ? # 為每個塊添加元數(shù)據(jù)
? ? ? ? chunk_docs = []
? ? ? ? for i, chunk in enumerate(chunks):
? ? ? ? ? ? chunk_doc = {
? ? ? ? ? ? ? ? "content": chunk,
? ? ? ? ? ? ? ? "title": document.title,
? ? ? ? ? ? ? ? "category": document.category,
? ? ? ? ? ? ? ? "chunk_index": i,
? ? ? ? ? ? ? ? "source": document.source,
? ? ? ? ? ? ? ? **document.metadata
? ? ? ? ? ? }
? ? ? ? ? ? chunk_docs.append(chunk_doc)
? ? ? ? return chunk_docs
```
### 向量化與索引構(gòu)建
```python
import chromadb
from sentence_transformers import SentenceTransformer
from typing import List
class VectorStoreManager:
? ? def __init__(self, model_name: str = "paraphrase-multilingual-MiniLM-L12-v2"):
? ? ? ? self.embedding_model = SentenceTransformer(model_name)
? ? ? ? self.chroma_client = chromadb.PersistentClient(path="./middle_earth_db")
? ? def create_collection(self, collection_name: str = "middle_earth_knowledge"):
? ? ? ? """創(chuàng)建向量存儲集合"""
? ? ? ? self.collection = self.chroma_client.create_collection(
? ? ? ? ? ? name=collection_name,
? ? ? ? ? ? metadata={"description": "中土世界知識庫"}
? ? ? ? )
? ? def add_documents(self, documents: List[Dict]):
? ? ? ? """添加文檔到向量數(shù)據(jù)庫"""
? ? ? ? ids = []
? ? ? ? embeddings = []
? ? ? ? metadatas = []
? ? ? ? for i, doc in enumerate(documents):
? ? ? ? ? ? # 生成嵌入向量
? ? ? ? ? ? embedding = self.embedding_model.encode(doc["content"]).tolist()
? ? ? ? ? ? ids.append(f"doc_{i}")
? ? ? ? ? ? embeddings.append(embedding)
? ? ? ? ? ? metadatas.append({
? ? ? ? ? ? ? ? "title": doc.get("title", ""),
? ? ? ? ? ? ? ? "category": doc.get("category", ""),
? ? ? ? ? ? ? ? "source": doc.get("source", ""),
? ? ? ? ? ? ? ? **{k: v for k, v in doc.items() if k not in ["content", "title", "category", "source"]}
? ? ? ? ? ? })
? ? ? ? # 批量添加
? ? ? ? self.collection.add(
? ? ? ? ? ? embeddings=embeddings,
? ? ? ? ? ? metadatas=metadatas,
? ? ? ? ? ? ids=ids
? ? ? ? )
? ? def similarity_search(self, query: str, k: int = 5):
? ? ? ? """相似度搜索"""
? ? ? ? query_embedding = self.embedding_model.encode(query).tolist()
? ? ? ? results = self.collection.query(
? ? ? ? ? ? query_embeddings=[query_embedding],
? ? ? ? ? ? n_results=k,
? ? ? ? ? ? include=["metadatas", "documents", "distances"]
? ? ? ? )
? ? ? ? return results
```
## 三、RAG檢索增強機制實現(xiàn)
### 混合檢索策略
針對中土世界知識的特殊性，實施混合檢索策略：
```python
class HybridRetriever:
? ? def __init__(self, vector_store: VectorStoreManager):
? ? ? ? self.vector_store = vector_store
? ? ? ? self.keyword_index = {}? # 關(guān)鍵詞倒排索引
? ? def build_keyword_index(self, documents: List[Dict]):
? ? ? ? """構(gòu)建中土世界關(guān)鍵詞索引"""
? ? ? ? middle_earth_keywords = {
? ? ? ? ? ? "種族": ["霍比特人", "精靈", "矮人", "人類", "奧克", "恩特"],
? ? ? ? ? ? "地理": ["夏爾", "瑞文戴爾", "摩瑞亞", "剛鐸", "魔多"],
? ? ? ? ? ? "人物": ["弗羅多", "甘道夫", "阿拉貢", "索倫", "咕嚕"],
? ? ? ? ? ? "物品": ["魔戒", "刺叮劍", "敵擊劍", "精靈寶鉆"]
? ? ? ? }
? ? ? ? for doc in documents:
? ? ? ? ? ? doc_keywords = []
? ? ? ? ? ? for category, keywords in middle_earth_keywords.items():
? ? ? ? ? ? ? ? for keyword in keywords:
? ? ? ? ? ? ? ? ? ? if keyword in doc["content"]:
? ? ? ? ? ? ? ? ? ? ? ? doc_keywords.append(keyword)
? ? ? ? ? ? if doc_keywords:
? ? ? ? ? ? ? ? for keyword in doc_keywords:
? ? ? ? ? ? ? ? ? ? if keyword not in self.keyword_index:
? ? ? ? ? ? ? ? ? ? ? ? self.keyword_index[keyword] = []
? ? ? ? ? ? ? ? ? ? self.keyword_index[keyword].append(doc["title"])
? ? def retrieve(self, query: str, use_hybrid: bool = True):
? ? ? ? """混合檢索"""
? ? ? ? # 語義檢索
? ? ? ? semantic_results = self.vector_store.similarity_search(query, k=10)
? ? ? ? if use_hybrid:
? ? ? ? ? ? # 關(guān)鍵詞檢索增強
? ? ? ? ? ? keyword_docs = []
? ? ? ? ? ? for keyword in self.keyword_index.keys():
? ? ? ? ? ? ? ? if keyword in query:
? ? ? ? ? ? ? ? ? ? keyword_docs.extend(self.keyword_index[keyword])
? ? ? ? ? ? # 合并結(jié)果
? ? ? ? ? ? all_docs = self._merge_results(semantic_results, keyword_docs)
? ? ? ? ? ? return all_docs
? ? ? ? return semantic_results
? ? def _merge_results(self, semantic_results, keyword_docs):
? ? ? ? """合并檢索結(jié)果"""
? ? ? ? # 實現(xiàn)結(jié)果去重和重排序
? ? ? ? pass
```
### 上下文優(yōu)化與重排序
```python
class ContextOptimizer:
? ? def __init__(self):
? ? ? ? self.relevance_threshold = 0.7
? ? def optimize_context(self, retrieved_docs, query: str):
? ? ? ? """優(yōu)化檢索到的上下文"""
? ? ? ? # 1. 相關(guān)性過濾
? ? ? ? filtered_docs = [
? ? ? ? ? ? doc for doc in retrieved_docs
? ? ? ? ? ? if self._calculate_relevance(doc["content"], query) > self.relevance_threshold
? ? ? ? ]
? ? ? ? # 2. 多樣性選擇
? ? ? ? selected_docs = self._select_diverse_docs(filtered_docs, max_docs=5)
? ? ? ? # 3. 構(gòu)建上下文
? ? ? ? context = self._construct_context(selected_docs, query)
? ? ? ? return context
? ? def _calculate_relevance(self, doc_content: str, query: str) -> float:
? ? ? ? """計算文檔與查詢的相關(guān)性"""
? ? ? ? # 使用更復(fù)雜的相關(guān)性計算
? ? ? ? query_terms = set(query.split())
? ? ? ? doc_terms = set(doc_content.split())
? ? ? ? intersection = query_terms.intersection(doc_terms)
? ? ? ? return len(intersection) / len(query_terms) if query_terms else 0
? ? def _select_diverse_docs(self, docs: List[Dict], max_docs: int):
? ? ? ? """選擇多樣化的文檔"""
? ? ? ? selected = []
? ? ? ? categories_seen = set()
? ? ? ? for doc in docs:
? ? ? ? ? ? category = doc.get("category", "unknown")
? ? ? ? ? ? if category not in categories_seen:
? ? ? ? ? ? ? ? selected.append(doc)
? ? ? ? ? ? ? ? categories_seen.add(category)
? ? ? ? ? ? if len(selected) >= max_docs:
? ? ? ? ? ? ? ? break
? ? ? ? return selected
? ? def _construct_context(self, docs: List[Dict], query: str) -> str:
? ? ? ? """構(gòu)建LLM可理解的上下文"""
? ? ? ? context_parts = []
? ? ? ? for i, doc in enumerate(docs, 1):
? ? ? ? ? ? context_parts.append(
? ? ? ? ? ? ? ? f"[文檔 {i}] 標(biāo)題: {doc.get('title', '未知')}\n"
? ? ? ? ? ? ? ? f"分類: {doc.get('category', '未知')}\n"
? ? ? ? ? ? ? ? f"內(nèi)容: {doc['content'][:300]}...\n"
? ? ? ? ? ? )
? ? ? ? context = "\n".join(context_parts)
? ? ? ? return context
```
## 四、智能問答系統(tǒng)實現(xiàn)
### 提示詞工程
針對中土世界特點設(shè)計專用提示詞模板：
```python
class MiddleEarthPromptEngineer:
? ? def __init__(self):
? ? ? ? self.system_prompt = """你是中土世界知識專家，專門回答關(guān)于托爾金創(chuàng)造的奇幻世界的問題。
? ? ? ? 你的回答需要基于提供的中土世界資料，確保準(zhǔn)確性和一致性。
? ? ? ? 中土世界特點：
? ? ? ? 1. 遵循托爾金原著設(shè)定
? ? ? ? 2. 注意不同種族的特性差異
? ? ? ? 3. 考慮歷史時間線的準(zhǔn)確性
? ? ? ? 4. 使用正確的專有名詞和術(shù)語
? ? ? ? 請以專業(yè)但易懂的方式回答問題。"""
? ? def build_rag_prompt(self, query: str, context: str) -> str:
? ? ? ? """構(gòu)建RAG提示詞"""
? ? ? ? prompt = f"""{self.system_prompt}
? ? ? ? 相關(guān)背景資料：
? ? ? ? {context}
? ? ? ? 用戶問題：{query}
? ? ? ? 請根據(jù)以上資料回答問題。如果資料中沒有相關(guān)信息，請明確說明不清楚，不要編造信息。
? ? ? ? 回答時請：
? ? ? ? 1. 引用資料中的具體信息
? ? ? ? 2. 如果涉及不同來源有沖突，指出可能的差異
? ? ? ? 3. 保持回答結(jié)構(gòu)清晰
? ? ? ? 回答："""
? ? ? ? return prompt
```
### 問答系統(tǒng)主流程
```python
from typing import Optional
import openai? # 或使用本地LLM
class MiddleEarthQA:9K.E2C.HK|2G.W4E.HK|YX.E8P.HK|PM.R6T.HK
? ? def __init__(self,
? ? ? ? ? ? ? ? retriever: HybridRetriever,
? ? ? ? ? ? ? ? llm_config: Optional[Dict] = None):
? ? ? ? self.retriever = retriever
? ? ? ? self.context_optimizer = ContextOptimizer()
? ? ? ? self.prompt_engineer = MiddleEarthPromptEngineer()
? ? ? ? # 配置LLM
? ? ? ? if llm_config:
? ? ? ? ? ? openai.api_key = llm_config.get("api_key")
? ? ? ? ? ? self.llm_model = llm_config.get("model", "gpt-3.5-turbo")
? ? ? ? else:
? ? ? ? ? ? # 本地LLM配置
? ? ? ? ? ? self.llm_model = self._setup_local_llm()
? ? def ask(self, question: str) -> Dict:
? ? ? ? """問答主流程"""
? ? ? ? try:
? ? ? ? ? ? # 1. 檢索相關(guān)文檔
? ? ? ? ? ? retrieved_docs = self.retriever.retrieve(question)
? ? ? ? ? ? # 2. 優(yōu)化上下文
? ? ? ? ? ? context = self.context_optimizer.optimize_context(
? ? ? ? ? ? ? ? retrieved_docs, question
? ? ? ? ? ? )
? ? ? ? ? ? # 3. 構(gòu)建提示詞
? ? ? ? ? ? prompt = self.prompt_engineer.build_rag_prompt(question, context)
? ? ? ? ? ? # 4. 生成回答
? ? ? ? ? ? response = self._generate_response(prompt)
? ? ? ? ? ? # 5. 添加溯源信息
? ? ? ? ? ? response_with_sources = self._add_source_attribution(
? ? ? ? ? ? ? ? response, retrieved_docs
? ? ? ? ? ? )
? ? ? ? ? ? return {
? ? ? ? ? ? ? ? "answer": response_with_sources,
? ? ? ? ? ? ? ? "sources": retrieved_docs[:3],? # 顯示主要來源
? ? ? ? ? ? ? ? "context_used": context[:500]? # 顯示使用的上下文摘要
? ? ? ? ? ? }
? ? ? ? except Exception as e:
? ? ? ? ? ? return {
? ? ? ? ? ? ? ? "error": str(e),
? ? ? ? ? ? ? ? "answer": "抱歉，回答問題時出現(xiàn)錯誤。"
? ? ? ? ? ? }
? ? def _generate_response(self, prompt: str) -> str:
? ? ? ? """調(diào)用LLM生成回答"""
? ? ? ? if hasattr(self, 'llm_model'):
? ? ? ? ? ? # 使用OpenAI API
? ? ? ? ? ? response = openai.ChatCompletion.create(
? ? ? ? ? ? ? ? model=self.llm_model,
? ? ? ? ? ? ? ? messages=[
? ? ? ? ? ? ? ? ? ? {"role": "system", "content": self.prompt_engineer.system_prompt},
? ? ? ? ? ? ? ? ? ? {"role": "user", "content": prompt}
? ? ? ? ? ? ? ? ],
? ? ? ? ? ? ? ? temperature=0.3,
? ? ? ? ? ? ? ? max_tokens=1000
? ? ? ? ? ? )
? ? ? ? ? ? return response.choices[0].message.content
? ? ? ? else:
? ? ? ? ? ? # 本地LLM調(diào)用
? ? ? ? ? ? return self._call_local_llm(prompt)
? ? def _add_source_attribution(self, response: str, sources: List[Dict]) -> str:
? ? ? ? """在回答中添加來源標(biāo)注"""
? ? ? ? source_refs = []
? ? ? ? for i, source in enumerate(sources[:3], 1):
? ? ? ? ? ? source_refs.append(
? ? ? ? ? ? ? ? f"[{i}] {source.get('title', '未知標(biāo)題')} "
? ? ? ? ? ? ? ? f"({source.get('category', '未知分類')})"
? ? ? ? ? ? )
? ? ? ? if source_refs:
? ? ? ? ? ? response += f"\n\n---\n參考資料：\n" + "\n".join(source_refs)
? ? ? ? return response
```
## 五、系統(tǒng)部署與評估
### Web接口實現(xiàn)
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="中土世界知識問答系統(tǒng)")
class QuestionRequest(BaseModel):
? ? question: str
? ? use_hybrid_search: bool = True
class AnswerResponse(BaseModel):
? ? answer: str
? ? sources: List[Dict]
? ? processing_time: float
# 全局系統(tǒng)實例
qa_system = None
@app.on_event("startup")
async def startup_event():
? ? """系統(tǒng)啟動時初始化"""
? ? global qa_system
? ? # 初始化所有組件
? ? vector_store = VectorStoreManager()
? ? retriever = HybridRetriever(vector_store)
? ? qa_system = MiddleEarthQA(retriever)
@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
? ? """問答接口"""
? ? if not qa_system:
? ? ? ? raise HTTPException(status_code=503, detail="系統(tǒng)未就緒")
? ? import time
? ? start_time = time.time()
? ? result = qa_system.ask(request.question)
? ? processing_time = time.time() - start_time
? ? return AnswerResponse(
? ? ? ? answer=result.get("answer", ""),
? ? ? ? sources=result.get("sources", []),
? ? ? ? processing_time=processing_time
? ? )
@app.get("/health")
async def health_check():
? ? """健康檢查"""
? ? return {"status": "healthy", "service": "middle_earth_qa"}
```
### 評估指標(biāo)與測試
```python
class SystemEvaluator:
? ? def __init__(self, qa_system: MiddleEarthQA):
? ? ? ? self.qa_system = qa_system
? ? ? ? self.test_cases = [
? ? ? ? ? ? {
? ? ? ? ? ? ? ? "question": "霍比特人的特點是什么？",
? ? ? ? ? ? ? ? "expected_keywords": ["夏爾", "熱愛和平", "小個子"]
? ? ? ? ? ? },
? ? ? ? ? ? {
? ? ? ? ? ? ? ? "question": "魔戒的歷史是怎樣的？",
? ? ? ? ? ? ? ? "expected_keywords": ["索倫", "精靈", "鑄造", "銷毀"]
? ? ? ? ? ? },
? ? ? ? ? ? # 更多測試用例...
? ? ? ? ]
? ? def evaluate(self) -> Dict:
? ? ? ? """系統(tǒng)評估"""
? ? ? ? results = {
? ? ? ? ? ? "accuracy": 0,
? ? ? ? ? ? "relevance": 0,
? ? ? ? ? ? "completeness": 0,
? ? ? ? ? ? "response_time": 0
? ? ? ? }
? ? ? ? total_tests = len(self.test_cases)
? ? ? ? successful_tests = 0
? ? ? ? for test_case in self.test_cases:
? ? ? ? ? ? response = self.qa_system.ask(test_case["question"])
? ? ? ? ? ? # 檢查是否包含預(yù)期關(guān)鍵詞
? ? ? ? ? ? answer = response.get("answer", "").lower()
? ? ? ? ? ? expected_keywords = [kw.lower() for kw in test_case["expected_keywords"]]
? ? ? ? ? ? keyword_matches = sum(1 for kw in expected_keywords if kw in answer)
? ? ? ? ? ? keyword_coverage = keyword_matches / len(expected_keywords)
? ? ? ? ? ? if keyword_coverage > 0.5:
? ? ? ? ? ? ? ? successful_tests += 1
? ? ? ? results["accuracy"] = successful_tests / total_tests
? ? ? ? return results
```
## 六、擴展與應(yīng)用展望
### 多模態(tài)擴展
未來可擴展支持中土世界地圖、族譜圖譜、人物關(guān)系圖等多模態(tài)數(shù)據(jù)：
```python
class MultimodalRAG:
? ? def __init__(self):
? ? ? ? self.image_retriever = None? # 圖像檢索器
? ? ? ? self.map_analyzer = None? # 地圖分析器
? ? def process_map_question(self, question: str, map_image: bytes):
? ? ? ? """處理地圖相關(guān)問題"""
? ? ? ? # 結(jié)合文本和圖像信息進(jìn)行回答
? ? ? ? pass
```
### 對話歷史與上下文管理
```python
class ConversationManager:
? ? def __init__(self):
? ? ? ? self.conversation_history = []
? ? def manage_context(self, user_input: str, chat_history: List) -> str:
? ? ? ? """管理對話上下文"""
? ? ? ? # 分析歷史對話，提取相關(guān)信息
? ? ? ? # 優(yōu)化當(dāng)前查詢
? ? ? ? pass
```
通過上述架構(gòu)和實現(xiàn)，我們構(gòu)建了一個專門針對中土世界知識的智能問答系統(tǒng)。該系統(tǒng)能夠準(zhǔn)確理解復(fù)雜的奇幻世界設(shè)定，提供基于可靠來源的準(zhǔn)確回答。RAG技術(shù)的應(yīng)用確保了回答的準(zhǔn)確性和可追溯性，為托爾金作品的愛好者和研究者提供了一個有價值的工具。
這種垂直領(lǐng)域智能體的構(gòu)建方法，同樣可以應(yīng)用于其他特定領(lǐng)域的知識問答系統(tǒng)構(gòu)建，具有很好的可擴展性和借鑒意義。
