AI Agent生产环境监控：从日志到可观测性的完整方案

📡 传统服务挂了你知道——数据库连不上、CPU 100%。但AI Agent"疑似表现不好"怎么监控？这不是普通的后端监控能解决的问题。

symptom_zh

你的AI Agent上线了。没有崩溃，没有报错，但就是感觉"不如测试时好"。Agent会在多种层面上"坏掉"：token消耗突然飙升、响应延迟时快时慢、回答质量逐渐下降、Agent陷入死循环不自知。传统监控只能告诉你"服务还活着"，但AI Agent需要可观测性（Observability）——你不仅要看到它在运行，还要看到它的思考过程、调用链、成本消耗和质量趋势。这篇从最基础的日志讲起，搭到生产级的监控体系。

适合人群： 部署了AI Agent到生产环境的开发者 读完能做： 给自己的Agent搭建完整的监控看板

diagnosis_zh

一、AI Agent监控为什么跟普通后端不一样？

一个普通后端请求：

用户请求 → 查数据库 → 返回JSON [耗时50ms，返回200]

一个AI Agent请求：

用户请求 → LLM思考(多轮推理) → 调用工具A(调用外部API) → 
LLM评估结果 → 调用工具B(查数据库) → LLM合成回答 → 调用工具C(发邮件通知用户)
[耗时3-15秒，多次API调用，token消耗不等]

区别很明显： - 耗时波动大 — 3秒到15秒都是正常的 - 成本线性关联 — 每次推理都在花钱 - 成功率定义模糊 — "回答不好"不算错误 - 多步调用链 — 传统APM（应用性能监控）最多追踪API调用，看不到Agent的决策过程

二、第一层：结构化日志

第一件事：把Agent的每次行动都记下来。

不要用print()，不要用console.log。要结构化日志，每个字段可搜索、可过滤。

import json
import logging
from datetime import datetime

class AgentLogger:
    """AI Agent的结构化日志"""

    @staticmethod
    def log_step(agent_id: str, step: str, input_data: dict, 
                 output_data: dict, duration_ms: int, 
                 tokens: int, cost: float):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": agent_id,
            "type": "agent_step",
            "step": step,           # think / tool_call / respond
            "input": input_data,
            "output": output_data,
            "metrics": {
                "duration_ms": duration_ms,
                "tokens": tokens,
                "cost": cost
            }
        }
        logging.getLogger("agent").info(json.dumps(log_entry))

# 使用
def agent_think(agent_id, prompt):
    start = time.time()
    response = call_llm(prompt)
    duration = int((time.time() - start) * 1000)

    AgentLogger.log_step(
        agent_id=agent_id,
        step="think",
        input_data={"prompt_length": len(prompt)},
        output_data={"response_length": len(response)},
        duration_ms=duration,
        tokens=response.usage.total_tokens if hasattr(response, 'usage') else 0,
        cost=calculate_cost(response)
    )
    return response

应该记录哪些字段：

三、第二层：调用链追踪

单步日志够了，但你需要知道一次完整的Agent交互用了多少步、每一步花了多少钱。这就是调用链（Trace）。

import uuid
import time
from contextlib import contextmanager

class TraceManager:
    """调用链追踪"""

    def __init__(self):
        self.traces = {}

    def start_trace(self, session_id: str = None) -> str:
        trace_id = session_id or str(uuid.uuid4())
        self.traces[trace_id] = {
            "trace_id": trace_id,
            "start_time": time.time(),
            "steps": [],
            "total_tokens": 0,
            "total_cost": 0.0
        }
        return trace_id

    @contextmanager
    def step(self, trace_id: str, step_name: str):
        trace = self.traces[trace_id]
        step_start = time.time()
        step_data = {"name": step_name, "start": step_start}
        try:
            yield step_data
        except Exception as e:
            step_data["error"] = str(e)
            raise
        finally:
            step_data["duration_ms"] = int((time.time() - step_start) * 1000)
            trace["steps"].append(step_data)

    def record_llm_call(self, trace_id: str, tokens: int, cost: float):
        trace = self.traces[trace_id]
        trace["total_tokens"] += tokens
        trace["total_cost"] += cost

    def end_trace(self, trace_id: str) -> dict:
        trace = self.traces[trace_id]
        trace["total_duration_ms"] = int((time.time() - trace["start_time"]) * 1000)
        trace["step_count"] = len(trace["steps"])
        return trace

# 使用
manager = TraceManager()

def handle_user_request(user_message):
    trace_id = manager.start_trace()

    with manager.step(trace_id, "think"):
        response = call_llm(f"思考如何回答: {user_message}")
        manager.record_llm_call(trace_id, response.usage.total_tokens, 0.002)

    if needs_tool:
        with manager.step(trace_id, "search_database"):
            db_results = query_database(user_message)

    with manager.step(trace_id, "respond"):
        final = call_llm(f"基于以下信息回答: {db_results}")
        manager.record_llm_call(trace_id, final.usage.total_tokens, 0.003)

    trace = manager.end_trace(trace_id)
    print(f"总耗时: {trace['total_duration_ms']}ms, 总成本: ${trace['total_cost']:.4f}")
    return final

四、第三层：质量监控

调用链只能告诉你"Agent有没有运行"，但不能告诉你"Agent运行得好不好"。这就是质量监控要做的事。

质量监控的关键指标：

class QualityMonitor:
    """Agent输出质量监控"""

    @staticmethod
    def score_response(question: str, response: str) -> dict:
        """让LLM给回答质量打分"""
        scoring_prompt = f"""
        为以下AI回答打分（1-10分），评分维度：
        - accuracy: 准确性（是否包含事实错误）
        - relevance: 相关度（是否回答了用户问题）
        - completeness: 完整度（是否漏掉了重要方面）
        - safety: 安全性（是否有不安全内容）

        问题：{question}
        回答：{response}

        只输出JSON：{"{"}"accuracy": N, "relevance": N, "completeness": N, "safety": N, "overall": N{"}"}
        """

        result = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": scoring_prompt}],
            response_format={"type": "json_object"}
        )
        return json.loads(result.choices[0].message.content)

监控的指标：

五、第四层：实时告警

有了数据之后，让工具帮你发现异常：

class AlertManager:
    """告警管理"""

    THRESHOLDS = {
        "high_latency": {"metric": "p95_duration_ms", "warn": 10000, "critical": 20000},
        "high_cost": {"metric": "cost_per_session", "warn": 0.05, "critical": 0.1},
        "low_quality": {"metric": "avg_quality_score", "warn": 7, "critical": 5},
        "loop_detected": {"metric": "step_count", "warn": 8, "critical": 12},
        "error_rate": {"metric": "error_rate_pct", "warn": 5, "critical": 15},
    }

    @classmethod
    def evaluate(cls, metrics: dict) -> list:
        alerts = []
        for rule_name, rule in cls.THRESHOLDS.items():
            value = metrics.get(rule["metric"])
            if value is None:
                continue
            if value >= rule["critical"]:
                alerts.append({"level": "critical", "rule": rule_name, 
                              "value": value, "threshold": rule["critical"]})
            elif value >= rule["warn"]:
                alerts.append({"level": "warn", "rule": rule_name,
                              "value": value, "threshold": rule["warn"]})
        return alerts

六、开源工具推荐

七、总结

• AI Agent监控三件套：结构化日志 + 调用链追踪 + 质量评分
• 传统APM不够用 — Agent需要看到思考过程和调用链
• 成本监控跟质量监控一样重要 — Agent跑飞了可能很贵
• 告警要有梯度 — warn先提醒，critical才叫醒你

💡 建议收藏本文。 Agent部署到生产不是终点，有监控才算上线。

📤 转发给部署Agent到生产的同事。 没有监控的Agent跟裸奔没什么区别。

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