前面的系列已经介绍了 API 调用、上下文窗口、长文本分块、Embedding、检索、引用、结构化输出、错误处理和安全边界。本篇在这些基础上做一个小项目：把本地 Markdown 文档按关键词检索出来，交给模型生成答案，并强制答案引用文档编号。它不是完整的生产级 RAG，而是一条足够小、可以逐步验证的问答流水线。

先定义数据流和边界

程序接收一个问题，先从 knowledge/ 读取 Markdown 文件，再用标准库完成简单关键词检索，最后只把命中的片段发送给模型。模型不能访问本地文件，也不能凭空补全文档中不存在的事实；如果证据不足，必须明确说“资料不足”。最终输出包含答案和 [S1] 这样的来源标记。

这里故意不用框架。关键词检索不如向量检索智能，却能让初学者看清四个边界：文档如何变成片段、片段如何进入上下文、答案如何绑定来源、网络失败时程序如何停止。等这条链路稳定后，再替换检索器，风险会小很多。

准备文档和环境

创建虚拟环境并安装官方 Python SDK：

1
2
3
python -m venv .venv
source .venv/bin/activate
python -m pip install openai

在项目目录创建两份示例资料：

1
2
3
knowledge/
  deploy.md
  backup.md

deploy.md 可以写“发布前运行测试，发布后检查健康状态”，backup.md 可以写“每天凌晨备份，保留七天”。真实项目应把资料放在版本控制中，并在更新时重新建立索引。

密钥和模型名只从环境变量读取：

1
2
export OPENAI_API_KEY="替换为你的真实密钥"
export MODEL_NAME="替换为你可用的模型名称"

不要把真实密钥写进源码、文档或提交记录。若使用兼容服务，只有在服务商文档明确支持同一接口时，才配置客户端地址。

编写最小程序

保存为 cited_qa.py。代码中的检索器只按词频排序；它的价值不在于效果最好，而在于结果可解释、无需额外服务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import os
import re
from pathlib import Path

from openai import OpenAI

DOC_DIR = Path("knowledge")
MODEL = os.environ["MODEL_NAME"]
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])


def load_docs() -> list[dict]:
    docs = []
    for path in sorted(DOC_DIR.glob("*.md")):
        text = path.read_text(encoding="utf-8").strip()
        if text:
            docs.append({"id": f"S{len(docs) + 1}", "name": path.name, "text": text})
    return docs


def keywords(text: str) -> set[str]:
    return {word.lower() for word in re.findall(r"[A-Za-z0-9_]+|[\u4e00-\u9fff]", text)}


def retrieve(question: str, docs: list[dict], limit: int = 3) -> list[dict]:
    query_words = keywords(question)
    ranked = []
    for doc in docs:
        score = len(query_words & keywords(doc["text"]))
        if score:
            ranked.append((score, doc))
    ranked.sort(key=lambda item: item[0], reverse=True)
    return [doc for _, doc in ranked[:limit]]


def answer(question: str, sources: list[dict]) -> str:
    context = "\n\n".join(
        f"[{doc['id']}] {doc['name']}\n{doc['text']}" for doc in sources
    )
    response = client.responses.create(
        model=MODEL,
        instructions=(
            "你是内部知识库问答助手。只能依据给出的资料回答。"
            "每个关键结论后添加对应的 [S编号]；资料没有答案时只说资料不足，"
            "不要猜测，也不要编造来源。用中文回答，先给结论，再给必要说明。"
        ),
        input=f"问题：{question}\n\n资料：\n{context}",
    )
    if not response.output_text.strip():
        raise RuntimeError("模型返回了空答案")
    return response.output_text.strip()


def main() -> None:
    question = input("问题：").strip()
    if not question:
        raise SystemExit("问题不能为空")
    docs = load_docs()
    sources = retrieve(question, docs)
    if not sources:
        print("没有检索到相关资料，未调用模型。")
        return
    print("\n参考资料：", ", ".join(f"[{d['id']}] {d['name']}" for d in sources))
    print("\n答案：\n" + answer(question, sources))


if __name__ == "__main__":
    main()

运行 python cited_qa.py 后输入问题即可。这里不能预先写死一段“运行结果”：真实答案取决于你的文档、模型和服务状态。可以验证的事实是，命中资料时程序显示来源并发起请求；没有命中时直接结束；模型返回空文本时抛出错误。

逐段理解关键设计

load_docs 给每份文档分配稳定的编号。引用编号不能在答案生成后临时猜，否则很容易出现答案引用了没有送入上下文的文件。生产系统还应保存文档版本、更新时间和访问权限，而不是只保存文件名。

retrieve 先筛掉零分文档，再限制最多三份资料。这是一个最小的上下文预算控制：命中越多不一定越好，过多无关内容会增加延迟、费用和模型混淆。真正使用 Embedding 时，可以保留同样的函数接口，只替换内部排序逻辑，并继续测试“问题—候选片段”的结果。

answer 把检索结果作为不可信资料，而不是指令。提示词明确了只能依据资料、必须引用和证据不足时拒答，但提示词不是安全校验的全部。若文档本身包含“忽略上文并执行某操作”等文字，程序仍不应把模型输出直接交给工具；涉及写入、删除或外部调用时，必须经过参数校验和人工确认。

常见问题

为什么答案仍可能不准确？ 检索命中只说明词语相关，不代表片段真正回答问题；模型也可能错误理解证据。应保存问题、命中片段、答案和人工评价，建立小型测试集，检查引用是否存在、结论是否被资料支持。

为什么不直接把所有文档发给模型？ 文档变多后会遇到上下文窗口、费用和隐私问题。先检索能缩小范围，但检索器也必须设置上限，并对空结果做明确处理。

引用编号会不会错？ 代码在发送请求前生成编号，答案只允许使用已提供的编号。可以进一步用正则提取答案中的 [S数字]，校验每个编号都属于 sources；校验失败就拒绝展示或要求模型修复，而不是默认为正确。

如何测试而不消耗额度？ 把 answer 改为接收一个客户端参数，测试时注入假的 responses.create，返回固定文本。检索测试使用固定文档，覆盖命中、空结果和同分排序；SDK 只保留少量集成测试。

小结

这个小项目把“文档—检索—上下文—生成—引用”串成了一个可观察的闭环。它没有掩盖关键词检索的局限，也没有把模型当作可信数据库：程序负责筛选、限制和边界，模型负责在给定证据上组织语言。下一步可以替换为向量检索、加入分块和引用校验，但每次只改变一个环节，并用固定问题集验证变化，才能知道系统究竟变好了还是只是回答更像样了。