lambdagent/agentexample/agentbuilder67/tools/run_stats.py

"""
run_stats.py — 运行统计工具
=============================

为 Optimizer agent 提供运行数据分析和配置更新能力。
连接 AgentPaaS API 获取 agent 运行数据。
"""
from __future__ import annotations

import json
import os
import urllib.request
import urllib.error
from pathlib import Path
from typing import Dict, List, Optional

PAAS_URL = os.environ.get("AGENTPAAS_URL", "http://127.0.0.1:8000")
API_KEY = os.environ.get("AGENTPAAS_API_KEY", "")


def _api_call(method: str, path: str, data: dict = None) -> dict:
    """调用 AgentPaaS API。"""
    url = f"{PAAS_URL}/api/v1{path}"
    body = json.dumps(data).encode("utf-8") if data else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Content-Type", "application/json")
    if API_KEY:
        req.add_header("Authorization", f"Bearer {API_KEY}")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        return {"error": f"HTTP {e.code}: {e.read().decode()[:300]}"}
    except urllib.error.URLError as e:
        return {"error": f"Connection error: {e}"}


def get_run_stats(agent_id: str, last_n: int = 50) -> Dict:
    """
    获取 agent 最近 N 次运行的统计数据。

    Args:
        agent_id: Agent ID
        last_n: 最近 N 次运行

    Returns:
        {
            "agent_id": str,
            "total_runs": int,
            "success_rate": float,
            "avg_steps": float,
            "avg_duration_ms": float,
            "avg_input_tokens": float,
            "avg_output_tokens": float,
            "tool_usage": {tool: count},
            "error_patterns": [str],
            "runs": [简要运行记录]
        }
    """
    result = _api_call("GET", f"/agents/{agent_id}/runs?limit={last_n}")
    if "error" in result:
        return result

    runs = result.get("runs", [])
    if not runs:
        return {
            "agent_id": agent_id,
            "total_runs": 0,
            "message": "No runs found",
        }

    total = len(runs)
    successful = [r for r in runs if r.get("status") == "completed"]
    failed = [r for r in runs if r.get("status") == "failed"]

    # 步数统计
    steps = [r.get("steps", 0) for r in runs if r.get("steps")]
    durations = [r.get("duration_ms", 0) for r in runs if r.get("duration_ms")]
    input_tokens = [r.get("input_tokens", 0) for r in runs if r.get("input_tokens")]
    output_tokens = [r.get("output_tokens", 0) for r in runs if r.get("output_tokens")]

    # 错误模式
    error_patterns = []
    for r in failed:
        err = r.get("error", "")
        if err and err not in error_patterns:
            error_patterns.append(err[:200])

    return {
        "agent_id": agent_id,
        "total_runs": total,
        "success_rate": len(successful) / total if total > 0 else 0,
        "failure_count": len(failed),
        "avg_steps": sum(steps) / len(steps) if steps else 0,
        "avg_duration_ms": sum(durations) / len(durations) if durations else 0,
        "avg_input_tokens": sum(input_tokens) / len(input_tokens) if input_tokens else 0,
        "avg_output_tokens": sum(output_tokens) / len(output_tokens) if output_tokens else 0,
        "error_patterns": error_patterns[:5],
        "runs_summary": [
            {
                "run_id": r.get("id"),
                "status": r.get("status"),
                "steps": r.get("steps"),
                "duration_ms": r.get("duration_ms"),
            }
            for r in runs[:10]
        ],
    }


def get_user_feedback(agent_id: str, last_n: int = 20) -> Dict:
    """
    获取 agent 的用户反馈数据。

    从 runs 的 output 和重试模式中推断用户满意度。
    （未来可接入显式评分 API）

    Returns:
        {
            "agent_id": str,
            "total_feedbacks": int,
            "avg_rating": float | null,
            "retry_rate": float,
            "common_complaints": [str]
        }
    """
    result = _api_call("GET", f"/agents/{agent_id}/runs?limit={last_n}")
    if "error" in result:
        return result

    runs = result.get("runs", [])
    if not runs:
        return {"agent_id": agent_id, "total_feedbacks": 0}

    # 推断重试：相似 input 在短时间内出现多次
    inputs_seen = {}
    retries = 0
    for r in runs:
        inp = r.get("input", "")[:100]
        if inp in inputs_seen:
            retries += 1
        inputs_seen[inp] = True

    return {
        "agent_id": agent_id,
        "total_feedbacks": len(runs),
        "avg_rating": None,  # 暂无显式评分
        "retry_rate": retries / len(runs) if runs else 0,
        "retry_count": retries,
        "common_complaints": [],  # 需要 NLP 分析 output
    }


def get_current_config(agent_id: str) -> Dict:
    """
    获取 agent 当前配置。

    Returns:
        agent 详情（含 config）
    """
    return _api_call("GET", f"/agents/{agent_id}")


def submit_update(agent_id: str, config: Dict, changelog: str) -> Dict:
    """
    提交 agent 配置更新（自动版本+1）。

    Args:
        agent_id: Agent ID
        config: 新配置
        changelog: 变更说明

    Returns:
        更新结果
    """
    return _api_call("PUT", f"/agents/{agent_id}", {
        "config": config,
        "changelog": changelog,
    })