AI Agent监控实战：追踪Token消耗、延迟与异常行为

💸 一个Agent跑飞一晚上花了$200的token费。监控延时是"帮你省钱"，监控异常行为才是"帮你保命"。

symptom_zh

上篇讲了监控体系怎么搭。这篇讲具体怎么用——监控看板怎么做、异常行为怎么检测、Bill Shock（天价账单）怎么防、死循环怎么自动打断。 全是实战代码和踩坑经验。不管是用LangSmith还是自建方案，这篇的指标和告警规则都能直接用。

读完能做： 搭建一个能自动发现Agent异常、自动告警的监控系统

diagnosis_zh

一、一个真实的Bill Shock案例

朋友做AI客服，上线第一个月，某天晚上Agent突然"疯了"——对同一个用户的问题反复调用LLM推理，每轮都生成超长回复。

第二天看账单：一晚$47。 正常情况一个月也就$200。

根因分析： 用户问了一个边界问题，Agent无法做出决策，于是陷入"再想一下→再调用工具→得到不明确结果→再想一下"的死循环。没有步数限制，没有成本告警。

从那以后，他的Agent加了三道安全锁： 1. 每轮对话最多15步 2. 单会话成本上限$0.05 3. token消耗突变检测

二、Token消耗追踪

Token是Agent的钱。要按Agent、按会话、按时段三种粒度追踪。

import time
from collections import defaultdict
from datetime import datetime, timedelta

class TokenTracker:
    """Token消耗追踪器"""

    def __init__(self):
        self.agent_budgets = {}  # agent_id -> 每日预算
        self.session_tokens = defaultdict(list)  # session_id -> [token数]

    def set_budget(self, agent_id: str, daily_budget: float):
        self.agent_budgets[agent_id] = daily_budget

    def record_usage(self, agent_id: str, session_id: str, 
                     tokens: int, cost: float):
        """记录一次LLM调用"""
        now = datetime.utcnow()
        self.session_tokens[session_id].append({
            "tokens": tokens,
            "cost": cost,
            "timestamp": now
        })

        # 实时检查：单会话成本是否超限
        session_cost = sum(s["cost"] for s in self.session_tokens[session_id])
        if session_cost > 0.05:  # 单会话超过$0.05预警
            print(f"⚠️ 会话 {session_id} 成本已达 ${session_cost:.4f}")

        # 实时检查：当日Agent总成本
        today_cost = self._get_daily_cost(agent_id)
        budget = self.agent_budgets.get(agent_id, float('inf'))
        if today_cost > budget * 0.8:
            print(f"⚠️ Agent {agent_id} 当日已用 ${today_cost:.2f}，预算 ${budget:.2f}")

    def _get_daily_cost(self, agent_id: str) -> float:
        today = datetime.utcnow().date()
        total = 0.0
        for sessions in self.session_tokens.values():
            for s in sessions:
                if s["timestamp"].date() == today:
                    total += s["cost"]
        return total

    def anomaly_detection(self, agent_id: str, window_minutes: int = 10):
        """检测异常Token消耗：过去window_minutes的消耗是否超过平均的3倍"""
        now = datetime.utcnow()
        cutoff = now - timedelta(minutes=window_minutes)

        recent = []
        all_time = []

        for sessions in self.session_tokens.values():
            for s in sessions:
                if s["timestamp"] >= cutoff:
                    recent.append(s["tokens"])
                all_time.append(s["tokens"])

        if not all_time or not recent:
            return None

        avg_tokens = sum(all_time) / len(all_time)
        recent_avg = sum(recent) / len(recent)

        if recent_avg > avg_tokens * 3:
            return {
                "alert": "异常Token消耗",
                "window": f"{window_minutes}分钟",
                "recent_avg": recent_avg,
                "historical_avg": avg_tokens,
                "ratio": recent_avg / avg_tokens
            }
        return None

# 使用
tracker = TokenTracker()
tracker.set_budget("customer-support", 10.0)  # 每日预算$10
tracker.record_usage("customer-support", "session-123", 500, 0.002)
alert = tracker.anomaly_detection("customer-support")
if alert:
    print(f"🚨 {alert['alert']}: {alert['ratio']:.1f}x 正常水平")

三、延迟分析（P50/P95/P99）

Agent的延迟分布跟普通API完全不同。普通API的P99可能是P50的2-3倍，但Agent的P99可能是P50的10倍——因为一些复杂问题需要更多推理步骤。

import numpy as np
from collections import deque

class LatencyAnalyzer:
    """延迟分析器"""

    def __init__(self, window_size: int = 100):
        self.latencies = deque(maxlen=window_size)

    def record(self, agent_id: str, step: str, duration_ms: int):
        self.latencies.append({
            "agent_id": agent_id,
            "step": step,
            "duration_ms": duration_ms,
            "timestamp": datetime.utcnow().isoformat()
        })

    def report(self) -> dict:
        if not self.latencies:
            return {}
        durations = sorted([l["duration_ms"] for l in self.latencies])
        n = len(durations)
        return {
            "p50": durations[int(n * 0.50)],
            "p95": durations[int(n * 0.95)],
            "p99": durations[int(n * 0.99)],
            "mean": np.mean(durations),
            "max": max(durations),
            "min": min(durations),
            "sample_count": n
        }

# 使用
analyzer = LatencyAnalyzer(window_size=1000)
# recording...
report = analyzer.report()
print(f"P50: {report['p50']}ms | P95: {report['p95']}ms | P99: {report['p99']}ms")

延迟监控告警规则： - P95 > 15秒 → warn - P95 > 30秒 → critical - 单个请求 > 60秒 → critical（触发自动kill）

四、异常行为检测

Token异常和延迟异常还算好发现。最难的是"行为异常"——Agent的决策逻辑出了问题。

三种常见的行为异常：

class BehaviorMonitor:
    """Agent行为监控"""

    def __init__(self):
        self.tool_call_history = defaultdict(list)
        self.response_length_history = defaultdict(list)

    def record_tool_call(self, agent_id: str, tool_name: str, step_number: int):
        """记录工具调用"""
        self.tool_call_history[agent_id].append({
            "tool": tool_name,
            "step": step_number,
            "time": time.time()
        })

    def detect_loop(self, agent_id: str, threshold: int = 5, window: int = 3) -> bool:
        """检测死循环：最近window秒内同一工具调用超过threshold次"""
        calls = self.tool_call_history[agent_id]
        if len(calls) < threshold:
            return False

        recent = [c for c in calls if time.time() - c["time"] < window]
        if len(recent) < threshold:
            return False

        # 检查最近的调用是否集中在同一工具
        tool_counts = defaultdict(int)
        for c in recent:
            tool_counts[c["tool"]] += 1

        for tool, count in tool_counts.items():
            if count >= threshold:
                return {
                    "type": "tool_loop",
                    "tool": tool,
                    "count": count,
                    "window": window
                }
        return None

    def detect_quality_degradation(self, agent_id: str, 
                                    new_response_length: int,
                                    window: int = 10) -> bool:
        """检测质量退化：回答长度趋势下降"""
        self.response_length_history[agent_id].append(new_response_length)
        history = self.response_length_history[agent_id]

        if len(history) < window * 2:
            return None

        recent = history[-window:]
        earlier = history[-(window*2):-window]

        if sum(recent) / len(recent) < sum(earlier) / len(earlier) * 0.5:
            return {
                "type": "quality_degradation",
                "recent_avg": sum(recent) / len(recent),
                "earlier_avg": sum(earlier) / len(earlier),
                "ratio": sum(recent) / len(recent) / (sum(earlier)/len(earlier))
            }
        return None

# Agent运行时注入检测
monitor = BehaviorMonitor()

# 每次工具调用
monitor.record_tool_call("support-bot", "search_db", current_step)
loop = monitor.detect_loop("support-bot")
if loop:
    print(f"🚨 检测到死循环: {loop['tool']} 调用了{loop['count']}次")
    # 执行中断逻辑：kill当前Agent会话
    kill_agent_session("support-bot")

五、监控看板搭建

把上面的数据聚合到看板上。如果不想自己搭，推荐用Grafana + Prometheus + 自定义exporter：

from prometheus_client import start_http_server, Gauge, Histogram, Counter
import random
import time

# Prometheus指标定义
agent_duration = Histogram(
    'agent_step_duration_ms', 'Agent step duration in ms',
    ['agent_id', 'step'], buckets=[100, 500, 1000, 3000, 5000, 10000, 20000]
)

agent_tokens = Gauge(
    'agent_tokens_total', 'Total tokens used',
    ['agent_id', 'session_id']
)

agent_cost = Gauge(
    'agent_cost_usd', 'Total cost in USD',
    ['agent_id']
)

agent_tool_calls = Counter(
    'agent_tool_calls_total', 'Total tool calls',
    ['agent_id', 'tool_name']
)

agent_loops_detected = Counter(
    'agent_loops_detected_total', 'Loop detection count',
    ['agent_id']
)

# 启动Prometheus exporter（Grafana可以抓取）
start_http_server(8000)
print("Prometheus metrics available at :8000/metrics")

# 在Agent运行时记录指标
def agent_step_callback(agent_id, step_name, duration_ms, tokens, tool_name=None):
    agent_duration.labels(agent_id=agent_id, step=step_name).observe(duration_ms)
    if tool_name:
        agent_tool_calls.labels(agent_id=agent_id, tool_name=tool_name).inc()

在Grafana上建三个核心看板：

看板1：成本（Cost Dashboard） - 各Agent每日/每小时token消耗趋势图 - 单会话成本分布直方图 - 成本Top 10会话列表

看板2：性能（Performance Dashboard） - P50/P95/P99延迟趋势 - 各步骤耗时占比（pie chart） - 错误率和成功率趋势

看板3：健康（Health Dashboard） - 活跃会话数 - 死循环告警计数 - 质量评分趋势 - 各Agent状态指示灯

六、自动熔断机制

当异常检测触发时，自动执行熔断（Circuit Breaker）：

class CircuitBreaker:
    """Agent自动熔断"""

    def __init__(self):
        self.failures = defaultdict(int)
        self.last_trip_time = {}

    def record_failure(self, agent_id: str):
        self.failures[agent_id] += 1
        if self.failures[agent_id] >= 3:
            self.trip(agent_id)

    def trip(self, agent_id: str):
        """熔断：暂停Agent"""
        print(f"🔴 熔断 Agent {agent_id}，停止处理新请求")
        self.last_trip_time[agent_id] = time.time()
        # 实际操作：关掉consumer、暂停队列等
        disable_agent_queue(agent_id)

    def try_reset(self, agent_id: str):
        """30秒后尝试恢复"""
        if agent_id in self.last_trip_time:
            elapsed = time.time() - self.last_trip_time[agent_id]
            if elapsed > 30:
                print(f"🟢 尝试恢复 Agent {agent_id}")
                enable_agent_queue(agent_id)
                self.failures[agent_id] = 0
                return True
        return False

七、总结

• Token监控是第一优先级 — 跑飞一晚可能$200+，预算+突变检测必须有
• 延迟分析用P95不要用平均值 — 平均值会被大量正常请求拉低，P95才能反映真实问题
• 行为异常比性能异常更危险 — 死循环和幻觉问题修复成本远超多花几块钱
• 熔断是最后的保险 — 检测到异常后自动暂停Agent，防止损失扩大

💡 建议收藏本文。 把这些监控和告警规则加到你的Agent部署清单上。

📤 转发给负责AI Agent运维的同事。 一个晚上$47账单的教训，值得每个人看完。

📚 AI Agent运维系列 (一) 从日志到可观测性的完整方案 → (二) 追踪Token消耗、延迟与异常行为（本篇）