MCP Agent 的来源感知验证：事实正确还不够，证据必须对得上

2026-09-29 23 预计阅读时间: 1 分钟
来源: huggingface.co AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：10 分钟

当 Agent 通过 MCP 调用搜索、数据库、知识库或业务系统时，“答案看起来正确”并不等于“答案得到了正确来源的支持”。同一句事实可能来自官方文档、过期缓存、二手博客，甚至来自与结论无关的页面。真正可靠的验证，需要同时检查结论、证据和来源之间的绑定关系。

两种正确性不能混为一谈

事实验证通常只问一个问题：这句话是真的吗？来源感知验证还要继续追问：

  • 证据是否来自任务允许使用的来源？
  • 来源是官方文档、内部数据库，还是未经审核的网页？
  • 证据是否直接支持当前结论，而不只是主题相近？
  • 内容是否足够新，适用于指定版本或时间点？
  • Agent 展示的引用，是否就是生成答案时实际使用的材料？

例如，“某 API 支持流式响应”可能确实为真，但如果任务要求核对当前生产环境的能力，引用两年前的博客并不合格。问题不在事实真假，而在来源的权限、时效性和适用范围。

因此，验证结果不应只有一个布尔值。更实用的模型至少包含：

claim correctness     结论是否被证据支持
source authority      来源是否具有足够权威性
source eligibility    来源是否符合当前任务策略
freshness             内容是否仍在有效期内
claim-source binding  引文是否直接对应结论
traceability          能否追溯到工具调用和原始记录

把来源当作结构化数据，而不是答案末尾的一串 URL

MCP 服务器的具体返回结构会因实现而异。可以这样实践：在应用层为工具结果增加统一的“证据信封”。下面的字段并非声称属于 MCP 协议标准，而是一种可改造的设计：

{
  "tool_call_id": "call-0182",
  "server": "internal-docs",
  "resource_uri": "kb://platform/api/streaming",
  "retrieved_at": "2025-03-08T10:30:00Z",
  "content_updated_at": "2025-02-21T09:00:00Z",
  "authority": "official",
  "scope": {
    "product": "gateway",
    "version": "4.2",
    "environment": "production"
  },
  "content": "Gateway 4.2 supports server-sent event streaming...",
  "content_sha256": "..."
}

这里有几个关键设计决定：

  1. resource_uri 标识真正被读取的资源，而不是搜索结果页。
  2. retrieved_at 和 content_updated_at 分开记录，避免把“刚刚抓取”误认为“内容最新”。
  3. scope 描述证据适用的产品、版本和环境。
  4. tool_call_id 将引用绑定到一次实际调用，便于审计。
  5. 内容摘要用于检测证据在验证后是否发生变化，但它不能证明来源本身可信。

如果 MCP 工具只返回自然语言，客户端很难稳定执行上述检查。更好的做法是让服务器返回文本内容与来源元数据，并在进入模型上下文前保留原始响应。

一个可运行的来源策略检查器

下面的 Python 示例只使用标准库，可以直接保存为 verify_sources.py 后运行。它验证来源类型、域名、时效性和版本范围。它不会判断自然语言是否真的支持结论；语义蕴含检查应作为下一层处理。

from datetime import datetime, timezone
from urllib.parse import urlparse

POLICY = {
    "allowed_authorities": {"official", "internal"},
    "allowed_hosts": {"docs.example.com", "kb.example.internal"},
    "max_age_days": 180,
    "required_version": "4.2",
}

EVIDENCE = [
    {
        "claim_id": "claim-1",
        "resource_uri": "https://docs.example.com/gateway/streaming",
        "authority": "official",
        "content_updated_at": "2025-02-21T09:00:00Z",
        "scope": {"version": "4.2"},
        "quote": "Gateway 4.2 supports server-sent event streaming.",
    },
    {
        "claim_id": "claim-2",
        "resource_uri": "https://random-blog.example/gateway",
        "authority": "community",
        "content_updated_at": "2023-01-10T09:00:00Z",
        "scope": {"version": "3.1"},
        "quote": "Streaming is available in some gateway releases.",
    },
]


def parse_time(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def verify(item: dict, now: datetime) -> list[str]:
    errors = []
    host = urlparse(item["resource_uri"]).hostname

    if item["authority"] not in POLICY["allowed_authorities"]:
        errors.append(f"authority not allowed: {item['authority']}")

    if host not in POLICY["allowed_hosts"]:
        errors.append(f"host not allowed: {host}")

    age = now - parse_time(item["content_updated_at"])
    if age.days > POLICY["max_age_days"]:
        errors.append(f"source is stale: {age.days} days old")

    version = item.get("scope", {}).get("version")
    if version != POLICY["required_version"]:
        errors.append(
            f"version mismatch: expected {POLICY['required_version']}, got {version}"
        )

    if not item.get("quote", "").strip():
        errors.append("missing supporting quote")

    return errors


if __name__ == "__main__":
    # 固定时间便于复现实例；生产环境应改为 datetime.now(timezone.utc)
    now = datetime(2025, 3, 8, tzinfo=timezone.utc)

    for item in EVIDENCE:
        errors = verify(item, now)
        status = "PASS" if not errors else "FAIL"
        print(f"{item['claim_id']}: {status}")
        for error in errors:
            print(f"  - {error}")

运行命令：

python verify_sources.py

预期结果是第一条证据通过，第二条因为来源类型、域名、时效性和版本不符合策略而失败。接入真实 MCP Agent 时，可以把 EVIDENCE 替换为工具调用日志，并让策略来自 YAML 配置或集中式策略服务。

不要让模型自己给自己的引用打分

让同一个模型生成答案、挑选引用并宣布“验证通过”，容易形成自证循环。更稳妥的执行链可以拆成四步：

  1. 采集：保存 MCP 工具的原始响应和调用参数。
  2. 规范化：提取资源标识、更新时间、权限级别、版本和原文片段。
  3. 确定性检查：用代码验证域名白名单、时效、访问权限和作用域。
  4. 语义检查：逐条判断引文是否支持 claim，并明确标记“支持”“矛盾”或“证据不足”。

语义检查可以使用模型，但输入应限制为单条结论和对应证据，输出采用结构化格式。例如：

你是证据核验器。只依据给定引文判断结论，不使用外部知识。

结论：Gateway 4.2 支持 SSE 流式响应。
引文：Gateway 4.2 supports server-sent event streaming.

仅输出 JSON：
{"verdict":"supported|contradicted|insufficient","reason":"..."}

即使语义结果为 supported，来源策略失败时也不应将其升级为可信结论。反过来，官方来源也不必然支持具体说法：权威性不能替代文本蕴含关系。

上线前应明确的边界

来源感知验证会增加延迟、存储和实现成本，不必对所有回答使用同一强度。天气闲聊可以采用宽松策略，支付操作、合规查询和生产变更则应要求官方或内部来源，并保存完整审计轨迹。

上线时可以检查以下事项：

  • 每个重要 claim 是否都有独立证据，而不是整段答案共享一个引用；
  • 来源白名单是否按任务、租户和环境区分；
  • 是否同时记录检索时间与内容更新时间；
  • 引用能否追溯到 MCP 服务器、工具调用和资源标识；
  • 缺少合格证据时，Agent 是否会明确拒答或降低置信度；
  • 日志中是否避免保存密钥、个人信息和无关敏感内容；
  • 缓存内容更新后，旧答案是否能够被重新验证或失效。

可靠的 MCP Agent 不只是“说对了”，还要能够解释它依据了什么、为什么这个来源适用，以及证据不足时为何选择停下来。把来源元数据、确定性策略和逐结论验证纳入执行链，才能让引用从展示装饰变成真正的信任边界。

# AI # Hugging Face # LLM

相关推荐