目录

  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 文档

# 人工智能 # 信息可视化 # 数据分析 # 语言模型 # 数据挖掘
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

SpringBoot+Vue 墙绘产品展示交易平台平台完整项目源码+SQL脚本+接口文档【Java Web毕设】

avatar 2048 AI社区
cover

【2025最新】基于SpringBoot+Vue的Web足球青训俱乐部管理后台系统管理系统源码+MyBatis+MySQL

avatar 2048 AI社区
cover

Java SpringBoot+Vue3+MyBatis 海滨体育馆管理系统系统源码｜前后端分离+MySQL数据库

avatar 2048 AI社区