从零构建生产级RAG服务--AI大模型应用开发
# 1. 为什么RAG需要“应用开发”思维 大模型API的调用门槛已经极低，但真正的落地挑战在于：**如何将模型与私有数据、业务逻辑、运维体系结合**。RAG（检索增强生成）是目前最成熟的范式，但多数demo止步于Notebook。本文将展示一个可部署、可观测、可扩展的RAG服务骨架，技术栈选用 **LangChain**（编排）、**FastAPI**（接口）、**Qdrant**（向量库）、**OpenAI**（生成与嵌入，可替换为本地模型）。 --- ## 2. 系统架构与核心流程 ``` 用户查询 → API网关 → 检索器（向量相似度+重排序）→ 上下文组装 → LLM生成 → 流式响应 ↑ 文档切片（递归字符分割）→ 向量化 → 存入Qdrant ``` 关键设计决策： - **异步非阻塞**：FastAPI + async/await，适配高并发I/O - **多级召回**：向量检索先取Top-K，再用交叉编码器重排（提升精度） - **流式输出**：SSE（Server-Sent Events）降低首字延迟 --- ## 3. 环境准备与依赖 ``` pip install langchain langchain-openai qdrant-client fastapi uvicorn sse-starlette ``` 项目结构： ``` rag_service/ ├── config.py # 环境变量与模型参数 ├── vector_store.py # Qdrant初始化与CRUD ├── retriever.py # 检索+重排逻辑 ├── chain.py # 构建RAG链 ├── main.py # FastAPI路由与事件循环 └── requirements.txt ``` --- ## 4. 核心代码实现 ### 4.1 向量存储封装（`vector_store.py`） 使用Qdrant的本地模式（也可切换为云端）。重点在于**批量写入与索引优化**。 ``` from qdrant_client import QdrantClient, models from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class VectorStore: def __init__(self, collection_name: str = "doc_chunks"): self.client = QdrantClient(path="./qdrant_data") # 持久化本地 self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small") self.collection = collection_name self._ensure_collection() def _ensure_collection(self): if not self.client.collection_exists(self.collection): self.client.create_collection( collection_name=self.collection, vectors_config=models.VectorParams( size=1536, # text-embedding-3-small 维度 distance=models.Distance.COSINE, ), optimizers_config=models.OptimizersConfigDiff( default_segment_number=2, # 加速构建 ), ) def add_documents(self, texts: list[str], metadatas: list[dict] = None): """切片+向量化+入库""" splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", " ", ""] ) chunks = splitter.create_documents(texts, metadatas) points = [] for idx, chunk in enumerate(chunks): vector = self.embeddings.embed_query(chunk.page_content) points.append( models.PointStruct( id=idx, # 生产环境建议UUID vector=vector, payload={"text": chunk.page_content, **chunk.metadata} ) ) self.client.upsert(collection_name=self.collection, points=points) def search(self, query: str, k: int = 10) -> list[dict]: q_vec = self.embeddings.embed_query(query) hits = self.client.search( collection_name=self.collection, query_vector=q_vec, limit=k, with_payload=True, ) return [{"text": hit.payload["text"], "score": hit.score} for hit in hits] ``` ### 4.2 检索增强链（`chain.py`） 引入**重排序**（用交叉编码器Cohere/OpenAI，这里模拟为按分数二次排序）和**提示模板**。 ``` from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import RunnablePassthrough from vector_store import VectorStore class RAGChain: def __init__(self, store: VectorStore): self.store = store self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, streaming=True) self.prompt = ChatPromptTemplate.from_template(""" 你是一个专业的助手。请基于以下上下文回答用户问题。 如果上下文不足以回答，请坦诚说“不知道”。 上下文： {context} 问题：{question} 回答： """) # 构建LCEL链 self.chain = ( {"context": self._retrieve, "question": RunnablePassthrough()} | self.prompt | self.llm ) def _retrieve(self, inputs: dict) -> str: question = inputs["question"] raw = self.store.search(question, k=10) # 重排序：这里简易按分数降序，实际可调用专门的reranker API ranked = sorted(raw, key=lambda x: x["score"], reverse=True)[:5] return "\n\n".join([f"- {item['text']}" for item in ranked]) async def astream(self, question: str): async for chunk in self.chain.astream(question): yield chunk.content # 流式输出文本片段 ``` ### 4.3 FastAPI 服务（`main.py`） 集成流式响应与健康检查。注意**启动时加载向量库**，避免每次请求重连。 ``` from fastapi import FastAPI, Request from sse_starlette.sse import EventSourceResponse from contextlib import asynccontextmanager from vector_store import VectorStore from chain import RAGChain store = None chain = None @asynccontextmanager async def lifespan(app: FastAPI): global store, chain store = VectorStore() # 示例：初始化时注入一些文档（实际可从外部加载） store.add_documents( ["LangChain是构建LLM应用的框架。", "Qdrant是高性能向量数据库。"], [{"source": "doc1"}, {"source": "doc2"}] ) chain = RAGChain(store) yield # 清理资源（可选） app = FastAPI(lifespan=lifespan) @app.get("/health") async def health(): return {"status": "ok"} @app.post("/ask") async def ask(request: Request): data = await request.json() question = data.get("question", "") if not question: return {"error": "missing question"} async def event_generator(): async for token in chain.astream(question): yield {"event": "message", "data": token} yield {"event": "done", "data": "[END]"} return EventSourceResponse(event_generator()) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` --- ## 5. 性能优化与观测性 ### 5.1 缓存层 对高频查询（如FAQ）使用Redis缓存完整生成结果，TTL=3600s。在`_retrieve`前加一层`@lru_cache`不适合分布式，改用`aiocache`。 ### 5.2 异步批处理 若文档量极大，使用`asyncio.gather`并发调用嵌入API（注意速率限制）。 ### 5.3 监控指标 添加`prometheus-fastapi-instrumentator`记录请求耗时、检索召回率（通过日志分析）。关键指标：**首字延迟（TTFT）**和**生成吞吐量**。 --- ## 6. 测试与调用示例 启动服务后，使用curl测试流式效果： ``` curl -N -X POST http://localhost:8000/ask \ -H "Content-Type: application/json" \ -d '{"question":"LangChain有什么用？"}' ``` 应返回SSE数据流，逐步输出生成内容。 --- ## 7. 扩展与替代方案 - **向量库**：可换为Milvus或Pgvector（需调整配置） - **嵌入与LLM**：本地化使用`llama.cpp`或`vLLM`，替换`OpenAI`类即可 - **重排序**：集成`CohereRerank`或`SentenceTransformer`交叉编码器提升精度 - **权限隔离**：为不同租户创建独立collection，实现多租户 --- ## 8. 总结 本文提供了一个**生产就绪的RAG服务原型**，核心代码约200行，但涵盖了连接管理、异步流式、向量检索与重排序等关键环节。真正的工业级系统还需补充**灰度发布、回退策略、成本控制（Token计数）**等模块。但此骨架足以作为团队内部POC的起点，并可平滑演进至分布式架构。 **关键教训**：不要忽略检索质量——80%的RAG效果取决于切片策略和向量模型，而非LLM本身。建议在`add_documents`中增加多粒度切片（段落+句子），并利用用户反馈数据微调重排序器。
