从检索到推理:用知识图谱构建可投入生产的 Agentic AI 系统

2026-09-12 36 预计阅读时间: 1 分钟
来源: infoq.com 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.

预计阅读时间:11 分钟

基础 RAG 解决了“把相关文档找出来”的问题,但生产级 Agentic AI 还需要回答几个更难的问题:上下文为什么这样组合?系统依据了哪些事实?决策能否追溯?工程师能否看见代理正在做什么?Cassie Shum 的分享把知识图谱放在这些问题的交汇处,并总结了四种实用架构模式:上下文打包、决策溯源、代码即事实,以及代理可见性。

RAG 之后,系统缺的是什么

传统 RAG 通常围绕一条相似度检索链路展开:用户提问、召回文本、拼接上下文、生成答案。这种方式适合知识问答,但当代理需要规划任务、调用工具、评估结果并继续行动时,单纯的文本片段不够用了。

代理还需要知道实体之间的关系、事实的来源、约束条件和当前任务状态。知识图谱可以把这些信息组织成节点与边,例如:

  • 一个服务依赖哪些数据库和外部 API;
  • 一条业务规则适用于哪些客户或环境;
  • 一个决策引用了哪些事实和代码版本;
  • 某次工具调用产生了什么结果,并影响了后续哪一步。

这里的重点不是把所有数据都搬进图数据库,而是为代理提供比“相似文本列表”更稳定的结构化工作记忆。

四种架构模式

1. 上下文打包:按任务组装上下文

上下文打包不是把召回结果全部塞进提示词,而是围绕当前任务收集一组相关事实。一个上下文包可以包含实体属性、关系、约束、最近操作和来源信息。这样做有两个直接收益:减少无关 token,同时降低代理因为缺少关键关系而做出错误判断的概率。

2. 决策溯源:让每个结论都有来路

生产系统不能只记录最终答案。代理为什么选择某个工具、为什么拒绝某个操作、为什么认为某项检查已经完成,都应当保留可查询的证据链。

知识图谱适合表达这种链路:决策 -> 依据 -> 事实 -> 来源。来源可以是文档、数据库记录、代码提交、测试结果或人工反馈。出现问题时,工程师可以沿着关系回溯,而不必只依赖一段难以解释的模型输出。

3. 代码即事实:把可执行规则纳入知识体系

很多系统把业务规则写在文档里,却把真正决定行为的逻辑放在代码中,最终导致“文档说一套、运行时做另一套”。将关键策略、校验器、工具契约和测试结果作为图谱中的一等信息,可以让代理引用更接近运行时真实行为的事实。

这并不意味着让模型直接生成并执行任意代码。更稳妥的方式是:将经过审查的代码、版本、接口契约和测试状态登记到图谱中,代理只能在明确的权限边界内调用它们。

4. 代理可见性:把反馈回路变成系统能力

代理系统的可观测性应当覆盖计划、检索、工具调用、结果评估和重试,而不只是记录一次请求的输入输出。将这些事件关联到任务、决策和事实后,工程师可以观察:

  • 哪些节点经常导致检索失败;
  • 哪些工具调用消耗了大量 token 或时间;
  • 哪些反馈会触发重复规划;
  • 哪些事实已经过期或缺少来源。

这类可见性直接服务于工程反馈回路。系统优化不再只是调整提示词,还可以从图谱中发现上下文组织、工具设计和知识更新的问题。

一个可运行的最小工程模型

下面的示例只使用 Python 标准库,演示如何把四种模式压缩到一个小型内存图谱中。它不是完整的知识图谱数据库或 Agent 框架,但可以作为设计原型:运行后会根据任务打包上下文,记录决策依据,并输出代理事件。

将代码保存为 agent_graph_demo.py,然后运行 python agent_graph_demo.py

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any


@dataclass
class KnowledgeGraph:
    nodes: dict[str, dict[str, Any]] = field(default_factory=dict)
    edges: list[tuple[str, str, str]] = field(default_factory=list)

    def add_node(self, node_id: str, kind: str, **data: Any) -> None:
        self.nodes[node_id] = {"kind": kind, **data}

    def link(self, source: str, relation: str, target: str) -> None:
        self.edges.append((source, relation, target))

    def neighbors(self, node_id: str, relation: str | None = None) -> list[str]:
        return [
            target
            for source, edge_relation, target in self.edges
            if source == node_id and (relation is None or edge_relation == relation)
        ]


def build_graph() -> KnowledgeGraph:
    graph = KnowledgeGraph()
    graph.add_node("incident-42", "task", question="Can we restart payments-api?")
    graph.add_node("payments-api", "service", environment="production")
    graph.add_node("restart-policy-v3", "policy", action="restart", allowed=False)
    graph.add_node("policy.py@8f31c2", "code", path="policies/policy.py", tests="passed")
    graph.add_node("runbook-17", "document", title="Payments incident runbook")

    graph.link("incident-42", "targets", "payments-api")
    graph.link("payments-api", "governed_by", "restart-policy-v3")
    graph.link("restart-policy-v3", "implemented_by", "policy.py@8f31c2")
    graph.link("restart-policy-v3", "documented_by", "runbook-17")
    return graph


def bundle_context(graph: KnowledgeGraph, task_id: str) -> dict[str, Any]:
    target = graph.neighbors(task_id, "targets")
    if not target:
        raise ValueError(f"task has no target: {task_id}")

    service_id = target[0]
    policy_ids = graph.neighbors(service_id, "governed_by")
    policy_id = policy_ids[0] if policy_ids else None
    evidence = []
    if policy_id:
        evidence.extend(graph.neighbors(policy_id, "implemented_by"))
        evidence.extend(graph.neighbors(policy_id, "documented_by"))

    return {
        "task": graph.nodes[task_id],
        "service": graph.nodes[service_id],
        "policy": graph.nodes.get(policy_id),
        "evidence": [{"id": item, **graph.nodes[item]} for item in evidence],
    }


def decide(context: dict[str, Any]) -> dict[str, Any]:
    policy = context["policy"] or {}
    allowed = policy.get("action") == "restart" and policy.get("allowed") is True
    return {
        "action": "restart" if allowed else "request-human-approval",
        "allowed": allowed,
        "reason": "policy evidence does not allow an automatic restart",
        "evidence": [item["id"] for item in context["evidence"]],
    }


def emit(event: str, **data: Any) -> None:
    record = {
        "time": datetime.now(timezone.utc).isoformat(),
        "event": event,
        **data,
    }
    print(record)


if __name__ == "__main__":
    graph = build_graph()
    emit("context_bundled", task_id="incident-42")
    context = bundle_context(graph, "incident-42")
    decision = decide(context)
    emit("decision_made", decision=decision)
    print("Context:", context)

在真实系统中,可以把 KnowledgeGraph 替换为 Neo4j、Amazon Neptune 或现有的图存储,把 emit 接入日志、Tracing 或事件总线,再让代理通过受控工具读取上下文和提交决策。示例中的 policy.py@8f31c2 也可以扩展为代码提交、测试报告和部署环境之间的关系。

Token 优化不是简单截断

上下文窗口变大并不等于代理更可靠。把大量相似文档塞进提示词,可能会稀释关键约束,也会增加成本和延迟。知识图谱提供了一种更精确的压缩方式:从任务实体出发,沿着有限的关系展开,只保留与当前决策相关的节点。

实践中可以为不同任务定义不同的关系白名单。例如故障处理任务优先读取 targetsgoverned_byimplemented_bydocumented_by;报表生成任务则可能优先读取数据资产、指标定义和权限关系。上下文打包器应当同时返回事实和来源,避免“压缩后只剩结论”。

落地时的边界

知识图谱不会自动解决模型幻觉、权限控制或数据过期问题。生产部署至少需要检查以下事项:

  • 图谱中的事实是否带有来源、版本和更新时间;
  • 代理是否只能调用经过授权的工具和代码;
  • 每次决策是否记录输入上下文、依据和工具结果;
  • 图谱查询是否设置深度、数量和 token 上限;
  • 人工反馈是否能回写为可追踪的事件;
  • 规则变化后,相关代码、文档和测试状态是否同步更新。

从小范围任务开始通常更容易验证价值,例如只为一个运维流程建立任务、服务、策略、代码和运行手册之间的关系。等上下文打包和决策溯源稳定后,再扩展到更复杂的规划与多工具协作。知识图谱真正带来的改变,不是让代理“知道更多”,而是让它在行动时拥有更清晰的结构、更可查的证据和更短的反馈路径。


相关推荐