用 SigNoz + OpenTelemetry 给 AI Agen

给 AI Agent 数据库写权限简直像在给熊孩子一把剪刀。很多人习惯用 Human-in-the-loop（人工审核）来兜底，但实测发现这根本没法规模化，凌晨三点没人会盯着每一个 Tool Call 去点 approve，最后要么变成盲目点击，要么流程卡死。

我尝试把 SRE 里的“错误预算（Error Budget）”概念搬到 Agent 管理上：权限不应该是静态的，而应该根据近期执行的成功率动态波动。如果 Agent 最近在疯狂报错，就直接降级它的权限，直到指标恢复。

这就是我用 SigNoz 和 OpenTelemetry 搭建的 LEASH 机制。

核心逻辑：权限分级

我把 Agent 的权限分成了三个 Tier，由 SigNoz 根据实时遥测数据决定当前等级：

• T3 (全权)：读写随意，允许执行破坏性清理。
• T2/T1 (受限)：权限逐步收缩。
• T1 (只读)：任何写操作直接被 Gateway 拦截。

系统架构实操

为了保证权限校验的独立性，我没有把逻辑写在 Agent 内部，而是拆成了四个独立服务：

• agent-runner：执行任务的 Agent。
• leash-broker：核心网关。所有 Tool Call 必须经过这里，校验当前 Tier 后决定 Allow 或 Deny，并接收 SigNoz 的 Webhook。
• migration-tool / resource-tool：实际执行操作的下游服务。

全链路使用 OpenTelemetry Python SDK 埋点，所有 Span 和 Metric 通过 OTLP 协议实时发送到 SigNoz。

# 权限校验伪代码逻辑
def handle_tool_call(request):
    current_tier = redis.get("agent_tier") # 由SigNoz Webhook动态更新
    required_tier = tool_permissions[request.tool_name]
    
    if current_tier >= required_tier:
        return execute_tool(request)
    else:
        return "Permission Denied: Tier too low"

关键点：让 Trace 成为证据

这套方案最硬核的地方在于，我把权限决策过程全部记录在 Span 属性里。在 SigNoz 的 Trace 视图中，leash.policy.decision 会直接显示：当前等级、所需等级、最终裁决。

这样即便不看代码，只要打开 Trace，就能瞬间明白某个操作为什么被拦截，而不是在几万行日志里搜索。

自动化降级链路

我配置了一个基于 leash_tool_calls_total 指标的告警：

• 过滤条件：tool_name = apply_migration 且 outcome = error。
• 触发动作：一旦 5 分钟内错误数超标 → SigNoz 触发告警 → 调用 leash-broker 的 Webhook → 将 Agent 强制降级至 T1。

这种基于实时遥测的闭环控制，比手动设阈值要可靠得多。