telemetry.py

import json
import time
from pathlib import Path
from typing import List

LOG_FILE = Path("agent_metrics.jsonl")


def log_agent_turn(
    session_id: str,
    prompt: str,
    tool_calls: List[dict],
    tokens_used: int,
    ast_intercepted: bool,
    context_size_before: int,
    context_size_after: int,
):
    """
    在 Agent 的核心循环（每轮大模型返回结果后）调用此函数埋点。

    参数说明：
    - session_id: 会话唯一 id
    - prompt: 模型输入文本（或摘要）
    - tool_calls: 模型请求的工具调用列表（字典，包含 name/args）
    - tokens_used: 本轮消耗的 token 数（若可用）
    - ast_intercepted: 是否触发 AST 语法拦截
    - context_size_before/after: 压缩前后的上下文大小（字符数或 token）
    """

    try:
        ratio = (
            (context_size_before - context_size_after) / context_size_before
            if context_size_before > 0
            else 0
        )
    except Exception:
        ratio = 0

    log_entry = {
        "timestamp": time.time(),
        "session_id": session_id,
        "prompt_length": len(prompt) if prompt is not None else 0,
        "tools_called": [t.get("name") for t in tool_calls] if tool_calls else [],
        "tokens_used": tokens_used,
        "ast_intercepted": bool(ast_intercepted),
        "context_compression_ratio": ratio,
    }

    try:
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
    except Exception as e:
        # 埋点不应抛出异常影响主流程，故捕获并打印
        print(f"[telemetry] 写入失败: {e}")


def read_all_logs() -> List[dict]:
    """返回所有已记录的埋点（仅供离线分析/调试）。"""
    if not LOG_FILE.exists():
        return []
    out = []
    with open(LOG_FILE, "r", encoding="utf-8") as f:
        for line in f:
            try:
                out.append(json.loads(line))
            except Exception:
                continue
    return out