【RAG扫盲系列·3】从零开始构建你的RAG项目第二弹：API 调用大模型问答

在上一篇文章中，我们构建了基础的向量检索系统，将文档切分、向量化并存储到FAISS中。但检索只是RAG的一半，真正的核心在于如何让大模型理解检索到的上下文并生成准确回答。今天，我们将通过API调用大模型，实现完整的问答流程。### 1. 为什么需要API调用大模型？在实际生产环境中，我们通常不会本地部署大模型（成本高、维护复杂），而是通过云API调用成熟的商业模型，如OpenAI的GPT系列、Claude或国内的通义千问等。这样做的好处是：- 开箱即用：无需搭建GPU环境- 弹性伸缩：按需付费，支持高并发- 持续迭代：模型自动升级，保持最新能力- 专业服务：提供完善的API文档和SDK### 2. 准备工作：安装依赖与获取API密钥首先，我们需要安装OpenAI的Python库（也是兼容大多数API的标准接口）：bashpip install openai python-dotenv创建一个 .env 文件存储密钥（不要提交到版本控制）：OPENAI_API_KEY=sk-your-api-key-hereOPENAI_BASE_URL=https://api.openai.com/v1 # 可替换为其他兼容API的地址### 3. 第一个实战：单轮问答（无上下文）我们先从最基础的API调用开始，验证连接是否正常：pythonimport osfrom dotenv import load_dotenvfrom openai import OpenAI# 加载环境变量load_dotenv()# 初始化客户端client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"))def ask_without_context(question: str) -> str: """ 直接向大模型提问，不提供任何上下文 """ try: response = client.chat.completions.create( model="gpt-3.5-turbo", # 可替换为其他模型 messages=[ {"role": "system", "content": "你是一个知识渊博的助手，请用中文回答用户问题。"}, {"role": "user", "content": question} ], temperature=0.7, # 控制回答的创造性 max_tokens=500 # 限制回答长度 ) return response.choices[0].message.content except Exception as e: return f"API调用失败: {str(e)}"# 测试if __name__ == "__main__": result = ask_without_context("什么是RAG？") print(f"无上下文的回答:\n{result}\n")运行结果示例（实际输出取决于模型版本）：RAG（Retrieval-Augmented Generation）是一种结合检索和生成的AI技术...它通过先检索相关文档，再基于这些信息生成回答...可以看到，模型能回答一般性问题，但如果我们需要它回答特定公司内部的知识库问题，它就会“胡编乱造”。### 4. 第二个实战：RAG完整流程（检索+生成）现在我们把之前的向量检索和API调用结合起来，实现真正的RAG问答系统：pythonimport osimport jsonfrom typing import List, Dictfrom dotenv import load_dotenvfrom openai import OpenAIfrom sentence_transformers import SentenceTransformerimport faissimport numpy as np# 加载环境变量load_dotenv()class RAGSystem: """完整的RAG问答系统""" def __init__(self, embedding_model_name: str = "all-MiniLM-L6-v2"): # 初始化嵌入模型 self.embedding_model = SentenceTransformer(embedding_model_name) # 初始化OpenAI客户端 self.client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL") ) # 存储文档和索引的容器 self.documents: List[str] = [] self.index: faiss.Index = None def add_documents(self, docs: List[str]): """ 添加文档到知识库 """ # 生成所有文档的向量 embeddings = self.embedding_model.encode(docs, convert_to_numpy=True) # 初始化或合并到FAISS索引 if self.index is None: dimension = embeddings.shape[1] self.index = faiss.IndexFlatL2(dimension) self.index.add(embeddings) self.documents.extend(docs) print(f"已添加 {len(docs)} 个文档，当前总文档数: {len(self.documents)}") def retrieve_context(self, query: str, top_k: int = 3) -> List[str]: """ 检索与查询最相关的文档片段 """ # 将查询转换为向量 query_vector = self.embedding_model.encode([query], convert_to_numpy=True) # 执行检索（返回距离和索引） distances, indices = self.index.search(query_vector, top_k) # 提取对应的文档内容 context_docs = [] for idx in indices[0]: if idx != -1: # FAISS返回-1表示无效索引 context_docs.append(self.documents[idx]) return context_docs def generate_answer(self, question: str, context: List[str]) -> str: """ 基于检索到的上下文生成回答 """ # 构建带上下文的提示词 context_text = "\n\n".join(context) system_prompt = """你是一个专业的知识库问答助手。请根据以下提供的上下文内容，准确回答用户的问题。 规则： 1. 如果上下文中包含答案，请直接引用并给出准确回答 2. 如果上下文中没有相关信息，请明确告知"知识库中未找到相关信息" 3. 不要编造或扩展上下文中不存在的信息 4. 用中文回答，保持简洁明了""" user_prompt = f"""上下文内容：{context_text}用户问题：{question}请基于以上上下文回答问题：""" try: response = self.client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.3, # 较低的温度使回答更准确 max_tokens=300 ) return response.choices[0].message.content except Exception as e: return f"生成回答时出错: {str(e)}" def ask(self, question: str) -> Dict: """ 完整的RAG问答流程 """ # 步骤1：检索上下文 context = self.retrieve_context(question) # 步骤2：生成回答 answer = self.generate_answer(question, context) return { "question": question, "answer": answer, "context_used": context # 返回使用的上下文以便调试 }# 示例：构建一个小型知识库并提问if __name__ == "__main__": # 初始化RAG系统 rag = RAGSystem() # 添加一些文档（模拟公司内部知识库） documents = [ "RAG（检索增强生成）是一种结合信息检索和文本生成的技术架构。它首先从知识库中检索相关文档，然后基于这些文档生成回答。", "RAG的核心优势在于：1. 减少大模型的幻觉问题 2. 支持实时更新知识库 3. 可追溯信息来源 4. 降低训练成本。", "实现RAG系统通常需要以下组件：嵌入模型、向量数据库（如FAISS）、大语言模型（如GPT-4）。", "公司的年假政策：入职满1年可享受5天年假，满5年可享受10天年假，满10年可享受15天年假。" ] rag.add_documents(documents) # 测试1：知识库内问题 result1 = rag.ask("RAG系统有哪些核心优势？") print(f"问题1: {result1['question']}") print(f"回答: {result1['answer']}") print(f"使用的上下文: {result1['context_used']}\n") # 测试2：知识库外问题 result2 = rag.ask("公司的年假政策是什么？") print(f"问题2: {result2['question']}") print(f"回答: {result2['answer']}\n") # 测试3：超出知识库范围的问题 result3 = rag.ask("公司的股权激励计划是什么？") print(f"问题3: {result3['question']}") print(f"回答: {result3['answer']}\n")运行结果示例：已添加 4 个文档，当前总文档数: 4问题1: RAG系统有哪些核心优势？回答: RAG的核心优势包括：1. 减少大模型的幻觉问题 2. 支持实时更新知识库 3. 可追溯信息来源 4. 降低训练成本。问题2: 公司的年假政策是什么？回答: 根据知识库信息，公司的年假政策为：入职满1年可享受5天年假，满5年可享受10天年假，满10年可享受15天年假。问题3: 公司的股权激励计划是什么？回答: 知识库中未找到相关信息。### 5. 进阶优化建议1. 上下文窗口管理：当检索结果过多时，需要裁剪或压缩，避免超出模型token限制2. 重排序（Re-ranking）：使用cross-encoder模型对检索结果重新排序，提升相关性3. 流式响应：使用 stream=True 参数实现逐token输出，提升用户体验4. 错误重试：为API调用添加指数退避重试机制，应对临时故障5. 缓存机制：对相同问题缓存结果，减少API调用成本### 6. 总结通过本文的实战，我们完成了RAG项目的关键一环——将检索结果与大模型API结合，实现了带上下文的智能问答系统。你学会了：- 如何通过OpenAI SDK调用大模型API- 如何构建检索提示词（prompt engineering）- 如何将向量检索与文本生成无缝集成- 如何处理知识库内和知识库外的不同场景下一步，你可以考虑：- 替换为国内大模型API（如通义千问、文心一言），降低延迟和成本- 集成更强大的向量数据库（如Pinecone、Weaviate）- 添加用户反馈机制，持续优化回答质量RAG的世界才刚刚开始，下一篇文章我们将探讨如何评估和优化RAG系统的性能，敬请期待！