基础 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 优化不是简单截断
上下文窗口变大并不等于代理更可靠。把大量相似文档塞进提示词,可能会稀释关键约束,也会增加成本和延迟。知识图谱提供了一种更精确的压缩方式:从任务实体出发,沿着有限的关系展开,只保留与当前决策相关的节点。
实践中可以为不同任务定义不同的关系白名单。例如故障处理任务优先读取 targets、governed_by、implemented_by 和 documented_by;报表生成任务则可能优先读取数据资产、指标定义和权限关系。上下文打包器应当同时返回事实和来源,避免“压缩后只剩结论”。
落地时的边界
知识图谱不会自动解决模型幻觉、权限控制或数据过期问题。生产部署至少需要检查以下事项:
- 图谱中的事实是否带有来源、版本和更新时间;
- 代理是否只能调用经过授权的工具和代码;
- 每次决策是否记录输入上下文、依据和工具结果;
- 图谱查询是否设置深度、数量和 token 上限;
- 人工反馈是否能回写为可追踪的事件;
- 规则变化后,相关代码、文档和测试状态是否同步更新。
从小范围任务开始通常更容易验证价值,例如只为一个运维流程建立任务、服务、策略、代码和运行手册之间的关系。等上下文打包和决策溯源稳定后,再扩展到更复杂的规划与多工具协作。知识图谱真正带来的改变,不是让代理“知道更多”,而是让它在行动时拥有更清晰的结构、更可查的证据和更短的反馈路径。