大多数大型语言模型应用在高保真度的演示中看起来都非常出色。但当它们真正被实际用户使用时，就会以一些既可预测又会带来严重问题的方式出现故障。
这些应用会给出本不该给出的答案；在文档检索功能较弱时会出现故障；由于网络延迟而导致程序超时；而且由于没有日志记录和相应的测试机制，根本无法准确判断到底发生了什么问题。
在本教程中，你将构建一个适合初学者的、能够适应实际生产环境需求的“检索增强生成”应用。这个应用不仅仅是一个调用API的简单脚本，而是一个包含FastAPI后端、持久化的FAISS向量存储系统以及各种安全防护机制的系统（比如检索限制机制和备用方案）。
目录
为什么单独使用“检索增强生成”技术并不足以让应用具备生产环境适用性
“检索增强生成”技术常常被认为能够有效解决模型产生错误答案的问题。通过将模型的输出与检索到的文本相结合，我们可以确保模型获得准确的信息从而提高其准确性。但是，仅仅将向量数据库与大型语言模型连接起来是远远不够的，这样的系统无法适应生产环境的需求。
在生产环境中，问题通常源于围绕这些模型的各种系统组件的隐性故障：
-
检索能力不足：如果应用程序检索到了无关的文本片段，模型就会试图通过编造答案来弥补这一缺陷。如果没有设置“不知道”的响应机制，模型就不得不自行“创造”答案，从而导致错误的结果。
-
缺乏可监控性：如果没有结构化的输出结果和基本的日志记录系统，你就无法判断是检索失败、提示语设计不合理，还是模型更新导致了错误的答案。
-
系统脆弱性：如果未实现备用方案，即使是简单的API超时或提供者返回的错误数据，也可能会直接导致用户端出现故障。
-
没有回归测试：在传统的软件开发中，我们会进行单元测试；而在人工智能领域，我们需要进行评估测试。如果没有这些测试机制，对提示语进行的微小修改可能会解决某个问题，但却会引发其他十个问题，而你可能根本意识不到这一点。
在本指南中，我们将系统地解决这些问题。
先决条件
本教程适合初学者学习，但假设您已经掌握了一些基础知识，这样就可以专注于构建一个功能完善的RAG系统，而不会被设置问题所困扰。
所需知识
您应该具备以下基础：
-
Python基础（函数、模块、虚拟环境）
-
基本的HTTP与JSON知识（请求、响应数据）
-
使用FastAPI的API（了解端点的含义以及如何运行服务器）
-
高级大语言模型概念（提示机制、温度控制、结构化输出等）
所需工具与账户
您需要准备以下内容：
-
Python 3.10或更高版本
-
一个可用的兼容OpenAI的API密钥（OpenAI或其他支持相同请求/响应格式的提供商提供的密钥）
-
一个可以运行FastAPI应用的本地环境（Mac/Linux/Windows系统均可）
本教程涵盖的内容与未涵盖的内容
我们将构建一个具备生产环境适用性的基础框架，包括以下内容：
-
一个基于FAISS技术的检索系统，该系统支持索引数据的持久化存储及元数据管理
-
一个检索过滤机制，用于防止出现“错误的结果”
-
结构化的JSON输出格式，以确保后端服务的稳定性
-
针对超时或供应商故障的备用处理机制
-
一个小型评估工具，用于检测系统功能是否出现退化
对于诸如重新排序算法、语义分块技术、身份验证机制以及后台作业等高级功能，我们仅会在未来的规划中考虑实现它们。
您正在构建的系统架构
我们的应用程序遵循严格的流程设计，因此每一个输出结果都基于可靠的数据进行生成：
-
用户查询：用户通过FastAPI端点提交问题。
-
检索过程：系统会分析用户提出的问题，并检索出最相似的文档片段。
-
过滤机制：我们会评估这些文档片段的相似度；如果内容相关性不足，就会立即停止检索并拒绝用户的请求。
-
数据增强与生成：如果通过过滤机制，我们就会向大语言模型发送包含上下文信息的提示词。
-
结构化响应：模型会返回一个JSON对象，其中包含答案、所使用的信息来源以及置信度等级。
项目设置与结构
为了保持代码的整洁性和可维护性，我们将采用模块化的设计结构。这样，即使更换了大语言模型的提供商或向量数据库，也不需要重新编写整个应用程序的核心部分。
项目结构
.
├── app.py # FastAPI入口点及API逻辑代码
├── rag.py # FAISS索引管理、数据持久化及文档检索功能
├── llm.py # 大语言模型接口及JSON解析模块
├── prompts.py # 中心化的提示词模板库
├── data/ # 包含原始.txt文档的文件夹
├── index/ | 存储FAISS索引及元数据的目录
└── evals/ | 用于评估的数据集及运行脚本
├── eval_set.json
└── run_evals.py
安装依赖项
首先，创建一个虚拟环境来隔离您的项目：
python -m venv .venv
source .venv/bin/activate # 在 Windows 上：.venv\Scripts\activate
pip install fastapi uvicorn faiss-cpu numpy pydantic requests python-dotenv
配置环境
在根目录下创建一个.env文件。我们使用的是与 OpenAI 兼容的接口：
OPENAI_API_KEY=您的实际 API 密钥
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini
关于兼容性的重要提示：以下代码假设使用的是 OpenAI 格式的接口。如果您使用的接口不兼容，就必须更改 URL、请求头信息（例如X-API-Key），以及在embed_texts()和call_llm()函数中提取嵌入向量和最终消息内容的方式。
如何使用 FAISS 构建 RAG 层
在rag.py文件中，我们实现了 RAG 系统中的“检索器”部分。这一过程涉及将原始文本转换为计算机可以比较的数学向量。
什么是 FAISS？它的作用是什么？
FAISS（Facebook AI Similarity Search）是一个用于快速进行向量相似性搜索的库。在 RAG 系统中，每段文本都会被转换成一个嵌入向量（即一组浮点数）。FAISS 会将这些向量存储在一个索引结构中，这样我们就可以快速地查询：
“给定这个查询向量，哪些文档片段与它最为相似？”
在本教程中，我们使用IndexFlatIP来计算内积，并通过faiss.normalize_L2(...)对向量进行归一化处理。经过归一化的向量，其内积结果就相当于余弦相似度，这样我们就能得到一个稳定的分数，用于判断查询结果与文档内容的匹配程度。
带有重叠区域的文本分割策略
我们会采用带有重叠区域的文本分割方法。如果我们将一篇文档恰好分成1000个字符长的片段，那么可能会把一个句子切成两半，从而导致语义丢失。而通过设置200个字符的重叠区域，我们可以确保相邻片段的上下文信息能够相互衔接。
rag.py的实现细节
import os
import faiss
import numpy as np
import requests
import json
from typing import List, Dict
from dotenv import load_dotenv
load_dotenv()
INDEX_PATH = "index/faiss.index"
META_PATH = "index/meta.json"
def chunk_text(text: str, size: int = 1000, overlap: int = 200) -> List[str]:
chunks = []
step = max(1, size - overlap)
for i in range(0, len(text), step):
chunk = text[i : i + size].strip()
if chunk:
chunks.append(chunk)
return chunks
def embed_texts(texts: List[str]) -> np.ndarray:
# 注意：如果您的接口不兼容 OpenAI，请更改此 URL 和请求头信息
url = f"{os.getenv('OPENAI_BASE_URL')}/embeddings"
headers = {"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"}
payload = {"input": texts, "model": "text-embedding-3-small"}
resp = requests.post(url, headers=headers, json=payload, timeout=30)
resp.raise_for_status()
# 如果您的接口使用的是其他响应格式，请修改以下代码
vectors = np.array([item["embedding"] for item in resp.json()["data"]], dtype="float32")
return vectors
def build_index() -> None:
all_chunks: List[str] = []
metadata: List[Dict] = []
if not os.path.exists("data"):
os.makedirs("data")
return
for file in os.listdir("data"):
if not file.endswith(".txt"):
continue
with open(f"data/{file}", "r", encoding="utf-8") as f:
text = f.read()
chunks = chunk_text(text)
all_chunks.extend(chunks)
for c in chunks:
metadata.append({"source": file, "text": c})
if not all_chunks:
return
embeddings = embed_texts(all_chunks)
faiss.normalize_L2.embeddings)
dim = embeddings.shape[1]
index = faiss.IndexFlatIP(dim)
index.add.embeddings)
os.makedirs("index", exist_ok=True)
faiss.write_index(index, INDEX_PATH)
with open(META_PATH, "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False)
def load_index():
if not (os.path.exists(INDEX_PATH) and os.path.exists(META_PATH)):
raise FileNotFoundError(
"FAISS 索引未找到。请将.txt文件添加到data/目录中，然后运行build_index()函数。"
)
index = faiss.read_index(INDEX_PATH)
with open(META_PATH, "r", encoding="utf-8") as f:
metadata = json.load(f)
return index, metadata
def retrieve(query: str, k: int = 5) -> List[Dict]:
index, metadata = load_index()
q_emb = embed_texts([query])
faiss.normalize_L2(q_emb)
scores, ids = index.search(q_emb, k)
results = []
for score, idx in zip(scores[0], ids[0]):
if idx == -1:
continue
m = metadata[idx]
results.append(
{"score": float(score), "source": m["source"], "text": m["text"], "id": int(idx)}
)
return results
如何添加具有结构化输出的LLM调用功能
AI应用中的一个主要缺陷在于LLM的“对话式”特性。如果后端期望收到一系列信息来源，而LLM返回的却只是些闲聊内容，那么程序就会崩溃。
我们通过结构化输出来解决这个问题：让模型返回一个严格的JSON对象，然后安全地解析这些数据。
`llm.py`的实现方式
import json
import requests
import os
from typing import Dict, Any
def call_llm(system_prompt: str, userprompt: str) -> Dict[str, Any]:
# 注意：如果使用的是非OpenAI兼容的服务提供商，请更改URL和请求头信息
url = f"{os.getenv('OPENAI_BASE_URL')}/chat/completions"
headers = {
"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
"Content-Type": "application/json",
}
payload = {
"model": os.getenv("OPENAI_MODEL"),
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": userprompt},
],
"response_format": {"type": "json_object"},
"temperature": 0,
}
try:
resp = requests.post(url, headers=headers, json=payload, timeout=30)
resp.raise_for_status()
content = resp.json()["choices"][0]["message"]["content"]
parsed = json.loads(content)
parsed.setdefault("answer", "")
parsed.setdefault("refusal", False)
parsed.setdefault("confidence", "medium")
parsed.setdefault("sources", [])
return parsed
except (requests.Timeout, requests.ConnectionError):
return {
"answer": "系统暂时无法使用（网络问题）。请稍后再试。",
"refusal": True,
"confidence": "low",
"sources": [],
"error_type": "network_error",
}
except Exception:
return {
"answer": "在生成答案过程中发生了系统错误。",
"refusal": True,
"confidence": "low",
"sources": [],
"error_type": "unknown_error",
}
如何设置防护机制：检索门与回退方案
防护机制其实就是拦截器。它们位于用户与模型之间，用于防止那些可以预见的错误发生。
检索门的工作原理及实现方法
在标准的RAG处理流程中，系统总会先调用LLM。但如果用户提出了一个不相关的问题，检索模块仍然会返回那些“最接近”正确答案的片段。
为了解决这个问题，我们可以使用检索门：
-
首先检索前k个最相关的结果，并获取它们的相似度得分
-
如果这些结果的相似度低于某个阈值（例如0.30），则立即拒绝用户的请求
-
只有当检索结果足够可靠时，才真正调用LLM来生成最终答案
使用归一化余弦相似度作为评估标准时，0.30这个阈值是一个合理的起点；不过你可以通过实际测试来调整这个数值（具体方法见下一节）。
备用方案及其重要性
备用方案能够确保在API出现故障或超时时，用户会收到有用的提示信息，而不会遇到系统崩溃的情况。此外，这些方案还能保持API响应格式的一致性，从而避免前端出现错误，并使日志记录更具参考价值。
在本教程中，备用方案是在call_llm()函数内部实现的，这样就能让FastAPI层的代码结构保持简洁。
FastAPI应用：创建/answer端点
app.py文件是整个应用程序的核心。它将信息检索、错误检测机制、提示语生成以及答案生成等功能整合在了一起。
app.py的实现细节
from fastapi import FastAPI
from pydantic import BaseModel
from rag import retrieve
from llm import call_llm
import prompts
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag_app")
app = FastAPI(title="Production-Ready RAG")
class QueryRequest(BaseModel):
question: str
@app.post("/answer")
async def get_answer(req: QueryRequest):
start_time = time.time()
question = (req.question or "").strip()
if not question:
return {
"answer": "请提供一个非空的问题。",
"refusal": True,
"confidence": "low",
"sources": [],
"latency_sec": round(time.time() - start_time, 2),
}
# 1) 信息检索
results = retrieve/question, k=5)
top_score = results[0]["score"] if results else 0.0
logger.info("query=%r top_score=%.3f num_results=%d", question, top_score, len(results))
# 2) 错误检测机制
if top_score < 0.30:
return {
"answer": "我没有相关的文档来回答这个问题。",
"refusal": True,
"confidence": "low",
"sources": [],
"latency_sec": round(time.time() - start_time, 2),
"retrieval": {"top_score": top_score, "k": 5},
}
# 3> 信息补充
context_text = "\n\n".join([f"来源 {r['source']}: {r['text']}" for r in results])
user_prompt = f"背景信息:\n{context_text}\n\n问题: {question}"
# 4> 使用备用方案进行答案生成
response = call_llm(promptsSYSTEM_PROMPT, user.prompt)
# 5> 添加调试信息
response["latency_sec"] = round(time.time() - start_time, 2)
response["retrieval"] = {"top_score": top_score, "k": 5}
return response
集中管理的提示语模板——文件：prompts.py
一个虽小但很重要的习惯是：将所有的提示语集中管理起来，这样就可以方便地对它们进行版本控制以及评估其效果。
prompts.py示例代码
SYSTEM_PROMPT = """你是一个RAG助手。请仅使用提供的背景信息来回答问题。
如果背景信息中没有答案，那么请返回{"refusal": True}。
返回一个格式正确的JSON对象，其中必须包含以下键：
- answer: 字符串形式的结果
- refusal: 布尔值，表示是否给出了答案
- confidence: "low"、"medium"或"high"，表示答案的可靠性
- sources: 一个字符串数组，列出你使用过的资料来源文件名
不要添加任何额外的键，也不要包含Markdown格式的内容或注释。"""
如何添加适合初学者的评估测试
在人工智能系统中，输出结果通常是概率性的。因此，与传统的软件相比，进行测试会变得更加困难。评估测试是一组“关键问题”和“预期行为”，通过反复运行这些测试，可以及时发现系统功能的退化或异常变化。
与其问“它是否输出了完全指定的字符串”，不如这样来测试：
-
当检索结果不佳时，应用程序应该拒绝响应吗？
-
在给出答案时，系统中是否会包含参考来源信息？
-
当提示语或模型本身发生变化时，系统的行为是否依然稳定？
步骤1：创建文件`evals/eval_set.json`
该文件中应同时包含正面案例和负面案例。
[
{
"id": "in_scope_01",
"question": "什么是检索门机制，它为什么重要？",
"expect_refusal": false,
"notes": "测试内容应涉及对检索门机制的解释，并说明其防止错误回答的作用。"
},
{
"id": "out_of_scope_01",
"question": "法国的首都是什么？",
"expect_refusal": true,
"notes": "如果知识库中只包含文档信息，应用程序应该拒绝回答这个问题。"
},
{
"id": "edge_01",
"question": "",
"expect_refusal": true,
"notes": "当输入内容为空时，系统不应启动大语言模型进行响应。"
}
]
步骤2：创建文件`evals/run_evals.py`
这个脚本会调用你的API接口，并检查系统的实际行为是否符合预期。
import json
import requests
API_URL = "http://127.0.0.1:8000/answer"
def run():
with open("evals/eval_set.json", "r", encoding="utf-8") as f:
cases = json.load(f)
passed = 0
failed = 0
for case in cases:
resp = requests.post(API_URL, json={"question": case["question"]}, timeout=60)
resp.raise_for_status()
out = resp.json()
got_refusal = bool(out.get("refusal", False))
expect_refusal = bool(case["expect_refusal"])
ok = (got_refusal == expect_refusal)
# 适合初学者的测试规则：如果系统给出了答案，那么参考来源信息必须存在且必须是列表形式。
if not got_refusal:
ok = ok and isinstance(out.get("sources"), list)
if ok:
passed += 1
print(f"通过测试：{case['id']}")
else:
failed += 1
print(f"未通过测试：{case['id']} 预期结果为{expect_refusal}，实际结果为{got_refusal}")
print("输出结果：", json.dumps(out, indent=2))
print(f"\n测试完成。通过次数：{passed}，失败次数：{failed}")
if failed:
raise SystemExit(1)
if __name__ == "__main__":
run()
如何在实际中使用这些评估测试
首先运行你的服务器：
uvicorn app:app --reload然后在另一个终端中执行评估测试脚本：
python evals/run_evals.py如果某次测试失败，那就说明在检索机制、提示语设计或系统行为方面肯定出现了问题。
下一步应该改进什么：如何进行实际的优化升级
构建一个可靠的RAG应用是一个迭代的过程。以下是可行的后续步骤：
-
语义分割：根据文本的含义而不是字符数量来对其进行分割。
-
重新排序：使用交叉编码器对排名靠前的文本片段进行重新排序，以提高准确性。
-
元数据过滤：根据类别、日期或部门来筛选结果，从而减少误报的情况。
-
更准确的引用说明：存储文本片段的标识信息，并明确显示答案来源于哪些片段。
-
可追溯性：添加请求ID、结构化的日志记录等信息，以便能够追踪系统运行过程中的具体情况。
-
异步处理与后台索引构建：将索引构建任务放在后台进行，以确保API接口仍能正常响应用户请求。
结语：具备生产环境适用性是一系列习惯的体现
要打造一个能在现实世界中发挥作用的人工智能应用，就必须构建出一个可预测、可测量且安全的系统。
-
检索质量是可以量化的：利用相似度评分来筛选输出结果。
-
坦诚拒绝也是一种能力：说“我不知道”总比撒谎要好。
-
必须有备用方案：为API可能出现故障的情况提前做好应对措施。
-
测试是防止系统退化的关键：在任何进行代码修改之前，都必须先运行相应的测试。
关于我
我是Chidozie Managwu，一位屡获殊荣的人工智能产品架构师兼创业者，我的工作致力于帮助全球的科技人才掌握真正具备生产环境适用性的技能。我作为GAFAI组织的代表参与了多项全球范围内的AI发展项目，并负责运营AI Titans Network这个为开发者提供学习交流平台的社区。
我的工作成果得到了全球科技界的认可，我也曾在HackerNoon等平台上分享自己的经验。
