如何用 OpenAI API 构建自己的智能问答系统（含完整代码）

目录

1. 整体架构（检索增强生成 — RAG）

2. 准备工作与依赖

3. 文档入库（生成 embeddings 并存入 FAISS）——ingest.py

4. 后端（FastAPI）：查询路由 + 检索 + 调用 OpenAI Responses ——app.py

5. 示例运行 & 测试

6. 调优建议、成本/安全注意、常见问题

---

1. 整体架构（RAG — Retrieval-Augmented Generation）

我们用检索增强生成(RAG)：

1. 离线把知识（文档、FAQ、代码片段等）拆分成段落并用 OpenAI Embeddings 转成向量

2. 把向量保存到本地向量库（示例用 FAISS）。

3. 用户问题到后端 → 先用同样的 embedding 将问题向量化 → 在向量库里检索 top-k 相关段落 → 把这些相关上下文拼接进 prompt（或 input）发送给 OpenAI Responses（或 Chat Completions）。

优点：回答能基于你自己的知识库，且更可控、不容易 hallucinate（同时要在 prompt 里用“source”策略减少错误）。

---

2. 准备工作与依赖

示例用 Python + FastAPI + FAISS + OpenAI 官方 Python SDK（示例基于官方 quickstart）。你需要先在 OpenAI 控制台创建 API Key 并导出环境变量（或用 .env 存）。官方文档说明了如何创建与使用 API Key。

依赖（示例）

python >= 3.10 pip install openai faiss-cpu fastapi uvicorn python-dotenv tiktoken aiofiles # 若想用更方便的文本分段工具，可加: pip install nltk sentencepiece

注意：示例使用from openai import OpenAI的新式 SDK 客户端；若你用旧版openai包（openai.ChatCompletion.create）请参考官方迁移说明。OpenAI 平台+1

在项目根目录建立.env：

OPENAI_API_KEY=sk-... EMBED_MODEL=text-embedding-3-small GEN_MODEL=gpt-5.2 FAISS_INDEX_PATH=./data/faiss_index.bin DOCS_DIR=./docs TOP_K=4 MAX_CONTEXT_TOKENS=2000

---

3. 文档入库脚本ingest.py（把文档拆段 -> 生成 embeddings -> 存 FAISS）

下面脚本做三件事：读取docs/下的文本文件、按段落分割、用 OpenAI embeddings 生成向量并保存到 FAISS，并把元数据（文本、来源）保存在并行 JSON 文件里，便于后续返回来源。

# ingest.py import os, json, glob from openai import OpenAI import numpy as np import faiss from dotenv import load_dotenv load_dotenv() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) EMBED_MODEL = os.getenv("EMBED_MODEL","text-embedding-3-small") DOCS_DIR = os.getenv("DOCS_DIR","./docs") INDEX_PATH = os.getenv("FAISS_INDEX_PATH","./data/faiss_index.bin") META_PATH = "./data/metadata.json" def chunk_text(text, max_chars=1000): # 简单按段落切分，可换更复杂的 chunker（根据句子边界 / token） parts = [] for para in text.split("\n\n"): para = para.strip() if not para: continue if len(para) <= max_chars: parts.append(para) else: # 强制切分 for i in range(0, len(para), max_chars): parts.append(para[i:i+max_chars]) return parts def collect_docs(): files = glob.glob(os.path.join(DOCS_DIR, "**/*.txt"), recursive=True) docs = [] for f in files: with open(f, "r", encoding="utf-8") as fh: text = fh.read() chunks = chunk_text(text) for i,ch in enumerate(chunks): docs.append({ "id": f"{os.path.basename(f)}_chunk{i}", "source": f, "text": ch }) return docs def embed_texts(texts): # texts: list[str] resp = client.embeddings.create(model=EMBED_MODEL, input=texts) # 返回 embeddings 列表 embeddings = [e.embedding for e in resp.data] return embeddings def build_faiss(docs, embeddings, dim): xb = np.array(embeddings).astype("float32") index = faiss.IndexFlatIP(dim) # 余弦相似度 (先 normalize) faiss.normalize_L2(xb) index.add(xb) faiss.write_index(index, INDEX_PATH) # 保存 metadata meta = {i: {"id": docs[i]["id"], "source": docs[i]["source"], "text": docs[i]["text"]} for i in range(len(docs))} os.makedirs(os.path.dirname(META_PATH), exist_ok=True) with open(META_PATH, "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) print(f"Saved index to {INDEX_PATH}, metadata to {META_PATH}") def main(): docs = collect_docs() texts = [d["text"] for d in docs] batch = 64 embeddings = [] for i in range(0, len(texts), batch): slice_texts = texts[i:i+batch] embs = embed_texts(slice_texts) embeddings.extend(embs) print(f"embedded {i+len(slice_texts)}/{len(texts)}") dim = len(embeddings[0]) build_faiss(docs, embeddings, dim) if __name__ == "__main__": main()

说明与来源：上面用法参考官方 Embeddings 文档与 Web-QA 教程示例流程（先嵌入再检索）。

---

4. 后端实现app.py（FastAPI）：查询 → 检索 top-k → 拼接上下文 → 调用 OpenAI Responses

注意：这里用 Responses API（官方推荐给新项目），把生成 prompt 的上下文直接作为input传入。你也可以改为 chat 风格 messages（看你用的 SDK 版本）。OpenAI 平台+1

# app.py import os, json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from openai import OpenAI import faiss, numpy as np from dotenv import load_dotenv load_dotenv() app = FastAPI() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) GEN_MODEL = os.getenv("GEN_MODEL","gpt-5.2") EMBED_MODEL = os.getenv("EMBED_MODEL","text-embedding-3-small") INDEX_PATH = os.getenv("FAISS_INDEX_PATH","./data/faiss_index.bin") META_PATH = "./data/metadata.json" TOP_K = int(os.getenv("TOP_K","4")) MAX_CONTEXT_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS","2000")) # 加载 index 与 metadata if not os.path.exists(INDEX_PATH) or not os.path.exists(META_PATH): raise RuntimeError("请先运行 ingest.py 生成向量索引与 metadata") index = faiss.read_index(INDEX_PATH) with open(META_PATH, "r", encoding="utf-8") as f: metadata = json.load(f) def embed_query(q): resp = client.embeddings.create(model=EMBED_MODEL, input=q) return np.array(resp.data[0].embedding, dtype="float32") def search_top_k(query_embedding, k=TOP_K): # faiss.IndexFlatIP 需要先 normalize q = query_embedding.reshape(1, -1) faiss.normalize_L2(q) D, I = index.search(q, k) hits = [] for score, idx in zip(D[0], I[0]): if idx == -1: continue meta = metadata[str(idx)] hits.append({"score": float(score), "id": meta["id"], "source": meta["source"], "text": meta["text"]}) return hits def build_prompt(question, hits): # 把 top-k 文本拼成上下文（截断以满足 token 限制） context_parts = [] token_count = 0 for h in hits: context_parts.append(f"来源: {h['source']}\n内容:\n{h['text']}\n---") context = "\n".join(context_parts) prompt = f"""你是一个知识型问答助手。请基于下列提供的 “来源内容” 回答用户的问题。 如果来源无法支持问题的具体答案，请诚实说明并尽可能给出基于常识的推断（标注为推断）。 不要编造不在来源中的事实。 来源内容： {context} 用户问题： {question} 请给出简明直接的答案，并在答案后列出引用到的来源（按文件名或路径）。""" return prompt class QueryIn(BaseModel): question: str @app.post("/query") async def query(qin: QueryIn): q = qin.question.strip() if not q: raise HTTPException(status_code=400, detail="问题不能为空") q_emb = embed_query(q) hits = search_top_k(q_emb, TOP_K) prompt = build_prompt(q, hits) # 调用 Responses API resp = client.responses.create( model=GEN_MODEL, input=prompt, # 可加参数控制长度/温度 max_output_tokens=800, temperature=0.0 ) # 官方 Responses 返回结构中，输出文本可从 resp.output 或 resp.output_text 获取（取决 SDK 版本） answer = getattr(resp, "output_text", None) or "".join([o.get("content", "") for o in resp.output]) if hasattr(resp, "output") else str(resp) return {"answer": answer, "sources": hits}

说明与要点：

• 检索使用 FAISS 的内积（IP）+ normalize，等同余弦相似度。FAISS 配置与用法来源广泛（示例简洁）。

• 在build_prompt中把检索到的段落以“来源 + 内容”风格传入，明确要求模型“不要捏造、引用来源”。这是减少 hallucination 的常见做法（RAG 实践）。可进一步在 prompt 中要求模型把答案分成“答复 + 引用”两部分。OpenAI 平台

• temperature=0.0更偏确定性，适合知识型问答。可根据需求调整。

• Responses API 的字段结构可能随 SDK/version 有微差异，示例使用通用访问方式（参考官方 quickstart）。OpenAI 平台

---

5. 示例运行 & 测试

1. 把你的知识文档放到docs/（支持.txt）。

2. 运行入库：

python ingest.py # 会在 ./data/ 生成 faiss_index.bin 和 metadata.json

1. 启动后端：

uvicorn app:app --reload --port 8000

1. 测试（curl）：

curl -X POST "http://127.0.0.1:8000/query" -H "Content-Type: application/json" \ -d '{"question":"XXX 你的测试问题"}'

6. 调优建议、成本与安全注意

调优建议

• 向量切分策略：短段（200–800 字符）通常效果好；确保不把重要上下文切断。可以用 sentence tokenizer 进行智能切分。

• Top-k 与 prompt 长度：k 越大，上下文越多但 token 也越多；你可以把检索到的段落再做简单过滤（基于相似度阈值）或摘要后再传给生成模型。官方 Web-QA 教程也提到类似策略。

• 使用检索结果做“tool”或“function call”结构（进阶）：当需要从多个来源调用外部工具时，可考虑 agent/函数调用框架（Agents），但复杂度更高。

成本 & token 管理

• Embeddings 与生成均会产生成本。embedding 逐段做批处理以节省时间与 cost。

• 使用 Responses API 时注意max_output_tokens与上下文 token 数量。官方文档解释了模型与 token 限制与计费

安全与隐私

• 切勿把你的 OPENAI_API_KEY 放到前端或公开仓库。只在后端服务器加载。官方文档强调 API key 的保密与认证措施。

• 若你的知识库包含敏感数据（个人信息、机密代码等），请在入库前脱敏或加访问控制。

---

7. FAQ（常见问题）

Q：为什么有时模型还是会“胡扯”？
A：RAG 可以显著降低 hallucination，但不能完全消除。注意 prompt 明确要求“只用来源回答／无法回答请说明”，并尽量把检索到的最相关片段放在前面。若仍有问题，可把temperature设为 0 并限制max_output_tokens。

Q：能否使用 Milvus / Weaviate 等向量库？
A：可以。本文示例用 FAISS（本地、轻量）。如果需要分布式、持久化、并发大流量，推荐 Milvus / Weaviate / Pinecone 等生产级向量数据库。

Q：如何把系统部署到线上并保证可扩展性？
A：把 embedding 与构建 index 的任务放到离线批处理（或异步 Worker）。后端只做检索与调用模型。使用 Docker、Kubernetes 与云向量 DB（或托管 FAISS 服务）提升可用性与扩展性。

---

参考与文档

• OpenAI Developer quickstart（Python SDK 示例、Responses API）：官方

• Embeddings 指南（模型与使用示例）：官方

• Web QA with embeddings（官方 tutorial，完整 RAG 示例）：官方

• Chat/Responses API 参考：官方 API 文档