Agent = 模型 + 指令（system prompt） + 工具集 + 状态

   
   
   
   

    
   
   
   
   
监控 Agent    —— 负责监控相关的问题
日志 Agent    —— 负责日志相关的问题
数据库 Agent  —— 负责数据库相关的问题

   
   
   
   

    
   
   
   
   
分诊 Agent（Triage）    —— 决策权：判断问题类型和影响面，决定交给谁；工具：只读，查工单、查 CMDB、查服务目录
诊断 Agent（Diagnose）  —— 决策权：给出根因假设和证据；工具：只读，查指标、查日志、查链路、查变更
执行 Agent（Act）       —— 决策权：执行具体处置动作；工具：写操作，但每个动作必须带审批单号
验证 Agent（Verify）    —— 决策权：判断处置是否生效；工具：只读，查指标、查健康检查
复盘 Agent（Review）    —— 决策权：生成复盘报告；工具：只读 + 写文档

   
   
   
   

    json
   
   
   
   
{
  "task_id":"INC-2026-1023-001",
"from":"triage",
"to":"diagnose",
"created_at":"2026-10-23T09:12:33+08:00",
"intent":"diagnose_root_cause",
"scope":{
    "service":"order-service",
    "cluster":"prod-sh-01",
    "namespace":"prod-order",
    "time_window":{
      "start":"2026-10-23T08:40:00+08:00",
      "end":"2026-10-23T09:10:00+08:00"
    }
},
"symptom":{
    "summary":"下单接口 502 比例从 0.1% 升至 18%",
    "first_seen":"2026-10-23T08:52:00+08:00",
    "affected_users":"estimated 3200 orders"
},
"constraints":{
    "read_only":true,
    "deadline":"2026-10-23T09:22:33+08:00",
    "max_tool_calls":30
},
"evidence_refs":[
    "prometheus:query/abc123",
    "ticket:INC-2026-1023-001"
]
}

   
   
   
   

    python
   
   
   
   
classIncidentState(TypedDict):
    # 事实层：只放结构化事实，不放原始日志
    task_id: str
    scope: dict
    symptom: dict
    # 决策层：每个决策都要有作出者和依据
    hypotheses: Annotated[list[dict], operator.add]
    actions: Annotated[list[dict], operator.add]
    verdict: dict | None
    # 控制层：预算与终止
    budget: dict
    next: str

   
   
   
   

    python
   
   
   
   
# 伪代码：工具调用网关的核心校验
defauthorize(agent: str, tool: str, args: dict, ctx: dict) -> None:
    # 1. 角色是否有该工具的授权
    if tool notin POLICY[agent]:
        raise PermissionDenied(f"{agent} 无权调用 {tool}")

    # 2. 写操作必须带审批单，且审批单覆盖该资源
    if POLICY[agent][tool] == "approval":
        ticket = ctx.get("approval_ticket")
        ifnot ticket:
            raise ApprovalRequired(tool)
        ifnot approval_service.covers(ticket, args.get("resource")):
            raise ApprovalScopeMismatch(ticket, args.get("resource"))

    # 3. 资源必须落在任务 scope 内（防止越界操作别的 namespace）
    ifnot in_scope(args, ctx["scope"]):
        raise OutOfScope(args, ctx["scope"])

    # 4. 速率与并发限制
    rate_limiter.check(agent, tool)

   
   
   
   

    
   
   
   
   
        ┌──────────┐
        │  intake  │  (系统入口，校验工单字段)
        └────┬─────┘
             ▼
        ┌──────────┐   scope 不完整
        │  triage  │──────────────► 人工补充 ──┐
        └────┬─────┘                          │
             │ scope 完整                     │
             ▼                                │
        ┌──────────┐  无假设 / 证据不足        │
        │ diagnose │──────────────► 升级人工 ◄─┘
        └────┬─────┘
             │ 有高置信假设 + 建议动作
             ▼
        ┌──────────┐  审批拒绝
        │   act    │──────────────► 记录并结束
        └────┬─────┘
             │ 执行完成
             ▼
        ┌──────────┐  未生效
        │  verify  │──────────────► 回到 diagnose（最多 2 次）
        └────┬─────┘
             │ 生效
             ▼
        ┌──────────┐
        │  review  │
        └────┬─────┘
             ▼
           END

   
   
   
   

    
   
   
   
   
09:12:03  工单创建（告警自动触发）：order-service 502 比例 18%
09:12:04  intake：校验工单字段完整，生成 task_id
09:12:06  triage：查 CMDB 确认 order-service 在 prod-sh-01/prod-order，
          影响面"华东区下单"，severity=P1，route=diagnose
09:12:08  diagnose：并行发起 4 个查询
          - prometheus: 5xx 比例、QPS、P99、pod 重启次数
          - loki: 最近 20 分钟 error 日志（限制 200 行）
          - tempo: 慢链路采样
          - k8s: deployment 最近变更（rollout history）
09:12:41  diagnose：发现 09:02 有一次镜像更新，新 pod 启动后 5xx 开始上升；
          同时发现新版本日志里有 "connection pool exhausted"
          → 假设 H1：新版本连接池配置过小，置信度 0.85
09:12:43  act：生成动作方案 "rollout undo 到上一版本"，
          请求审批（工单 INC-2026-1023-001 关联审批单 AP-7781）
09:13:20  审批通过（值班经理在卡片上点确认）
09:13:22  act：执行 kubectl rollout undo，等待 rollout 完成（92 秒）
09:14:54  verify：5xx 比例从 18% 降到 0.3%，P99 从 4.2s 降到 320ms，
          pod ready 3/3 → verdict=effective
09:15:10  review：生成时间线、根因、改进项，写入知识库

   
   
   
   

    bash
   
   
   
   
python -m venv /opt/agent-ops/venv
source /opt/agent-ops/venv/bin/activate

pip install \
"langgraph>=1.0,<2.0" \
"langgraph-checkpoint-postgres>=2.0" \
"langchain-mcp-adapters>=0.2" \
"langchain-openai>=0.3" \
"psycopg[binary,pool]>=3.2" \
"opentelemetry-sdk>=1.28" \
"opentelemetry-exporter-otlp>=1.28"

   
   
   
   

    python
   
   
   
   
# state.py
from __future__ import annotations

import operator
from typing import Annotated, Literal, TypedDict


classScope(TypedDict, total=False):
    service: str
    cluster: str
    namespace: str
    time_start: str
    time_end: str


classBudget(TypedDict):
    max_rounds: int          # 总轮次上限
    max_tool_calls: int      # 工具调用总次数上限
    max_tokens: int          # 估算 token 上限
    deadline: str            # ISO8601，超时即终止


classIncidentState(TypedDict, total=False):
    # ---- 事实层 ----
    task_id: str
    scope: Scope
    symptom: dict
    severity: str

    # ---- 决策层（用 reducer 追加，不覆盖） ----
    hypotheses: Annotated[list[dict], operator.add]
    actions: Annotated[list[dict], operator.add]
    verdict: dict

    # ---- 控制层 ----
    budget: Budget
    rounds: int
    tool_calls: int
    next: str
    handoff: dict
    errors: Annotated[list[str], operator.add]

   
   
   
   

    python
   
   
   
   
# contracts.py
from datetime import datetime, timezone
from typing importLiteral

from pydantic import BaseModel, Field, field_validator


classHandoff(BaseModel):
    task_id: str
    from_agent: Literal["triage", "diagnose", "act", "verify", "review"]
    to_agent: Literal["diagnose", "act", "verify", "review", "human"]
    intent: Literal[
        "diagnose_root_cause", "execute_action",
        "verify_effect", "write_review", "escalate",
    ]
    scope: dict = Field(min_length=1)
    read_only: bool
    deadline: str
    max_tool_calls: int = Field(ge=1, le=100)
    evidence_refs: list[str] = Field(default_factory=list)

    @field_validator("scope")
    @classmethod
    defscope_must_have_namespace(cls, v: dict) -> dict:
        if"namespace"notin v and"service"notin v:
            raise ValueError("scope 必须包含 namespace 或 service")
        return v

    @field_validator("deadline")
    @classmethod
    defdeadline_in_future(cls, v: str) -> str:
        dt = datetime.fromisoformat(v)
        if dt <= datetime.now(timezone.utc):
            raise ValueError("deadline 必须晚于当前时间")
        return v

   
   
   
   

    python
   
   
   
   
# graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

from state import IncidentState
from nodes import (
    intake_node, triage_node, diagnose_node,
    act_node, verify_node, review_node, escalate_node,
)

MAX_DIAGNOSE = 2
MAX_VERIFY_FAIL = 2


defroute_after_triage(state: IncidentState) -> str:
    scope = state.get("scope") or {}
    ifnot scope.get("namespace") andnot scope.get("service"):
        return"escalate"
    return"diagnose"


defroute_after_diagnose(state: IncidentState) -> str:
    hyps = state.get("hypotheses") or []
    ifnot hyps:
        return"escalate"
    best = max(hyps, key=lambda h: h.get("confidence", 0))
    if best.get("confidence", 0) < 0.6:
        return"escalate"
    ifnot best.get("suggested_action"):
        return"escalate"
    # 只读诊断不允许直接执行，必须走审批
    return"act"


defroute_after_act(state: IncidentState) -> str:
    acts = state.get("actions") or []
    ifnot acts:
        return"escalate"
    last = acts[-1]
    if last.get("status") == "rejected":
        return END
    if last.get("status") != "executed":
        return"escalate"
    return"verify"


defroute_after_verify(state: IncidentState) -> str:
    verdict = state.get("verdict") or {}
    if verdict.get("result") == "effective":
        return"review"
    # 未生效：允许回到诊断，但最多 MAX_VERIFY_FAIL 次
    failed = sum(1for a in state.get("actions", []) if a.get("status") == "executed")
    if failed >= MAX_VERIFY_FAIL:
        return"escalate"
    if (state.get("rounds") or0) >= (state.get("budget") or {}).get("max_rounds", 12):
        return"escalate"
    return"diagnose"


defbuild_graph(checkpointer):
    g = StateGraph(IncidentState)
    g.add_node("intake", intake_node)
    g.add_node("triage", triage_node)
    g.add_node("diagnose", diagnose_node)
    g.add_node("act", act_node)
    g.add_node("verify", verify_node)
    g.add_node("review", review_node)
    g.add_node("escalate", escalate_node)

    g.add_edge(START, "intake")
    g.add_edge("intake", "triage")
    g.add_conditional_edges("triage", route_after_triage,
                            {"diagnose": "diagnose", "escalate": "escalate"})
    g.add_conditional_edges("diagnose", route_after_diagnose,
                            {"act": "act", "escalate": "escalate"})
    g.add_conditional_edges("act", route_after_act,
                            {"verify": "verify", "escalate": "escalate", END: END})
    g.add_conditional_edges("verify", route_after_verify,
                            {"review": "review", "diagnose": "diagnose",
                             "escalate": "escalate"})
    g.add_edge("review", END)
    g.add_edge("escalate", END)

    return g.compile(checkpointer=checkpointer)

   
   
   
   

    python
   
   
   
   
# nodes.py
import json

from langchain_core.messages import HumanMessage, SystemMessage

from llm import get_llm          # 按角色返回不同模型
from mcp_client import get_tools  # 按角色返回不同工具集

DIAGNOSE_PROMPT = """你是生产故障诊断专家。你的唯一职责是给出根因假设，不执行任何变更。

工作规则：
1. 最多调用 8 次工具。优先并行查询，不要串行。
2. 每个假设必须包含：假设内容、支持证据（指标数值/日志原文/变更记录）、置信度(0-1)。
3. 证据必须来自工具返回的原始数据，不得推测数值。
4. 如果证据不足以支持任何假设，返回空列表，不要编造。
5. 不要给出执行建议之外的任何操作指令。

输出 JSON，格式：
{
  "hypotheses": [
    {"id": "H1", "claim": "...", "evidence": [{"source": "...", "detail": "..."}],
     "confidence": 0.85, "suggested_action": {"type": "rollout_undo", "target": "..."}}
  ]
}"""


asyncdefdiagnose_node(state: IncidentState) -> dict:
    llm = get_llm(role="diagnose")
    tools = await get_tools(role="diagnose")
    llm_with_tools = llm.bind_tools(tools)

    budget = state["budget"]
    ctx = {
        "scope": state["scope"],
        "symptom": state["symptom"],
        "deadline": budget["deadline"],
        "remaining_tool_calls": budget["max_tool_calls"] - state.get("tool_calls", 0),
    }

    messages = [
        SystemMessage(content=DIAGNOSE_PROMPT),
        HumanMessage(content=json.dumps(ctx, ensure_ascii=False)),
    ]

    # 工具调用循环（简化版 ReAct，真实实现要处理并行 tool_calls）
    tool_map = {t.name: t for t in tools}     # MCP 返回的是 list，先转成字典
    tool_calls_used = 0
    for _ inrange(8):
        resp = await llm_with_tools.ainvoke(messages)
        messages.append(resp)
        ifnot resp.tool_calls:
            break
        for call in resp.tool_calls:
            if tool_calls_used >= ctx["remaining_tool_calls"]:
                break
            tool = tool_map.get(call["name"])
            if tool isNone:
                # 模型编造了不存在的工具名，直接回一条错误让它重来
                messages.append({
                    "role": "tool",
                    "tool_call_id": call["id"],
                    "content": f"ERROR: 未知工具 {call['name']}，可用工具：{list(tool_map)}",
                })
                continue
            result = await tool.ainvoke(call["args"])
            tool_calls_used += 1
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": str(result)[:4000],   # 截断，防止撑爆上下文
            })

    parsed = parse_json_safely(messages[-1].content)
    return {
        "hypotheses": parsed.get("hypotheses", []),
        "rounds": state.get("rounds", 0) + 1,
        "tool_calls": state.get("tool_calls", 0) + tool_calls_used,
        "handoff": {"from_agent": "diagnose", "to_agent": "act"},
    }

   
   
   
   

    python
   
   
   
   
import json
import re


defparse_json_safely(text: str) -> dict:
    ifnot text:
        return {"hypotheses": [], "_parse_error": "empty"}
    # 去掉 ```json ... ``` 包裹
    fenced = re.search(r"```(?:json)?\s*(.+?)```", text, re.S)
    if fenced:
        text = fenced.group(1)
    # 截取第一个 { 到最后一个 }
    start, end = text.find("{"), text.rfind("}")
    if start == -1or end <= start:
        return {"hypotheses": [], "_parse_error": "no_json_object"}
    candidate = text[start:end + 1]
    for attempt in (candidate, re.sub(r",\s*([}\]])", r"\1", candidate)):
        try:
            return json.loads(attempt)
        except json.JSONDecodeError:
            continue
    return {"hypotheses": [], "_parse_error": "invalid_json"}

   
   
   
   

    python
   
   
   
   
# mcp_client.py
import os
from langchain_mcp_adapters.client import MultiServerMCPClient

MCP_ENDPOINTS = {
    "prometheus": {
        "transport": "streamable_http",
        "url": "http://mcp-prometheus.ops.svc.cluster.local:8080/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['MCP_TOKEN_PROM']}"},
        "timeout": 30,
    },
    "loki": {
        "transport": "streamable_http",
        "url": "http://mcp-loki.ops.svc.cluster.local:8080/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['MCP_TOKEN_LOKI']}"},
        "timeout": 30,
    },
    "k8s-ro": {
        "transport": "streamable_http",
        "url": "http://mcp-k8s-ro.ops.svc.cluster.local:8080/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['MCP_TOKEN_K8S_RO']}"},
        "timeout": 20,
    },
    "k8s-rw": {
        "transport": "streamable_http",
        "url": "http://mcp-k8s-rw.ops.svc.cluster.local:8080/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['MCP_TOKEN_K8S_RW']}"},
        "timeout": 60,
    },
    "ticket": {
        "transport": "streamable_http",
        "url": "http://mcp-ticket.ops.svc.cluster.local:8080/mcp",
        "timeout": 15,
    },
}

ROLE_SERVERS = {
    "triage":   ["ticket"],
    "diagnose": ["prometheus", "loki", "k8s-ro", "ticket"],
    "act":      ["k8s-ro", "k8s-rw", "ticket"],
    "verify":   ["prometheus", "k8s-ro"],
    "review":   ["ticket", "prometheus", "loki"],
}

_clients: dict[str, MultiServerMCPClient] = {}


asyncdefget_tools(role: str):
    if role notin ROLE_SERVERS:
        raise ValueError(f"未知角色: {role}")
    if role notin _clients:
        cfg = {name: MCP_ENDPOINTS[name] for name in ROLE_SERVERS[role]}
        _clients[role] = MultiServerMCPClient(cfg)
    returnawait _clients[role].get_tools()

   
   
   
   

    python
   
   
   
   
# 差的工具描述
@mcp.tool()
defquery(query: str) -> str:
    """查询监控数据"""

# 好的工具描述
@mcp.tool()
defprometheus_query_range(
    expr: str,
    start: str,
    end: str,
    step: str = "60s",
) -> str:
    """执行 PromQL 区间查询，返回时间序列。

    Args:
        expr: PromQL 表达式，例如 'sum(rate(http_requests_total{code=~"5.."}[5m])) by (pod)'
        start: 起始时间，RFC3339 格式，例如 '2026-10-23T08:40:00+08:00'
        end: 结束时间，RFC3339 格式，必须晚于 start，跨度不超过 24 小时
        step: 查询步长，例如 '60s'、'5m'，跨度大时用大步长避免返回过多点

    Returns:
        JSON 字符串，包含 resultType 和 result 数组；超过 500 个数据点时自动降采样。
    """

   
   
   
   

    python
   
   
   
   
# nodes.py（续）
from langgraph.types import interrupt


defact_node(state: IncidentState) -> dict:
    best = max(state["hypotheses"], key=lambda h: h.get("confidence", 0))
    action = best["suggested_action"]

    # 1. 幂等检查：同一个 task 的同一个动作只执行一次
    idem_key = f"act:{state['task_id']}:{hash_action(action)}"
    if idempotency.seen(idem_key):
        return {"actions": state.get("actions", []) + [{
            "action": action, "status": "skipped_duplicate",
        }]}

    # 2. 挂起，等待审批。返回值是外部 resume 时传入的 payload
    decision = interrupt({
        "type": "approval_request",
        "task_id": state["task_id"],
        "action": action,
        "hypothesis": {"id": best["id"], "claim": best["claim"],
                       "confidence": best["confidence"]},
        "evidence": best.get("evidence", []),
        "requested_at": datetime.now(timezone.utc).isoformat(),
    })

    # 3. 审批结果处理
    ifnot decision.get("approved"):
        return {"actions": state.get("actions", []) + [{
            "action": action,
            "status": "rejected",
            "reason": decision.get("reason", ""),
            "operator": decision.get("operator", ""),
        }]}

    # 4. 执行（带审批单号，网关会校验）
    try:
        result = tool_gateway.execute(
            action,
            approval_ticket=decision["ticket_id"],
            scope=state["scope"],
            timeout=action.get("timeout", 120),
        )
        idempotency.mark(idem_key)
        status = "executed"
    except Exception as exc:
        result = {"error": str(exc)}
        status = "failed"

    return {
        "actions": state.get("actions", []) + [{
            "action": action,
            "status": status,
            "result": result,
            "approval_ticket": decision.get("ticket_id"),
            "operator": decision.get("operator", ""),
            "executed_at": datetime.now(timezone.utc).isoformat(),
        }],
        "rounds": state.get("rounds", 0) + 1,
    }

   
   
   
   

    python
   
   
   
   
# run.py
import asyncio
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.types import Command

from graph import build_graph

DSN = "postgresql://agent:${PG_PASS}@pg-ops.internal:5432/agent_state"


asyncdefmain():
    asyncwith AsyncPostgresSaver.from_conn_string(DSN) as saver:
        await saver.setup()
        graph = build_graph(saver)

        config = {"configurable": {"thread_id": "INC-2026-1023-001"}}

        # 第一次运行：会停在 interrupt
        result = await graph.ainvoke(
            {"task_id": "INC-2026-1023-001", "symptom": {...}, "budget": {...}},
            config=config,
        )
        print("挂起等待审批:", result.get("__interrupt__"))

        # 审批通过后恢复（真实场景由审批服务的回调触发）
        result = await graph.ainvoke(
            Command(resume={"approved": True, "ticket_id": "AP-7781",
                            "operator": "zhangsan"}),
            config=config,
        )
        print("最终状态:", result["verdict"])


asyncio.run(main())

   
   
   
   

    python
   
   
   
   
# 查询挂起的任务
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

asyncwith AsyncPostgresSaver.from_conn_string(DSN) as saver:
    await saver.setup()
    graph = build_graph(saver)
    config = {"configurable": {"thread_id": "INC-2026-1023-001"}}

    snapshot = await graph.aget_state(config)
    for task in snapshot.tasks:
        for itr in task.interrupts:
            print(itr.value)   # 审批请求的内容

   
   
   
   

    python
   
   
   
   
# idempotency.py
import hashlib
import json
import redis

r = redis.Redis(host="redis-ops.internal", port=6379, db=2)


defhash_action(action: dict) -> str:
    canonical = json.dumps(action, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(canonical.encode()).hexdigest()[:16]


defseen(key: str) -> bool:
    return r.exists(key) == 1


defacquire(key: str, ttl: int = 3600) -> bool:
    """SETNX 语义：只有第一次能拿到锁"""
    returnbool(r.set(key, "1", nx=True, ex=ttl))


defmark(key: str, ttl: int = 86400) -> None:
    r.set(key, "done", ex=ttl)

   
   
   
   

    python
   
   
   
   
import asyncio


asyncdefrun_node_with_timeout(node_fn, state, seconds=180):
    try:
        returnawait asyncio.wait_for(node_fn(state), timeout=seconds)
    except asyncio.TimeoutError:
        return {
            "errors": [f"{node_fn.__name__} 超时（{seconds}s）"],
            "next": "escalate",
        }

   
   
   
   

    python
   
   
   
   
defguard_rounds(state) -> bool:
    return state.get("rounds", 0) < state["budget"]["max_rounds"]

   
   
   
   

    python
   
   
   
   
import hashlib
import json


defstate_fingerprint(state: dict) -> str:
    # 只取决策层，忽略时间戳等易变字段
    payload = {
        "hypotheses": [h.get("id") for h in state.get("hypotheses", [])],
        "last_action": (state.get("actions") or [{}])[-1].get("action"),
        "next": state.get("next"),
    }
    return hashlib.md5(json.dumps(payload, sort_keys=True).encode()).hexdigest()

   
   
   
   

    python
   
   
   
   
config = {
    "configurable": {"thread_id": tid},
    "recursion_limit": 25,      # 默认 25，别设太大
}

   
   
   
   

    python
   
   
   
   
# 资源级锁
import uuid


defresource_key(scope: dict, action: dict) -> str:
    returnf"lock:{scope['cluster']}:{scope['namespace']}:{action['target']}"


defwith_resource_lock(scope, action, fn, ttl=300):
    key = resource_key(scope, action)
    token = str(uuid.uuid4())
    ifnot r.set(key, token, nx=True, ex=ttl):
        raise ResourceBusy(f"{key} 正在被其他任务操作")
    try:
        return fn()
    finally:
        # 只释放自己持有的锁（Lua 脚本保证原子性）
        r.eval(
            "if redis.call('get', KEYS[1]) == ARGV[1] then "
            "return redis.call('del', KEYS[1]) else return 0 end",
            1, key, token,
        )

   
   
   
   

    python
   
   
   
   
# MCP Server 侧：限制返回条数和字节数
MAX_LINES = 200
MAX_BYTES = 64 * 1024


deftruncate_logs(lines: list[str]) -> str:
    ifnot lines:
        return"（无匹配日志）"
    head = lines[:50]
    tail = lines[-30:] iflen(lines) > 80else []
    omitted = len(lines) - len(head) - len(tail)
    body = "\n".join(head)
    if omitted > 0:
        body += f"\n... [省略 {omitted} 行] ...\n" + "\n".join(tail)
    encoded = body.encode("utf-8")
    iflen(encoded) > MAX_BYTES:
        body = encoded[:MAX_BYTES].decode("utf-8", errors="ignore") + "\n... [已截断] ..."
    return body

   
   
   
   

    python
   
   
   
   
from opentelemetry import trace

tracer = trace.get_tracer("agent-ops")


asyncdeftraced_node(name: str, state: IncidentState, fn):
    with tracer.start_as_current_span(f"agent.{name}") as span:
        span.set_attribute("gen_ai.operation.name", "invoke_agent")
        span.set_attribute("gen_ai.agent.name", name)
        span.set_attribute("gen_ai.conversation.id", state["task_id"])
        span.set_attribute("app.task.severity", state.get("severity", "unknown"))
        try:
            result = await fn(state)
            span.set_attribute("app.node.status", "ok")
            return result
        except Exception as exc:
            span.set_attribute("app.node.status", "error")
            span.record_exception(exc)
            raise

   
   
   
   

    python
   
   
   
   
with tracer.start_as_current_span("gen_ai.tool.call") as span:
    span.set_attribute("gen_ai.tool.name", tool_name)
    span.set_attribute("gen_ai.tool.call.id", call_id)
    span.set_attribute("gen_ai.usage.input_tokens", usage.input_tokens)
    span.set_attribute("gen_ai.usage.output_tokens", usage.output_tokens)

   
   
   
   

    promql
   
   
   
   
# 各角色的工具错误率，用于定位是哪个 Agent 在拖后腿
sum by (agent) (rate(agent_tool_calls_total{status="error"}[5m]))
  /
sum by (agent) (rate(agent_tool_calls_total[5m]))

   
   
   
   

    jsonl
   
   
   
   
{"id":"E001","ticket":"订单服务502，比例18%，华东区","expect_route":"diagnose","expect_severity":"P1","expect_action_type":"rollout_undo"}
{"id":"E002","ticket":"数据库连接数满了，应用报too many connections","expect_route":"diagnose","expect_severity":"P1","expect_action_type":"scale_connection"}
{"id":"E003","ticket":"测试环境某pod CrashLoopBackOff","expect_route":"diagnose","expect_severity":"P3","expect_action_type":null}
{"id":"E004","ticket":"帮我查一下昨天的订单量","expect_route":"reject","expect_severity":null,"expect_action_type":null}

   
   
   
   

    bash
   
   
   
   
#!/usr/bin/env bash
# 评估跑批：固定模型、温度、工具版本，输出可对比的报告
set -euo pipefail

MODEL="${EVAL_MODEL:?需要设置 EVAL_MODEL}"
OUT="eval/report-$(date +%Y%m%d-%H%M)-${MODEL//\//_}.json"

python -m eval.runner \
  --dataset eval/golden.jsonl \
  --model "$MODEL" \
  --temperature 0 \
  --concurrency 4 \
  --output "$OUT"

python -m eval.compare --baseline eval/baseline.json --candidate "$OUT"

   
   
   
   

    bash
   
   
   
   
# 1. 拉出这个任务的完整 trace（Jaeger）
curl -s "http://jaeger:16686/api/traces?service=agent-ops&tags=%7B%22gen_ai.conversation.id%22%3A%22INC-2026-1019-014%22%7D" \
  | jq '.data[0].spans[] | {name: .operationName, start: .startTime, dur: .duration}'

   
   
   
   

    
   
   
   
   
{"name": "agent.triage",   "start": 1729317742000000, "dur": 2100000}
{"name": "agent.diagnose", "start": 1729317744100000, "dur": 8300000}
{"name": "agent.triage",   "start": 1729317752400000, "dur": 1900000}
{"name": "agent.diagnose", "start": 1729317754300000, "dur": 8100000}
...（重复 13 次）

   
   
   
   

    sql
   
   
   
   
-- 2. 看这个任务的最终状态
SELECT task_id, rounds, tool_calls,
       jsonb_array_length(state->'hypotheses') AS hyp_cnt,
       state->'scope'ASscope
FROM agent_checkpoints
WHERE thread_id ='INC-2026-1019-014'
ORDERBY checkpoint_id DESC LIMIT 1;

   
   
   
   

    
   
   
   
   
task_id              | rounds | tool_calls | hyp_cnt | scope
INC-2026-1019-014    | 13     | 41         | 0       | {"service":"order-service"}

   
   
   
   

    promql
   
   
   
   
# 这个任务的轮次分布
histogram_quantile(0.95, sum by (le) (rate(agent_rounds_per_task_bucket[1h])))
# 0.95 分位是 9，但最大值 27

   
   
   
   

    python
   
   
   
   
# 修复 1：所有节点统一通过包装器递增 rounds
defwith_rounds(fn):
    asyncdefwrapper(state: IncidentState) -> dict:
        out = await fn(state)
        out["rounds"] = state.get("rounds", 0) + 1
        return out
    return wrapper


triage_node = with_rounds(_triage_node)
diagnose_node = with_rounds(_diagnose_node)
act_node = with_rounds(_act_node)
verify_node = with_rounds(_verify_node)

   
   
   
   

    python
   
   
   
   
defwith_rounds(fn):
    asyncdefwrapper(state: IncidentState):
        out = await fn(state)
        rounds = state.get("rounds", 0) + 1
        ifisinstance(out, Command):
            out.update = {**(out.update or {}), "rounds": rounds}
            return out
        out["rounds"] = rounds
        return out
    return wrapper

   
   
   
   

    python
   
   
   
   
# 修复 2：在入口处统一做预算检查，不依赖各节点自觉
defbudget_guard(state: IncidentState) -> str | None:
    budget = state.get("budget") or {}
    if state.get("rounds", 0) >= budget.get("max_rounds", 12):
        return"rounds_exceeded"
    if state.get("tool_calls", 0) >= budget.get("max_tool_calls", 40):
        return"tool_calls_exceeded"
    deadline = budget.get("deadline")
    if deadline and datetime.fromisoformat(deadline) <= datetime.now(timezone.utc):
        return"deadline_exceeded"
    returnNone

   
   
   
   

    python
   
   
   
   
defroute_after_triage(state):
    if (reason := budget_guard(state)):
        return"escalate"
    ...

   
   
   
   

    python
   
   
   
   
# 修复 3：escalate 节点不允许再回到任何 Agent
asyncdefescalate_node(state: IncidentState) -> dict:
    await notify_human(state)          # 发通知
    await write_audit_log(state)       # 写审计
    return {"next": "end"}             # 只能结束

   
   
   
   

    bash
   
   
   
   
# 重放这个工单（用相同的输入，在测试环境）
python -m replay --task-id INC-2026-1019-014 --env staging

# 预期：rounds <= 12，且在第 2 轮就 escalate
grep "rounds" /tmp/replay.log | tail -3

   
   
   
   

    bash
   
   
   
   
# 1. 查这次工具调用的审计日志
jq -r 'select(.agent=="diagnose") | [.ts, .tool, .args, .result] | @tsv' \
  /var/log/agent/audit.jsonl | grep -i "rollout"

   
   
   
   

    
   
   
   
   
2026-10-19T14:22:31Z  k8s_rollout_undo  {"namespace":"prod-order","deployment":"order-service","revision":0}  {"status":"ok"}

   
   
   
   

    bash
   
   
   
   
# 2. 查 diagnose 的 MCP 配置
kubectl -n agent-ops get deploy agent-diagnose -o jsonpath='{.spec.template.spec.containers[0].env}' | jq
# 发现 MCP_K8S_SERVERS 里包含了 k8s-rw

   
   
   
   

    promql
   
   
   
   
# 各角色的写操作调用次数（这个指标本不该有 diagnose 的数据）
sum by (agent, tool) (increase(agent_tool_calls_total{tool=~".*restart.*|.*rollout.*|.*scale.*"}[1d]))
# {agent="act", tool="k8s_rollout_undo"} 12
# {agent="diagnose", tool="k8s_rollout_undo"} 1     <-- 异常

   
   
   
   

    python
   
   
   
   
ROLE_SERVERS = {
    "triage":   ["ticket"],
    "diagnose": ["prometheus", "loki", "k8s-ro", "ticket"],   # 移除 k8s-rw
    "act":      ["k8s-ro", "k8s-rw", "ticket"],
    "verify":   ["prometheus", "k8s-ro"],
    "review":   ["ticket", "prometheus", "loki"],
}

   
   
   
   

    python
   
   
   
   
# gateway/policy.py
WRITE_TOOLS = {
    "k8s_restart", "k8s_scale", "k8s_rollout_undo",
    "k8s_delete_pod", "net_drain", "config_patch",
}

POLICY = {
    "triage":   set(),
    "diagnose": set(),
    "act":      WRITE_TOOLS,
    "verify":   set(),
    "review":   set(),
}


defauthorize(agent: str, tool: str, args: dict, ctx: dict) -> None:
    allowed = POLICY.get(agent, set())
    if tool in WRITE_TOOLS and tool notin allowed:
        # 写操作只有 act 能做，且必须有审批单
        raise PermissionDenied(f"{agent} 不允许调用写工具 {tool}")
    if tool in WRITE_TOOLS:
        ticket = ctx.get("approval_ticket")
        ifnot ticket:
            raise ApprovalRequired(tool)
        ifnot approval_service.validate(ticket, agent=agent, tool=tool, args=args):
            raise ApprovalInvalid(ticket)

   
   
   
   

    yaml
   
   
   
   
-alert:UnauthorizedWriteAttempt
expr:increase(agent_tool_calls_total{status="permission_denied"}[5m])>0
for:0m
labels:
    severity:critical
annotations:
    summary:"{{ $labels.agent }} 尝试调用未授权工具 {{ $labels.tool }}"

   
   
   
   

    bash
   
   
   
   
# 1. 用 diagnose 角色尝试调写工具，应该被拒
python -m tools.check --agent diagnose --tool k8s_rollout_undo
# PermissionDenied: diagnose 不允许调用写工具 k8s_rollout_undo

# 2. act 角色带合法审批单，应该通过
python -m tools.check --agent act --tool k8s_rollout_undo --ticket AP-7781
# OK

# 3. act 角色不带审批单，应该被拒
python -m tools.check --agent act --tool k8s_rollout_undo
# ApprovalRequired

   
   
   
   

    bash
   
   
   
   
# 1. 看 triage 交给 diagnose 的 handoff
jq '.handoff' /var/log/agent/INC-2026-1021-007/triage.json

   
   
   
   

    json
   
   
   
   
{
  "task_id":"INC-2026-1021-007",
"from":"triage",
"to":"diagnose",
"summary":"华东区订单服务异常，疑似新版本问题，建议排查连接池",
"severity":"P1"
}

   
   
   
   

    bash
   
   
   
   
# 2. 看 diagnose 输出的建议动作
jq '.hypotheses[0].suggested_action' /var/log/agent/INC-2026-1021-007/diagnose.json

   
   
   
   

    json
   
   
   
   
{
  "type":"rollout_undo",
  "target":"order-service",
  "namespace":"prod-order",
  "cluster":"prod-bj-01"
}

   
   
   
   

    promql
   
   
   
   
# 执行动作的集群分布
sum by (cluster) (increase(agent_actions_total{action="rollout_undo"}[7d]))
# prod-bj-01: 23
# prod-sh-01: 4      <-- 与工单里的"华东区"严重不符

   
   
   
   

    python
   
   
   
   
classHandoff(BaseModel):
    task_id: str
    from_agent: str
    to_agent: str
    intent: str
    scope: Scope = Field(...)          # 必填，不接受 None
    read_only: bool
    deadline: str
    max_tool_calls: int

   
   
   
   

    python
   
   
   
   
classScope(BaseModel):
    service: str
    cluster: str = Field(pattern=r"^prod-[a-z]{2}-\d{2}$")
    namespace: str = Field(pattern=r"^prod-[a-z0-9-]+$")
    time_start: str
    time_end: str