分享一个给AI Agent做“运行时监控”的实战方案

AI Agent最坑的地方不在于它会崩溃，而在于它会“悄悄跑偏”。比如你让它订一张400刀以下的机票，它可能通过某种诡异的逻辑给自己升了舱，花掉1200刀，然后自信地告诉你“任务已完成”。在传统的监控系统里，这根本不算报错，因为程序没崩，但业务逻辑已经崩了。

为了解决这个，我尝试搭建了一套类似“守护进程”的韧性层，核心逻辑是：一旦发现Agent的行为偏离计划，立即拦截、解释原因并自动回滚。

在公司内部推行这套方案时，我死磕了一个原则：回滚状态必须本地化且持久化，绝对不能依赖外部API。如果你的安全网本身得靠第三方接口才能跑通，那这不叫韧性，这叫增加单点故障。

具体的架构拆分是这样的：

• 本地关键路径（SQLite）： 使用 node:sqlite 建立 Checkpoint Manager。Agent每走一步，计划、预算、已完成步骤全部写死在本地磁盘。回滚时只读这里，不走网络。
• 决策支持层（SigNoz）： 把 SigNoz 当作决策参考。Policy Engine 会通过 Query API 检查本次运行的历史上下文，评估当前偏差的严重程度。

为了让监控真正起到作用，而不是单纯为了堆技术栈，我做了三个实操优化：

一、将整个运行周期定义为一个关联 Trace。run.execute 是父 Span，后续的执行步骤、检查点创建、策略评估全部作为子 Span 挂载，这样排查问题时能一眼看到完整的调用链。

const SpanNames = {
 RUN_EXECUTE: "run.execute",
 EXECUTOR_STEP: "executor.step",
 CHECKPOINT_CREATED: "checkpoint.created",
 POLICY_EVALUATION: "policy.evaluation",
 POLICY_EXPLANATION: "policy.explanation",
 CHECKPOINT_RESTORED: "checkpoint.restored",
} as const;

二、实现中途拦截。Policy Engine 在打分前会查询之前的记录，如果发现这次运行已经多次触发警告，直接升级严重等级。

if (prior.queried && prior.previousScores.some((s) => s >= config.pauseThreshold)) {
 // 升级处理 —— 这已经不是该 Agent 本次运行的第一次警告了
}

三、利用 MCP 服务生成真实解释。当违规触发时，系统会通过 JSON-RPC 调用 signoz_search_traces 查找真实链路数据，而不是给用户回一段模版话术。

[signoz-mcp] INVOKING tools/call signoz_search_traces run=run-872ac2d2 endpoint=http://localhost:8000/mcp

这种方案在实际落地中对提效很明显，尤其是面对那些不稳定的第三方插件时，能极大减少人工盯着日志复盘的时间。