Skip to content

RAG 基础

RAG 全称 Retrieval-Augmented Generation,中文叫检索增强生成。放到脚本里看就两步:先从资料里找几段相关内容,再把这些内容连同问题一起交给模型。

普通调用是直接问模型:

text
密码记错了怎么登录?

RAG 会多做一步:

text
先在 FAQ 里找到“忘记密码怎么办?”
再让模型根据这条 FAQ 回答。

这一步很关键。模型不是凭空记住本地 FAQ,而是 Python 把检索结果放进请求里,模型才能看着资料回答。

一、从检索到回答

06 里已经有一个 FAQ 索引:

text
outputs/faq-index.json

搜索脚本能把相近 FAQ 打出来:

text
0.8421 faq-001 忘记密码怎么办?
  可以在登录页点击忘记密码,通过邮箱验证码重置密码。

检索结果还只是资料片段。要变成一段能直接看的回答,还要把这条资料交给模型,让模型用自然语言组织出来。

RAG 里要分清三个东西:

  • 原始问题:用户问了什么。
  • 检索结果:本地资料里找到了哪些相关片段。
  • 回答要求:模型只能根据资料回答,资料里没有就说没有。

这三样混在一起,模型就容易自由发挥。分开写,请求会清楚很多。

二、拼接资料

先看资料怎么拼。假设检索出了两条 FAQ:

python
matches = [
    {
        "id": "faq-001",
        "question": "忘记密码怎么办?",
        "answer": "可以在登录页点击忘记密码,通过邮箱验证码重置密码。",
        "score": 0.8421,
    },
    {
        "id": "faq-002",
        "question": "如何修改绑定邮箱?",
        "answer": "登录后进入账号设置,在安全信息里修改邮箱。",
        "score": 0.5120,
    },
]

可以把它们整理成这样的文本:

text
[faq-001]
问题: 忘记密码怎么办?
答案: 可以在登录页点击忘记密码,通过邮箱验证码重置密码。

[faq-002]
问题: 如何修改绑定邮箱?
答案: 登录后进入账号设置,在安全信息里修改邮箱。

[faq-001] 这种编号不是给模型装饰用的。它让回答有来源,后面保存结果或排查回答时,能知道模型看的是哪几条资料。

拼接函数可以写成:

python
def format_sources(matches):
    blocks = []
    for item in matches:
        blocks.append(
            f"[{item['id']}]\n"
            f"问题: {item['question']}\n"
            f"答案: {item['answer']}"
        )
    return "\n\n".join(blocks)

三、回答约束

RAG 里提示词最重要的不是写得花,而是把回答边界写清楚。

比如:

text
根据提供的 FAQ 资料回答用户问题。
只能使用 FAQ 资料里的信息。
如果资料里没有答案,就回答“资料里没有找到相关答案”。
回答末尾列出使用的 FAQ id。

这几句解决几个实际问题:

  • 不让模型脱离资料自由发挥。
  • 找不到资料时有固定说法。
  • 回答里带来源 id,方便回看检索结果。

提示词不能当安全锁,但能让模型少飘。资料范围还是由 Python 控制:检索出来几条,就只把这几条放进请求。

四、完整脚本

新增文件:scripts/answer_faq.py

python
#!/usr/bin/env python3
"""检索 FAQ 后,让模型根据资料回答问题。"""

import argparse
import json
import math
import os
import sys
from pathlib import Path

from openai import OpenAI


def parse_args():
    parser = argparse.ArgumentParser(description="answer question with FAQ RAG")
    parser.add_argument("query", help="用户问题")
    parser.add_argument("--index", default="outputs/faq-index.json", help="FAQ 索引文件")
    parser.add_argument("--top-k", type=int, default=3, help="放进回答上下文的 FAQ 条数")
    return parser.parse_args()


def cosine_similarity(left, right):
    dot = sum(a * b for a, b in zip(left, right))
    left_norm = math.sqrt(sum(a * a for a in left))
    right_norm = math.sqrt(sum(b * b for b in right))
    if left_norm == 0 or right_norm == 0:
        return 0.0
    return dot / (left_norm * right_norm)


def search_faq(client, embedding_model, index, query, top_k):
    response = client.embeddings.create(
        model=embedding_model,
        input=query,
    )
    query_embedding = response.data[0].embedding

    results = []
    for item in index:
        score = cosine_similarity(query_embedding, item["embedding"])
        results.append({
            "score": score,
            "id": item["id"],
            "question": item["question"],
            "answer": item["answer"],
        })

    results.sort(key=lambda item: item["score"], reverse=True)
    return results[:top_k]


def format_sources(matches):
    blocks = []
    for item in matches:
        blocks.append(
            f"[{item['id']}]\n"
            f"问题: {item['question']}\n"
            f"答案: {item['answer']}"
        )
    return "\n\n".join(blocks)


def main():
    args = parse_args()
    api_key = os.getenv("OPENAI_API_KEY")
    chat_model = os.getenv("OPENAI_MODEL")
    embedding_model = os.getenv("OPENAI_EMBEDDING_MODEL")

    if not api_key:
        print("缺少环境变量: OPENAI_API_KEY", file=sys.stderr)
        return 2
    if not chat_model:
        print("缺少环境变量: OPENAI_MODEL", file=sys.stderr)
        return 2
    if not embedding_model:
        print("缺少环境变量: OPENAI_EMBEDDING_MODEL", file=sys.stderr)
        return 2

    index_path = Path(args.index)
    if not index_path.exists():
        print(f"索引文件不存在: {index_path}", file=sys.stderr)
        print("先运行: uv run python scripts/build_faq_index.py", file=sys.stderr)
        return 2

    index = json.loads(index_path.read_text(encoding="utf-8"))
    client = OpenAI(api_key=api_key, timeout=60.0)

    matches = search_faq(
        client=client,
        embedding_model=embedding_model,
        index=index,
        query=args.query,
        top_k=args.top_k,
    )

    sources = format_sources(matches)
    model_input = f"""用户问题:
{args.query}

FAQ 资料:
---
{sources}
---"""

    response = client.responses.create(
        model=chat_model,
        instructions=(
            "根据提供的 FAQ 资料回答用户问题。"
            "只能使用 FAQ 资料里的信息。"
            "如果资料里没有答案,就回答“资料里没有找到相关答案”。"
            "回答末尾列出使用的 FAQ id。"
        ),
        input=model_input,
        max_output_tokens=500,
    )

    print(response.output_text.strip())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

运行前要先有索引文件:

bash
uv run python scripts/build_faq_index.py

然后问一个问题:

bash
uv run python scripts/answer_faq.py "密码记错了怎么登录"

输出可能像这样:

text
可以在登录页点击“忘记密码”,然后通过邮箱验证码重置密码。

使用的 FAQ id: faq-001

这个回答不是模型自己凭空记起来的,而是来自检索出来的 faq-001

五、保存检索证据

RAG 脚本只打印回答,排查时会少一块信息:这次到底检索到了哪些 FAQ。

把检索结果也打印到 stderr

python
for item in matches:
    print(
        f"match {item['score']:.4f} {item['id']} {item['question']}",
        file=sys.stderr,
    )

放在调用模型前:

python
    matches = search_faq(
        client=client,
        embedding_model=embedding_model,
        index=index,
        query=args.query,
        top_k=args.top_k,
    )

    for item in matches:
        print(
            f"match {item['score']:.4f} {item['id']} {item['question']}",
            file=sys.stderr,
        )

这样运行时会看到:

text
match 0.8421 faq-001 忘记密码怎么办?
match 0.5120 faq-002 如何修改绑定邮箱?
match 0.4018 faq-004 如何关闭消息通知?

如果回答不对,先看这里。检索结果已经错了,问题在检索;检索结果对但回答乱了,问题在提示词或生成阶段。

六、找不到答案

RAG 最容易让人误会的一点是:检索到了相近资料,不等于资料里有答案。

比如用户问:

text
发票抬头能不能改?

当前 FAQ 里只有:

text
发票在哪里下载?

embedding 可能会把这条排上来,因为都和发票有关。但“在哪里下载”和“能不能改抬头”不是一个问题。模型如果只看关键词,可能会硬凑一个答案。

提示词里写“资料里没有答案就说明没有”,就是为了处理这种场景。更稳一点,还可以给检索分数加一个很粗的门槛:

python
if matches and matches[0]["score"] < 0.5:
    print("资料里没有找到相关答案")
    return 0

0.5 不是通用标准,只是本地 FAQ 的一个起点。分数阈值要靠实际问题观察,不能拿一个数字到处套。

七、引用片段

回答里只列 FAQ id 有时还不够。更好排查的做法是把命中的原文也保存下来。

可以把回答和检索结果整理成 JSON:

python
result = {
    "query": args.query,
    "answer": response.output_text.strip(),
    "matches": [
        {
            "score": item["score"],
            "id": item["id"],
            "question": item["question"],
            "answer": item["answer"],
        }
        for item in matches
    ],
}

写入文件:

python
output_path = Path("outputs/rag-answer.json")
output_path.write_text(
    json.dumps(result, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

保存后的结构大概是:

json
{
  "query": "密码记错了怎么登录",
  "answer": "可以在登录页点击“忘记密码”,然后通过邮箱验证码重置密码。\n\n使用的 FAQ id: faq-001",
  "matches": [
    {
      "score": 0.8421,
      "id": "faq-001",
      "question": "忘记密码怎么办?",
      "answer": "可以在登录页点击忘记密码,通过邮箱验证码重置密码。"
    }
  ]
}

这样回看一条回答时,不只看回答文字,还能看到它当时引用了哪些资料。

八、上下文长度

--top-k 不是越大越好。放进去的 FAQ 越多,模型看到的资料越杂,也越容易被不相关内容干扰。

比如问题是:

text
密码记错了怎么登录

前 1 条 FAQ 已经能回答:

text
忘记密码怎么办?

如果把 20 条 FAQ 全塞进去,里面可能有邮箱、发票、通知、订单,各种内容都混在一起。模型虽然能读,但回答时反而更容易绕。

小 FAQ 里 top-k=3 比较好观察:第一条是不是命中,后两条是不是明显偏了。资料多起来后,通常还要配合分数阈值、分类过滤、来源过滤这些规则。

九、最小 RAG 结构

这个 FAQ RAG 脚本只有几个对象:

text
faq.json
outputs/faq-index.json
scripts/build_faq_index.py
scripts/answer_faq.py

它们的关系很清楚:

build_faq_index.py 负责把资料变成索引,资料改了就重新跑。answer_faq.py 负责拿问题去索引里找相近 FAQ,再把命中的内容交给模型回答。

这一版没有数据库,没有向量库,也没有网页界面。FAQ 只有几条时,用 JSON 文件就能看清楚 RAG 的形状:检索在生成之前,模型回答只能看到 Python 放进去的资料。