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

src/eval/test_user.py
418 строк13 KB
Your Name
added come roles
28 май 2026, 15:16
28 май 2026, 15:16745363b
"""Test user — LLM-based task and request generator.

Generates eval tasks from high-level topics and formulates requests
for interactive eval mode. Supports:
- Task generation from topic/domain descriptions
- Clarifying questions for ambiguous tasks
- Multi-turn interactive request refinement
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any, Optional

from src.eval.models import EvalTask
from src.services.async_smart_client import AsyncSmartOpenAI
from src.services.telemetry import telemetry, EventType


@dataclass(frozen=True)
class GeneratedTask:
    """A task generated by LLM from a topic.

    Attributes:
        id: Unique task identifier.
        description: Natural language task description.
        expected_behavior: Expected outcome description.
        reference_code: Optional reference implementation.
        suggested_lint_rules: List of linters to apply.
        difficulty: Estimated difficulty level ("easy", "medium", "hard").
    """

    id: str
    description: str
    expected_behavior: str
    reference_code: Optional[str] = None
    suggested_lint_rules: list[str] = field(default_factory=lambda: ["luacheck"])
    difficulty: str = "medium"

    def to_eval_task(self) -> EvalTask:
        """Convert to EvalTask for use in TestRunner.

        Returns:
            EvalTask instance.
        """
        return EvalTask(
            id=self.id,
            description=self.description,
            expected_behavior=self.expected_behavior,
            reference_code=self.reference_code,
            lint_rules=self.suggested_lint_rules,
            mode="batch",
        )


@dataclass(frozen=True)
class ClarifyingQuestion:
    """A clarifying question for interactive mode.

    Attributes:
        question: The question to ask the user.
        expected_answer_type: Type of expected answer ("text", "code", "yes_no").
        context: Why this question is being asked.
    """

    question: str
    expected_answer_type: str
    context: str


# System prompts
_TASK_GEN_SYSTEM_PROMPT = """\
You are an expert test designer specializing in creating programming tasks \
for code generation eval systems. Your tasks are clear, specific, and \
measurable.

## Task Generation Guidelines

1. **Clarity**: Each task should have a single, unambiguous description.
2. **Measurability**: Expected behavior should be verifiable (output, return value).
3. **Progressive difficulty**: Generate tasks from easy to hard.
4. **Realistic**: Tasks should reflect real-world programming scenarios.
5. **Language-specific**: Adapt tasks to the target language idioms.

## Output Format

Respond with valid JSON only. No markdown, no explanations.

[
    {
        "id": "<unique_id>",
        "description": "<task description>",
        "expected_behavior": "<expected outcome>",
        "reference_code": "<optional reference implementation or null>",
        "suggested_lint_rules": ["luacheck"],
        "difficulty": "easy|medium|hard"
    }
]
"""

_CLARIFYING_QUESTION_SYSTEM_PROMPT = """\
You help clarify ambiguous programming tasks by asking focused questions. \
Your questions should:
- Resolve ambiguity in task requirements
- Clarify edge cases and error handling expectations
- Determine expected input/output behavior
- Be answerable in 1-2 sentences

Respond with valid JSON only.

{
    "question": "<the clarifying question>",
    "expected_answer_type": "text|code|yes_no",
    "context": "<why this question matters>"
}
"""


class TaskGenerator:
    """TaskGenerator — LLM-based task generator and clarifier.

    Args:
        client: AsyncSmartOpenAI for LLM calls.
        model: LLM model to use (default: "gpt-4o").
        temperature: LLM temperature (default: 0.7 for creative tasks).

    Example:
        user = TaskGenerator(client)
        tasks = await user.generate_tasks("Lua file I/O", count=5)
    """

    def __init__(
        self,
        client: AsyncSmartOpenAI,
        model: str = "gpt-4o",
        temperature: float = 0.7,
    ) -> None:
        self.client = client
        self.model = model
        self.temperature = temperature

    async def generate_tasks(
        self,
        topic: str,
        count: int = 5,
        language: str = "Lua",
        difficulty_distribution: Optional[dict[str, int]] = None,
    ) -> list[GeneratedTask]:
        """Generate eval tasks from a topic.

        Args:
            topic: High-level topic (e.g. "Lua file I/O", "string parsing").
            count: Number of tasks to generate.
            language: Target programming language.
            difficulty_distribution: Optional {"easy": N, "medium": N, "hard": N}.

        Returns:
            List of GeneratedTask instances.
        """
        difficulty_info = ""
        if difficulty_distribution:
            parts = []
            for level, cnt in difficulty_distribution.items():
                parts.append(f"{cnt} {level}")
            difficulty_info = f"\nDifficulty distribution: {', '.join(parts)}"

        user_prompt = (
            f"Generate {count} programming tasks for {language} language "
            f"related to: {topic}{difficulty_info}\n\n"
            f"Ensure tasks are distinct and cover different aspects of {topic}."
        )

        messages = [
            {"role": "system", "content": _TASK_GEN_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ]

        telemetry.emit(
            EventType.LLM_REQ,
            "Task Generation Request",
            {"topic": topic, "count": count, "language": language},
        )

        try:
            response = await self.client.chat_completion(
                messages=messages,
                model=self.model,
                temperature=self.temperature,
                max_tokens=3000,
            )

            content = response.choices[0].message.content.strip()
            tasks = self._parse_tasks(content, topic)

            telemetry.emit(
                EventType.LLM_RES,
                "Task Generation Complete",
                {"tasks_generated": len(tasks)},
            )

            return tasks

        except Exception as e:
            telemetry.emit(
                EventType.ERROR,
                "Task Generation Failed",
                {"error": str(e)},
            )
            return []

    async def generate_clarifying_question(
        self,
        task_description: str,
        context: str = "",
    ) -> Optional[ClarifyingQuestion]:
        """Generate a clarifying question for an ambiguous task.

        Args:
            task_description: The task description to clarify.
            context: Additional context about the task.

        Returns:
            ClarifyingQuestion or None if no clarification needed.
        """
        user_prompt = f"Task: {task_description}"
        if context:
            user_prompt += f"\nContext: {context}"
        user_prompt += "\n\nWhat question would help clarify this task?"

        messages = [
            {"role": "system", "content": _CLARIFYING_QUESTION_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ]

        try:
            response = await self.client.chat_completion(
                messages=messages,
                model=self.model,
                temperature=0.3,
                max_tokens=1300,
            )

            content = response.choices[0].message.content.strip()
            return self._parse_clarifying_question(content)

        except Exception as e:
            telemetry.emit(
                EventType.ERROR,
                "Clarifying Question Failed",
                {"error": str(e)},
            )
            return None

    async def refine_task(
        self,
        original_description: str,
        answers: dict[str, str],
    ) -> GeneratedTask:
        """Refine a task based on clarifying Q&A.

        Args:
            original_description: Original task description.
            answers: Dict of question -> answer pairs.

        Returns:
            Refined GeneratedTask.
        """
        qa_block = "\n".join(f"Q: {q}\nA: {a}" for q, a in answers.items())

        user_prompt = (
            f"Original task: {original_description}\n\n"
            f"Clarifying Q&A:\n{qa_block}\n\n"
            f"Generate an improved task description that incorporates these answers. "
            f"Return a single task in JSON format matching this schema:\n"
            f'{{"id": "...", "description": "...", "expected_behavior": "...", '
            f'"reference_code": null, "suggested_lint_rules": ["luacheck"], '
            f'"difficulty": "easy|medium|hard"}}'
        )

        messages = [
            {"role": "system", "content": _TASK_GEN_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ]

        try:
            response = await self.client.chat_completion(
                messages=messages,
                model=self.model,
                temperature=0.3,
                max_tokens=1500,
            )

            content = response.choices[0].message.content.strip()
            tasks = self._parse_tasks(content, "refined")
            return tasks[0] if tasks else GeneratedTask(
                id="refined",
                description=original_description,
                expected_behavior="See original task",
            )

        except Exception as e:
            telemetry.emit(
                EventType.ERROR,
                "Task Refinement Failed",
                {"error": str(e)},
            )
            return GeneratedTask(
                id="refined",
                description=original_description,
                expected_behavior="See original task",
            )

    def _parse_tasks(self, content: str, topic: str) -> list[GeneratedTask]:
        """Parse LLM response into GeneratedTask list.

        Args:
            content: Raw LLM response.
            topic: Topic prefix for ID generation.

        Returns:
            List of GeneratedTask instances.

        Raises:
            ValueError: If response is not valid JSON.
        """
        cleaned = self._extract_json(content)
        try:
            data = json.loads(cleaned)
        except json.JSONDecodeError as e:
            raise ValueError(f"Invalid JSON in task generation: {e}") from e

        if not isinstance(data, list):
            data = [data]

        tasks = []
        for i, item in enumerate(data):
            task_id = item.get("id", f"{topic}_{i + 1:03d}")
            tasks.append(GeneratedTask(
                id=task_id,
                description=item.get("description", ""),
                expected_behavior=item.get("expected_behavior", ""),
                reference_code=item.get("reference_code"),
                suggested_lint_rules=item.get("suggested_lint_rules", ["luacheck"]),
                difficulty=item.get("difficulty", "medium"),
            ))

        return tasks

    def _parse_clarifying_question(self, content: str) -> Optional[ClarifyingQuestion]:
        """Parse LLM response into ClarifyingQuestion.

        Args:
            content: Raw LLM response.

        Returns:
            ClarifyingQuestion or None if parsing fails.
        """
        cleaned = self._extract_json(content)
        try:
            data = json.loads(cleaned)
        except json.JSONDecodeError:
            return None

        return ClarifyingQuestion(
            question=data.get("question", ""),
            expected_answer_type=data.get("expected_answer_type", "text"),
            context=data.get("context", ""),
        )

    @staticmethod
    def _extract_json(content: str) -> str:
        """Extract JSON from LLM response (strip markdown fences).

        Args:
            content: Raw LLM response.

        Returns:
            Clean JSON string.
        """
        content = content.strip()
        if content.startswith("```"):
            lines = content.splitlines()
            json_lines = []
            in_code_block = False
            for line in lines:
                if line.startswith("```"):
                    in_code_block = not in_code_block
                    continue
                if in_code_block:
                    json_lines.append(line)
            return "\n".join(json_lines)
        return content


# Module-level convenience function
async def generate_eval_tasks(
    topic: str,
    client: AsyncSmartOpenAI,
    count: int = 5,
    language: str = "Lua",
    model: str = "gpt-4o",
) -> list[EvalTask]:
    """Generate eval tasks and return as EvalTask list.

    Args:
        topic: High-level topic.
        client: AsyncSmartOpenAI instance.
        count: Number of tasks.
        language: Target language.
        model: LLM model.

    Returns:
        List of EvalTask instances ready for TestRunner.
    """
    user = TaskGenerator(client=client, model=model)
    generated = await user.generate_tasks(topic, count=count, language=language)
    return [t.to_eval_task() for t in generated]