{ "cells": [ { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "# Imports\n", "import json\n", "import concurrent.futures\n", "import re\n", "from textwrap import dedent\n", "from statistics import mean\n", "from dotenv import load_dotenv\n", "from anthropic import Anthropic" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "# Client Initialization and helper functions\n", "\n", "load_dotenv()\n", "\n", "client = Anthropic()\n", "model = \"claude-3-5-haiku-latest\"\n", "\n", "\n", "def add_user_message(messages, text):\n", " user_message = {\"role\": \"user\", \"content\": text}\n", " messages.append(user_message)\n", "\n", "\n", "def add_assistant_message(messages, text):\n", " assistant_message = {\"role\": \"assistant\", \"content\": text}\n", " messages.append(assistant_message)\n", "\n", "\n", "def chat(messages, system=None, temperature=1.0, stop_sequences=[]):\n", " params = {\n", " \"model\": model,\n", " \"max_tokens\": 1000,\n", " \"messages\": messages,\n", " \"temperature\": temperature,\n", " \"stop_sequences\": stop_sequences,\n", " }\n", "\n", " if system:\n", " params[\"system\"] = system\n", "\n", " message = client.messages.create(**params)\n", " return message.content[0].text" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "# Report Builder\n", "def generate_prompt_evaluation_report(evaluation_results):\n", " total_tests = len(evaluation_results)\n", " scores = [result[\"score\"] for result in evaluation_results]\n", " avg_score = mean(scores) if scores else 0\n", " max_possible_score = 10\n", " pass_rate = (\n", " 100 * len([s for s in scores if s >= 7]) / total_tests\n", " if total_tests\n", " else 0\n", " )\n", "\n", " html = f\"\"\"\n", " \n", " \n", " \n", " \n", " \n", " Prompt Evaluation Report\n", " \n", " \n", " \n", "

\n", "

Prompt Evaluation Report

\n", "

Total Test Cases

\n", "

{total_tests}

\n", "

Average Score

\n", "

{avg_score:.1f} / {max_possible_score}

\n", "

Pass Rate (≥7)

\n", "

{pass_rate:.1f}%

\n", "

\n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \"\"\"\n", "\n", " for result in evaluation_results:\n", " prompt_inputs_html = \"
\".join(\n", " [\n", " f\"{key}: {value}\"\n", " for key, value in result[\"test_case\"][\"prompt_inputs\"].items()\n", " ]\n", " )\n", "\n", " criteria_string = \"
• \".join(\n", " result[\"test_case\"][\"solution_criteria\"]\n", " )\n", "\n", " score = result[\"score\"]\n", " if score >= 8:\n", " score_class = \"score-high\"\n", " elif score <= 5:\n", " score_class = \"score-low\"\n", " else:\n", " score_class = \"score-medium\"\n", "\n", " html += f\"\"\"\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \"\"\"\n", "\n", " html += \"\"\"\n", " \n", "

Scenario	Prompt Inputs	Solution Criteria	Output	Score	Reasoning
{result[\"test_case\"][\"scenario\"]}	{prompt_inputs_html}	• {criteria_string}	{result[\"output\"]}	{score}	{result[\"reasoning\"]}

\n", " \n", " \n", " \"\"\"\n", "\n", " return html" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "# PromptEvaluator Implementation\n", "class PromptEvaluator:\n", " def __init__(self, max_concurrent_tasks=3):\n", " self.max_concurrent_tasks = max_concurrent_tasks\n", "\n", " def render(self, template_string, variables):\n", " placeholders = re.findall(r\"{([^{}]+)}\", template_string)\n", "\n", " result = template_string\n", " for placeholder in placeholders:\n", " if placeholder in variables:\n", " result = result.replace(\n", " \"{\" + placeholder + \"}\", str(variables[placeholder])\n", " )\n", "\n", " return result.replace(\"{{\", \"{\").replace(\"}}\", \"}\")\n", "\n", " def generate_unique_ideas(\n", " self, task_description, prompt_inputs_spec, num_cases\n", " ):\n", " \"\"\"Generate a list of unique ideas for test cases based on the task description\"\"\"\n", "\n", " prompt = \"\"\"\n", " Generate {num_cases} unique, diverse ideas for testing a prompt that accomplishes this task:\n", " \n", " \n", " {task_description}\n", " \n", "\n", " The prompt will receive the following inputs\n", " \n", " {prompt_inputs_spec}\n", " \n", " \n", " Each idea should represent a distinct scenario or example that tests different aspects of the task.\n", " \n", " Output Format:\n", " Provide your response as a structured JSON array where each item is a brief description of the idea.\n", " \n", " Example:\n", " ```json\n", " [\n", " \"Testing with technical computer science terminology\",\n", " \"Testing with medical research findings\",\n", " \"Testing with complex mathematical concepts\",\n", " ...\n", " ]\n", " ```\n", " \n", " Ensure each idea is:\n", " - Clearly distinct from the others\n", " - Relevant to the task description\n", " - Specific enough to guide generation of a full test case\n", " - Quick to solve without requiring extensive computation or multi-step processing\n", " - Solvable with no more than 400 tokens of output\n", "\n", " Remember, only generate {num_cases} unique ideas\n", " \"\"\"\n", "\n", " system_prompt = \"You are a test scenario designer specialized in creating diverse, unique testing scenarios.\"\n", "\n", " example_prompt_inputs = \"\"\n", " for key, value in prompt_inputs_spec.items():\n", " val = value.replace(\"\\n\", \"\\\\n\")\n", " example_prompt_inputs += f'\"{key}\": str # {val},'\n", "\n", " rendered_prompt = self.render(\n", " dedent(prompt),\n", " {\n", " \"task_description\": task_description,\n", " \"num_cases\": num_cases,\n", " \"prompt_inputs\": example_prompt_inputs,\n", " },\n", " )\n", "\n", " messages = []\n", " add_user_message(messages, rendered_prompt)\n", " add_assistant_message(messages, \"```json\")\n", " text = chat(\n", " messages,\n", " stop_sequences=[\"```\"],\n", " system=system_prompt,\n", " temperature=1.0,\n", " )\n", "\n", " return json.loads(text)\n", "\n", " def generate_test_case(self, task_description, idea, prompt_inputs_spec={}):\n", " \"\"\"Generate a single test case based on the task description and a specific idea\"\"\"\n", "\n", " example_prompt_inputs = \"\"\n", " for key, value in prompt_inputs_spec.items():\n", " val = value.replace(\"\\n\", \"\\\\n\")\n", " example_prompt_inputs += f'\"{key}\": \"EXAMPLE_VALUE\", // {val}\\n'\n", "\n", " allowed_keys = \", \".join(\n", " [f'\"{key}\"' for key in prompt_inputs_spec.keys()]\n", " )\n", "\n", " prompt = \"\"\"\n", " Generate a single detailed test case for a prompt evaluation based on:\n", " \n", " \n", " {task_description}\n", " \n", " \n", " \n", " {idea}\n", " \n", " \n", " \n", " {allowed_keys}\n", " \n", " \n", " Output Format:\n", " ```json\n", " {{\n", " \"prompt_inputs\": {{\n", " {example_prompt_inputs}\n", " }},\n", " \"solution_criteria\": [\"criterion 1\", \"criterion 2\", ...] // Concise list of criteria for evaluating the solution, 1 to 4 items\n", " }}\n", " ```\n", " \n", " IMPORTANT REQUIREMENTS:\n", " - You MUST ONLY use these exact input keys in your prompt_inputs: {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\n", " - Make the test case realistic and practically useful\n", " - Include measurable, concise solution criteria\n", " - The solution criteria should ONLY address the direct requirements of the task description and the generated prompt_inputs\n", " - Avoid over-specifying criteria with requirements that go beyond the core task\n", " - Keep solution criteria simple, focused, and directly tied to the fundamental task\n", " - The test case should be tailored to the specific idea provided\n", " - Quick to solve without requiring extensive computation or multi-step processing\n", " - Solvable with no more than 400 tokens of output\n", " - DO NOT include any fields beyond those specified in the output format\n", "\n", " Here's an example of a sample input with an ideal output:\n", " \n", " \n", " Extract topics out of a passage of text\n", " \n", " \n", " 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)\n", " \n", "\n", " \n", " \"content\"\n", " \n", " \n", " \n", " ```json\n", " {\n", " \"prompt_inputs\": {\n", " \"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.\"\n", " },\n", " \"solution_criteria\": [\n", " \"Includes all topics mentioned\" \n", " ]\n", " }\n", " ```\n", " \n", " This is ideal output because the solution criteria is concise and doesn't ask for anything outside of the scope of the task description.\n", " \"\"\"\n", "\n", " system_prompt = \"You are a test case creator specializing in designing evaluation scenarios.\"\n", "\n", " rendered_prompt = self.render(\n", " dedent(prompt),\n", " {\n", " \"allowed_keys\": allowed_keys,\n", " \"task_description\": task_description,\n", " \"idea\": idea,\n", " \"example_prompt_inputs\": example_prompt_inputs,\n", " },\n", " )\n", "\n", " messages = []\n", " add_user_message(messages, rendered_prompt)\n", " add_assistant_message(messages, \"```json\")\n", " text = chat(\n", " messages,\n", " stop_sequences=[\"```\"],\n", " system=system_prompt,\n", " temperature=0.7,\n", " )\n", "\n", " test_case = json.loads(text)\n", " test_case[\"task_description\"] = task_description\n", " test_case[\"scenario\"] = idea\n", "\n", " return test_case\n", "\n", " def generate_dataset(\n", " self,\n", " task_description,\n", " prompt_inputs_spec={},\n", " num_cases=1,\n", " output_file=\"dataset.json\",\n", " ):\n", " \"\"\"Generate test dataset based on task description and save to file\"\"\"\n", " ideas = self.generate_unique_ideas(\n", " task_description, prompt_inputs_spec, num_cases\n", " )\n", "\n", " dataset = []\n", " completed = 0\n", " total = len(ideas)\n", " last_reported_percentage = 0\n", "\n", " with concurrent.futures.ThreadPoolExecutor(\n", " max_workers=self.max_concurrent_tasks\n", " ) as executor:\n", " future_to_idea = {\n", " executor.submit(\n", " self.generate_test_case,\n", " task_description,\n", " idea,\n", " prompt_inputs_spec,\n", " ): idea\n", " for idea in ideas\n", " }\n", "\n", " for future in concurrent.futures.as_completed(future_to_idea):\n", " try:\n", " result = future.result()\n", " completed += 1\n", " current_percentage = int((completed / total) * 100)\n", " milestone_percentage = (current_percentage // 20) * 20\n", "\n", " if milestone_percentage > last_reported_percentage:\n", " print(f\"Generated {completed}/{total} test cases\")\n", " last_reported_percentage = milestone_percentage\n", "\n", " dataset.append(result)\n", " except Exception as e:\n", " print(f\"Error generating test case: {e}\")\n", "\n", " with open(output_file, \"w\") as f:\n", " json.dump(dataset, f, indent=2)\n", "\n", " return dataset\n", "\n", " def grade_output(self, test_case, output, extra_criteria):\n", " \"\"\"Grade the output of a test case using the model\"\"\"\n", "\n", " prompt_inputs = \"\"\n", " for key, value in test_case[\"prompt_inputs\"].items():\n", " val = value.replace(\"\\n\", \"\\\\n\")\n", " prompt_inputs += f'\"{key}\":\"{val}\",\\n'\n", "\n", " extra_criteria_section = \"\"\n", " if extra_criteria:\n", " extra_criteria_template = \"\"\"\n", " Mandatory Requirements - ANY VIOLATION MEANS AUTOMATIC FAILURE (score of 3 or lower):\n", " \n", " {extra_criteria}\n", " \n", " \"\"\"\n", " extra_criteria_section = self.render(\n", " dedent(extra_criteria_template),\n", " {\"extra_criteria\": extra_criteria},\n", " )\n", "\n", " eval_template = \"\"\"\n", " Your task is to evaluate the following AI-generated solution with EXTREME RIGOR.\n", "\n", " Original task description:\n", " \n", " {task_description}\n", " \n", "\n", " Original task inputs:\n", " \n", " {{ {prompt_inputs} }}\n", " \n", "\n", " Solution to Evaluate:\n", " \n", " {output}\n", " \n", "\n", " Criteria you should use to evaluate the solution:\n", " \n", " {solution_criteria}\n", " \n", "\n", " {extra_criteria_section}\n", "\n", " Scoring Guidelines:\n", " * Score 1-3: Solution fails to meet one or more MANDATORY requirements\n", " * Score 4-6: Solution meets all mandatory requirements but has significant deficiencies in secondary criteria\n", " * Score 7-8: Solution meets all mandatory requirements and most secondary criteria, with minor issues\n", " * Score 9-10: Solution meets all mandatory and secondary criteria\n", "\n", " IMPORTANT SCORING INSTRUCTIONS:\n", " * Grade the output based ONLY on the listed criteria. Do not add your own extra requirements.\n", " * If a solution meets all of the mandatory and secondary criteria give it a 10\n", " * 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.\n", " * ANY violation of a mandatory requirement MUST result in a score of 3 or lower\n", " * The full 1-10 scale should be utilized - don't hesitate to give low scores when warranted\n", "\n", " Output Format\n", " Provide your evaluation as a structured JSON object with the following fields, in this specific order:\n", " - \"strengths\": An array of 1-3 key strengths\n", " - \"weaknesses\": An array of 1-3 key areas for improvement\n", " - \"reasoning\": A concise explanation of your overall assessment\n", " - \"score\": A number between 1-10\n", "\n", " Respond with JSON. Keep your response concise and direct.\n", " Example response shape:\n", " {{\n", " \"strengths\": string[],\n", " \"weaknesses\": string[],\n", " \"reasoning\": string,\n", " \"score\": number\n", " }}\n", " \"\"\"\n", "\n", " eval_prompt = self.render(\n", " dedent(eval_template),\n", " {\n", " \"task_description\": test_case[\"task_description\"],\n", " \"prompt_inputs\": prompt_inputs,\n", " \"output\": output,\n", " \"solution_criteria\": \"\\n\".join(test_case[\"solution_criteria\"]),\n", " \"extra_criteria_section\": extra_criteria_section,\n", " },\n", " )\n", "\n", " messages = []\n", " add_user_message(messages, eval_prompt)\n", " add_assistant_message(messages, \"```json\")\n", " eval_text = chat(\n", " messages,\n", " stop_sequences=[\"```\"],\n", " temperature=0.0,\n", " )\n", " return json.loads(eval_text)\n", "\n", " def run_test_case(\n", " self, test_case, run_prompt_function, extra_criteria=None\n", " ):\n", " \"\"\"Run a test case and grade the result\"\"\"\n", " output = run_prompt_function(test_case[\"prompt_inputs\"])\n", "\n", " model_grade = self.grade_output(test_case, output, extra_criteria)\n", " model_score = model_grade[\"score\"]\n", " reasoning = model_grade[\"reasoning\"]\n", "\n", " return {\n", " \"output\": output,\n", " \"test_case\": test_case,\n", " \"score\": model_score,\n", " \"reasoning\": reasoning,\n", " }\n", "\n", " def run_evaluation(\n", " self,\n", " run_prompt_function,\n", " dataset_file,\n", " extra_criteria=None,\n", " json_output_file=\"output.json\",\n", " html_output_file=\"output.html\",\n", " ):\n", " \"\"\"Run evaluation on all test cases in the dataset\"\"\"\n", " with open(dataset_file, \"r\") as f:\n", " dataset = json.load(f)\n", "\n", " results = []\n", " completed = 0\n", " total = len(dataset)\n", " last_reported_percentage = 0\n", "\n", " with concurrent.futures.ThreadPoolExecutor(\n", " max_workers=self.max_concurrent_tasks\n", " ) as executor:\n", " future_to_test_case = {\n", " executor.submit(\n", " self.run_test_case,\n", " test_case,\n", " run_prompt_function,\n", " extra_criteria,\n", " ): test_case\n", " for test_case in dataset\n", " }\n", "\n", " for future in concurrent.futures.as_completed(future_to_test_case):\n", " result = future.result()\n", " completed += 1\n", " current_percentage = int((completed / total) * 100)\n", " milestone_percentage = (current_percentage // 20) * 20\n", "\n", " if milestone_percentage > last_reported_percentage:\n", " print(f\"Graded {completed}/{total} test cases\")\n", " last_reported_percentage = milestone_percentage\n", " results.append(result)\n", "\n", " average_score = mean([result[\"score\"] for result in results])\n", " print(f\"Average score: {average_score}\")\n", "\n", " with open(json_output_file, \"w\") as f:\n", " json.dump(results, f, indent=2)\n", "\n", " html = generate_prompt_evaluation_report(results)\n", " with open(html_output_file, \"w\", encoding=\"utf-8\") as f:\n", " f.write(html)\n", "\n", " return results" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Create an instance of PromptEvaluator\n", "# Increase `max_concurrent_tasks` for greater concurrency, but beware of rate limit errors!\n", "evaluator = PromptEvaluator(max_concurrent_tasks=1)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "dataset = evaluator.generate_dataset(\n", " # Describe the purpose or goal of the prompt you're trying to test\n", " task_description=\"\",\n", " # Describe the different inputs that your prompt requires\n", " prompt_inputs_spec={},\n", " # Where to write the generated dataset\n", " output_file=\"dataset.json\",\n", " # Number of test cases to generate (recommend keeping this low if you're getting rate limit errors)\n", " num_cases=3,\n", ")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Define and run the prompt you want to evaluate, returning the raw model output\n", "# This function is executed once for each test case\n", "def run_prompt(prompt_inputs):\n", " prompt = f\"\"\"\n", " \n", " \"\"\"\n", "\n", " messages = []\n", " add_user_message(messages, prompt)\n", " return chat(messages)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "results = evaluator.run_evaluation(\n", " run_prompt_function=run_prompt, dataset_file=\"dataset.json\"\n", ")" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.6" } }, "nbformat": 4, "nbformat_minor": 2 }