监控

为什么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 可观测性的核心在于全链路关联： ...

AI 应用为什么需要专门的监控 传统 APM（应用性能监控）关注的是 CPU、内存、响应时间和错误率。但 AI 应用引入了全新的维度： Token 维度：每次请求消耗的 Token 数直接影响成本 模型质量：响应不仅要快，还要准确、安全、无幻觉 长尾延迟：流式输出可能持续数分钟，P99 指标失真 多模型路由：请求分散在不同模型上，需要统一视图 如果你的监控面板还只有"QPS + 延迟 + 错误率"三件套，你实际上是在"盲飞"。 监控指标体系 四层指标架构 ┌─────────────────────────────────────────────┐ │ 业务指标层 (Business) │ │ 任务完成率 · 用户满意度 · 成本/请求 │ ├─────────────────────────────────────────────┤ │ AI 质量指标层 (Quality) │ │ 幻觉率 · 意图准确率 · 安全违规率 │ ├─────────────────────────────────────────────┤ │ 模型性能指标层 (Model) │ │ 首 Token 延迟 · 生成速率 · KV Cache 命中率 │ ├─────────────────────────────────────────────┤ │ 基础设施指标层 (Infra) │ │ GPU 利用率 · 显存使用 · CPU/内存 · 网络 │ └─────────────────────────────────────────────┘ 完整指标清单 层级 指标名 类型 说明 告警阈值 Infra gpu_utilization Gauge GPU 计算利用率 <20% 或 >95% Infra gpu_memory_used_ratio Gauge GPU 显存使用率 >0.95 Infra gpu_temperature Gauge GPU 温度 (°C) >85 Model ttft_seconds Histogram 首 Token 延迟 (TTFT) P95 >2s Model tokens_per_second Gauge Token 生成速率 <30 tokens/s Model kv_cache_hit_rate Gauge KV Cache 命中率 <0.3 Model active_sequences Gauge 活跃推理序列数 >max_seqs×0.9 Model queue_depth Gauge 排队请求数 >10 Quality hallucination_score Gauge 幻觉评分 (0-1) >0.15 Quality safety_violation_count Counter 安全违规次数 >0/h Quality user_feedback_score Histogram 用户评分 (1-5) 均值 <3.5 Business cost_per_request Gauge 单请求成本 ($) >$0.05 Business task_completion_rate Gauge 任务完成率 <80% Business daily_spend Counter 日累计花费 >预算 80% Prometheus 指标采集 应用侧指标暴露 from prometheus_client import Counter, Histogram, Gauge, generate_latest from prometheus_client import CollectorRegistry, CONTENT_TYPE_LATEST from fastapi import FastAPI, Response import time app = FastAPI() registry = CollectorRegistry() # === 基础设施指标 === gpu_utilization = Gauge( "ai_gpu_utilization", "GPU utilization percentage", ["gpu_id", "model_name"], registry=registry ) gpu_memory_used = Gauge( "ai_gpu_memory_used_bytes", "GPU memory used in bytes", ["gpu_id"], registry=registry ) # === 模型性能指标 === request_duration = Histogram( "ai_request_duration_seconds", "Total request duration", ["model_name", "endpoint"], buckets=[0.5, 1, 2, 5, 10, 30, 60, 120, 300], registry=registry, ) ttft = Histogram( "ai_time_to_first_token_seconds", "Time to first token", ["model_name"], buckets=[0.1, 0.25, 0.5, 1, 2, 5, 10], registry=registry, ) tokens_generated = Counter( "ai_tokens_generated_total", "Total tokens generated", ["model_name", "type"], # type: input|output registry=registry, ) # === 质量指标 === hallucination_score = Gauge( "ai_hallucination_score", "Hallucination score (0=good, 1=bad)", ["model_name"], registry=registry, ) safety_violations = Counter( "ai_safety_violations_total", "Safety violation count", ["model_name", "violation_type"], registry=registry, ) # === 业务指标 === request_cost = Gauge( "ai_request_cost_usd", "Cost per request in USD", ["model_name"], registry=registry, ) daily_spend = Counter( "ai_daily_spend_usd", "Daily total spend in USD", ["model_name"], registry=registry, ) # === 中间件：自动采集请求指标 === @app.middleware("http") async def metrics_middleware(request, call_next): start_time = time.time() response = await call_next(request) duration = time.time() - start_time model = request.headers.get("X-Model-Name", "unknown") request_duration.labels( model_name=model, endpoint=request.url.path, ).observe(duration) return response # === 业务逻辑中手动埋点 === async def chat_completion(messages, model="gpt-4o-mini"): start = time.time() # 调用 LLM response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, stream_options={"include_usage": True}, ) first_token_time = None total_output_tokens = 0 async for chunk in response: if first_token_time is None and chunk.choices[0].delta.content: first_token_time = time.time() ttft.labels(model_name=model).observe(first_token_time - start) if chunk.usage: total_output_tokens = chunk.usage.completion_tokens input_tokens = chunk.usage.prompt_tokens tokens_generated.labels(model_name=model, type="input").inc(input_tokens) tokens_generated.labels(model_name=model, type="output").inc(total_output_tokens) # 计算成本 cost = calculate_cost(model, input_tokens, total_output_tokens) request_cost.labels(model_name=model).set(cost) daily_spend.labels(model_name=model).inc(cost) return response # === 异步质量评估 === async def evaluate_quality(response: str, context: str, model: str): """异步评估响应质量""" # 幻觉检测 score = await hallucination_detector.check(response, context) hallucination_score.labels(model_name=model).set(score) # 安全检查 violations = safety_checker.check(response) for v_type, count in violations.items(): safety_violations.labels( model_name=model, violation_type=v_type, ).inc(count) # === Prometheus 指标暴露端点 === @app.get("/metrics") async def metrics(): return Response( generate_latest(registry), media_type=CONTENT_TYPE_LATEST, ) GPU 指标采集（DCGM Exporter） # gpu-metrics-stack.yaml apiVersion: v1 kind: Namespace metadata: name: gpu-monitoring --- apiVersion: apps/v1 kind: DaemonSet metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter template: metadata: labels: app: dcgm-exporter spec: nodeSelector: accelerator: nvidia-t4 # 调度到 GPU 节点 tolerations: - key: nvidia.com/gpu operator: Exists containers: - name: dcgm-exporter image: nvcr.io/nvidia/k8s/dcgm-exporter:3.3.0-ubuntu22.04 ports: - containerPort: 9400 name: metrics env: - name: DCGM_EXPORTER_LISTEN value: ":9400" - name: DCGM_EXPORTER_KUBERNETES value: "true" securityContext: capabilities: add: ["SYS_ADMIN"] --- # ServiceMonitor 让 Prometheus 自动发现 apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter endpoints: - port: metrics path: "/metrics" interval: 10s DCGM 关键 GPU 指标 指标 说明 用途 DCGM_FI_DEV_GPU_UTIL GPU 计算利用率 (%) 判断是否充分使用 GPU DCGM_FI_DEV_MEM_COPY_UTIL 显存带宽利用率 (%) 判断是否显存带宽瓶颈 DCGM_FI_DEV_FB_USED 已用显存 (MB) 判断显存是否够用 DCGM_FI_DEV_FB_FREE 空闲显存 (MB) 可调度的余量 DCGM_FI_DEV_GPU_TEMP GPU 温度 (°C) 散热健康 DCGM_FI_DEV_POWER_USAGE 功耗 (W) 能耗成本 DCGM_FI_PROF_PIPE_TENSOR_ACTIVE Tensor Core 利用率 推理效率 Grafana 仪表盘构建 仪表盘布局设计 ┌──────────────────────────────────────────────────────────┐ │ AI Application Overview │ ├────────────┬────────────┬────────────┬───────────────────┤ │ QPS │ P95 TTFT │ Error Rate│ Daily Spend │ │ (Stat) │ (Stat) │ (Stat) │ (Stat) │ ├────────────┴────────────┼────────────┴───────────────────┤ │ Request Latency │ Token Throughput │ │ (Time Series) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ GPU Utilization │ GPU Memory Usage │ │ (Time Series, per GPU) │ (Time Series, per GPU) │ ├──────────────────────────┼───────────────────────────────┤ │ Model Distribution │ Queue Depth │ │ (Pie Chart) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ Hallucination Score │ Safety Violations │ │ (Gauge + Time Series) │ (Alert Table) │ ├──────────────────────────┴───────────────────────────────┤ │ Request Logs (Table) │ └──────────────────────────────────────────────────────────┘ Grafana Dashboard JSON { "dashboard": { "title": "AI Application Monitor", "tags": ["ai", "llm", "production"], "timezone": "browser", "refresh": "10s", "panels": [ { "id": 1, "title": "Requests Per Second", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 0, "y": 0}, "targets": [{ "expr": "sum(rate(ai_request_duration_seconds_count[1m]))", "legendFormat": "QPS" }], "fieldConfig": { "defaults": { "thresholds": { "steps": [ {"color": "red", "value": null}, {"color": "yellow", "value": 10}, {"color": "green", "value": 50} ] } } } }, { "id": 2, "title": "P95 Time to First Token", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 6, "y": 0}, "targets": [{ "expr": "histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le))", "legendFormat": "P95 TTFT" }], "fieldConfig": { "defaults": { "unit": "s", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 1}, {"color": "red", "value": 2} ] } } } }, { "id": 3, "title": "Request Latency Distribution", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 0, "y": 4}, "targets": [ { "expr": "histogram_quantile(0.50, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P50 {{model_name}}" }, { "expr": "histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P95 {{model_name}}" }, { "expr": "histogram_quantile(0.99, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P99 {{model_name}}" } ], "fieldConfig": { "defaults": {"unit": "s"} } }, { "id": 4, "title": "GPU Utilization & Memory", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 12, "y": 4}, "targets": [ { "expr": "DCGM_FI_DEV_GPU_UTIL", "legendFormat": "GPU {{gpu}} Util %" }, { "expr": "DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL * 100", "legendFormat": "GPU {{gpu}} Mem %" } ], "fieldConfig": { "defaults": { "min": 0, "max": 100, "custom": {"fillOpacity": 10} } } }, { "id": 5, "title": "Daily Spend by Model", "type": "bargauge", "gridPos": {"h": 6, "w": 12, "x": 0, "y": 12}, "targets": [{ "expr": "sum by (model_name) (increase(ai_daily_spend_usd_total[24h]))", "legendFormat": "{{model_name}}" }], "fieldConfig": { "defaults": { "unit": "currencyUSD", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 50}, {"color": "red", "value": 100} ] } } } }, { "id": 6, "title": "Hallucination Score Trend", "type": "timeseries", "gridPos": {"h": 6, "w": 12, "x": 12, "y": 12}, "targets": [{ "expr": "avg_over_time(ai_hallucination_score[10m])", "legendFormat": "Hallucination Score" }], "fieldConfig": { "defaults": { "min": 0, "max": 1, "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 0.1}, {"color": "red", "value": 0.2} ] } } } } ] } } 告警体系 多级告警规则 # alerting-rules.yaml groups: - name: ai_infra_alerts interval: 30s rules: # === 基础设施告警 === - alert: GPUHighTemperature expr: DCGM_FI_DEV_GPU_TEMP > 85 for: 5m labels: severity: critical team: infra annotations: summary: "GPU 温度过高: {{ $value }}°C" description: "GPU {{ $labels.gpu }} on {{ $labels.instance }} 温度超过 85°C" - alert: GPUHighMemoryUsage expr: DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL > 0.95 for: 2m labels: severity: warning team: infra annotations: summary: "GPU 显存使用率 >95%" - alert: GPULowUtilization expr: avg_over_time(DCGM_FI_DEV_GPU_UTIL[10m]) < 10 for: 10m labels: severity: info team: infra annotations: summary: "GPU 利用率过低 (<10%)，可能资源浪费" - name: ai_model_alerts interval: 15s rules: # === 模型性能告警 === - alert: HighTTFT expr: | histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le, model_name) ) > 2 for: 5m labels: severity: warning team: ai annotations: summary: "首 Token 延迟 P95 > 2s ({{ $labels.model_name }})" - alert: HighErrorRate expr: | sum(rate(ai_request_duration_seconds_count{status="error"}[5m])) by (model_name) / sum(rate(ai_request_duration_seconds_count[5m])) by (model_name) > 0.05 for: 3m labels: severity: critical team: ai annotations: summary: "错误率 >5% ({{ $labels.model_name }})" - alert: ModelQueueBacklog expr: ai_queue_depth > 20 for: 2m labels: severity: warning team: ai annotations: summary: "请求排队 >20 ({{ $labels.model_name }})" - name: ai_quality_alerts interval: 60s rules: # === 质量告警 === - alert: HighHallucinationRate expr: avg_over_time(ai_hallucination_score[15m]) > 0.15 for: 10m labels: severity: critical team: ai annotations: summary: "幻觉评分 >0.15 持续 10 分钟" - alert: SafetyViolationDetected expr: increase(ai_safety_violations_total[1h]) > 0 for: 0m labels: severity: critical team: security annotations: summary: "检测到安全违规 ({{ $labels.violation_type }})" - name: ai_business_alerts interval: 60s rules: # === 业务告警 === - alert: DailyBudgetExceeded expr: sum(increase(ai_daily_spend_usd_total[24h])) > 500 for: 1m labels: severity: critical team: business annotations: summary: "日预算超出 (${{ $value }})" - alert: CostPerRequestSpike expr: ai_request_cost_usd > 0.10 for: 5m labels: severity: warning team: business annotations: summary: "单请求成本 >$0.10 ({{ $labels.model_name }})" 告警通知渠道配置 # alertmanager-config.yaml apiVersion: v1 kind: ConfigMap metadata: name: alertmanager-config namespace: monitoring data: config.yml: | route: group_by: ["alertname", "model_name"] group_wait: 30s group_interval: 5m repeat_interval: 4h receiver: "default" routes: - matchers: ["severity=critical"] receiver: "critical" group_wait: 10s repeat_interval: 1h - matchers: ["team=security"] receiver: "security" group_wait: 0s receivers: - name: "default" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-alerts" send_resolved: true - name: "critical" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-critical" send_resolved: true pagerduty_configs: - routing_key: "..." severity: critical - name: "security" webhook_configs: - url: "https://internal.security.com/ai-alert-webhook" 告警分级与响应 级别 响应时间 通知渠道 升级策略 示例 P0 Critical 5 分钟 PagerDuty + 电话 15分钟升级到团队负责人 安全违规、服务宕机 P1 Warning 30 分钟 Slack @channel 2小时升级 延迟超标、队列积压 P2 Info 4 小时 Slack 普通消息 无升级 GPU 利用率低 P3 Notice 24 小时 邮件 无升级 日花费趋势异常 日志聚合与链路追踪 结构化日志标准 import structlog import json logger = structlog.get_logger() async def handle_chat_request(request: ChatRequest): """带全链路日志的请求处理""" request_id = generate_request_id() logger.info("chat_request_start", request_id=request_id, user_id=request.user_id, model=request.model, message_length=len(request.message), conversation_id=request.conversation_id, ) try: # 模型路由 routed_model = router.route(request.message) logger.info("model_routed", request_id=request_id, original_model=request.model, routed_model=routed_model, ) # RAG 检索 if needs_retrieval(request.message): retrieved = await retriever.search(request.message) logger.info("rag_retrieval_complete", request_id=request_id, chunks_retrieved=len(retrieved), top_score=retrieved[0]["score"] if retrieved else 0, ) # LLM 调用 response = await llm_client.chat.completions.create(...) logger.info("chat_request_complete", request_id=request_id, model=routed_model, input_tokens=response.usage.prompt_tokens, output_tokens=response.usage.completion_tokens, ttft_ms=first_token_latency, total_latency_ms=total_latency, cost_usd=cost, ) return response except Exception as e: logger.error("chat_request_failed", request_id=request_id, error_type=type(e).__name__, error_message=str(e), ) raise 成本异常检测 class CostAnomalyDetector: """基于统计方法的成本异常检测""" def __init__(self, window_size: int = 168): # 7 天小时数据 self.window_size = window_size self.cost_history: list[float] = [] def update(self, hourly_cost: float) -> dict | None: self.cost_history.append(hourly_cost) if len(self.cost_history) < self.window_size: return None # 使用滑动窗口统计 recent = self.cost_history[-self.window_size:] mean = sum(recent) / len(recent) variance = sum((x - mean) ** 2 for x in recent) / len(recent) std = variance ** 0.5 # Z-score 异常检测 z_score = abs(hourly_cost - mean) / max(std, 0.01) if z_score > 3: return { "type": "cost_spike", "current_cost": hourly_cost, "expected_mean": mean, "z_score": z_score, "deviation_pct": (hourly_cost - mean) / mean * 100, } return None 结语 AI 应用监控的核心理念是从基础设施到业务价值的全链路可观测。GPU 利用率告诉你硬件是否健康，TTFT 告诉你用户是否在等待，幻觉率告诉你 AI 是否可信，成本指标告诉你系统是否可持续。 ...

监控分层架构 AI 应用监控需要三层视角，每层关注不同问题： ┌─────────────────────────────────────────────────┐ │ Layer 3: 模型质量监控 │ │ - 回答正确率 / 幻觉率 / 毒性检测 / 偏见 │ │ - 用户满意度反馈 │ ├─────────────────────────────────────────────────┤ │ Layer 2: 应用性能监控 │ │ - 首字延迟 (TTFT) / 完整延迟 / 吞吐量 │ │ - Token 消耗 / 成本 / 缓存命中率 │ ├─────────────────────────────────────────────────┤ │ Layer 1: 基础设施监控 │ │ - GPU 利用率 / 内存 / 网络 / 磁盘 │ │ - API 错误率 / 限流次数 / 超时 │ └─────────────────────────────────────────────────┘ 指标设计 核心指标清单 层级 指标 类型 告警阈值 模型 回答质量评分 Gauge < 0.8 模型 幻觉率 Gauge > 5% 模型 用户点赞率 Gauge < 70% 应用 TTFT (首字延迟) Histogram P99 > 2s 应用 完整延迟 Histogram P99 > 15s 应用 Token 消耗 Counter 日预算 120% 应用 缓存命中率 Gauge < 15% 基础设施 GPU 利用率 Gauge > 90% 基础设施 API 错误率 Gauge > 1% 基础设施 限流次数 Counter > 100/min Prometheus 指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # === 模型质量指标 === llm_quality_score = Gauge( "llm_quality_score", "Average answer quality score (0-1)", ["model", "task_type"] ) llm_hallucination_rate = Gauge( "llm_hallucination_rate", "Hallucination rate detected", ["model"] ) llm_user_feedback = Counter( "llm_user_feedback_total", "User feedback counts", ["model", "feedback"] # feedback: positive/negative ) # === 应用性能指标 === llm_ttft = Histogram( "llm_time_to_first_token_seconds", "Time to first token in seconds", ["model"], buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10] ) llm_total_latency = Histogram( "llm_total_latency_seconds", "Total request latency in seconds", ["model"], buckets=[0.5, 1, 2, 5, 10, 20, 30, 60] ) llm_tokens_total = Counter( "llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) llm_cost_total = Counter( "llm_cost_usd_total", "Total cost in USD", ["model"] ) llm_cache_hit_rate = Gauge( "llm_cache_hit_rate", "Cache hit rate", ["cache_type"] # cache_type: exact/semantic ) # === 基础设施指标 === gpu_utilization = Gauge( "gpu_utilization_percent", "GPU utilization percentage", ["gpu_id", "node"] ) api_error_rate = Gauge( "llm_api_error_rate", "LLM API error rate", ["provider", "error_type"] ) 指标采集 中间件式采集 import time from contextlib import asynccontextmanager class LLMInstrumentation: def __init__(self, registry): self.registry = registry @asynccontextmanager async def trace_request(self, model: str, task_type: str): request_id = generate_id() start = time.monotonic() first_token_time = None async def on_first_token(): nonlocal first_token_time first_token_time = time.monotonic() try: yield {"request_id": request_id, "on_first_token": on_first_token} # 请求成功 elapsed = time.monotonic() - start llm_total_latency.labels(model=model).observe(elapsed) if first_token_time: ttft = first_token_time - start llm_ttft.labels(model=model).observe(ttft) except Exception as e: api_error_rate.labels( provider=model, error_type=type(e).__name__ ).inc() raise def record_tokens(self, model, input_tokens, output_tokens): llm_tokens_total.labels( model=model, direction="input" ).inc(input_tokens) llm_tokens_total.labels( model=model, direction="output" ).inc(output_tokens) # 成本计算 cost = self._calc_cost(model, input_tokens, output_tokens) llm_cost_total.labels(model=model).inc(cost) def _calc_cost(self, model, in_tok, out_tok): prices = { "gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006), "claude-sonnet": (0.003, 0.015), } in_price, out_price = prices.get(model, (0, 0)) return (in_tok * in_price + out_tok * out_price) / 1000 使用方式 instrument = LLMInstrumentation(registry) async def chat_completion(messages, model="gpt-4o"): async with instrument.trace_request(model, "chat") as ctx: response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, ) async for chunk in response: if chunk.choices[0].delta.content: if not ctx["first_token_called"]: await ctx["on_first_token"]() ctx["first_token_called"] = True yield chunk.choices[0].delta.content instrument.record_tokens(model, input_count, output_count) 质量监控：LLM-as-a-Judge class QualityMonitor: def __init__(self, judge_model="gpt-4o", sample_rate=0.1): self.judge_model = judge_model self.sample_rate = sample_rate async def maybe_evaluate(self, query, response, context=None): """按采样率随机评估回答质量""" import random if random.random() > self.sample_rate: return score, issues = await self._judge(query, response, context) llm_quality_score.labels( model=self.judge_model, task_type="auto" ).set(score) if "hallucination" in issues: llm_hallucination_rate.labels( model=self.judge_model ).inc() async def _judge(self, query, response, context): prompt = f"""评估以下回答的质量。 问题：{query} 回答：{response} 参考资料：{context or '无'} 请检查： 1. 事实准确性（是否有幻觉） 2. 回答完整性 3. 逻辑一致性 输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}""" result = await llm_client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) data = json.loads(result.choices[0].message.content) return data["score"], data.get("issues", []) Grafana 仪表板 关键 Panel 配置 Panel 1: 延迟分布（P50/P90/P99） ...