从零搭建 | 硅基 AGI · 智能体学习与测评

引言理论讲了很多，但真正从零搭建一个生产级RAG系统，需要考虑很多工程细节。本文将带你从零开始，一步步搭建一个完整的RAG系统。一、系统架构 ┌─────────────────────────────────────────────┐ │ 用户接口 │ ├─────────────────────────────────────────────┤ │ API网关 │ ├──────────┬──────────┬───────────────────────┤ │ 查询处理 │ 检索引擎 │ 生成引擎 │ ├──────────┴──────────┴───────────────────────┤ │ 数据处理流水线 │ ├─────────────────────────────────────────────┤ │ 向量数据库 │ 文档存储 │ 缓存层 │ └─────────────────────────────────────────────┘ 二、技术选型 # 2026年推荐技术栈 tech_stack = { "embedding_model": "text-embedding-3-large", # 或BGE-large-zh "vector_db": "Qdrant", # 或Milvus "reranker": "bge-reranker-large", "llm": "GPT-4o-mini", # 或开源模型 "framework": "LangChain", # 或LlamaIndex "cache": "Redis", "document_store": "PostgreSQL", } 三、实现 3.1 文档处理 class DocumentProcessor: def __init__(self): self.chunker = RecursiveChunker(max_tokens=500) self.embedder = EmbeddingModel("text-embedding-3-large") async def process(self, documents): chunks = [] for doc in documents: # 1. 解析文档 text = await self.parse(doc) # 2. 分块 doc_chunks = self.chunker.chunk(text) # 3. 添加元数据 for i, chunk in enumerate(doc_chunks): chunks.append({ "id": f"{doc.id}-chunk-{i}", "text": chunk, "embedding": await self.embedder.embed(chunk), "metadata": { "doc_id": doc.id, "doc_title": doc.title, "chunk_index": i, "source": doc.source } }) return chunks 3.2 检索引擎 class RetrievalEngine: def __init__(self): self.vector_store = QdrantClient(host="localhost", port=6333) self.reranker = Reranker("bge-reranker-large") async def search(self, query, top_k=20, rerank_top_k=5): # 1. 向量检索 query_embedding = await self.embedder.embed(query) results = self.vector_store.search( collection_name="documents", query_vector=query_embedding, limit=top_k ) # 2. 重排序 reranked = await self.reranker.rerank(query, results, top_k=rerank_top_k) return reranked 3.3 生成引擎 class GenerationEngine: def __init__(self): self.llm = LLM("gpt-4o-mini") self.cache = RedisCache() async def generate(self, query, retrieved_docs): # 1. 检查缓存 cache_key = hash(query + str([d.id for d in retrieved_docs])) cached = await self.cache.get(cache_key) if cached: return cached # 2. 构建prompt context = self.format_context(retrieved_docs) prompt = f""" 基于以下参考信息回答问题。如果参考信息不足以回答，请说明。参考信息: {context} 问题: {query} 回答: """ # 3. 生成 answer = await self.llm.generate(prompt) # 4. 缓存 await self.cache.set(cache_key, answer, ttl=3600) return answer 3.4 完整系统 class RAGSystem: def __init__(self): self.processor = DocumentProcessor() self.retriever = RetrievalEngine() self.generator = GenerationEngine() async def ingest(self, documents): """导入文档""" chunks = await self.processor.process(documents) await self.retriever.vector_store.upsert(chunks) async def query(self, question): """查询""" # 1. 检索 docs = await self.retriever.search(question) # 2. 生成 answer = await self.generator.generate(question, docs) return { "answer": answer, "sources": [{"title": d.metadata["doc_title"], "text": d.text[:200]} for d in docs] } 四、优化 4.1 性能优化 # 1. 缓存热门查询 # 2. 预计算embedding # 3. 并行检索 # 4. 流式生成 async def query_stream(self, question): docs = await self.retriever.search(question) async for token in self.generator.generate_stream(question, docs): yield token 4.2 质量优化 # 1. 查询改写 query_rewritten = await self.rewrite_query(question) # 2. 多路检索 vector_results = await self.vector_search(query) keyword_results = await self.keyword_search(query) fused = self.fuse(vector_results, keyword_results) # 3. 自适应检索 if self.needs_multi_hop(question): docs = await self.multi_hop_retrieve(question) else: docs = await self.simple_retrieve(question) 五、部署 # docker-compose.yml version: '3.8' services: rag-api: build: . ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} - QDRANT_URL=http://qdrant:6333 - REDIS_URL=redis://redis:6379 depends_on: - qdrant - redis qdrant: image: qdrant/qdrant:latest ports: - "6333:6333" volumes: - qdrant_data:/qdrant/storage redis: image: redis:7-alpine ports: - "6379:6379" volumes: qdrant_data: 六、监控 # 关键监控指标 metrics = { "query_latency_p50": "中位查询延迟", "query_latency_p99": "99%查询延迟", "retrieval_accuracy": "检索准确率", "answer_quality": "回答质量评分", "cache_hit_rate": "缓存命中率", "error_rate": "错误率", "token_cost": "Token消耗" } 结语搭建一个RAG系统不难，但搭建一个生产级RAG系统需要考虑很多细节——分块策略、检索质量、生成质量、缓存、监控、成本控制。 ...