搭一个完全在本机运行的知识库问答：用 Qwen3-Embedding 0.6B 计算向量、检索 top-3 片段

搭一个完全在本机运行的知识库问答：用 Qwen3-Embedding 0.6B 计算向量、检索 top-3 片段，用 Qwen3 生成回答，并用 [1][2] 标注来源。附带一个检索评测：8 个问题，对比查询时“加指令前缀”和“不加”的命中率。新旧规定冲突时，按生效日期把新规定排在前面。

先看清成品（全部在 box 上实测：Ollama 加 Qwen3-Embedding-0.6B Q8_0 和 Qwen3-0.6B）

• 测试： pytest 2 项通过。

• 检索评测：

  方式	top-1 命中	top-3 命中
  不加指令	6/8	8/8
  加查询指令	8/8（早一次运行是 7/8）	8/8

• 问答： 问“现在住宿一线城市每晚最多报多少”，回答 700 元，来源是新的补充规定。第一版答成了旧规定的 600 元（见调试示例）。

合规提醒：Qwen3 系列开源模型是 Apache-2.0 许可（以 Hugging Face 模型页为准），商用前请自己再确认一遍。本地部署时资料不出本机，但 Ollama 的 11434 和 vLLM 的 8000 端口不要暴露到公网；vLLM 默认不鉴权，本文强制设置 --api-key。用百炼（DashScope）云端 API 时，数据会发到阿里云，敏感数据先脱敏。价格以阿里云百炼的计费页为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

Qwen3-Embedding 模型卡片的用法：查询前面加 Instruct: 任务描述\nQuery: 前缀，文档不加。

文件结构

44-qwen-A-src/
├── rag.py
├── eval.json
├── test_rag.py
└── docs/（4 份 .md）

第 1 步：代码和资料

rag.py：

"""本地知识库问答（全部在自己电脑上运行，资料不出本机）：Qwen3-Embedding 做向量检索 + Qwen3 生成回答。
依赖 Ollama：嵌入模型和对话模型的名字用环境变量 EMBED_MODEL / CHAT_MODEL 指定。
用法：
  python rag.py index docs/            切块、计算向量，存到 index.json
  python rag.py ask "问题"              检索 top-3 片段并回答，回答里用 [1][2] 标注来源
  python rag.py eval eval.json          只评测检索：正确文件是否出现在 top-3（对比加/不加查询指令）"""
import json, math, os, re, sys
from pathlib import Path
import requests

OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
EMBED = os.environ.get("EMBED_MODEL", "qwen3-embedding-local:0.6b")
CHAT = os.environ.get("CHAT_MODEL", "qwen3-local:0.6b")
TASK = "根据问题，检索能回答该问题的公司制度或客服知识片段"

def embed(texts):
    r = requests.post(f"{OLLAMA}/api/embed", json={"model": EMBED, "input": texts}, timeout=600); r.raise_for_status()
    return r.json()["embeddings"]

def query_text(q, instruct=True):
    # Qwen3-Embedding 模型卡：查询一侧加 “Instruct: 任务\nQuery:问题”，文档一侧不加
    return f"Instruct: {TASK}\nQuery:{q}" if instruct else q

def chunks(path):
    """按 Markdown 小标题切块，每块带上文档标题，避免“1. 每晚不超过 600 元”这种脱离上下文的碎片"""
    text = Path(path).read_text(encoding="utf-8"); title = text.splitlines()[0].lstrip("# ").strip()
    for part in re.split(r"\n(?=## )", text):
        body = part.strip()
        if body and not body.startswith("# "): yield f"{title} / {body}"
        elif body.count("\n") > 0 and not re.search(r"\n## ", text): yield body   # 没有小标题的短文档，整篇一块

def effective_date(text):
    """从“自 2026 年 9 月 1 日起”这类句子里取生效日期，没有就返回空字符串"""
    m = re.search(r"自\s*(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日起", text)
    return f"{m[1]}-{int(m[2]):02d}-{int(m[3]):02d}" if m else ""

def cos(a, b): return sum(x * y for x, y in zip(a, b)) / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))

def search(index, q, k=3, instruct=True):
    qv = embed([query_text(q, instruct)])[0]
    return sorted(index, key=lambda c: -cos(qv, c["vec"]))[:k]

def main():
    cmd = sys.argv[1]
    if cmd == "index":
        items = [{"src": p.name, "text": c} for p in sorted(Path(sys.argv[2]).glob("*.md")) for c in chunks(p)]
        for it, v in zip(items, embed([it["text"] for it in items])): it["vec"] = v
        json.dump(items, open("index.json", "w"), ensure_ascii=False); print(f"✓ {len(items)} 个片段已写入 index.json")
    index = json.load(open("index.json")) if cmd != "index" else None
    if cmd == "ask":
        # 有生效日期的（较新的补充规定）排在前面并明确标注，小模型才不会拿旧条款回答
        hits = sorted(search(index, sys.argv[2]), key=lambda h: effective_date(h["text"]), reverse=True)
        label = lambda h: f"（来自 {h['src']}，{'自 ' + effective_date(h['text']) + ' 起生效的新规定，和旧条款冲突时以此为准' if effective_date(h['text']) else '未注明日期'}）"
        ctx = "\n\n".join(f"[{i}] {label(h)}{h['text']}" for i, h in enumerate(hits, 1))
        r = requests.post(f"{OLLAMA}/v1/chat/completions", timeout=600, json={"model": CHAT, "temperature": 0, "messages": [
            {"role": "system", "content": "只根据给出的资料回答，用 [编号] 标出依据；资料之间冲突时以日期更新的为准；资料里没有就说不知道。/no_think"},
            {"role": "user", "content": f"资料：\n{ctx}\n\n问题：{sys.argv[2]}"}]}).json()
        ans = re.sub(r"<think>.*?</think>", "", r["choices"][0]["message"]["content"], flags=re.S).strip()
        print(ans); print("\n检索到的片段：" + "；".join(f"[{i}] {h['src']}" for i, h in enumerate(hits, 1)))
        bad = [n for n in re.findall(r"\[(\d+)\]", ans) if not 1 <= int(n) <= len(hits)]
        if bad: print(f"⚠ 回答引用了不存在的编号 {bad}")
    if cmd == "eval":
        cases = json.load(open(sys.argv[2], encoding="utf-8"))
        for instruct in (False, True):
            ok = sum(any(h["src"] == c["src"] for h in search(index, c["q"], 3, instruct)) for c in cases)
            top1 = sum(search(index, c["q"], 1, instruct)[0]["src"] == c["src"] for c in cases)
            print(f"{'加查询指令' if instruct else '不加指令  '}：top-1 命中 {top1}/{len(cases)}，top-3 命中 {ok}/{len(cases)}")

if __name__ == "__main__":
    main()

eval.json：

[{"q": "去北京出差酒店一晚能报多少钱", "src": "报销补充说明.md"},
 {"q": "工作满三年有多少天年假", "src": "员工手册.md"},
 {"q": "密码多久要改一次", "src": "IT支持.md"},
 {"q": "周末买的东西什么时候寄出", "src": "网店客服FAQ.md"},
 {"q": "在家连不上内网怎么办", "src": "IT支持.md"},
 {"q": "生病请假要什么材料", "src": "员工手册.md"},
 {"q": "可以开发票吗", "src": "网店客服FAQ.md"},
 {"q": "迟到了怎么补救", "src": "员工手册.md"}]

test_rag.py：

from rag import effective_date, query_text

def test_effective_date():
    assert effective_date("自 2026 年 9 月 1 日起，出差住宿标准调整为") == "2026-09-01"
    assert effective_date("一线城市每晚不超过 600 元") == ""

def test_query_instruction_only_on_query():
    assert query_text("年假").startswith("Instruct: ") and query_text("年假", False) == "年假"

docs/IT支持.md：

# IT 支持说明（虚构，仅用于测试）
## 账号
1. 新员工入职当天由 IT 开通邮箱和 OA 账号，初始密码通过短信发送。
2. 密码每 90 天必须修改一次，不能与最近 3 次相同。
## 设备
1. 笔记本电脑由公司统一采购，离职时需归还。
2. 打印机在 3 楼茶水间旁，扫描件会发到个人邮箱。
## VPN
在家办公需使用公司 VPN，客户端在内网门户下载；遇到连接失败先检查系统时间是否准确。

docs/员工手册.md：

# 示例公司员工手册（虚构，仅用于测试）

## 一、考勤
1. 工作时间为周一至周五 9:30–18:30，午休 12:00–13:30。
2. 每月可以申请 2 次“弹性迟到”，每次不超过 30 分钟，需在当天 10:00 前在 OA 提交。

## 二、休假
1. 入职满 1 年的员工，每年带薪年假 10 天；满 5 年为 15 天。
2. 年假可以顺延到次年 3 月 31 日，逾期作废。
3. 病假需提供医院证明，3 天以内由直属主管审批。

## 三、报销
1. 出差住宿标准：一线城市每晚不超过 600 元，其他城市不超过 400 元。
2. 报销需在费用发生后 30 天内提交，附电子发票。

docs/报销补充说明.md：

# 报销补充说明（虚构，仅用于测试）
自 2026 年 9 月 1 日起，出差住宿标准调整为：一线城市每晚不超过 700 元，其他城市不超过 450 元。其余规定不变。

docs/网店客服FAQ.md：

# 示例小店客服知识库（虚构）
- 发货：工作日 16:00 前下单当天发货，周末订单周一发货。默认中通快递。
- 退换：签收 7 天内可无理由退货，商品需保持完好；运费由买家承担，质量问题除外。
- 发票：可开电子普通发票，在订单页“申请开票”填写抬头和税号。
- 营业时间：人工客服 9:00–21:00。

第 2 步：运行

$ python rag.py index docs/
✓ 8 个片段已写入 index.json
$ python rag.py eval eval.json
不加指令  ：top-1 命中 6/8，top-3 命中 8/8
加查询指令：top-1 命中 8/8，top-3 命中 8/8
$ CHAT_MODEL=qwen3-local:0.6b python rag.py ask "现在出差住宿一线城市每晚最多报多少？"
[1] 一线城市每晚不超过 700 元。

检索到的片段：[1] 报销补充说明.md；[2] 员工手册.md；[3] 员工手册.md

检查点：用你自己的资料时，也先写 10 个左右的评测问题。改了切块方式或换了模型，先看命中率有没有下降。

调试示例：新旧规定冲突

第一版检索到的 top-3 片段里，旧手册（600 元）排在补充规定（700 元）前面，qwen3 0.6B 回答了 600 元。检索本身没有错，两份文件都相关，问题在于模型不知道哪一份更新。

改法：effective_date() 从片段里提取“自 X 年 X 月 X 日起”这样的生效日期，生效日期新的排在前面，并在提示词里注明“同一事项以生效日期最新的为准”。改完后 qwen3 0.6B 回答 700 元。

但 qwen2.5 1.5B 在同样的提示下只输出了“[1]”。可见本地小模型对提示词的遵守程度差别很大，换模型就要重新测一遍。

常见问题

模板规则

• 先有评测问题，再调整检索。
• 新旧规定要能按生效日期排序。
• 资料和模型都留在本机。

项目资源

本地 RAG 的效果不靠感觉判断，靠那几十个评测问题的命中率。