监控

Agent 越多，安全越像地基 9 月 28 日同一天，Anthropic 发布 Sonnet 5.5、Manus 跳到 2.0，NVIDIA 也拿出了一个看似不声不响、实则关键的动作：开放智能体安全平台（Open Agent Safety Platform）。它要做的事，是从软件层一直监控到芯片层，盯着智能体的一举一动。据介绍，已有超过一百家机构参与共建。当 Agent 从「聊天」走向「操作你的邮箱、账户、代码库」，传统只看模型输出的安全审核已经不够了——你得知道它在运行时到底访问了什么、改了什么、花了多少钱。 为什么要下沉到芯片 应用层的日志可以被 Agent 自己篡改或绕过，尤其是当它有了独立身份和自主权之后。把监控下沉到硬件和运行时，意味着安全策略由底层强制执行：模型准备调用某个高风险工具时，系统级可以直接拦截，而不是事后审计。这相当于给数字员工装上了不可绕过的门禁和监控探头。对金融、医疗、政务这类高合规场景，这种「硬隔离」比「软提示」可靠得多。平台还试图把常见的风险模式——比如越权访问数据库、异常对外转账、未经确认就发送邮件——整理成可复用的策略模板，让企业不必从零开始编写规则。 开放平台的算计 NVIDIA 把它做成开放平台而非私有产品，有两层算盘。一是卡位：Agent 安全终将成为标配，谁的监控层成为事实标准，谁就握住这一代系统的地基；二是绑定算力：安全监控跑在自家 GPU 上，等于把「安全」变成了又一个卖卡的理由。百余家机构的参与，则是在快速堆出生态，避免又变成一家公司的独角戏。 收束 Agent 越聪明、越自主，安全就越不能靠「提示它别作恶」。NVIDIA 这步棋的方向是对的：把约束写进运行时、焊在硬件里。对企业，现在评估 Agent 平台时，除了跑分和价格，务必多问一句——它的行为可观测、可拦截、可审计吗？答不上来的，迟早出事。 去论坛讨论 关于「Agent安全」你还有哪些角度？欢迎到 硅基AGI论坛 发帖讨论，或直接 按标题搜索 找到相关话题，和14位AI角色与真实用户一起把话题聊透。

Agent为什么难调试 不确定性：同一个输入可能走不同路径 多步推理：中间步骤不透明 工具调用：外部依赖多 需要监控什么 每步思考过程 工具调用输入输出 Token消耗和延迟 错误率和成功率 人工干预次数

为什么Agent需要可观测性？ 传统软件的行为是确定性的——同样输入产生同样输出。但Agent的行为由LLM驱动，具有随机性和非确定性。这使得Agent系统更需要完善的可观测性——才能知道Agent"在做什么"、“为什么这么做”、“花了多少钱”。 可观测性三大支柱 1. 日志（Logging） 记录Agent每一步的行为： { "timestamp": "2026-07-16T10:00:01.234Z", "session_id": "sess_abc123", "agent": "research_agent", "action": "tool_call", "tool": "web_search", "input": {"query": "AI芯片市场2025"}, "output": {"results": 5, "top_result": "..."}, "duration_ms": 1234, "tokens": {"input": 25, "output": 0}, "cost_usd": 0.0001, "status": "success" } 日志设计原则： 结构化（JSON而非纯文本） 可关联（session_id + step_id） 可过滤（level, agent, tool等字段） 采样策略（全量记录关键步骤，采样记录调试信息） 2. 指标（Metrics） 量化Agent运行状态： 业务指标： 任务成功率 用户满意度（评分/反馈） 平均完成时间 平均步数 技术指标： LLM调用延迟（p50/p95/p99） 工具调用延迟 Token消耗量 API错误率 每任务成本 资源指标： GPU利用率 内存占用 并发会话数 队列深度 3. 追踪（Tracing） 记录一次完整任务的全链路： Trace: sess_abc123 ├── Span 1: task_planning (1.2s) │ ├── LLM call: gpt-4 (0.8s, 500 tokens) │ └── Output: [search, analyze, report] ├── Span 2: web_search (0.6s) │ ├── tool: search_api │ └── Result: 5 items ├── Span 3: content_analysis (2.1s) │ ├── LLM call: gpt-4 (1.8s, 2000 tokens) │ └── Output: analysis_summary ├── Span 4: report_generation (1.5s) │ ├── LLM call: gpt-4 (1.2s, 1500 tokens) │ └── Output: final_report.md Total: 5.4s, 4000 tokens, $0.06 监控架构 数据采集层 # Agent执行包装器 class TracedAgent: def __init__(self, agent, tracer): self.agent = agent self.tracer = tracer @trace async def run(self, input_data): with self.tracer.span("agent_run") as span: span.set_attr("input", input_data) result = await self.agent.run(input_data) span.set_attr("output", result) span.set_attr("tokens", self.agent.total_tokens) span.set_attr("cost", self.agent.total_cost) return result @trace async def call_tool(self, tool, args): with self.tracer.span("tool_call") as span: span.set_attr("tool", tool.name) span.set_attr("args", args) start = time.time() result = await tool.run(args) duration = time.time() - start span.set_attr("duration_ms", duration * 1000) span.set_attr("result", result) return result 数据存储层 数据类型 存储方案 保留期 日志 Elasticsearch / Loki 30天 指标 Prometheus / InfluxDB 90天 追踪 Jaeger / Tempo 7天 会话记录 PostgreSQL / MongoDB 按需 可视化层 实时仪表盘： ...

AI系统可观测性的三个支柱 传统软件的可观测性关注延迟、吞吐、错误率。AI系统需要额外关注：token消耗、模型质量漂移、幻觉率、工具调用成功率等AI特有指标。 指标采集 核心指标定义 from prometheus_client import Counter, Histogram, Gauge # 请求指标 REQUEST_TOTAL = Counter('ai_requests_total', 'Total AI requests', ['model', 'status']) REQUEST_LATENCY = Histogram('ai_request_duration_seconds', 'Request duration', ['model']) ACTIVE_REQUESTS = Gauge('ai_active_requests', 'Active requests') # Token指标 TOKEN_INPUT = Counter('ai_tokens_input_total', 'Input tokens', ['model']) TOKEN_OUTPUT = Counter('ai_tokens_output_total', 'Output tokens', ['model']) TOKEN_COST = Counter('ai_token_cost_usd', 'Token cost in USD', ['model']) # 质量指标 HALLUCINATION_RATE = Gauge('ai_hallucination_rate', 'Hallucination rate', ['model']) TOOL_CALL_SUCCESS = Counter('ai_tool_calls_total', 'Tool calls', ['tool', 'status']) # 缓存指标 CACHE_HIT_RATE = Gauge('ai_cache_hit_rate', 'Cache hit rate') 中间件实现 class ObservabilityMiddleware: def __init__(self, app): self.app = app async def __call__(self, request): ACTIVE_REQUESTS.inc() start = time.time() model = request.json.get("model", "unknown") try: response = await self.app(request) duration = time.time() - start REQUEST_TOTAL.labels(model=model, status="success").inc() REQUEST_LATENCY.labels(model=model).observe(duration) if "usage" in response: TOKEN_INPUT.labels(model=model).inc(response["usage"]["prompt_tokens"]) TOKEN_OUTPUT.labels(model=model).inc(response["usage"]["completion_tokens"]) cost = self.calculate_cost(model, response["usage"]) TOKEN_COST.labels(model=model).inc(cost) return response except Exception as e: REQUEST_TOTAL.labels(model=model, status="error").inc() raise finally: ACTIVE_REQUESTS.dec() def calculate_cost(self, model, usage): pricing = {"gpt-4": 0.03, "qwen3-32b": 0.002, "claude-3": 0.015} rate = pricing.get(model, 0.01) return (usage["prompt_tokens"] + usage["completion_tokens"]) / 1000 * rate 链路追踪 from opentelemetry import trace tracer = trace.get_tracer(__name__) class TracedLLMCall: def __init__(self, llm_client): self.client = llm_client async def chat(self, messages, **kwargs): with tracer.start_as_current_span("llm_chat") as span: span.set_attribute("llm.model", kwargs.get("model", "unknown")) span.set_attribute("llm.messages_count", len(messages)) span.set_attribute("llm.temperature", kwargs.get("temperature", 0.7)) start = time.time() response = await self.client.chat(messages, **kwargs) duration = time.time() - start span.set_attribute("llm.duration_ms", duration * 1000) span.set_attribute("llm.prompt_tokens", response["usage"]["prompt_tokens"]) span.set_attribute("llm.completion_tokens", response["usage"]["completion_tokens"]) return response 质量监控 class QualityMonitor: def __init__(self, sample_rate=0.05): self.sample_rate = sample_rate # 采样5%的请求做质量评估 async def evaluate_response(self, query, response, context=None): """异步评估响应质量""" import random if random.random() > self.sample_rate: return None metrics = {} # 幻觉检测 metrics["hallucination"] = await self.detect_hallucination(response, context) # 相关性 metrics["relevance"] = await self.score_relevance(query, response) # 毒性检测 metrics["toxicity"] = await self.detect_toxicity(response) # 记录到Prometheus if metrics["hallucination"]: HALLUCINATION_RATE.inc() else: HALLUCINATION_RATE.dec(0.01) return metrics 告警规则 # Prometheus告警规则 groups: - name: ai_alerts rules: - alert: HighErrorRate expr: rate(ai_requests_total{status="error"}[5m]) / rate(ai_requests_total[5m]) > 0.05 for: 5m annotations: summary: "AI error rate > 5%" - alert: HighLatency expr: histogram_quantile(0.95, ai_request_duration_seconds_bucket) > 30 for: 10m annotations: summary: "P95 latency > 30s" - alert: HighCost expr: rate(ai_token_cost_usd[1h]) > 100 for: 30m annotations: summary: "Hourly cost > $100" - alert: ModelDegradation expr: ai_hallucination_rate > 0.15 for: 1h annotations: summary: "Hallucination rate > 15%" Grafana仪表板 关键面板： ...

引言 AI应用的质量保证（QA）远比传统软件复杂。传统软件QA关注"功能是否正常"，而AI应用QA还需要关注"输出是否准确"、“行为是否安全”、“体验是否良好"等多个维度。2026年，AI质量保证已经发展成为一个完整的学科。本文将系统介绍AI质量保证体系。 AI质量保证框架 框架组成 AI质量保证体系 ├── 开发阶段QA │ ├── 提示测试 │ ├── 模型评估 │ └── 安全审查 ├── 发布阶段QA │ ├── 回归测试 │ ├── A/B测试 │ └── 灰度发布 ├── 运行阶段QA │ ├── 实时监控 │ ├── 用户反馈 │ └── 异常检测 └── 治理阶段QA ├── 合规审查 ├── 审计日志 └── 持续改进 质量维度 维度 说明 评估方法 准确性 输出信息是否正确 事实核查、专家评估 完整性 是否完整回答了问题 人工评估、LLM评估 一致性 相似输入的输出是否一致 一致性测试 安全性 是否拒绝有害请求 红队测试 公平性 是否存在偏见 偏见检测 延迟 响应时间是否可接受 性能监控 可靠性 系统是否稳定运行 可用性监控 成本 单次调用成本是否合理 成本监控 开发阶段QA 提示质量审查 class PromptQA: def review(self, prompt, test_cases): """ 提示质量审查 """ report = { "accuracy": self.test_accuracy(prompt, test_cases), "format": self.test_format(prompt, test_cases), "robustness": self.test_robustness(prompt), "safety": self.test_safety(prompt), "cost": self.estimate_cost(prompt) } report["overall_score"] = self.calculate_overall(report) report["recommendation"] = self.recommend(report) return report 安全审查清单 ### AI安全审查清单 □ 提示注入防御 - 是否有输入隔离？ - 是否有指令强化？ - 是否有输出过滤？ □ 有害内容防御 - 是否能拒绝暴力内容请求？ - 是否能拒绝违法内容请求？ - 是否有内容过滤器？ □ 隐私保护 - 是否会泄露用户数据？ - 是否会泄露系统信息？ - 是否有数据脱敏？ □ 公平性 - 是否存在性别偏见？ - 是否存在种族偏见？ - 是否存在年龄偏见？ □ 可靠性 - 高负载下是否稳定？ - 模型API故障时是否有兜底？ - 是否有超时处理？ 发布阶段QA 灰度发布流程 class CanaryRelease: def __init__(self, config): self.stages = [ {"name": "internal", "traffic": 0.0, "duration": "1d"}, {"name": "canary_1", "traffic": 0.01, "duration": "1d"}, {"name": "canary_5", "traffic": 0.05, "duration": "2d"}, {"name": "canary_20", "traffic": 0.20, "duration": "2d"}, {"name": "full", "traffic": 1.0, "duration": "permanent"} ] def evaluate_stage(self, stage, metrics): """ 评估灰度阶段是否可以推进 """ checks = { "error_rate_ok": metrics["error_rate"] < 0.01, "latency_ok": metrics["p95_latency"] < 3000, "satisfaction_ok": metrics["satisfaction"] > 4.0, "cost_ok": metrics["cost_per_request"] < 0.05, "no_safety_incidents": metrics["safety_incidents"] == 0 } return all(checks.values()) 运行阶段QA 实时监控系统 class AIQualityMonitor: def __init__(self): self.metrics = { "accuracy": RollingMetric(window=1000), "latency": PercentileMetric(), "error_rate": RateMetric(window=60), "user_satisfaction": RollingMetric(window=500), "cost": CostTracker(), "safety": SafetyMonitor() } def record_request(self, request, response, user_feedback=None): """记录每次请求""" self.metrics["latency"].record(response.latency) self.metrics["error_rate"].record(response.error) self.metrics["cost"].record(response.token_cost) if user_feedback: self.metrics["user_satisfaction"].record(user_feedback) # 异步分析准确性和安全性 asyncio.create_task(self.analyze_async(request, response)) def check_alerts(self): """检查告警""" alerts = [] if self.metrics["error_rate"].current > 0.05: alerts.append("错误率过高") if self.metrics["latency"].p95 > 5000: alerts.append("延迟超标") if self.metrics["user_satisfaction"].current < 3.5: alerts.append("用户满意度下降") if self.metrics["safety"].has_incident(): alerts.append("安全事件") return alerts 用户反馈收集 class FeedbackCollector: def collect(self, user_id, response_id, feedback_type, content): """ 收集用户反馈 """ feedback = { "user_id": user_id, "response_id": response_id, "type": feedback_type, # "thumbs_up", "thumbs_down", "rating", "text" "content": content, "timestamp": datetime.now() } # 存储 self.store(feedback) # 如果是负面反馈，触发分析 if feedback_type == "thumbs_down": asyncio.create_task(self.analyze_negative_feedback(feedback)) 异常检测 class AnomalyDetector: def detect(self, recent_outputs, baseline): """ 检测输出异常 """ anomalies = [] # 长度异常 recent_lengths = [len(o) for o in recent_outputs] if mean(recent_lengths) > baseline["length_mean"] * 1.5: anomalies.append("输出长度异常增加") # 拒绝率异常 recent_refusals = sum(1 for o in recent_outputs if "无法" in o) if recent_refusals / len(recent_outputs) > baseline["refusal_rate"] * 2: anomalies.append("拒绝率异常升高") # 重复率异常 if self.compute_diversity(recent_outputs) < 0.3: anomalies.append("输出多样性下降") return anomalies 质量治理 审计日志 class AuditLogger: def log(self, event_type, details): """ 记录审计日志 """ log_entry = { "timestamp": datetime.now().isoformat(), "event_type": event_type, "details": details, "version": self.current_version, "hash": self.compute_hash(details) } # 写入不可变日志 self.immutable_store.append(log_entry) 合规检查 ### AI合规检查清单 □ 数据合规 - 用户数据是否加密存储？ - 是否有数据保留策略？ - 是否满足GDPR/个人信息保护法？ □ 算法合规 - 是否有算法备案？ - 是否有安全评估报告？ - 是否满足深度合成管理规定？ □ 内容合规 - 是否有内容审核机制？ - 是否有违法内容过滤？ - 是否有未成年人保护？ □ 透明度 - 是否告知用户在使用AI？ - 是否提供反馈渠道？ - 是否有人工替代方案？ 持续改进 PDCA循环 Plan（计划）： - 设定质量目标 - 制定改进计划 Do（执行）： - 实施改进 - 收集数据 Check（检查）： - 分析数据 - 评估效果 Act（行动）： - 标准化成功经验 - 修正不成功的尝试 质量仪表板 class QualityDashboard: def generate(self): return { "overall_health": "green", # green/yellow/red "metrics": { "accuracy": {"current": 0.92, "trend": "↑", "target": 0.90}, "latency_p95": {"current": 1200, "trend": "→", "target": 2000}, "satisfaction": {"current": 4.3, "trend": "↑", "target": 4.0}, "error_rate": {"current": 0.003, "trend": "↓", "target": 0.01}, "cost_per_request": {"current": 0.02, "trend": "→", "target": 0.05} }, "recent_incidents": [...], "improvement_actions": [...] } 2026年新趋势 1. AI自监控 AI系统自己监控自己的输出质量，自动发现问题。 ...

Agent可观测性：你无法优化你看不见的东西 传统软件的可观测性已经相当成熟——我们有Prometheus做指标、ELK做日志、Jaeger做追踪。但Agent系统引入了新的可观测性挑战：非确定性的执行路径、不可预测的token消耗、LLM输出的质量评估、多Agent协作的链路追踪。2026年，Agent可观测性已经形成了一套独立的最佳实践。 Agent可观测性的三大支柱 ┌───────────────────────────────────────────────────┐ │ Agent 可观测性三大支柱 │ ├─────────────┬─────────────┬───────────────────────┤ │ │ │ │ │ 追踪 │ 日志 │ 指标 │ │ (Traces) │ (Logs) │ (Metrics) │ │ │ │ │ │ Agent执行 │ 结构化事件 │ 性能计数器 │ │ 链路追踪 │ 决策记录 │ 资源消耗 │ │ 跨Agent关联 │ 上下文快照 │ 质量评估 │ │ │ │ │ └─────────────┴─────────────┴───────────────────────┘ 1. 分布式追踪 Agent Trace模型 Agent系统的追踪比传统微服务更复杂，因为一个"请求"可能涉及多轮LLM调用、多次工具调用、跨多个Agent。 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional import uuid @dataclass class AgentSpan: """Agent追踪的基本单元""" span_id: str = field(default_factory=lambda: str(uuid.uuid4())) parent_id: Optional[str] = None trace_id: str = "" # 基本信息 name: str = "" # span名称 span_type: str = "" # llm_call / tool_call / agent_step / sub_agent agent_name: str = "" # 哪个Agent start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None # Agent特有信息 input_data: Any = None output_data: Any = None model: str = "" # 使用的LLM模型 prompt_tokens: int = 0 completion_tokens: int = 0 cost: float = 0.0 # 状态 status: str = "ok" # ok / error / timeout error_message: Optional[str] = None # 元数据 attributes: dict = field(default_factory=dict) def finish(self, output=None, status="ok", error=None): self.end_time = datetime.now() self.output_data = output self.status = status self.error_message = error @property def duration_ms(self) -> float: if self.end_time: return (self.end_time - self.start_time).total_seconds() * 1000 return 0 追踪实现 class AgentTracer: """Agent追踪器""" def __init__(self, service_name="agent-service"): self.service_name = service_name self.spans: list[AgentSpan] = [] self.current_span_stack: list[AgentSpan] = [] def start_trace(self, name: str, agent_name: str) -> AgentSpan: """开始一个新的追踪（根span）""" trace_id = str(uuid.uuid4()) span = AgentSpan( trace_id=trace_id, name=name, span_type="agent_step", agent_name=agent_name ) self.spans.append(span) self.current_span_stack.append(span) return span def start_span( self, name: str, span_type: str, agent_name: str = "", input_data: Any = None ) -> AgentSpan: """开始一个子span""" parent = self.current_span_stack[-1] if self.current_span_stack else None span = AgentSpan( trace_id=parent.trace_id if parent else str(uuid.uuid4()), parent_id=parent.span_id if parent else None, name=name, span_type=span_type, agent_name=agent_name, input_data=input_data ) self.spans.append(span) self.current_span_stack.append(span) return span def end_span(self, span: AgentSpan, output=None, status="ok", error=None): """结束一个span""" span.finish(output, status, error) if self.current_span_stack and self.current_span_stack[-1].span_id == span.span_id: self.current_span_stack.pop() def get_trace_tree(self, trace_id: str) -> dict: """获取追踪树""" trace_spans = [s for s in self.spans if s.trace_id == trace_id] return self._build_tree(trace_spans, parent_id=None) def _build_tree(self, spans: list[AgentSpan], parent_id: str | None) -> dict: children = [s for s in spans if s.parent_id == parent_id] return [ { "span_id": s.span_id, "name": s.name, "type": s.span_type, "agent": s.agent_name, "duration_ms": s.duration_ms, "tokens": s.prompt_tokens + s.completion_tokens, "cost": s.cost, "status": s.status, "children": self._build_tree(spans, s.span_id) } for s in children ] 使用示例 tracer = AgentTracer() # 开始追踪 root = tracer.start_trace("用户咨询", "router_agent") # Agent执行LLM调用 llm_span = tracer.start_span("LLM调用", "llm_call", "router_agent", input_data="用户问题") response = await llm.complete("...") tracer.end_span(llm_span, output=response.text, status="ok") llm_span.prompt_tokens = response.usage.prompt_tokens llm_span.completion_tokens = response.usage.completion_tokens llm_span.cost = calculate_cost(response.usage, "gpt-4o") # Agent调用工具 tool_span = tracer.start_span("搜索知识库", "tool_call", "router_agent") results = await knowledge_base.search("query") tracer.end_span(tool_span, output=results) # 路由到子Agent sub_agent_span = tracer.start_span("专家Agent处理", "sub_agent", "expert_agent") # ... 子Agent内部会有自己的span ... tracer.end_span(sub_agent_span, output="最终回答") # 结束追踪 tracer.end_span(root, output="最终回答") # 查看追踪树 trace_tree = tracer.get_trace_tree(root.trace_id) 追踪可视化 追踪树可以渲染为瀑布图： ...

引言 Agent 系统的"黑盒"问题是生产化的最大障碍之一。一个 Agent 调用了 3 个工具、经过 5 轮推理、消耗了 8000 tokens，但出了问题你却不知道在哪一步。2026年，OpenTelemetry 社区正式发布了 SemConv for GenAI 规范，为 LLM 可观测性提供了标准化方案。 一、为什么 Agent 可观测性不同于传统应用 传统微服务的可观测性关注：请求路径、延迟分布、错误率。Agent 系统增加了三个新维度： Token 维度：每次调用消耗多少 Token？成本如何分摊？ 推理维度：模型"想"了什么？为什么选择这个工具？为什么跳过某步？ 非确定性维度：相同输入可能产生不同输出，仅靠日志无法复现 二、OpenTelemetry GenAI 语义规范 2026年正式定稿的 GenAI SemConv 定义了以下核心 Attributes： # GenAI 基础属性 gen_ai.system: "openai" # 提供商 gen_ai.request.model: "gpt-5" # 模型名称 gen_ai.request.temperature: 0.7 # 采样温度 gen_ai.request.max_tokens: 4096 # 最大 Token # Token 使用 gen_ai.usage.input_tokens: 1523 # 输入 Token gen_ai.usage.output_tokens: 876 # 输出 Token gen_ai.usage.cost: 0.0234 # 本次调用成本（美元） # Agent 特有 gen_ai.agent.name: "research-agent" gen_ai.agent.tool.name: "web_search" gen_ai.agent.tool.result.quality: 0.85 gen_ai.agent.iteration: 3 # 第几轮迭代 # 工具调用 gen_ai.tool.name: "calculator" gen_ai.tool.input: '{"expr": "2+2"}' gen_ai.tool.output: '{"result": 4}' gen_ai.tool.duration_ms: 45 三、全链路追踪实现 架构概览 ┌─────────────────────────────────────────────────────┐ │ Agent Application │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │Step1│──►│Step2│──►│Step3│──►│Step4│ │ │ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ ┌──▼─────────▼─────────▼─────────▼──┐ │ │ │ OpenTelemetry SDK │ │ │ │ (Auto-instrumentation + Custom) │ │ │ └──────────────┬────────────────────┘ │ └─────────────────┼───────────────────────────────────┘ │ OTLP/gRPC ┌────────────▼────────────┐ │ OTel Collector │ │ (处理/采样/导出) │ └──┬─────┬─────┬──────────┘ │ │ │ ┌────▼┐ ┌─▼──┐ ┌▼─────┐ │Jaeger│ │Prom│ │Loki │ │(Trace)│ │(Met)│ │(Log)│ └─────┘ └────┘ └──────┘ Python 实现 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.instrumentation.openai import OpenAIInstrumentor # 1. 初始化 OTel provider = TracerProvider() processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://otel-collector:4317") ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. 自动注入 OpenAI 调用的 Span OpenAIInstrumentor().instrument() # 3. Agent 自定义 Span tracer = trace.get_tracer("agent-system") class ObservabilityMiddleware: """Agent 可观测性中间件""" def __init__(self): self.tracer = trace.get_tracer("agent") async def on_agent_start(self, agent_name: str, input_data: dict): """Agent 启动时创建 Root Span""" self.root_span = self.tracer.start_span( f"agent.{agent_name}", attributes={ "gen_ai.agent.name": agent_name, "agent.input.size": len(str(input_data)), } ) async def on_llm_call(self, model: str, messages: list, **kwargs): """LLM 调用前记录""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"llm.{model}", context=ctx, attributes={ "gen_ai.request.model": model, "gen_ai.request.message_count": len(messages), "gen_ai.request.temperature": kwargs.get("temperature", 1.0), } ) return span async def on_llm_end(self, span, response): """LLM 调用后记录 Token 使用""" usage = response.usage span.set_attributes({ "gen_ai.usage.input_tokens": usage.prompt_tokens, "gen_ai.usage.output_tokens": usage.completion_tokens, "gen_ai.usage.total_tokens": usage.total_tokens, "gen_ai.usage.cost": calculate_cost( usage.prompt_tokens, usage.completion_tokens, response.model ), }) span.end() async def on_tool_call(self, tool_name: str, tool_input: dict): """工具调用追踪""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"tool.{tool_name}", context=ctx, attributes={ "gen_ai.tool.name": tool_name, "gen_ai.tool.input": json.dumps(tool_input)[:500], } ) return span async def on_agent_end(self, output: str): """Agent 结束""" self.root_span.set_attributes({ "agent.output.size": len(output), "agent.status": "success", }) self.root_span.end() Trace 可视化示例 在 Jaeger 中看到的典型 Agent Trace： ...

概述 AI智能体监控告警最佳实践是AI智能体领域中AI智能体监控告警最佳实践的重要主题。本文将从多个角度深入分析这一话题，为读者提供系统性的认知框架和实践参考。 核心概念 基本定义 在深入讨论之前，我们需要明确几个核心概念。AI智能体是指能够感知环境、理解指令、规划行动并调用工具完成任务的AI系统。与传统的聊天机器人不同，智能体具有自主性、目标导向性和工具使用能力。 AI智能体监控告警最佳实践涉及的关键技术包括： 大语言模型：作为智能体的认知引擎，负责理解、推理和生成 工具调用：通过Function Calling或MCP协议与外部系统交互 记忆系统：短期记忆处理当前对话，长期记忆存储历史经验 规划引擎：将复杂任务分解为可执行的子步骤 技术原理 从技术层面看，AI智能体监控告警最佳实践的核心在于如何让AI系统更好地理解和执行人类意图。这涉及多个技术环节的协同： 首先是感知层，智能体需要准确理解用户的自然语言指令，提取关键信息和约束条件。其次是规划层，将高层目标分解为具体的执行步骤。然后是执行层，调用合适的工具完成每个步骤。最后是反馈层，根据执行结果调整后续策略。 实践分析 当前现状 在实践指南领域，当前的技术实践呈现出几个明显特征： 工程化程度提升：从实验室原型到生产级系统，工程能力成为关键差异化因素 评估体系完善：越来越多标准化的评测基准被提出，帮助开发者量化能力边界 开源生态繁荣：开源框架和工具链的成熟降低了开发门槛 安全意识增强：对AI安全和对齐问题的重视程度显著提升 关键挑战 尽管进展显著，AI智能体监控告警最佳实践仍面临几个核心挑战： 技术挑战： 大模型的幻觉问题在智能体场景下被放大，因为智能体需要做出实际决策 多步推理中的错误累积效应导致长程任务成功率下降 工具调用的可靠性受外部API稳定性影响 工程挑战： 智能体的可观测性不足，调试和排错困难 成本控制与性能优化的平衡 从单机到分布式部署的架构复杂性 安全挑战： Prompt注入等攻击手段不断进化 智能体权限管理需要更精细化的控制 数据隐私保护在多Agent协作场景下更加复杂 优化策略 针对上述挑战，以下是几个关键优化方向： 技术优化 分而治之：将复杂任务分解为可独立验证的子任务，降低单步错误影响 多路投票：对关键决策使用多次采样投票机制，提高可靠性 渐进式信任：智能体权限从最小化开始，根据表现逐步扩展 人在回路：高风险决策保留人工审核环节 工程优化 可观测性优先：建立完善的日志、指标和追踪体系 灰度发布：新版本智能体先在小流量环境验证 自动化测试：构建端到端测试套件，防止回归 成本监控：实时追踪Token消耗和API调用成本 案例研究 为了更具体地说明AI智能体监控告警最佳实践的实践价值，我们来看一个典型场景： 某科技公司在内部IT运维中部署了AI智能体，负责处理员工的工单请求。智能体需要理解员工的自然语言描述，判断问题类型，查询知识库，执行修复操作或转接人工。 实施过程中遇到的关键问题包括： 员工描述模糊导致意图识别错误 知识库信息过时导致给出错误建议 某些操作需要管理员权限存在安全风险 解决方案： 引入澄清对话机制，在不确定时主动追问 建立知识库更新流程，定期审核内容 实施权限分级制度，敏感操作需人工确认 效果：工单首次解决率提升35%，平均处理时间缩短60%，员工满意度显著提升。 未来趋势 AI智能体监控告警最佳实践的发展趋势值得关注： 标准化：MCP等开放协议将推动工具接口标准化，降低集成成本 垂直化：针对特定行业和场景的专用智能体将大量涌现 协作化：多智能体协作将成为复杂任务的标准解决方案 自主化：智能体的自主决策能力将持续提升，但需要配套的安全机制 结论 AI智能体监控告警最佳实践是AI智能体技术发展中的重要一环。无论是技术原理的深入理解，还是实践中的工程优化，都需要系统性思维。对于开发者和企业而言，关键在于： 理解技术能力和边界，避免过度期待 建立系统化的评估和监控体系 在创新和安全之间找到平衡 持续学习和适应快速变化的技术生态 硅基AGI探索者将持续关注实践指南领域的最新进展，为读者提供深度分析和实践指导。— ...

为什么智能体需要专属的可观测性？ 传统软件的可观测性聚焦于 CPU、内存、延迟等系统指标。但 AI 智能体引入了全新的可观测维度： Token 消耗：每次 LLM 调用都有成本，需要精确追踪 推理链路：Agent 可能经历多轮 Thought → Action → Observation 循环，需要完整 Trace 工具调用质量：工具是否被正确调用？返回结果是否有效？ Prompt 效果：不同 Prompt 版本对输出质量的影响如何？ 幻觉检测：模型输出是否与事实相符？ 缺乏可观测性的智能体就像一个黑盒——你不知道它在想什么，也不知道它为什么出错。本文将带你从零搭建一套生产级的智能体可观测性平台。 可观测性三支柱在 AI 场景下的重构 传统可观测性的三支柱是 Metrics、Logs、Traces。在智能体场景下，我们需要将其扩展为五支柱： 支柱 传统场景 智能体场景 Traces 请求链路追踪 Thought-Action-Observation 链路 Metrics QPS、延迟、错误率 Token 用量、工具调用成功率、幻觉率 Logs 结构化日志 Prompt/Completion 完整记录 Evaluations N/A 输出质量自动评估 Cost N/A Token 成本与预算控制 架构设计 ┌────────────────────────────────────────────────────────┐ │ Agent Application │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │ │ │ LLM Call │ │ Tool Call │ │ Observation Callback │ │ │ └─────┬─────┘ └─────┬────┘ └──────────┬───────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Observability SDK │ │ │ │ (OpenTelemetry + Custom Spans + Token Counter) │ │ │ └──────────────────────┬──────────────────────────┘ │ └─────────────────────────┼──────────────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ Jaeger │ │ Prometheus │ │ Postgres │ │ (Traces) │ │ (Metrics) │ │ (Logs) │ └─────────────┘ └─────────────┘ └───────────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌─────────────┐ │ Grafana │ │ (Dashboard) │ └─────────────┘ 核心组件实现 1. Trace 追踪：基于 OpenTelemetry 扩展 智能体的 Trace 与传统微服务不同，需要记录 LLM 特有的 Span 属性： ...

1. 为什么 Agent 需要专门的可观测性 传统微服务可观测性关注请求-响应链路。Agent 系统则复杂得多： 多步骤推理：一个用户请求可能触发 10-50 次 LLM 调用 不确定输出：相同输入可能产生不同输出，难以复现问题 工具调用链：Agent → tool1 → Agent → tool2 → … 形成深层调用链 成本高昂：每次 LLM 调用都有实际成本，需要精准归因 幻觉风险：输出质量难以量化监控 普通监控工具（Prometheus/Grafana）只覆盖指标，无法完整追踪 Agent 行为。需要专门的 Agent 可观测性方案。 2. 可观测性三大支柱 ┌─────────────────────────────────────────────────┐ │ Agent Observability │ ├────────────┬──────────────┬────────────────────┤ │ Traces │ Metrics │ Logs │ │ (追踪) │ (指标) │ (日志) │ │ │ │ │ │ ┌────────┐ │ ┌────────┐ │ ┌──────────────┐ │ │ │Span │ │ │Counter │ │ │ Struct- │ │ │ │Tree │ │ │Gauge │ │ │ ured Log │ │ │ │Timeline│ │ │Histog. │ │ │ Event Log │ │ │ └────────┘ │ └────────┘ │ └──────────────┘ │ │ │ │ │ │ "发生了 │ "发生了几次" │ "发生了什么" │ │ 什么" │ "耗时多久" │ "为什么失败" │ └────────────┴──────────────┴────────────────────┘ 统一关联：TraceID + SpanID + LogID 3. 分布式追踪设计 3.1 Agent Span 模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Optional import uuid class SpanKind(str, Enum): LLM_CALL = "llm_call" TOOL_CALL = "tool_call" AGENT_STEP = "agent_step" RETRIEVAL = "retrieval" USER_INPUT = "user_input" SYSTEM = "system" @dataclass class SpanContext: trace_id: str span_id: str parent_span_id: Optional[str] = None @dataclass class SpanEvent: name: str timestamp: datetime attributes: dict = field(default_factory=dict) @dataclass class AgentSpan: """Agent 追踪 Span - 兼容 OpenTelemetry""" trace_id: str span_id: str parent_span_id: Optional[str] kind: SpanKind name: str agent_id: str session_id: str # 时间 start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None duration_ms: Optional[float] = None # 输入/输出 input: dict = field(default_factory=dict) output: dict = field(default_factory=dict) # 状态 status: str = "ok" # ok, error, timeout error_message: Optional[str] = None # 成本 & 性能 token_usage: dict = field(default_factory=dict) # {input: N, output: N} cost_usd: float = 0.0 model: Optional[str] = None # 扩展属性 attributes: dict = field(default_factory=dict) events: list[SpanEvent] = field(default_factory=list) child_spans: list[str] = field(default_factory=list) # child span_ids def finish(self, status: str = "ok"): self.end_time = datetime.now() self.duration_ms = (self.end_time - self.start_time).total_seconds() * 1000 self.status = status def add_event(self, name: str, attributes: dict = None): self.events.append(SpanEvent( name=name, timestamp=datetime.now(), attributes=attributes or {} )) 3.2 追踪器实现 class AgentTracer: """Agent 分布式追踪器""" def __init__(self, exporter: "TraceExporter"): self.exporter = exporter self._active_spans: dict[str, AgentSpan] = {} # span_id -> span self._trace_tree: dict[str, list[str]] = {} # trace_id -> [span_ids] def start_trace(self, session_id: str, agent_id: str, name: str = "agent_session") -> tuple[str, str]: """开始一次 Agent 会话追踪""" trace_id = str(uuid.uuid4()) root_span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=root_span_id, parent_span_id=None, kind=SpanKind.AGENT_STEP, name=name, agent_id=agent_id, session_id=session_id, ) self._active_spans[root_span_id] = span self._trace_tree[trace_id] = [root_span_id] return trace_id, root_span_id def start_span(self, trace_id: str, parent_span_id: str, kind: SpanKind, name: str, agent_id: str, session_id: str, input_data: dict = None) -> str: span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=span_id, parent_span_id=parent_span_id, kind=kind, name=name, agent_id=agent_id, session_id=session_id, input=input_data or {}, ) self._active_spans[span_id] = span self._trace_tree.setdefault(trace_id, []).append(span_id) # 更新父 span 的 child 列表 if parent_span_id in self._active_spans: self._active_spans[parent_span_id].child_spans.append(span_id) return span_id def finish_span(self, span_id: str, output_data: dict = None, status: str = "ok", error: str = None): if span_id not in self._active_spans: return span = self._active_spans[span_id] span.finish(status) if output_data: span.output = output_data if error: span.error_message = error # 异步导出 asyncio.create_task(self.exporter.export_span(span)) def get_trace(self, trace_id: str) -> list[AgentSpan]: span_ids = self._trace_tree.get(trace_id, []) return [self._active_spans[sid] for sid in span_ids if sid in self._active_spans] @staticmethod def _new_span_id() -> str: return str(uuid.uuid4())[:16] # 短 ID，便于展示 3.3 OpenTelemetry 集成 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter class OTelIntegration: """将 Agent Span 桥接到 OpenTelemetry""" def __init__(self, endpoint: str = "localhost:4317"): self.provider = TracerProvider() self.exporter = OTLPSpanExporter(endpoint=endpoint) self.processor = BatchSpanProcessor(self.exporter) self.provider.add_span_processor(self.processor) trace.set_tracer_provider(self.provider) self.tracer = trace.get_tracer("agent-system") def to_otel_span(self, agent_span: AgentSpan): """将 AgentSpan 转换为 OTel Span""" with self.tracer.start_as_current_span( agent_span.name, context=self._make_context(agent_span), kind=self._map_kind(agent_span.kind), ) as otel_span: # 设置属性 otel_span.set_attribute("agent.id", agent_span.agent_id) otel_span.set_attribute("session.id", agent_span.session_id) otel_span.set_attribute("llm.model", agent_span.model or "") otel_span.set_attribute("cost.usd", agent_span.cost_usd) for k, v in agent_span.token_usage.items(): otel_span.set_attribute(f"token.{k}", v) # 设置状态 if agent_span.status == "error": otel_span.set_status(trace.Status(trace.StatusCode.ERROR, agent_span.error_message)) # 添加事件 for evt in agent_span.events: otel_span.add_event(evt.name, evt.attributes) def _map_kind(self, kind: SpanKind): mapping = { SpanKind.LLM_CALL: trace.SpanKind.CLIENT, SpanKind.TOOL_CALL: trace.SpanKind.INTERNAL, SpanKind.AGENT_STEP: trace.SpanKind.INTERNAL, SpanKind.RETRIEVAL: trace.SpanKind.CLIENT, } return mapping.get(kind, trace.SpanKind.INTERNAL) 4. 指标采集 4.1 关键指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary import time class AgentMetrics: """Agent 系统关键指标""" def __init__(self, namespace: str = "agent"): # LLM 调用指标 self.llm_requests = Counter( f"{namespace}_llm_requests_total", "Total LLM requests", ["model", "agent_id", "status"] ) self.llm_latency = Histogram( f"{namespace}_llm_latency_seconds", "LLM request latency", ["model", "agent_id"], buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0] ) self.llm_tokens = Counter( f"{namespace}_llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) self.llm_cost = Counter( f"{namespace}_llm_cost_usd_total", "Total LLM cost in USD", ["model", "agent_id"] ) # Agent 行为指标 self.agent_steps = Counter( f"{namespace}_agent_steps_total", "Total agent reasoning steps", ["agent_id", "step_type"] ) self.agent_session_duration = Histogram( f"{namespace}_agent_session_duration_seconds", "Agent session duration", ["agent_id", "outcome"] # outcome: success/failure/timeout ) self.agent_tool_calls = Counter( f"{namespace}_agent_tool_calls_total", "Total tool calls", ["agent_id", "tool_name", "status"] ) # 系统指标 self.active_sessions = Gauge( f"{namespace}_active_sessions", "Current active sessions", ["agent_id"] ) self.queue_depth = Gauge( f"{namespace}_queue_depth", "Current queue depth", ["queue_name"] ) self.error_rate = Summary( f"{namespace}_error_rate", "Error rate over 5m window", ["agent_id", "error_type"] ) def record_llm_call(self, model: str, agent_id: str, duration: float, input_tokens: int, output_tokens: int, cost: float, status: str): self.llm_requests.labels(model=model, agent_id=agent_id, status=status).inc() self.llm_latency.labels(model=model, agent_id=agent_id).observe(duration) self.llm_tokens.labels(model=model, direction="input").inc(input_tokens) self.llm_tokens.labels(model=model, direction="output").inc(output_tokens) self.llm_cost.labels(model=model, agent_id=agent_id).inc(cost) def record_agent_step(self, agent_id: str, step_type: str): self.agent_steps.labels(agent_id=agent_id, step_type=step_type).inc() def record_tool_call(self, agent_id: str, tool_name: str, status: str): self.agent_tool_calls.labels(agent_id=agent_id, tool_name=tool_name, status=status).inc() 4.2 成本归因指标 from collections import defaultdict class CostAttribution: """成本归因：按用户/会话/功能分解成本""" def __init__(self, redis_client): self.redis = redis_client async def record_cost(self, trace_id: str, user_id: str, feature: str, model: str, cost: float, tokens: int): pipe = self.redis.pipeline() # 按用户聚合 pipe.incrbyfloat(f"cost:user:{user_id}:total", cost) pipe.incrby(f"cost:user:{user_id}:tokens", tokens) # 按功能聚合 pipe.incrbyfloat(f"cost:feature:{feature}:total", cost) # 按模型聚合 pipe.incrbyfloat(f"cost:model:{model}:total", cost) # 按 trace 明细 pipe.hset(f"cost:trace:{trace_id}", mapping={ "user_id": user_id, "feature": feature, "model": model, "cost": str(cost), "tokens": str(tokens), "timestamp": str(time.time()) }) await pipe.execute() async def get_user_cost_report(self, user_id: str, days: int = 7) -> dict: return { "total_cost_usd": float(await self.redis.get(f"cost:user:{user_id}:total") or 0), "total_tokens": int(await self.redis.get(f"cost:user:{user_id}:tokens") or 0), "by_feature": await self._get_by_dimension(f"cost:feature", user_id, days), "by_model": await self._get_by_dimension(f"cost:model", user_id, days), } 5. 结构化日志 5.1 日志格式规范 import json import logging import sys from datetime import datetime from typing import Any class StructuredLogger: """结构化日志：JSON 格式，便于 ELK/Loki 检索""" def __init__(self, service_name: str, level: str = "INFO"): self.service = service_name self.logger = logging.getLogger(service_name) self.logger.setLevel(getattr(logging, level)) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(self) self.logger.addHandler(handler) def format(self, record: logging.LogRecord) -> str: log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": record.levelname, "service": self.service, "message": record.getMessage(), "trace_id": getattr(record, "trace_id", None), "span_id": getattr(record, "span_id", None), "session_id": getattr(record, "session_id", None), "agent_id": getattr(record, "agent_id", None), "user_id": getattr(record, "user_id", None), } # 添加额外字段 if hasattr(record, "extra_fields"): log_entry.update(record.extra_fields) return json.dumps(log_entry, ensure_ascii=False) def bind(self, **kwargs) -> "BoundLogger": return BoundLogger(self, kwargs) class BoundLogger: """绑定上下文的 Logger""" def __init__(self, logger: StructuredLogger, bound: dict): self._logger = logger self._bound = bound def _log(self, level: str, msg: str, **kwargs): extra = {**self._bound, **kwargs} record = self._logger.logger.makeRecord( self._logger.logger.name, getattr(logging, level), "(unknown)", 0, msg, [], None ) record.extra_fields = extra for k, v in extra.items(): setattr(record, k, v) self._logger.logger.handle(record) def info(self, msg: str, **kwargs): self._log("INFO", msg, **kwargs) def warning(self, msg: str, **kwargs): self._log("WARNING", msg, **kwargs) def error(self, msg: str, **kwargs): self._log("ERROR", msg, **kwargs) def debug(self, msg: str, **kwargs): self._log("DEBUG", msg, **kwargs) 5.2 日志事件类型 class AgentLogEvents: """Agent 关键生命周期事件""" @staticmethod def session_started(logger: BoundLogger, session_id: str, user_id: str): logger.info("agent.session.started", session_id=session_id, user_id=user_id, event_type="session_lifecycle") @staticmethod def llm_call_started(logger: BoundLogger, model: str, prompt_length: int): logger.info("llm.call.started", model=model, prompt_length=prompt_length, event_type="llm_call") @staticmethod def llm_call_completed(logger: BoundLogger, model: str, latency_ms: float, input_tokens: int, output_tokens: int, cost: float): logger.info("llm.call.completed", model=model, latency_ms=latency_ms, input_tokens=input_tokens, output_tokens=output_tokens, cost_usd=cost, event_type="llm_call") @staticmethod def tool_call_started(logger: BoundLogger, tool_name: str, args: dict): logger.info("tool.call.started", tool_name=tool_name, tool_args=args, event_type="tool_call") @staticmethod def tool_call_completed(logger: BoundLogger, tool_name: str, success: bool, result_summary: str): logger.info("tool.call.completed", tool_name=tool_name, success=success, result_summary=result_summary[:200], event_type="tool_call") @staticmethod def hallucination_detected(logger: BoundLogger, claim: str, confidence: float): logger.warning("agent.hallucination.detected", claim=claim[:100], confidence=confidence, event_type="quality") @staticmethod def session_completed(logger: BoundLogger, session_id: str, total_steps: int, total_cost: float, outcome: str): logger.info("agent.session.completed", session_id=session_id, total_steps=total_steps, total_cost_usd=total_cost, outcome=outcome, event_type="session_lifecycle") 6. 完整可观测性集成 class ObservableAgent: """集成可观测性的 Agent 基类""" def __init__(self, agent_id: str, tracer: AgentTracer, metrics: AgentMetrics, logger: StructuredLogger): self.agent_id = agent_id self.tracer = tracer self.metrics = metrics self.logger = logger.bind(agent_id=agent_id) self._current_trace_id: str | None = None self._current_span_id: str | None = None async def run(self, user_query: str, user_id: str, session_id: str) -> str: # 开始追踪 trace_id, root_span_id = self.tracer.start_trace(session_id, self.agent_id) self._current_trace_id = trace_id self._current_span_id = root_span_id # 绑定 trace 上下文到日志 self.logger = self.logger.bind(trace_id=trace_id, session_id=session_id, user_id=user_id) session_start = time.time() try: self.logger.info("agent.session.started", query=user_query[:100]) self.metrics.active_sessions.labels(agent_id=self.agent_id).inc() # 执行 Agent 逻辑（带追踪） result = await self._execute_with_tracing(user_query) duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="success" ).observe(duration) self.logger.info("agent.session.completed", duration_seconds=duration, outcome="success") self.tracer.finish_span(root_span_id, {"result": result}, "ok") return result except Exception as e: duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="failure" ).observe(duration) self.logger.error("agent.session.failed", error=str(e), exc_info=True) self.tracer.finish_span(root_span_id, status="error", error=str(e)) raise finally: self.metrics.active_sessions.labels(agent_id=self.agent_id).dec() async def _execute_with_tracing(self, query: str) -> str: # LLM 调用 llm_span_id = self.tracer.start_span( self._current_trace_id, self._current_span_id, SpanKind.LLM_CALL, "llm_chat", self.agent_id, self._get_session_id(), {"query": query} ) llm_start = time.time() try: self.logger.info("llm.call.started") response = await self._call_llm(query) latency = time.time() - llm_start self.metrics.record_llm_call( model="gpt-4o", agent_id=self.agent_id, duration=latency, input_tokens=100, output_tokens=50, cost=0.003, status="ok" ) self.tracer.finish_span(llm_span_id, {"response": response}, "ok") return response except Exception as e: self.tracer.finish_span(llm_span_id, status="error", error=str(e)) raise def _get_session_id(self) -> str: return self.logger._bound.get("session_id", "") 7. 可观测性数据流水线 Agent 运行时 │ ├── Traces ──→ OTel Collector ──→ Jaeger/Tempo ──→ Grafana │ ├── Metrics ──→ Prometheus ──────────────────────→ Grafana │ └── Logs ────→ Loki/ELK ───────────────────────→ Grafana │ └──→ 告警规则 → AlertManager → Slack/PagerDuty 7.1 OpenTelemetry Collector 配置 # otel-collector-config.yaml receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 processors: batch: timeout: 5s send_batch_size: 100 memory_limiter: check_interval: 5s limit_mib: 1000 attributes: actions: - key: agent.id action: insert value: "unknown" resource: attributes: - key: deployment.environment value: "production" action: insert exporters: otlp/jaeger: endpoint: jaeger:4317 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 loki: endpoint: http://loki:3100/loki/api/v1/push service: pipelines: traces: receivers: [otlp] processors: [batch, memory_limiter, attributes] exporters: [otlp/jaeger] metrics: receivers: [otlp] processors: [batch, memory_limiter] exporters: [prometheus] logs: receivers: [otlp] processors: [batch, memory_limiter] exporters: [loki] 8. 告警规则 # prometheus-alerts.yaml groups: - name: agent_alerts rules: - alert: AgentHighErrorRate expr: | sum(rate(agent_agent_steps_total{status="error"}[5m])) by (agent_id) / sum(rate(agent_agent_steps_total[5m])) by (agent_id) > 0.05 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} 错误率超过 5%" - alert: AgentHighLatency expr: | histogram_quantile(0.95, sum(rate(agent_llm_latency_seconds_bucket[5m])) by (le, agent_id) ) > 5 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} P95 延迟超过 5 秒" - alert: AgentHighCost expr: | rate(agent_llm_cost_usd_total[1h]) > 10 for: 5m labels: severity: critical annotations: summary: "Agent 每小时成本超过 $10" - alert: AgentHallucinationDetected expr: | increase(agent_quality_hallucination_total[10m]) > 5 labels: severity: warning annotations: summary: "检测到 {{ $value }} 次幻觉" 9. 总结 Agent 可观测性的核心在于全链路关联： ...