AI Agent 可观测性与线上评测 SOP：链路追踪、成本与告警

from dataclasses import dataclass, field

@dataclass
class AgentMetrics:
    latencies: list = field(default_factory=list)
    token_costs: list = field(default_factory=list)
    tool_success: int = 0
    tool_total: int = 0
    errors: int = 0
    total_runs: int = 0

    def record_latency(self, seconds: float) -> None:
        self.latencies.append(seconds)

    def record_cost(self, cost_usd: float) -> None:
        self.token_costs.append(cost_usd)

    def record_tool_call(self, success: bool) -> None:
        self.tool_total += 1
        if success:
            self.tool_success += 1

    def record_run(self, error: bool = False) -> None:
        self.total_runs += 1
        if error:
            self.errors += 1

    @staticmethod
    def percentile(values: list, p: float) -> float:
        if not values:
            return 0.0
        sorted_vals = sorted(values)
        idx = min(int(len(sorted_vals) * p / 100), len(sorted_vals) - 1)
        return sorted_vals[idx]

    def summary(self) -> dict:
        return {
            "latency_p50": self.percentile(self.latencies, 50),
            "latency_p95": self.percentile(self.latencies, 95),
            "latency_p99": self.percentile(self.latencies, 99),
            "cost_avg": sum(self.token_costs) / len(self.token_costs)
                        if self.token_costs else 0,
            "tool_success_rate": self.tool_success / self.tool_total
                                 if self.tool_total else 0,
            "error_rate": self.errors / self.total_runs
                          if self.total_runs else 0,
        }

from langfuse import observe

@observe(name="agent_run")
def run_agent(query: str) -> str:
    research = search_tool(query)
    answer = draft_answer(research)
    return answer

@observe(name="search_tool")
def search_tool(query: str) -> str:
    # 替换为你的搜索 API 调用
    return f"搜索结果：{query}"

@observe(name="draft_answer")
def draft_answer(context: str) -> str:
    # 替换为你的 LLM 调用
    return f"基于 {context} 的回答"

from phoenix.otel import register
from opentelemetry import trace

# 需先 pip install arize-phoenix-otel
tracer_provider = register(
    project_name="my-agent",
    endpoint="http://localhost:6006/v1/traces",
    auto_instrument=True,
)
tracer = trace.get_tracer("agent")

def run_agent_phoenix(query: str) -> str:
    with tracer.start_as_current_span("agent_run") as span:
        span.set_attribute("input.value", query)
        result = f"回答：{query}"
        span.set_attribute("output.value", result)
        return result

# 定价表（USD per 1K tokens），按你的模型更新
PRICING = {
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
}

def compute_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    rates = PRICING.get(model)
    if rates is None:
        return 0.0
    return (input_tokens * rates["input"]
            + output_tokens * rates["output"]) / 1000

from langfuse import observe

@observe(name="llm_call", as_type="generation")
def call_llm(model: str, messages: list) -> dict:
    # 替换为你的 LLM SDK 调用（如 openai.chat.completions.create）
    response = {
        "content": "模型回复",
        "usage": {"prompt_tokens": 500, "completion_tokens": 200},
    }
    usage = response["usage"]
    cost = compute_cost(model, usage["prompt_tokens"],
                        usage["completion_tokens"])
    return {
        "content": response["content"],
        "cost_usd": cost,
        "input_tokens": usage["prompt_tokens"],
        "output_tokens": usage["completion_tokens"],
    }

import json
from langfuse import observe

@observe(name="web_search_tool")
def web_search(query: str) -> dict:
    # 替换为实际搜索 API 调用
    raw_result = {
        "title": "示例结果",
        "snippet": "相关内容摘要",
        "url": "https://example.com",
    }
    return {
        "result": raw_result["snippet"],
        # artifact 存原始返回，方便排查
        "artifact": json.dumps(raw_result, ensure_ascii=False),
    }

@observe(name="code_executor_tool")
def execute_code(code: str) -> dict:
    # 替换为沙箱化代码执行
    return {
        "stdout": "执行输出",
        "exit_code": 0,
        "artifact": code,  # 存执行的代码
    }

from langfuse import observe

@observe(name="llm_judge")
def evaluate_answer(query: str, answer: str) -> float:
    judge_prompt = (
        "请给以下回答打分（1-5 分），评估准确性和有用性。\n"
        f"问题：{query}\n"
        f"回答：{answer}\n"
        "只回复一个 1-5 的数字。"
    )
    # 替换为你的 judge LLM 调用
    # 建议 judge 用比生产更强的模型，避免自我偏好偏差
    score_text = "4"  # judge_response 解析后
    return float(score_text)

from langfuse import get_client

def submit_user_feedback(trace_id: str, thumbs_up: bool) -> None:
    langfuse = get_client()
    langfuse.create_score(
        trace_id=trace_id,
        name="user_feedback",
        value=1 if thumbs_up else 0,
        data_type="NUMERIC",
        comment="用户点赞/点踩",
    )

from dataclasses import dataclass

@dataclass
class AlertRule:
    metric: str
    threshold: float
    comparator: str  # "gt"（大于触发）或 "lt"（小于触发）

    def check(self, value: float) -> bool:
        if self.comparator == "gt":
            return value > self.threshold
        return value < self.threshold

RULES = [
    AlertRule("latency_p95", 10.0, "gt"),        # p95 延迟 > 10s
    AlertRule("error_rate", 0.05, "gt"),          # 错误率 > 5%
    AlertRule("tool_success_rate", 0.90, "lt"),   # 工具成功率 < 90%
    AlertRule("cost_avg", 0.50, "gt"),            # 平均成本 > $0.50/run
]

def check_alerts(snapshot: dict) -> list:
    """检查指标快照，返回触发的告警列表。"""
    triggered = []
    for rule in RULES:
        value = snapshot.get(rule.metric, 0.0)
        if rule.check(value):
            triggered.append(
                f"ALERT: {rule.metric}={value:.4f} "
                f"({rule.comparator} {rule.threshold})"
            )
    return triggered