19 | 可观测性与数据飞轮——从'能跑'到'越跑越好'

import json
import time
from datetime import datetime
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Any

@dataclass
class LogEntry:
    """一条结构化日志"""
    timestamp: str
    session_id: str
    step_type: str          # "llm_call" | "tool_call" | "user_input" | "final_output"
    content: dict[str, Any]
    duration_ms: float = 0
    token_usage: dict = field(default_factory=dict)

class AgentLogger:
    """Agent 结构化日志器"""

    def __init__(self, log_dir: str = "./agent_logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)
        self.entries: list[LogEntry] = []

    def log(self, session_id: str, step_type: str,
            content: dict, duration_ms: float = 0,
            token_usage: dict | None = None):
        entry = LogEntry(
            timestamp=datetime.utcnow().isoformat() + "Z",
            session_id=session_id,
            step_type=step_type,
            content=content,
            duration_ms=duration_ms,
            token_usage=token_usage or {},
        )
        self.entries.append(entry)

        # 按天写入文件，每行一条 JSON（方便后续用 grep/jq 搜索）
        log_file = self.log_dir / f"{datetime.utcnow():%Y-%m-%d}.jsonl"
        with open(log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(entry), ensure_ascii=False) + "\n")

    def query(self, session_id: str | None = None,
              step_type: str | None = None) -> list[LogEntry]:
        """按 session_id 或 step_type 过滤日志"""
        results = self.entries
        if session_id:
            results = [e for e in results if e.session_id == session_id]
        if step_type:
            results = [e for e in results if e.step_type == step_type]
        return results


# ── 使用示例 ──────────────────────────────────────
logger = AgentLogger()

# 记录用户输入
logger.log("sess-001", "user_input", {"query": "帮我查一下北京的天气"})

# 记录 LLM 调用
t0 = time.time()
# ... 实际调用 LLM ...
elapsed = (time.time() - t0) * 1000
logger.log("sess-001", "llm_call", {
    "model": "gpt-4o",
    "prompt_summary": "天气查询决策",
    "response": "需要调用 weather_api 工具",
}, duration_ms=elapsed, token_usage={"prompt": 180, "completion": 45})

# 记录工具调用
logger.log("sess-001", "tool_call", {
    "tool": "weather_api",
    "params": {"city": "Beijing"},
    "result": {"temp": "28C", "condition": "晴"},
})

import json
from collections import defaultdict
from pathlib import Path

def compute_daily_metrics(log_dir: str, date: str) -> dict:
    """从 JSONL 日志中计算当日核心指标"""
    log_file = Path(log_dir) / f"{date}.jsonl"
    entries = []
    with open(log_file, encoding="utf-8") as f:
        for line in f:
            entries.append(json.loads(line))

    # 按 session 分组
    sessions = defaultdict(list)
    for e in entries:
        sessions[e["session_id"]].append(e)

    total_requests = len(sessions)
    total_cost = 0.0
    durations = []
    success_count = 0

    for sid, events in sessions.items():
        # 计算单次请求的延迟（所有步骤耗时之和）
        session_duration = sum(e.get("duration_ms", 0) for e in events)
        durations.append(session_duration)

        # 计算 token 成本（以 GPT-4o 为例：input $2.5/1M, output $10/1M）
        for e in events:
            if e["step_type"] == "llm_call" and e.get("token_usage"):
                prompt_tokens = e["token_usage"].get("prompt", 0)
                completion_tokens = e["token_usage"].get("completion", 0)
                cost = (prompt_tokens * 2.5 + completion_tokens * 10) / 1_000_000
                total_cost += cost

        # 判断是否成功（有 final_output 且有明确的成功标记）
        final_events = [e for e in events if e["step_type"] == "final_output"]
        if final_events and final_events[0]["content"].get("success", False):
            success_count += 1

    durations.sort()
    p50 = durations[len(durations) // 2] if durations else 0
    p95 = durations[int(len(durations) * 0.95)] if durations else 0

    return {
        "date": date,
        "total_requests": total_requests,
        "avg_cost_per_request": round(total_cost / max(total_requests, 1), 4),
        "total_cost_usd": round(total_cost, 2),
        "p50_latency_ms": round(p50, 1),
        "p95_latency_ms": round(p95, 1),
        "success_rate": round(success_count / max(total_requests, 1) * 100, 1),
    }


# ── 使用示例 ──────────────────────────────────────
metrics = compute_daily_metrics("./agent_logs", "2026-07-09")
print(json.dumps(metrics, indent=2, ensure_ascii=False))
# 输出示例:
# {
#   "date": "2026-07-09",
#   "total_requests": 1024,
#   "avg_cost_per_request": 0.0385,
#   "total_cost_usd": 39.42,
#   "p50_latency_ms": 2100.0,
#   "p95_latency_ms": 4200.0,
#   "success_rate": 91.2
# }

[Root Span: 用户请求]  (总耗时 3200ms)
  ├── [Span: LLM 推理 #1]  (1800ms, 决定调用搜索工具)
  │     └── prompt_tokens: 320, completion_tokens: 85
  ├── [Span: 工具调用 search_api]  (600ms)
  │     └── query: "2026年AI发展趋势", results: 5条
  ├── [Span: LLM 推理 #2]  (700ms, 基于搜索结果生成回答)
  │     └── prompt_tokens: 580, completion_tokens: 210
  └── [Span: 最终输出]  (100ms)

import time
import uuid
from dataclasses import dataclass, field
from typing import Any
from contextlib import contextmanager

@dataclass
class Span:
    """一个操作跨度"""
    name: str
    span_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    parent_id: str | None = None
    start_time: float = 0
    end_time: float = 0
    attributes: dict[str, Any] = field(default_factory=dict)
    children: list["Span"] = field(default_factory=list)

    @property
    def duration_ms(self) -> float:
        return (self.end_time - self.start_time) * 1000

    def to_dict(self) -> dict:
        """序列化为字典，方便存储和传输"""
        d = {
            "span_id": self.span_id,
            "name": self.name,
            "parent_id": self.parent_id,
            "duration_ms": round(self.duration_ms, 1),
            "attributes": self.attributes,
        }
        if self.children:
            d["children"] = [c.to_dict() for c in self.children]
        return d


class Tracer:
    """Agent 追踪器"""

    def __init__(self):
        self.traces: list[Span] = []

    @contextmanager
    def trace(self, name: str, **attributes):
        """创建一个 Root Span 上下文"""
        root = Span(name=name, attributes=attributes)
        root.start_time = time.time()
        self.traces.append(root)
        self._current = root
        try:
            yield root
        finally:
            root.end_time = time.time()

    @contextmanager
    def span(self, name: str, **attributes):
        """在 Root Span 下创建子 Span"""
        child = Span(
            name=name,
            parent_id=self._current.span_id,
            attributes=attributes,
        )
        child.start_time = time.time()
        self._current.children.append(child)

        parent = self._current
        self._current = child
        try:
            yield child
        finally:
            child.end_time = time.time()
            self._current = parent

    def export_traces(self) -> list[dict]:
        """导出所有 Trace"""
        return [t.to_dict() for t in self.traces]


# ── 使用示例 ──────────────────────────────────────
tracer = Tracer()

with tracer.trace("user_request", query="帮我总结今天的新闻") as root:

    with tracer.span("llm_call", model="gpt-4o", step="planning") as s:
        # 模拟 LLM 调用
        time.sleep(0.8)
        s.attributes["response"] = "需要搜索新闻并总结"
        s.attributes["tokens"] = {"prompt": 150, "completion": 40}

    with tracer.span("tool_call", tool="news_search") as s:
        time.sleep(0.3)
        s.attributes["result_count"] = 5

    with tracer.span("llm_call", model="gpt-4o", step="generation") as s:
        time.sleep(0.6)
        s.attributes["tokens"] = {"prompt": 480, "completion": 200}

# 打印追踪结果
import json
print(json.dumps(tracer.export_traces(), indent=2, ensure_ascii=False))

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from langchain.tools import tool
import os

# ── 设置 LangFuse 环境变量 ──────────────────────────
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxxxxxxxxxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxxxxxxxxxxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

# ── 导入 LangFuse 的 Callback Handler ──────────────
from langfuse.callback import CallbackHandler

langfuse_handler = CallbackHandler(
    session_id="demo-session",
    user_id="user-001",
    tags=["production", "v2-agent"],
)

# ── 定义工具 ──────────────────────────────────────
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果: 关于'{query}'的最新信息..."

# ── 创建 Agent 并接入 LangFuse ────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")
tools = [search_web]

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行时传入 callbacks 参数即可
result = agent_executor.invoke(
    {"input": "2026年大语言模型的最新趋势是什么？"},
    config={"callbacks": [langfuse_handler]},
)

print(result["output"])

from langfuse.decorators import observe, langfuse_context

@observe()
def my_agent(query: str) -> str:
    """被 @observe 装饰的函数会自动创建 Trace"""
    plan = plan_step(query)       # 自动记录为子 Span
    result = execute_step(plan)   # 自动记录为子 Span
    return format_output(result)

@observe(as_type="generation")
def plan_step(query: str) -> str:
    """标记为 generation 类型，会自动记录 LLM 相关信息"""
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"为以下问题制定计划: {query}"}],
    )
    # 手动更新 Span 的元数据
    langfuse_context.update_current_observation(
        usage={
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
        },
        model="gpt-4o",
    )
    return response.choices[0].message.content

Agent 运行 → 产生 Trace 日志 → 筛选高质量轨迹 → 构建 SFT 训练数据
    → 微调专用模型 → 替换通用 LLM → Agent 表现更好
    → 产生更多日志 → ...（循环加速）

import json
from pathlib import Path

def trace_to_sft_example(trace: dict) -> dict:
    """将一条 Trace 转换成 SFT 训练格式"""
    # 从 Trace 中提取用户输入
    user_input = trace["attributes"].get("query", "")

    # 从 Trace 的子 Span 中构建完整的推理过程
    reasoning_steps = []
    for child in trace.get("children", []):
        if child["name"] == "llm_call":
            reasoning_steps.append(
                f"[思考] {child['attributes'].get('response', '')}"
            )
        elif child["name"] == "tool_call":
            tool_name = child["attributes"].get("tool", "unknown")
            tool_result = child["attributes"].get("result", "")
            reasoning_steps.append(
                f"[调用工具 {tool_name}] 结果: {tool_result}"
            )

    # 最终输出
    final_output = ""
    for child in trace.get("children", []):
        if child["name"] == "final_output":
            final_output = child["attributes"].get("output", "")

    # 组装成完整的训练样本
    full_response = "\n".join(reasoning_steps)
    if final_output:
        full_response += f"\n[最终回答] {final_output}"

    return {
        "instruction": user_input,
        "output": full_response,
    }


def build_sft_dataset(traces: list[dict], output_path: str):
    """批量转换并保存训练数据集"""
    examples = []
    for trace in traces:
        try:
            example = trace_to_sft_example(trace)
            if example["instruction"] and example["output"]:
                examples.append(example)
        except Exception as e:
            print(f"跳过转换失败的 Trace: {e}")

    # 保存为 JSONL（Hugging Face datasets 可直接加载）
    with open(output_path, "w", encoding="utf-8") as f:
        for ex in examples:
            f.write(json.dumps(ex, ensure_ascii=False) + "\n")

    print(f"已生成 {len(examples)} 条训练数据 -> {output_path}")
    return examples


# ── 使用示例 ──────────────────────────────────────
# 假设 filtered_traces 是从环节2筛选出的高质量 Trace 列表
# build_sft_dataset(filtered_traces, "./sft_dataset.jsonl")

{"instruction": "帮我查一下北京明天的天气", "output": "[思考] 用户需要查询天气，我需要调用天气API工具\n[调用工具 weather_api] 结果: 北京明天晴，最高温度32度\n[最终回答] 北京明天是晴天，最高温度32度，比较热，建议注意防晒。"}