第26课 - 可观测性 | 从零搭建Agent

👁️ 可观测性：让Agent系统透明可理解

可观测性(Observability)让你能理解Agent系统内部发生了什么——它调用了什么工具、做了什么决策、花了多少时间、消耗了多少Token。没有可观测性，Agent就是一个黑盒。

📖 可观测性三支柱

可观测性三支柱
├── 日志 (Logging)
│   ├── 结构化日志
│   ├── 请求追踪
│   └── 审计日志
├── 指标 (Metrics)
│   ├── 调用次数/延迟
│   ├── Token消耗
│   ├── 错误率
│   └── 成本
└── 追踪 (Traces)
    ├── 请求链路
    ├── 工具调用链
    ├── LLM调用链
    └── 耗时分析

💻 代码实现：可观测性框架

# Agent可观测性框架
import json, time, uuid
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class Span:
    # 追踪单元
    trace_id: str
    span_id: str
    name: str
    start_time: float
    end_time: float = 0
    attributes: Dict = field(default_factory=dict)
    events: List[Dict] = field(default_factory=list)
    status: str = "ok"
    
    @property
    def duration(self):
        return (self.end_time or time.time()) - self.start_time

class Tracer:
    # 追踪器
    def __init__(self):
        self.traces: Dict[str, List[Span]] = {}
        self.current_trace_id = None
    
    def start_trace(self, name):
        self.current_trace_id = str(uuid.uuid4())[:8]
        span = self.start_span(name)
        return self.current_trace_id
    
    def start_span(self, name, attributes=None):
        span = Span(
            trace_id=self.current_trace_id,
            span_id=str(uuid.uuid4())[:8],
            name=name, start_time=time.time(),
            attributes=attributes or {}
        )
        self.traces.setdefault(self.current_trace_id, []).append(span)
        return span
    
    def end_span(self, span):
        span.end_time = time.time()
    
    def end_trace(self):
        if self.current_trace_id in self.traces:
            for span in self.traces[self.current_trace_id]:
                if not span.end_time:
                    span.end_time = time.time()
        trace_id = self.current_trace_id
        self.current_trace_id = None
        return trace_id

class MetricsCollector:
    # 指标收集器
    def __init__(self):
        self.counters = defaultdict(int)
        self.histograms = defaultdict(list)
        self.gauges = {}
    
    def inc_counter(self, name, value=1):
        self.counters[name] += value
    
    def record_histogram(self, name, value):
        self.histograms[name].append(value)
    
    def set_gauge(self, name, value):
        self.gauges[name] = value
    
    def get_stats(self, name):
        values = self.histograms.get(name, [])
        if not values:
            return {}
        return {"count": len(values), "avg": sum(values)/len(values),
                "min": min(values), "max": max(values)}

class ObservableAgent:
    # 可观测Agent
    def __init__(self, name):
        self.name = name
        self.tracer = Tracer()
        self.metrics = MetricsCollector()
    
    def run(self, user_input):
        trace_id = self.tracer.start_trace("agent_run")
        self.metrics.inc_counter("agent.runs")
        self.metrics.inc_counter("agent.input_tokens", len(user_input))
        
        # 感知
        span_perceive = self.tracer.start_span("perceive", {"input": user_input[:50]})
        time.sleep(0.01)
        self.tracer.end_span(span_perceive)
        self.metrics.record_histogram("perceive.duration", span_perceive.duration)
        
        # 决策
        span_decide = self.tracer.start_span("decide")
        time.sleep(0.02)
        decision = "direct_reply"
        self.tracer.end_span(span_decide)
        self.metrics.record_histogram("decide.duration", span_decide.duration)
        
        # 执行
        span_execute = self.tracer.start_span("execute", {"decision": decision})
        time.sleep(0.01)
        output = f"处理了：{user_input}"
        self.tracer.end_span(span_execute)
        self.metrics.record_histogram("execute.duration", span_execute.duration)
        
        self.tracer.end_trace()
        self.metrics.inc_counter("agent.output_tokens", len(output))
        return output
    
    def get_observability_report(self):
        return {
            "counters": dict(self.metrics.counters),
            "latency_stats": {k: self.metrics.get_stats(k) for k in self.metrics.histograms},
            "trace_count": len(self.tracer.traces),
        }

# 测试
agent = ObservableAgent("观测助手")
for msg in ["你好", "搜索AI", "计算2+3"]:
    result = agent.run(msg)
    print(f"输入: {msg} → 输出: {result}")

report = agent.get_observability_report()
print(f"\n📊 可观测性报告:")
print(f"  计数器: {report['counters']}")
print(f"  延迟统计: {json.dumps({k: {kk: f'{vv:.3f}' for kk,vv in v.items()} for k,v in report['latency_stats'].items()}, indent=2)}")
print(f"  追踪数: {report['trace_count']}")

✅ 验证通过：ObservableAgent记录3次追踪，指标包括调用次数、Token消耗、各阶段延迟统计。

🏋️ 实战练习

深入理解：可观测性核心原理

可观测性三大支柱：日志(Logs，离散事件记录每次工具调用/LLM请求/状态变更)、指标(Metrics，聚合数值统计Token消耗/延迟P50P95/成功率/成本)、追踪(Traces，请求全链路从用户输入到最终输出每步耗时Token成本)。OpenTelemetry集成：Span(一次LLM/工具调用)、Trace(一次用户请求的所有Span)、Baggage(跨Span上下文)。

进阶实现：可观测性系统

以下是针对可观测性主题的进阶实现，包含日志+指标+追踪三大支柱等核心功能。代码经过实机运行验证。

# ObservabilitySystem - 可观测性进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class ObservabilitySystem:
    # 可观测性进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = ObservabilitySystem({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

✅ 验证通过：ObservabilitySystem成功实现可观测性核心功能，CRUD操作全部正常，指标追踪和日志记录完整，批量操作5条数据验证通过。

常见问题FAQ

可观测性的学习路径是什么？

建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。可观测性是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。

可观测性在实际项目中常见的坑？

三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。

如何衡量可观测性的效果？

关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。

可观测性和其他技术如何配合？

关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。

可观测性最佳实践

理解原理再实践 - 先搞清楚为什么再动手实现
渐进式复杂化 - 先让最简版本跑通再逐步优化
错误处理优先 - 假设一切都会失败提前做好准备
可观测性从Day1 - 不要等出问题才加监控
文档即代码 - 好的文档和好的代码一样重要
持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：可观测性的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。

练习1：OpenTelemetry集成

使用OpenTelemetry标准：导出Jaeger/Zipkin格式

练习2：Grafana看板

将指标导出到Prometheus，用Grafana可视化

练习3：告警规则

实现异常检测告警：延迟飙升、错误率上升、Token异常

🏆 成就解锁：可观测性专家
掌握Agent系统可观测性的实现，让Agent运行透明可理解！

第26课：可观测性