┌─────────────────────────────────────────────────────────────┐
│                    🔍 检索层                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  语义检索   │  │  关键词检索 │  │  混合检索   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    🧠 向量层                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  文档向量   │  │  查询向量   │  │  相似度计算 │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    📚 存储层                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  向量数据库 │  │  元数据库   │  │  文件存储   │        │
│  │ (ChromaDB)  │  │ (SQLite)    │  │ (本地/云端) │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    📄 文档层                                │
│        需求文档 + API文档 + 测试用例 + 经验总结             │
└─────────────────────────────────────────────────────────────┘

技术栈选择：最佳工具组合

核心框架：

LlamaIndex：专业的RAG框架，功能强大
ChromaDB：轻量级向量数据库，部署简单
Sentence-Transformers：高质量的文本向量化

文档处理：

Docling：IBM出品，支持PDF、Word等格式
Unstructured：处理各种非结构化文档
PyPDF2/pdfplumber：PDF文档解析

向量模型：

text-embedding-3-small：OpenAI的嵌入模型
bge-large-zh-v1.5：中文友好的开源模型
m3e-base：轻量级中文嵌入模型

📚 知识库设计实战

1. 文档分类体系

设计一个清晰的知识分类体系：

from enum import Enum
from typing import Dict, List
from pydantic import BaseModel

class DocumentType(Enum):
    """文档类型枚举"""
    REQUIREMENT = "requirement"      # 需求文档
    API_DOC = "api_doc"             # API文档
    TEST_CASE = "test_case"         # 测试用例
    TEST_PLAN = "test_plan"         # 测试计划
    BUG_REPORT = "bug_report"       # 缺陷报告
    BEST_PRACTICE = "best_practice" # 最佳实践
    KNOWLEDGE_BASE = "knowledge"    # 知识库文章

class DocumentMetadata(BaseModel):
    """文档元数据"""
    doc_id: str
    title: str
    doc_type: DocumentType
    project: str
    version: str
    author: str
    created_at: str
    updated_at: str
    tags: List[str]
    summary: str
    file_path: str
    
class KnowledgeCategory:
    """知识分类管理"""
    
    def __init__(self):
        self.categories = {
            "项目文档": {
                "需求文档": ["功能需求", "非功能需求", "业务规则"],
                "设计文档": ["架构设计", "接口设计", "数据库设计"],
                "API文档": ["接口规范", "参数说明", "示例代码"]
            },
            "测试资产": {
                "测试用例": ["功能测试", "性能测试", "安全测试"],
                "测试计划": ["测试策略", "测试范围", "资源安排"],
                "测试报告": ["执行报告", "缺陷报告", "总结报告"]
            },
            "经验知识": {
                "最佳实践": ["测试方法", "工具使用", "流程优化"],
                "踩坑记录": ["常见问题", "解决方案", "预防措施"],
                "技术分享": ["新技术", "工具推荐", "学习资料"]
            }
        }

2. 文档处理引擎

构建一个强大的文档处理引擎：

from llama_index.core import Document, SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter
from docling.document_converter import DocumentConverter
import os
from typing import List, Optional

class DocumentProcessor:
    """文档处理引擎 - 文档的'消化系统'"""
    
    def __init__(self):
        self.converter = DocumentConverter()
        self.splitter = SentenceSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separator=" "
        )
    
    async def process_file(self, file_path: str, metadata: DocumentMetadata) -> List[Document]:
        """处理单个文件"""
        try:
            # 根据文件类型选择处理方式
            if file_path.endswith('.pdf'):
                content = await self._process_pdf(file_path)
            elif file_path.endswith(('.doc', '.docx')):
                content = await self._process_word(file_path)
            elif file_path.endswith('.md'):
                content = await self._process_markdown(file_path)
            else:
                content = await self._process_text(file_path)
            
            # 创建文档对象
            document = Document(
                text=content,
                metadata={
                    "doc_id": metadata.doc_id,
                    "title": metadata.title,
                    "doc_type": metadata.doc_type.value,
                    "project": metadata.project,
                    "file_path": file_path,
                    "tags": metadata.tags
                }
            )
            
            # 文档分块
            nodes = self.splitter.get_nodes_from_documents([document])
            
            return nodes
            
        except Exception as e:
            print(f"处理文件 {file_path} 失败: {e}")
            return []
    
    async def _process_pdf(self, file_path: str) -> str:
        """处理PDF文档"""
        try:
            # 使用Docling处理PDF
            result = self.converter.convert(file_path)
            return result.document.export_to_markdown()
        except Exception as e:
            # 降级到PyPDF2
            import PyPDF2
            with open(file_path, 'rb') as file:
                reader = PyPDF2.PdfReader(file)
                text = ""
                for page in reader.pages:
                    text += page.extract_text()
                return text
    
    async def _process_word(self, file_path: str) -> str:
        """处理Word文档"""
        try:
            result = self.converter.convert(file_path)
            return result.document.export_to_markdown()
        except Exception as e:
            # 降级到python-docx
            from docx import Document
            doc = Document(file_path)
            return "\n".join([paragraph.text for paragraph in doc.paragraphs])
    
    async def _process_markdown(self, file_path: str) -> str:
        """处理Markdown文档"""
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read()
    
    async def _process_text(self, file_path: str) -> str:
        """处理纯文本文档"""
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read()
    
    def extract_metadata_from_content(self, content: str, file_path: str) -> dict:
        """从内容中提取元数据"""
        metadata = {
            "word_count": len(content.split()),
            "char_count": len(content),
            "file_size": os.path.getsize(file_path),
            "language": self._detect_language(content)
        }
        
        # 提取关键词
        keywords = self._extract_keywords(content)
        metadata["keywords"] = keywords
        
        return metadata
    
    def _detect_language(self, content: str) -> str:
        """检测文档语言"""
        # 简单的语言检测逻辑
        chinese_chars = len([c for c in content if '\u4e00' <= c <= '\u9fff'])
        total_chars = len(content)
        
        if chinese_chars / total_chars > 0.3:
            return "zh"
        else:
            return "en"
    
    def _extract_keywords(self, content: str, top_k: int = 10) -> List[str]:
        """提取关键词"""
        # 这里可以使用jieba、NLTK等工具进行关键词提取
        # 简化实现
        words = content.split()
        word_freq = {}
        for word in words:
            if len(word) > 2:  # 过滤短词
                word_freq[word] = word_freq.get(word, 0) + 1
        
        # 返回频率最高的词
        sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
        return [word for word, freq in sorted_words[:top_k]]

3. 向量存储系统

构建高效的向量存储和检索系统：

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
import chromadb
from typing import List, Dict, Any

class VectorStoreManager:
    """向量存储管理器 - 知识的'索引系统'"""
    
    def __init__(self, persist_directory: str = "./chroma_db"):
        self.persist_directory = persist_directory
        
        # 初始化ChromaDB
        self.chroma_client = chromadb.PersistentClient(path=persist_directory)
        
        # 初始化嵌入模型
        self.embed_model = OpenAIEmbedding(
            model="text-embedding-3-small",
            api_key="your-api-key"
        )
        
        # 创建不同类型的集合
        self.collections = {
            "requirements": self._get_or_create_collection("requirements"),
            "api_docs": self._get_or_create_collection("api_docs"),
            "test_cases": self._get_or_create_collection("test_cases"),
            "best_practices": self._get_or_create_collection("best_practices")
        }
    
    def _get_or_create_collection(self, name: str):
        """获取或创建集合"""
        try:
            return self.chroma_client.get_collection(name)
        except:
            return self.chroma_client.create_collection(
                name=name,
                metadata={"hnsw:space": "cosine"}
            )
    
    async def add_documents(self, documents: List[Document], collection_name: str = "default"):
        """添加文档到向量库"""
        try:
            # 选择集合
            if collection_name not in self.collections:
                self.collections[collection_name] = self._get_or_create_collection(collection_name)
            
            collection = self.collections[collection_name]
            
            # 创建向量存储
            vector_store = ChromaVectorStore(chroma_collection=collection)
            storage_context = StorageContext.from_defaults(vector_store=vector_store)
            
            # 创建索引
            index = VectorStoreIndex.from_documents(
                documents,
                storage_context=storage_context,
                embed_model=self.embed_model
            )
            
            print(f"成功添加 {len(documents)} 个文档到集合 {collection_name}")
            return index
            
        except Exception as e:
            print(f"添加文档失败: {e}")
            return None
    
    async def search_similar_documents(
        self, 
        query: str, 
        collection_name: str = "default",
        top_k: int = 5,
        filters: Dict[str, Any] = None
    ) -> List[Dict]:
        """搜索相似文档"""
        try:
            collection = self.collections.get(collection_name)
            if not collection:
                return []
            
            # 生成查询向量
            query_embedding = self.embed_model.get_text_embedding(query)
            
            # 构建查询条件
            where_clause = {}
            if filters:
                where_clause.update(filters)
            
            # 执行搜索
            results = collection.query(
                query_embeddings=[query_embedding],
                n_results=top_k,
                where=where_clause if where_clause else None
            )
            
            # 格式化结果
            formatted_results = []
            for i in range(len(results['documents'][0])):
                formatted_results.append({
                    "content": results['documents'][0][i],
                    "metadata": results['metadatas'][0][i],
                    "score": 1 - results['distances'][0][i],  # 转换为相似度分数
                    "id": results['ids'][0][i]
                })
            
            return formatted_results
            
        except Exception as e:
            print(f"搜索失败: {e}")
            return []
    
    def get_collection_stats(self) -> Dict[str, int]:
        """获取集合统计信息"""
        stats = {}
        for name, collection in self.collections.items():
            try:
                stats[name] = collection.count()
            except:
                stats[name] = 0
        return stats

4. 智能检索引擎

构建智能的检索和问答系统：

from llama_index.core import QueryBundle
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor

class IntelligentRetriever:
    """智能检索引擎 - 知识的'搜索专家'"""
    
    def __init__(self, vector_store_manager: VectorStoreManager):
        self.vector_store = vector_store_manager
        self.query_engines = {}
    
    async def setup_query_engine(self, collection_name: str, index):
        """设置查询引擎"""
        # 创建检索器
        retriever = VectorIndexRetriever(
            index=index,
            similarity_top_k=10
        )
        
        # 创建后处理器
        postprocessor = SimilarityPostprocessor(similarity_cutoff=0.7)
        
        # 创建查询引擎
        query_engine = RetrieverQueryEngine(
            retriever=retriever,
            node_postprocessors=[postprocessor]
        )
        
        self.query_engines[collection_name] = query_engine
    
    async def intelligent_search(
        self, 
        query: str, 
        collection_name: str = "default",
        search_type: str = "hybrid"
    ) -> Dict[str, Any]:
        """智能搜索"""
        
        if search_type == "semantic":
            return await self._semantic_search(query, collection_name)
        elif search_type == "keyword":
            return await self._keyword_search(query, collection_name)
        else:  # hybrid
            return await self._hybrid_search(query, collection_name)
    
    async def _semantic_search(self, query: str, collection_name: str) -> Dict[str, Any]:
        """语义搜索"""
        results = await self.vector_store.search_similar_documents(
            query=query,
            collection_name=collection_name,
            top_k=5
        )
        
        return {
            "search_type": "semantic",
            "query": query,
            "results": results,
            "total_found": len(results)
        }
    
    async def _keyword_search(self, query: str, collection_name: str) -> Dict[str, Any]:
        """关键词搜索"""
        # 提取关键词
        keywords = self._extract_query_keywords(query)
        
        # 基于关键词过滤
        filters = {"keywords": {"$in": keywords}} if keywords else None
        
        results = await self.vector_store.search_similar_documents(
            query=query,
            collection_name=collection_name,
            top_k=5,
            filters=filters
        )
        
        return {
            "search_type": "keyword",
            "query": query,
            "keywords": keywords,
            "results": results,
            "total_found": len(results)
        }
    
    async def _hybrid_search(self, query: str, collection_name: str) -> Dict[str, Any]:
        """混合搜索"""
        # 执行语义搜索
        semantic_results = await self._semantic_search(query, collection_name)
        
        # 执行关键词搜索
        keyword_results = await self._keyword_search(query, collection_name)
        
        # 合并和重排序结果
        combined_results = self._merge_search_results(
            semantic_results["results"],
            keyword_results["results"]
        )
        
        return {
            "search_type": "hybrid",
            "query": query,
            "results": combined_results,
            "total_found": len(combined_results),
            "semantic_count": len(semantic_results["results"]),
            "keyword_count": len(keyword_results["results"])
        }
    
    def _extract_query_keywords(self, query: str) -> List[str]:
        """从查询中提取关键词"""
        # 简化的关键词提取
        import re
        words = re.findall(r'\w+', query.lower())
        # 过滤停用词
        stop_words = {'的', '是', '在', '有', '和', '与', '或', '但', '如何', '什么', '怎么'}
        keywords = [word for word in words if word not in stop_words and len(word) > 1]
        return keywords
    
    def _merge_search_results(self, semantic_results: List, keyword_results: List) -> List:
        """合并搜索结果"""
        # 使用字典去重，保持顺序
        merged = {}
        
        # 语义搜索结果权重更高
        for result in semantic_results:
            doc_id = result.get("id")
            if doc_id:
                result["score"] = result["score"] * 1.2  # 语义搜索加权
                merged[doc_id] = result
        
        # 添加关键词搜索结果
        for result in keyword_results:
            doc_id = result.get("id")
            if doc_id and doc_id not in merged:
                merged[doc_id] = result
        
        # 按分数排序
        sorted_results = sorted(merged.values(), key=lambda x: x["score"], reverse=True)
        return sorted_results[:10]  # 返回前10个结果

🎯 实战案例：构建项目知识库

让我们看一个完整的实战案例：

async def build_project_knowledge_base():
    """构建项目知识库的完整流程"""
    
    # 1. 初始化系统
    processor = DocumentProcessor()
    vector_manager = VectorStoreManager("./project_kb")
    retriever = IntelligentRetriever(vector_manager)
    
    # 2. 准备文档
    documents_to_process = [
        {
            "file_path": "./docs/requirements.pdf",
            "metadata": DocumentMetadata(
                doc_id="req_001",
                title="用户管理系统需求文档",
                doc_type=DocumentType.REQUIREMENT,
                project="user_system",
                version="v1.0",
                author="产品经理",
                created_at="2024-01-01",
                updated_at="2024-01-15",
                tags=["用户管理", "权限", "登录"],
                summary="用户管理系统的详细需求说明",
                file_path="./docs/requirements.pdf"
            )
        },
        {
            "file_path": "./docs/api_spec.md",
            "metadata": DocumentMetadata(
                doc_id="api_001",
                title="用户管理API接口文档",
                doc_type=DocumentType.API_DOC,
                project="user_system",
                version="v1.0",
                author="后端工程师",
                created_at="2024-01-10",
                updated_at="2024-01-20",
                tags=["API", "接口", "用户"],
                summary="用户管理相关的API接口规范",
                file_path="./docs/api_spec.md"
            )
        }
    ]
    
    # 3. 处理文档
    all_documents = []
    for doc_info in documents_to_process:
        documents = await processor.process_file(
            doc_info["file_path"],
            doc_info["metadata"]
        )
        all_documents.extend(documents)
    
    # 4. 构建向量索引
    requirements_docs = [doc for doc in all_documents if doc.metadata["doc_type"] == "requirement"]
    api_docs = [doc for doc in all_documents if doc.metadata["doc_type"] == "api_doc"]
    
    req_index = await vector_manager.add_documents(requirements_docs, "requirements")
    api_index = await vector_manager.add_documents(api_docs, "api_docs")
    
    # 5. 设置查询引擎
    await retriever.setup_query_engine("requirements", req_index)
    await retriever.setup_query_engine("api_docs", api_index)
    
    # 6. 测试检索
    test_queries = [
        "用户登录的业务规则是什么？",
        "用户注册API的参数有哪些？",
        "密码复杂度要求是什么？"
    ]
    
    for query in test_queries:
        print(f"\n🔍 查询: {query}")
        
        # 在需求文档中搜索
        req_results = await retriever.intelligent_search(query, "requirements")
        print(f"📋 需求文档结果: {len(req_results['results'])} 条")
        
        # 在API文档中搜索
        api_results = await retriever.intelligent_search(query, "api_docs")
        print(f"🔌 API文档结果: {len(api_results['results'])} 条")
    
    # 7. 输出统计信息
    stats = vector_manager.get_collection_stats()
    print(f"\n📊 知识库统计:")
    for collection, count in stats.items():
        print(f"  {collection}: {count} 个文档")

# 运行示例
if __name__ == "__main__":
    import asyncio
    asyncio.run(build_project_knowledge_base())

🎉 总结

RAG知识库系统是AI测试平台的"记忆中枢"，通过合理的设计和实现，我们可以让AI助手：

记住项目信息：永久保存项目文档和经验
精准检索知识：快速找到相关的背景信息
提供个性化回答：基于项目特点给出针对性建议
持续学习优化：不断积累和完善知识库

下一篇我们将学习AI对话系统的开发，让用户能够自然地与知识库进行交互！

💡 RAG开发小贴士：构建知识库就像整理图书馆，分类要清晰，索引要准确，检索要快速。记住，好的知识库不是存得多，而是找得准！