MoFin/deploy/profile-scripts/json_failure_monitor.py

#!/usr/bin/env python3
"""json_failure_monitor.py — JSON解析失败监控 v1

监控知微pipeline中LLM输出JSON解析失败事件，记录模型名+原始输出前200字符。
配合 cron_health_monitor.py 使用。

运行方式：
  python3 json_failure_monitor.py           # 扫描agent.log最近的JSON解析失败
  python3 json_failure_monitor.py --watch    # 持续跟踪新条目

输出：JSON解析失败事件写入 ~/.hermes/profiles/position-analyst/logs/json_failures.jsonl
     每行一个JSON对象，字段：timestamp, model, provider, session_id, raw_preview(前200字)
"""

import json, os, re, sys, time
from pathlib import Path
from datetime import datetime, timezone

PROFILE_DIR = Path.home() / ".hermes" / "profiles" / "position-analyst"
LOG_DIR = PROFILE_DIR / "logs"
FAILURES_LOG = LOG_DIR / "json_failures.jsonl"

# 匹配JSON解析失败的模式（从 agent.log 中提取）
# 常见模式：json.loads/json.JSONDecodeError/SyntaxError/Invalid JSON
JSON_FAILURE_PATTERNS = [
    re.compile(r'json\.loads.*Error', re.I),
    re.compile(r'JSONDecodeError', re.I),
    re.compile(r'Expecting.*value.*char', re.I),
    re.compile(r'Invalid control character', re.I),
    re.compile(r'SyntaxError.*invalid syntax.*json', re.I),
    re.compile(r'not (valid |)JSON', re.I),
    re.compile(r'JSON schema.*mismatch', re.I),
    re.compile(r'fail.*parse.*JSON', re.I),
    re.compile(r'Failed to decode JSON', re.I),
    re.compile(r'output.*not.*valid.*json', re.I),
]

# 提取模型名
MODEL_PATTERN = re.compile(r'model=([\w.-]+)')
PROVIDER_PATTERN = re.compile(r'provider=([\w.-]+)')
SESSION_PATTERN = re.compile(r'\[([\w_-]+)\]')


def scan_log(log_path, since_pos=0):
    """扫描日志文件中的JSON解析失败事件"""
    if not log_path.exists():
        return [], since_pos
    
    events = []
    size = log_path.stat().st_size
    
    with open(log_path, 'r', errors='replace') as f:
        if since_pos > 0:
            f.seek(since_pos)
        
        for line in f:
            # 检查是否匹配JSON失败模式
            is_failure = any(p.search(line) for p in JSON_FAILURE_PATTERNS)
            if not is_failure:
                continue
            
            # 提取元数据
            model_m = MODEL_PATTERN.search(line)
            provider_m = PROVIDER_PATTERN.search(line)
            session_m = SESSION_PATTERN.search(line)
            
            event = {
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "model": model_m.group(1) if model_m else "unknown",
                "provider": provider_m.group(1) if provider_m else "unknown",
                "session_id": session_m.group(1) if session_m else "unknown",
                "raw_preview": line[:200].strip(),
            }
            events.append(event)
        
        since_pos = f.tell()
    
    return events, since_pos


def write_events(events):
    """写入JSON失败事件到jsonl文件"""
    if not events:
        return 0
    
    with open(FAILURES_LOG, 'a') as f:
        for ev in events:
            f.write(json.dumps(ev, ensure_ascii=False) + '\n')
    
    return len(events)


def summarize_recent(hours=24):
    """输出最近N小时的JSON解析失败统计"""
    if not FAILURES_LOG.exists():
        return "暂无JSON解析失败记录"
    
    cutoff = time.time() - hours * 3600
    events = []
    
    with open(FAILURES_LOG, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                ev = json.loads(line)
                events.append(ev)
            except json.JSONDecodeError:
                continue
    
    # 筛选近期
    recent = []
    for ev in events:
        try:
            t = datetime.fromisoformat(ev.get("timestamp", "")).timestamp()
            if t >= cutoff:
                recent.append(ev)
        except (ValueError, TypeError):
            continue
    
    if not recent:
        return f"最近{hours}小时内无JSON解析失败事件"
    
    # 按模型聚合
    by_model = {}
    for ev in recent:
        model = ev.get("model", "unknown")
        by_model.setdefault(model, {"count": 0, "sessions": set()})
        by_model[model]["count"] += 1
        by_model[model]["sessions"].add(ev.get("session_id", ""))
    
    lines = [f"最近{hours}h JSON解析失败统计：共{len(recent)}次"]
    for model, info in sorted(by_model.items(), key=lambda x: -x[1]["count"]):
        lines.append(f"  {model}: {info['count']}次 (涉及{len(info['sessions'])}个session)")
    
    return "\n".join(lines)


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--summarize":
        print(summarize_recent())
        sys.exit(0)
    
    # 扫描所有agent.log*
    log_files = sorted(LOG_DIR.glob("agent.log*"), reverse=True)
    total = 0
    state_file = PROFILE_DIR / ".json_failure_monitor.state"
    
    since_pos = 0
    if state_file.exists():
        try:
            since_pos = int(state_file.read_text().strip())
        except (ValueError, OSError):
            since_pos = 0
    
    for lf in log_files:
        events, since_pos = scan_log(lf, since_pos)
        n = write_events(events)
        total += n
        if n > 0:
            print(f"[{datetime.now().isoformat()}] {lf.name}: 发现{n}个JSON解析失败")
    
    # 保存扫描位置
    state_file.write_text(str(since_pos))
    
    if total == 0:
        print(f"[{datetime.now().isoformat()}] 未发现新的JSON解析失败")