# -*- coding: utf-8 -*-
"""Prompt evaluation utilities: dataset generation, execution, and model-as-judge grading."""
import asyncio
import concurrent.futures
import json
import logging
from collections.abc import Callable
from statistics import mean
from textwrap import dedent
from core_mixins.logger import get_logger
from core_genai.interfaces.agent import IAgent
from core_genai.prompt.report import generate_prompt_evaluation_report
from core_genai.prompt.types import EvaluationResult
from core_genai.prompt.types import TestCase
class _PassthroughDict(dict):
"""Preserves unknown {placeholders} intact when used with str.format_map."""
def __missing__(self, key: str) -> str:
return "{" + key + "}"
[docs]
def add_user_message(messages: list, text: str) -> None:
"""Append a user turn to the messages list."""
messages.append({"role": "user", "content": text})
[docs]
def add_assistant_message(messages: list, text: str) -> None:
"""Append an assistant turn to the messages list."""
messages.append({"role": "assistant", "content": text})
[docs]
class PromptEvaluator:
"""
Generates test datasets, runs prompts, and grades
outputs with a model-as-judge.
"""
[docs]
def __init__(
self,
agent: IAgent,
model: str,
max_concurrent_tasks: int = 3,
max_tokens: int = 4000,
logger: logging.Logger | None = None,
) -> None:
if not logger:
logger = get_logger(
logger_name=__name__,
reset_handlers=True,
propagate=False,
)
self._agent = agent
self._model = model
self.max_concurrent_tasks = max_concurrent_tasks
self.max_tokens = max_tokens
self.logger = logger
def _chat(
self,
messages: list,
system: str | None = None,
temperature: float = 1.0,
stop_sequences: list[str] | None = None,
) -> str:
"""Call agent.analyze synchronously and return the first text block."""
kwargs: dict = {
"temperature": temperature,
"stop_sequences": stop_sequences or [],
"max_tokens": self.max_tokens,
}
if system is not None:
kwargs["system"] = system
# asyncio.run() creates and tears down a fresh event loop on every call.
# This is intentional: run_test_case / generate_test_case execute inside
# ThreadPoolExecutor workers that have no running loop, so each worker
# bridges to async here independently. It relies on the agent's client
# building its transport per request rather than binding to one loop.
response = asyncio.run(
self._agent.analyze(
model=self._model,
prompt=messages,
**kwargs,
)
)
return self._agent.get_text(response)
[docs]
@staticmethod
def render(template_string: str, variables: dict) -> str:
"""Replace {key} placeholders in template_string with values from variables."""
return template_string.format_map(_PassthroughDict(variables))
[docs]
def generate_unique_ideas(
self,
task_description: str,
prompt_inputs_spec: dict[str, str],
num_cases: int,
) -> list[str]:
"""Ask the model for num_cases distinct scenario ideas for the given task."""
prompt = """
Generate {num_cases} unique, diverse ideas for testing a prompt that accomplishes this task:
<task_description>
{task_description}
</task_description>
The prompt will receive the following inputs
<prompt_inputs>
{prompt_inputs}
</prompt_inputs>
Each idea should represent a distinct scenario or example that tests different aspects of the task.
Output Format:
Provide your response as a structured JSON array where each item is a brief description of the idea.
Example:
```json
[
"Testing with technical computer science terminology",
"Testing with medical research findings",
"Testing with complex mathematical concepts",
...
]
```
Ensure each idea is:
- Clearly distinct from the others
- Relevant to the task description
- Specific enough to guide generation of a full test case
- Quick to solve without requiring extensive computation or multi-step processing
- Solvable with no more than 400 tokens of output
Remember, only generate {num_cases} unique ideas
"""
system_prompt = (
"You are a test scenario designer specialized in "
"creating diverse, unique testing scenarios."
)
parts: list[str] = []
for key, value in prompt_inputs_spec.items():
val = value.replace("\n", "\\n")
parts.append(f'"{key}": str # {val},')
example_prompt_inputs = "".join(parts)
rendered_prompt = self.render(
dedent(prompt),
{
"task_description": task_description,
"num_cases": num_cases,
"prompt_inputs": example_prompt_inputs,
},
)
messages: list = []
add_user_message(messages, rendered_prompt)
add_assistant_message(messages, "```json")
text = self._chat(
messages,
stop_sequences=["```"],
system=system_prompt,
)
return json.loads(text)
[docs]
def generate_test_case( # pylint: disable=too-many-locals
self,
task_description: str,
idea: str,
prompt_inputs_spec: dict[str, str] | None = None,
) -> TestCase:
"""Generate a single structured test case from a scenario idea."""
if prompt_inputs_spec is None:
prompt_inputs_spec = {}
json_input_lines: list[str] = []
description_lines: list[str] = []
for key, value in prompt_inputs_spec.items():
description = value.replace("\n", " ")
json_input_lines.append(f'"{key}": "EXAMPLE_VALUE"')
description_lines.append(f'- "{key}": {description}\n')
example_prompt_inputs = ",\n ".join(json_input_lines)
allowed_keys = ", ".join(f'"{key}"' for key in prompt_inputs_spec)
if prompt_inputs_spec:
input_keys_section = (
"<allowed_input_keys>\n"
f"{''.join(description_lines)}"
"</allowed_input_keys>"
)
key_constraints = (
f"- You MUST ONLY use these exact input keys: {allowed_keys}\n"
" - Do NOT add any additional keys to prompt_inputs\n"
" - All keys listed in allowed_input_keys must be included in your response"
)
else:
input_keys_section = ""
key_constraints = "- Use any prompt input keys appropriate for the scenario"
prompt = """
Generate a single detailed test case for a prompt evaluation based on:
<task_description>
{task_description}
</task_description>
<specific_idea>
{idea}
</specific_idea>
{input_keys_section}
Output Format (respond with valid JSON only, no comments or trailing commas):
```json
{{
"prompt_inputs": {{
{example_prompt_inputs}
}},
"solution_criteria": ["criterion 1", "criterion 2"]
}}
```
IMPORTANT REQUIREMENTS:
{key_constraints}
- Make the test case realistic and practically useful
- Include 1 to 4 measurable, concise solution criteria
- The solution criteria should ONLY address the direct requirements of the task description and the generated prompt_inputs
- Avoid over-specifying criteria with requirements that go beyond the core task
- Keep solution criteria simple, focused, and directly tied to the fundamental task
- The test case should be tailored to the specific idea provided
- Quick to solve without requiring extensive computation or multi-step processing
- Solvable with no more than 400 tokens of output
- DO NOT include any fields beyond those specified in the output format
Here's an example with a sample input and an ideal output:
<sample_input>
<sample_task_description>
Extract topics out of a passage of text
</sample_task_description>
<sample_specific_idea>
Testing with a text that contains multiple nested topics and subtopics (e.g., a passage about renewable energy that covers solar power economics, wind turbine technology, and policy implications simultaneously)
</sample_specific_idea>
<sample_allowed_input_keys>
"content"
</sample_allowed_input_keys>
</sample_input>
<ideal_output>
```json
{{
"prompt_inputs": {{
"content": "The transition to renewable energy encompasses numerous interdependent dimensions. Solar photovoltaic technology has seen dramatic cost reductions, with panel efficiency improving 24% since 2010 while manufacturing costs declined by 89%, making it economically competitive with fossil fuels in many markets. Concurrently, wind energy has evolved through innovative turbine designs featuring carbon-fiber composite blades and advanced control systems that increase energy capture by 35% in low-wind conditions."
}},
"solution_criteria": [
"Includes all topics mentioned"
]
}}
```
</ideal_output>
This is ideal output because the solution criteria is concise and doesn't ask for anything outside of the scope of the task description.
"""
system_prompt = (
"You are a test case creator specializing in "
"designing evaluation scenarios."
)
rendered_prompt = self.render(
dedent(prompt),
{
"task_description": task_description,
"idea": idea,
"example_prompt_inputs": example_prompt_inputs,
"input_keys_section": input_keys_section,
"key_constraints": key_constraints,
},
)
messages: list = []
add_user_message(messages, rendered_prompt)
add_assistant_message(messages, "```json")
text = self._chat(
messages,
stop_sequences=["```"],
system=system_prompt,
temperature=0.7,
)
test_case: TestCase = json.loads(text)
test_case["task_description"] = task_description
test_case["scenario"] = idea
return test_case
def _log_milestone(self, completed: int, total: int, last_reported: int, message: str) -> int:
"""Log progress message at every 20% milestone; return updated last-reported value."""
current_percentage = int((completed / total) * 100)
milestone = (current_percentage // 20) * 20
if milestone > last_reported:
self.logger.info(message, completed, total)
return milestone
return last_reported
[docs]
def generate_dataset( # pylint: disable=too-many-locals
self,
task_description: str,
prompt_inputs_spec: dict[str, str] | None = None,
num_cases: int = 1,
output_file: str = "dataset.json",
) -> list[TestCase]:
"""Generate and persist a test dataset for the given task description."""
if prompt_inputs_spec is None:
prompt_inputs_spec = {}
ideas = self.generate_unique_ideas(
task_description,
prompt_inputs_spec,
num_cases,
)
dataset: list[TestCase] = []
completed = 0
total = len(ideas)
last_reported_percentage = 0
with concurrent.futures.ThreadPoolExecutor(
max_workers=self.max_concurrent_tasks,
) as executor:
future_to_idea = {
executor.submit(
self.generate_test_case,
task_description,
idea,
prompt_inputs_spec,
): idea
for idea in ideas
}
for future in concurrent.futures.as_completed(future_to_idea):
try:
result = future.result()
completed += 1
last_reported_percentage = self._log_milestone(
completed, total, last_reported_percentage,
"Generated %d/%d test cases",
)
dataset.append(result)
except Exception as e: # pylint: disable=broad-exception-caught
self.logger.error("Error generating test case: %s", e)
with open(output_file, "w", encoding="utf-8") as f:
json.dump(dataset, f, indent=2)
return dataset
[docs]
def grade_output(
self,
test_case: TestCase,
output: str,
extra_criteria: str | None,
) -> dict:
"""Score a prompt output against the test case criteria using the model."""
parts: list[str] = []
for key, value in test_case["prompt_inputs"].items():
val = value.replace("\n", "\\n")
parts.append(f'"{key}":"{val}",\n')
prompt_inputs = "".join(parts)
extra_criteria_section = ""
if extra_criteria:
extra_criteria_template = """
Mandatory Requirements - ANY VIOLATION MEANS AUTOMATIC FAILURE (score of 3 or lower):
<extra_important_criteria>
{extra_criteria}
</extra_important_criteria>
"""
extra_criteria_section = self.render(
dedent(extra_criteria_template),
{"extra_criteria": extra_criteria},
)
eval_template = """
Your task is to evaluate the following AI-generated solution with EXTREME RIGOR.
Original task description:
<task_description>
{task_description}
</task_description>
Original task inputs:
<task_inputs>
{{ {prompt_inputs} }}
</task_inputs>
Solution to Evaluate:
<solution>
{output}
</solution>
Criteria you should use to evaluate the solution:
<criteria>
{solution_criteria}
</criteria>
{extra_criteria_section}
Scoring Guidelines:
* Score 1-3: Solution fails to meet one or more MANDATORY requirements
* Score 4-6: Solution meets all mandatory requirements but has significant deficiencies in secondary criteria
* Score 7-8: Solution meets all mandatory requirements and most secondary criteria, with minor issues
* Score 9-10: Solution meets all mandatory and secondary criteria
IMPORTANT SCORING INSTRUCTIONS:
* Grade the output based ONLY on the listed criteria. Do not add your own extra requirements.
* If a solution meets all of the mandatory and secondary criteria give it a 10
* Don't complain that the solution "only" meets the mandatory and secondary criteria. Solutions shouldn't go above and beyond - they should meet the exact listed criteria.
* ANY violation of a mandatory requirement MUST result in a score of 3 or lower
* The full 1-10 scale should be utilized - don't hesitate to give low scores when warranted
Output Format
Provide your evaluation as a structured JSON object with the following fields, in this specific order:
- "strengths": An array of 1-3 key strengths
- "weaknesses": An array of 1-3 key areas for improvement
- "reasoning": A concise explanation of your overall assessment
- "score": A number between 1-10
Respond with JSON. Keep your response concise and direct.
Example response shape:
{{
"strengths": string[],
"weaknesses": string[],
"reasoning": string,
"score": number
}}
"""
eval_prompt = self.render(
dedent(eval_template),
{
"task_description": test_case["task_description"],
"prompt_inputs": prompt_inputs,
"output": output,
"solution_criteria": "\n".join(test_case["solution_criteria"]),
"extra_criteria_section": extra_criteria_section,
},
)
messages: list = []
add_user_message(messages, eval_prompt)
add_assistant_message(messages, "```json")
eval_text = self._chat(
messages,
stop_sequences=["```"],
temperature=0.0,
)
return json.loads(eval_text)
[docs]
def run_test_case(
self,
test_case: TestCase,
run_prompt_function: Callable[[dict[str, str]], str],
extra_criteria: str | None = None,
) -> EvaluationResult:
"""Run run_prompt_function on a test case and return the graded result."""
output = run_prompt_function(test_case["prompt_inputs"])
model_grade = self.grade_output(test_case, output, extra_criteria)
return {
"output": output,
"test_case": test_case,
"score": int(model_grade["score"]),
"reasoning": str(model_grade["reasoning"]),
"strengths": [str(item) for item in model_grade.get("strengths", [])],
"weaknesses": [str(item) for item in model_grade.get("weaknesses", [])],
}
[docs]
def run_evaluation( # pylint: disable=too-many-locals,too-many-arguments,too-many-positional-arguments
self,
run_prompt_function: Callable[[dict[str, str]], str],
dataset_file: str,
extra_criteria: str | None = None,
json_output_file: str = "output.json",
html_output_file: str = "output.html",
pass_threshold: int = 7,
) -> list[EvaluationResult]:
"""Run the full evaluation pipeline on every test case in dataset_file."""
with open(dataset_file, "r", encoding="utf-8") as f:
dataset: list[TestCase] = json.load(f)
results: list[EvaluationResult] = []
completed = 0
total = len(dataset)
last_reported_percentage = 0
with concurrent.futures.ThreadPoolExecutor(
max_workers=self.max_concurrent_tasks
) as executor:
future_to_test_case = {
executor.submit(
self.run_test_case,
test_case,
run_prompt_function,
extra_criteria,
): test_case
for test_case in dataset
}
for future in concurrent.futures.as_completed(future_to_test_case):
try:
result = future.result()
completed += 1
last_reported_percentage = self._log_milestone(
completed, total, last_reported_percentage,
"Graded %d/%d test cases",
)
results.append(result)
except Exception as e: # pylint: disable=broad-exception-caught
self.logger.error("Error running test case: %s", e)
if results:
average_score = mean([result["score"] for result in results])
self.logger.info("Average score: %s", average_score)
with open(json_output_file, "w", encoding="utf-8") as f:
json.dump(results, f, indent=2)
html = generate_prompt_evaluation_report(results, pass_threshold)
with open(html_output_file, "w", encoding="utf-8") as f:
f.write(html)
return results