秘塔 AI：做一个“有出处的回答”，用本地模型生成并核对引用

做一个“有出处的回答”：先搜索，读取排名前 2 的网页正文，交给本地模型回答，并用 [n] 标注来源。程序再做两道核对：引用编号是否存在、每段是否都有引用；以及引用的那句话和对应资料的字面重合度（bigram 重合，阈值 0.35），以此识别“引错了来源”。

先看清成品

• 测试： pytest 5 项通过。
• 模拟服务器实测： 两个本地小模型都没能稳定通过核对，程序每次都标出了问题。
  • qwen2.5 1.5B 照抄资料，不写引用。
  • qwen3 0.6B 有时引错来源。
• 重合度检查的局限： 它只能发现字面上明显对不上的引用，不能证明引用一定正确。

没有在 box 上跑的部分： box 上没有可用的秘塔 API Key，唯一对真实接口做过的调用是用无效 Key 测错误返回。

• 实测结果： 错误也是 HTTP 200，body 里带 errCode：
  • search 返回 errCode 2005“API密钥无效”。
  • reader 返回 errCode 500“服务器错误”。
• 其余部分： 都在我自己写的模拟服务器 fake_metaso.py 上跑。它返回的是 3 条虚构结果。
• 字段名的来源： 响应字段（webpages 里的 title、link、snippet、score、date）参考的是开放平台 playground 页面和第三方集成的说明，没有找到正式的官方字段文档，接入前请先用真实 Key 打印一次完整响应。

合规提醒：METASO_API_KEY 只放在环境变量里。搜索结果和网页正文的版权属于原作者：笔记里保留来源链接，引用时注明出处，不要整篇转载。使用读取网页接口时，遵守对方网站的规则（服务条款、robots），不要批量抓取，也不要抓取需要登录才能看的页面。调用会消耗积分，价格和积分规则以秘塔开放平台的页面为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

同 45-metaso-B：秘塔 API Key 和本地 Ollama 模型（用 ANS_MODEL 指定）。

文件结构

45-metaso-C-src/
├── answer_with_sources.py
├── metaso_search.py
└── test_check.py

第 1 步：代码

answer_with_sources.py：

"""“自己的 Perplexity”：秘塔搜索拿来源 → 读取前几条网页正文 → 本地模型写带 [编号] 引用的回答 → 程序核对引用。
核对规则：引用编号必须存在；每个段落至少有一个引用；列出没被引用的来源。不合格就提示人工核对。
环境变量：METASO_API_KEY、METASO_BASE、OLLAMA_URL、ANS_MODEL（默认 qwen2.5-local:1.5b）
用法：python answer_with_sources.py "问题" [--read 2]"""
import argparse, os, re, sys
import requests
from metaso_search import MetasoError, post, search

OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("ANS_MODEL", "qwen2.5-local:1.5b")

def gather(q, n_read):
    pages = (search(q, size=5).get("webpages") or [])
    srcs = []
    for i, p in enumerate(pages):
        body = p.get("snippet") or ""
        if i < n_read:
            try: body = post("reader", {"url": p["link"]}, accept="text/plain")[:2000]
            except (MetasoError, requests.RequestException) as e: print(f"  读取 {p['link']} 失败，改用摘要：{e}", file=sys.stderr)
        srcs.append({"title": p.get("title", ""), "link": p.get("link", ""), "date": p.get("date") or "", "text": body})
    return srcs

def answer(q, srcs):
    ctx = "\n\n".join(f"[{i}] {s['title']}（{s['date'] or '日期未知'}）\n{s['text']}" for i, s in enumerate(srcs, 1))
    r = requests.post(f"{OLLAMA}/v1/chat/completions", timeout=600, json={"model": MODEL, "temperature": 0, "reasoning_effort": "none",
        "messages": [{"role": "system", "content": "根据给出的资料回答问题。写 2~4 个短段落，每个段落末尾用 [编号] 标出依据的资料，"
                                                   "可以写多个如 [1][3]。资料没有的内容不要写，也不要照抄资料的标题和列表。\n"
                                                   "格式：每句话写完后紧跟它依据的编号，例如“……的结论 [2]。”"},
                     {"role": "user", "content": f"资料：\n{ctx}\n\n问题：{q}"}]})
    return re.sub(r"<think>.*?</think>", "", r.json()["choices"][0]["message"]["content"], flags=re.S).strip()

def bigrams(t):
    t = re.sub(r"[\s\[\]\d，。、；：（）“”,.;:()!?！？-]", "", t)
    return {t[i:i + 2] for i in range(len(t) - 1)}

def support(ans, srcs, threshold=0.35):
    """逐句检查：句子的字二元组有多少出现在它引用的资料里。低于阈值 → 这条引用可能对不上（不是严格证明，只是筛查）"""
    weak = []
    # 行尾的 “……。[2]” 统一成 “……[2]。”，否则句号后的引用会被漏查；
    # 只处理行尾：句中的 “。 [2] 中提到……” 是下一句开头的引用，不能挪到上一句
    ans = re.sub(r"([。！？])[ \t]*((?:\[\d+\])+)[ \t]*(?=\n|$)", r"\2\1", ans)
    for sent in re.findall(r"[^。！？\n]+?(?:\[\d+\])+", ans):
        nums = [int(x) for x in re.findall(r"\[(\d+)\]", sent)]
        b = bigrams(re.sub(r"\[\d+\]", "", sent))
        if len(b) < 8: continue          # “资料 [3] 提到”这种引导语太短，不算一句话，跳过
        if not all(1 <= k <= len(srcs) for k in nums): continue
        best = max(len(b & bigrams(srcs[k - 1]["title"] + srcs[k - 1]["text"])) / len(b) for k in nums)
        if best < threshold: weak.append(f"“{sent.strip()[:40]}” 和资料 {nums} 的重合度只有 {best:.0%}")
    return weak

def check(ans, n):
    paras = [p for p in re.split(r"\n\s*\n", ans) if p.strip()]
    cited = {int(x) for x in re.findall(r"\[(\d+)\]", ans)}
    problems = [f"引用了不存在的 [{c}]" for c in sorted(cited) if not 1 <= c <= n]
    problems += [f"第 {i} 段没有引用：{p[:30]}…" for i, p in enumerate(paras, 1) if not re.search(r"\[\d+\]", p)]
    unused = [i for i in range(1, n + 1) if i not in cited]
    return problems, unused

if __name__ == "__main__":
    ap = argparse.ArgumentParser(); ap.add_argument("q"); ap.add_argument("--read", type=int, default=2); a = ap.parse_args()
    try: srcs = gather(a.q, a.read)
    except MetasoError as e: sys.exit(f"✗ {e}")
    if not srcs: sys.exit("没有搜到结果")
    ans = answer(a.q, srcs); problems, unused = check(ans, len(srcs)); problems += support(ans, srcs)
    print(ans + "\n\n来源：")
    for i, s in enumerate(srcs, 1): print(f"[{i}] {s['title']} {s['link']}" + ("（未被引用）" if i in unused else ""))
    print("\n" + ("✓ 引用核对通过" if not problems else "⚠ 引用核对没通过，请人工核对：\n  " + "\n  ".join(problems)))

test_check.py：

from answer_with_sources import check

def test_good():
    assert check("A 很好 [1]。\n\nB 也行 [2][3]。", 3) == ([], [])

def test_bad_citation_and_uncited_paragraph():
    problems, unused = check("A [5]。\n\n没有引用的段落。", 3)
    assert any("[5]" in p for p in problems) and any("第 2 段" in p for p in problems) and unused == [1, 2, 3]

def test_support_flags_wrong_source():
    from answer_with_sources import support
    srcs = [{"title": "选购", "text": "CPU：需要转码 4K 视频时选带核显的 Intel 型号。"},
            {"title": "RAID", "text": "RAID1 是镜像，两块盘存同样数据。"}]
    assert support("转码 4K 视频要选带核显的 Intel 型号 [1]。", srcs) == []
    assert len(support("转码 4K 视频要选带核显的 Intel 型号 [2]。", srcs)) == 1

def test_support_checks_citation_after_period():
    from answer_with_sources import support
    srcs = [{"title": "选购", "text": "CPU：需要转码 4K 视频时选带核显的 Intel 型号。"}, {"title": "RAID", "text": "RAID1 是镜像。"}]
    assert len(support("转码 4K 视频要选带核显的 Intel 型号。[2]", srcs)) == 1

def test_leading_citation_not_moved_to_previous_sentence():
    from answer_with_sources import support
    srcs = [{"title": "a", "text": "硬盘位至少两盘位，方便做 RAID1。"}, {"title": "b", "text": "RAID5 至少三块盘，可以容忍一块盘损坏。"}]
    assert support("硬盘位至少两盘位，方便做 RAID1。 [2] 中提到 RAID5 至少三块盘，可以容忍一块盘损坏[2]。", srcs) == []

第 2 步：运行

$ ANS_MODEL=qwen3-local:0.6b python answer_with_sources.py "家用 NAS 怎么选"
资料 [3] 提到，不要将 NAS 管理界面直接暴露到公网，优先使用 VPN 或 Tailscale 进行远程访问，以确保网络安全。……

来源：
[1] 示例：家用 NAS 选购要点 http://127.0.0.1:18400/page/1（未被引用）
[2] 示例：RAID1 和 RAID5 的区别 http://127.0.0.1:18400/page/2
[3] 示例：NAS 远程访问安全 http://127.0.0.1:18400/page/3

⚠ 引用核对没通过，请人工核对：
  第 1 段没有引用：家用 NAS 选购要点（2026-09-12）中提到，硬盘位…

$ ANS_MODEL=qwen2.5-local:1.5b python answer_with_sources.py "家用 NAS 怎么选"
家用 NAS 选购要点（2026-09-12）
……
⚠ 引用核对没通过，请人工核对：
  第 1 段没有引用：家用 NAS 选购要点（2026-09-12）…

检查点：通不过核对的回答不要直接用。程序的作用就是把这类回答拦下来，让人去看原文。

调试示例

1. 照抄资料，还抄了示例。 qwen2.5 1.5B 多次原样照抄资料 1，不写引用，被“每段需要有引用”这条规则标出。我在提示词里给了一个具体的示例句，模型干脆把示例句抄了进去。后来把示例改成抽象的格式说明。

2. 引错来源。 qwen3 0.6B 写出了“转码 4K 要选带核显的型号[2]”，但 [2] 讲的是 RAID。只检查编号是否存在是查不出来的，所以加了 support()：计算这句话和对应资料的 bigram 重合度，低于 0.35 就提示。

3. 自己的检查写错了两次。 用旧的错误回答回归测试 support() 时，它一条都没有报，原因是模型写的是“……。[2]”，引用在句号后面，被正则漏掉了。我加了一步统一格式，但又把句首的“。 [2] 中提到……”错误地挪到了上一句，导致误报。最后改成只处理行尾的引用，并且跳过“资料 [3] 提到”这种太短的引导语。这三种情况现在都有对应的测试。

常见问题

模板规则

• 回答必须带引用，由程序核对编号和每段的引用。
• 重合度检查只是报警，不是判决。
• 核对规则本身也要有回归测试。

项目资源

检查有没有出处的程序自己也会出错，所以它本身也得配上测试。