src/eval/error_analyzer.py - master - Watashicuvu/agentic-tools

src/eval/error_analyzer.py
184 строки7 KB
Your Name
added come roles
28 май 2026, 15:16
28 май 2026, 15:16745363b
"""
LLM-анализ ошибок для eval pipeline.

Анализирует ошибки генерации кода и диагностики,
возвращая структурированный диагноз через Pydantic модель.

Usage:
    from src.eval.error_analyzer import analyze_task_error

    analysis = await analyze_task_error(
        smart_client=client,
        model="gpt-4o-mini",
        task_description="Create a timer module",
        error="Tool call error: search.code_implementations failed",
        generated_code="",
        reference_code="-- reference lua code",
        tool_calls_log=[...],
    )
    print(analysis.root_cause)
    print(analysis.suggested_fix)
"""

import json
import logging
from typing import Optional, Any

from pydantic import BaseModel, Field, ValidationError

from src.services.async_smart_client import AsyncSmartOpenAI

logger = logging.getLogger(__name__)


class ErrorAnalysisResponse(BaseModel):
    """Структурированный LLM-анализ ошибки eval-задачи."""

    root_cause: str = Field(
        description="Краткое описание корневой причины ошибки (1-2 предложения)"
    )
    suggested_fix: str = Field(
        description="Конкретное предложение по исправлению"
    )
    confidence: float = Field(
        ge=0.0,
        le=1.0,
        description="Уверенность в диагнозе (0.0-1.0)",
    )
    error_category: str = Field(
        description="Категория ошибки: agent_error, syntax_error, runtime_error, logic_error, timeout"
    )
    recommended_actions: list[str] = Field(
        default_factory=list,
        description="Рекомендуемые действия для исправления",
    )


async def analyze_task_error(
    smart_client: AsyncSmartOpenAI,
    model: str,
    task_description: str,
    error: str,
    generated_code: str = "",
    reference_code: str = "",
    tool_calls_log: Optional[list[dict[str, Any]]] = None,
    task_role: str = "executor",
    target_language: str = "lua",
) -> Optional[ErrorAnalysisResponse]:
    """
    Анализирует ошибку eval-задачи через LLM.

    Args:
        smart_client: LLM клиент.
        model: Модель LLM.
        task_description: Описание задачи.
        error: Текст ошибки.
        generated_code: Сгенерированный код (если есть).
        reference_code: Эталонный код (если есть).
        tool_calls_log: Лог вызовов инструментов (если есть).
        task_role: Роль задачи (executor, strategist, reviewer).
        target_language: Язык генерации кода (lua, python и т.д.).

    Returns:
        ErrorAnalysisResponse или None при ошибке LLM.
    """
    # Формируем контекст
    context_parts = [
        f"Task Description:\n{task_description[:2000]}",
        f"Error:\n{error[:2000]}",
    ]

    if generated_code:
        context_parts.append(f"Generated Code:\n```\n{generated_code[:2000]}\n```")

    if reference_code:
        context_parts.append(f"Reference Code:\n```\n{reference_code[:2000]}\n```")

    if tool_calls_log:
        tool_log_lines = []
        for tc in tool_calls_log[:20]:
            tool_name = tc.get("tool_name", "unknown")
            args = str(tc.get("args", {}))[:200]
            success = tc.get("success", False)
            status = "✓" if success else "✗"
            tool_log_lines.append(f"  {status} {tool_name}({args})")
        context_parts.append(f"Tool Calls Log:\n" + "\n".join(tool_log_lines))

    context = "\n\n---\n\n".join(context_parts)

    # Адаптируем промпт под язык и роль
    role_descriptions = {
        "executor": "writing code that implements the task requirements",
        "strategist": "planning the architecture and approach for the task",
        "reviewer": "reviewing and auditing code for quality and correctness",
    }
    role_desc = role_descriptions.get(task_role, "working on the task")

    prompt = (
        f"You are an expert debugging analyst for an AI code generation eval system. "
        f"Your task is to diagnose why a code generation task failed.\n\n"
        f"The system generates {target_language} code. The agent was acting in the '{task_role}' role, "
        f"{role_desc}. Analyze the failure and provide:\n"
        f"1. root_cause: What went wrong? (1-2 sentences)\n"
        f"2. suggested_fix: How to fix it? (specific, actionable)\n"
        f"3. confidence: Your confidence level (0.0-1.0)\n"
        f"4. error_category: One of: agent_error, syntax_error, runtime_error, logic_error, timeout\n"
        f"5. recommended_actions: 2-4 specific steps to prevent this failure\n\n"
        f"Context:\n"
        f"{context}\n\n"
        f"Return ONLY valid JSON. No markdown fences.\n"
        f"Example:\n"
        "{\n"
        '  "root_cause": "The agent failed to call the required MCP tool...",\n'
        '  "suggested_fix": "Add explicit tool call instructions in the prompt...",\n'
        '  "confidence": 0.85,\n'
        '  "error_category": "agent_error",\n'
        '  "recommended_actions": ["Update system prompt to include tool usage", ...]\n'
        "}"
    )

    try:
        response = await smart_client.chat_completion(
            messages=[
                {
                    "role": "system",
                    "content": "You are a debugging expert. Return ONLY valid JSON. No markdown fences.",
                },
                {"role": "user", "content": prompt},
            ],
            model=model,
            temperature=0.2,
            max_tokens=800,
        )

        # Extract content
        content = _extract_content(response)
        if not content:
            return None

        # Parse JSON
        data = json.loads(content)
        return ErrorAnalysisResponse.model_validate(data)

    except (json.JSONDecodeError, ValidationError) as e:
        logger.warning(f"ErrorAnalyzer: JSON validation failed: {e}")
        return None
    except Exception as e:
        logger.warning(f"ErrorAnalyzer: LLM analysis failed: {e}")
        return None


def _extract_content(response: Any) -> Optional[str]:
    """Извлекает текст content из ответа LLM (разные форматы)."""
    if isinstance(response, dict):
        # Direct content
        if "content" in response:
            return response["content"]
        # OpenAI format
        choices = response.get("choices", [])
        if choices:
            return choices[0].get("message", {}).get("content", "")
    # Has .content attribute
    elif hasattr(response, "content"):
        return response.content
    return None