DevMastersZA/Marco_Professional_Profile
0
1{2 "cells": [3 {4 "cell_type": "markdown",5 "metadata": {},6 "source": [7 "# Lab 2: Parallelization and Evaluator-Optimizer Pattern\n",8 "\n",9 "This notebook implements the **Evaluator-Optimizer Pattern** with **Parallelization**:\n",10 "\n",11 "1. **Evaluator**: Gathers API keys and prepares model configurations\n",12 "2. **Parallel Execution**: All models run simultaneously using async/await\n",13 "3. **Aggregator**: Collects and formats all outputs for evaluation\n",14 "4. **Final Evaluator**: Judge model ranks all responses from best to worst\n",15 "\n",16 "## Pattern Flow\n",17 "\n",18 "```\n",19 "Evaluator (API Keys & Configs) \n",20 " ↓\n",21 "Parallel API Calls (All models run simultaneously)\n",22 " ↓\n",23 "Aggregator (Collect & Format outputs)\n",24 " ↓\n",25 "Final Evaluator (Judge ranks all outputs)\n",26 "```"27 ]28 },29 {30 "cell_type": "code",31 "execution_count": null,32 "metadata": {},33 "outputs": [],34 "source": [35 "# Start with imports - ask ChatGPT to explain any package that you don't know\n",36 "\n",37 "import os\n",38 "import json\n",39 "import asyncio\n",40 "import random\n",41 "from datetime import datetime, timedelta\n",42 "from typing import Dict, List, Tuple, Any\n",43 "from dotenv import load_dotenv\n",44 "from openai import OpenAI\n",45 "from anthropic import Anthropic\n",46 "from IPython.display import Markdown, display"47 ]48 },49 {50 "cell_type": "code",51 "execution_count": null,52 "metadata": {},53 "outputs": [],54 "source": [55 "# Always remember to do this!\n",56 "load_dotenv(override=True)"57 ]58 },59 {60 "cell_type": "code",61 "execution_count": null,62 "metadata": {},63 "outputs": [],64 "source": [65 "# Print the key prefixes to help with any debugging\n",66 "\n",67 "openai_api_key = os.getenv('OPENAI_API_KEY')\n",68 "anthropic_api_key = os.getenv('ANTHROPIC_API_KEY')\n",69 "google_api_key = os.getenv('GOOGLE_API_KEY')\n",70 "deepseek_api_key = os.getenv('DEEPSEEK_API_KEY')\n",71 "groq_api_key = os.getenv('GROQ_API_KEY')\n",72 "\n",73 "if openai_api_key:\n",74 " print(f\"OpenAI API Key exists and begins {openai_api_key[:8]}\")\n",75 "else:\n",76 " print(\"OpenAI API Key not set\")\n",77 " \n",78 "if anthropic_api_key:\n",79 " print(f\"Anthropic API Key exists and begins {anthropic_api_key[:7]}\")\n",80 "else:\n",81 " print(\"Anthropic API Key not set (and this is optional)\")\n",82 "\n",83 "if google_api_key:\n",84 " print(f\"Google API Key exists and begins {google_api_key[:2]}\")\n",85 "else:\n",86 " print(\"Google API Key not set (and this is optional)\")\n",87 "\n",88 "if deepseek_api_key:\n",89 " print(f\"DeepSeek API Key exists and begins {deepseek_api_key[:3]}\")\n",90 "else:\n",91 " print(\"DeepSeek API Key not set (and this is optional)\")\n",92 "\n",93 "if groq_api_key:\n",94 " print(f\"Groq API Key exists and begins {groq_api_key[:4]}\")\n",95 "else:\n",96 " print(\"Groq API Key not set (and this is optional)\")"97 ]98 },99 {100 "cell_type": "markdown",101 "metadata": {},102 "source": [103 "## Step 1: Generate Question\n",104 "\n",105 "First, we'll generate a challenging question to ask all the models."106 ]107 },108 {109 "cell_type": "code",110 "execution_count": null,111 "metadata": {},112 "outputs": [],113 "source": [114 "request = \"Please come up with a challenging, nuanced question that I can ask a number of LLMs to evaluate their intelligence. \"\n",115 "request += \"Answer only with the question, no explanation.\"\n",116 "messages = [{\"role\": \"user\", \"content\": request}]"117 ]118 },119 {120 "cell_type": "code",121 "execution_count": null,122 "metadata": {},123 "outputs": [],124 "source": [125 "openai = OpenAI()\n",126 "response = openai.chat.completions.create(\n",127 " model=\"gpt-5-mini\",\n",128 " messages=messages,\n",129 ")\n",130 "question = response.choices[0].message.content\n",131 "print(question)"132 ]133 },134 {135 "cell_type": "code",136 "execution_count": null,137 "metadata": {},138 "outputs": [],139 "source": [140 "# Prepare the question for all models\n",141 "competitors = []\n",142 "answers = []\n",143 "messages = [{\"role\": \"user\", \"content\": question}]"144 ]145 },146 {147 "cell_type": "markdown",148 "metadata": {},149 "source": [150 "## Step 2: Evaluator - Prepare API Keys and Configurations\n",151 "\n",152 "The **Evaluator** gathers API keys and prepares configurations for all models."153 ]154 },155 {156 "cell_type": "code",157 "execution_count": null,158 "metadata": {},159 "outputs": [],160 "source": [161 "# ==========================================\n",162 "# Evaluator: Gathers API keys and prepares model configurations\n",163 "# ==========================================\n",164 "# This function prepares all model configurations ready for parallel execution\n",165 "\n",166 "def evaluator_prepare_configs():\n",167 " \"\"\"\n",168 " Evaluator: Gathers API keys and prepares configurations for all models.\n",169 " Returns a list of model configurations ready for parallel execution.\n",170 " \"\"\"\n",171 " configs = []\n",172 " \n",173 " # Model 1: OpenAI\n",174 " configs.append({\n",175 " \"model_name\": \"gpt-5-nano\",\n",176 " \"provider\": \"openai\",\n",177 " \"client\": OpenAI(),\n",178 " \"call_type\": \"chat.completions\",\n",179 " \"extra_params\": {}\n",180 " })\n",181 " \n",182 " # Model 2: Anthropic\n",183 " configs.append({\n",184 " \"model_name\": \"claude-sonnet-4-5\",\n",185 " \"provider\": \"anthropic\",\n",186 " \"client\": Anthropic(),\n",187 " \"call_type\": \"messages.create\",\n",188 " \"extra_params\": {\"max_tokens\": 1000}\n",189 " })\n",190 " \n",191 " # Model 3: Gemini\n",192 " configs.append({\n",193 " \"model_name\": \"gemini-2.5-flash\",\n",194 " \"provider\": \"gemini\",\n",195 " \"client\": OpenAI(api_key=google_api_key, base_url=\"https://generativelanguage.googleapis.com/v1beta/openai/\"),\n",196 " \"call_type\": \"chat.completions\",\n",197 " \"extra_params\": {}\n",198 " })\n",199 " \n",200 " # Model 4: DeepSeek\n",201 " configs.append({\n",202 " \"model_name\": \"deepseek/deepseek-r1-0528:free\",\n",203 " \"provider\": \"deepseek\",\n",204 " \"client\": OpenAI(api_key=deepseek_api_key, base_url=\"https://openrouter.ai/api/v1\"),\n",205 " \"call_type\": \"chat.completions\",\n",206 " \"extra_params\": {}\n",207 " })\n",208 " \n",209 " # Model 5: Groq\n",210 " configs.append({\n",211 " \"model_name\": \"openai/gpt-oss-120b\",\n",212 " \"provider\": \"groq\",\n",213 " \"client\": OpenAI(api_key=groq_api_key, base_url=\"https://api.groq.com/openai/v1\"),\n",214 " \"call_type\": \"chat.completions\",\n",215 " \"extra_params\": {}\n",216 " })\n",217 " \n",218 " # Model 6: Ollama (if available)\n",219 " configs.append({\n",220 " \"model_name\": \"llama3.2\",\n",221 " \"provider\": \"ollama\",\n",222 " \"client\": OpenAI(base_url='http://localhost:11434/v1', api_key='ollama'),\n",223 " \"call_type\": \"chat.completions\",\n",224 " \"extra_params\": {}\n",225 " })\n",226 " \n",227 " print(f\"✅ Evaluator prepared {len(configs)} model configurations\")\n",228 " return configs\n",229 "\n",230 "# Prepare all configurations\n",231 "model_configs = evaluator_prepare_configs()"232 ]233 },234 {235 "cell_type": "markdown",236 "metadata": {},237 "source": [238 "## Step 3: Parallel Execution - Call All Models Simultaneously\n",239 "\n",240 "All models are called **in parallel** using async/await, making the process much faster than sequential calls."241 ]242 },243 {244 "cell_type": "code",245 "execution_count": null,246 "metadata": {},247 "outputs": [],248 "source": [249 "# ==========================================\n",250 "# Async function to call a single model\n",251 "# ==========================================\n",252 "\n",253 "async def call_model_async(config: Dict[str, Any], messages: List[Dict]) -> Tuple[str, str]:\n",254 " \"\"\"\n",255 " Call a single model asynchronously. Returns (model_name, answer) or (model_name, error_message).\n",256 " \"\"\"\n",257 " model_name = config[\"model_name\"]\n",258 " provider = config[\"provider\"]\n",259 " client = config[\"client\"]\n",260 " call_type = config[\"call_type\"]\n",261 " extra_params = config[\"extra_params\"]\n",262 " \n",263 " try:\n",264 " if provider == \"anthropic\":\n",265 " # Anthropic uses a different API structure\n",266 " response = await asyncio.to_thread(\n",267 " client.messages.create,\n",268 " model=model_name,\n",269 " messages=messages,\n",270 " **extra_params\n",271 " )\n",272 " answer = response.content[0].text\n",273 " else:\n",274 " # OpenAI-compatible APIs\n",275 " response = await asyncio.to_thread(\n",276 " client.chat.completions.create,\n",277 " model=model_name,\n",278 " messages=messages,\n",279 " **extra_params\n",280 " )\n",281 " answer = response.choices[0].message.content\n",282 " \n",283 " print(f\"✅ {model_name} completed\")\n",284 " return model_name, answer\n",285 " \n",286 " except Exception as e:\n",287 " error_msg = f\"Error calling {model_name}: {str(e)}\"\n",288 " print(f\"❌ {error_msg}\")\n",289 " return model_name, error_msg\n",290 "\n",291 "# ==========================================\n",292 "# Parallel execution function\n",293 "# ==========================================\n",294 "\n",295 "def format_bytes(size: int) -> str:\n",296 " \"\"\"Format bytes into a human-readable string (B, KB, MB).\"\"\"\n",297 " for unit in ['B', 'KB', 'MB']:\n",298 " if size < 1024.0:\n",299 " return f\"{size:.2f} {unit}\"\n",300 " size /= 1024.0\n",301 " return f\"{size:.2f} GB\"\n",302 "\n",303 "async def execute_models_in_parallel(configs: List[Dict[str, Any]], messages: List[Dict]) -> Tuple[List[str], List[str]]:\n",304 " # Overall execution start\n",305 " print(f\"\\n🚀 Starting parallel execution of {len(configs)} models...\\n\")\n",306 " \n",307 " # Track data for the table\n",308 " table_rows = []\n",309 " competitors = []\n",310 " answers = []\n",311 " \n",312 " async def call_with_metrics(config):\n",313 " model_name = config.get(\"model_name\", \"Unknown\")\n",314 " start_time = datetime.now()\n",315 " \n",316 " try:\n",317 " # Assumes call_model_async returns (model_name, answer)\n",318 " _, answer = await call_model_async(config, messages)\n",319 " end_time = datetime.now()\n",320 " \n",321 " # Check for error strings inside the success path\n",322 " if isinstance(answer, str) and answer.startswith(\"Error\"):\n",323 " status = \"❌ Error\"\n",324 " out_size = 0\n",325 " else:\n",326 " status = \"✅ Success\"\n",327 " out_size = len(str(answer).encode('utf-8'))\n",328 " \n",329 " except Exception as e:\n",330 " end_time = datetime.now()\n",331 " status = \"❌ Error\"\n",332 " answer = str(e)\n",333 " out_size = 0\n",334 "\n",335 " # Calculate duration\n",336 " duration = end_time - start_time\n",337 " total_seconds = int(duration.total_seconds())\n",338 " mm, ss = divmod(total_seconds, 60)\n",339 " hh, mm = divmod(mm, 60)\n",340 " dur_str = f\"{hh:02d}:{mm:02d}:{ss:02d}\" if hh > 0 else f\"{mm:02d}:{ss:02d}\"\n",341 "\n",342 " # Store metrics for table\n",343 " table_rows.append({\n",344 " \"model\": model_name,\n",345 " \"status\": status,\n",346 " \"start\": start_time.strftime(\"%H:%M:%S\"),\n",347 " \"end\": end_time.strftime(\"%H:%M:%S\"),\n",348 " \"duration\": dur_str,\n",349 " \"size\": format_bytes(out_size)\n",350 " })\n",351 " \n",352 " return model_name, answer, status\n",353 "\n",354 " # Run tasks in parallel\n",355 " tasks = [call_with_metrics(config) for config in configs]\n",356 " results = await asyncio.gather(*tasks)\n",357 "\n",358 " # Process final lists\n",359 " for model_name, answer, status in results:\n",360 " if status == \"✅ Success\":\n",361 " competitors.append(model_name)\n",362 " answers.append(answer)\n",363 "\n",364 " # Print Tabular Output\n",365 " header = f\"{'Model':<20} {'Status':<10} {'Start':<10} {'End':<10} {'Duration':<10} {'Size':<12}\"\n",366 " print(header)\n",367 " print(\"-\" * len(header))\n",368 " for row in table_rows:\n",369 " print(f\"{row['model']:<20} {row['status']:<10} {row['start']:<10} {row['end']:<10} {row['duration']:<10} {row['size']:<12}\")\n",370 " \n",371 " print(f\"\\n✅ Completed. {len(competitors)}/{len(configs)} models successful.\")\n",372 " return competitors, answers\n",373 "\n",374 "async def mock_execute_models_in_parallel(configs: List[Dict[str, Any]]) -> Tuple[List[str], List[str]]:\n",375 " \"\"\"\n",376 " Mocks parallel API calls to display timing and size metrics in a table.\n",377 " No actual API calls are made.\n",378 " \"\"\"\n",379 " print(f\"\\n🚀 Starting MOCK execution of {len(configs)} models...\\n\")\n",380 " \n",381 " table_rows = []\n",382 " competitors = []\n",383 " answers = []\n",384 "\n",385 " async def mock_api_call(config):\n",386 " model_name = config.get(\"model_name\", \"Unknown-Model\")\n",387 " start_time = datetime.now()\n",388 " \n",389 " # Simulate varying network latency (0.5 to 2.5 seconds)\n",390 " await asyncio.sleep(random.uniform(0.5, 2.5))\n",391 " \n",392 " # Randomly decide if this mock call \"fails\" (10% chance)\n",393 " is_success = random.random() > 0.1\n",394 " \n",395 " if is_success:\n",396 " status = \"✅ Success\"\n",397 " # Mock a response string of random length\n",398 " mock_answer = \"Mock response data \" * random.randint(5, 500)\n",399 " out_size = len(mock_answer.encode('utf-8'))\n",400 " else:\n",401 " status = \"❌ Error\"\n",402 " mock_answer = \"Error: Mocked API failure\"\n",403 " out_size = 0\n",404 " \n",405 " end_time = datetime.now()\n",406 " \n",407 " # Calculate duration in mm:ss or hh:mm:ss\n",408 " duration = end_time - start_time\n",409 " total_seconds = int(duration.total_seconds())\n",410 " mm, ss = divmod(total_seconds, 60)\n",411 " hh, mm = divmod(mm, 60)\n",412 " dur_str = f\"{hh:02d}:{mm:02d}:{ss:02d}\" if hh > 0 else f\"{mm:02d}:{ss:02d}\"\n",413 "\n",414 " # Record metrics for the final table\n",415 " metrics = {\n",416 " \"model\": model_name,\n",417 " \"status\": status,\n",418 " \"start\": start_time.strftime(\"%H:%M:%S\"),\n",419 " \"end\": end_time.strftime(\"%H:%M:%S\"),\n",420 " \"duration\": dur_str,\n",421 " \"size\": format_bytes(out_size)\n",422 " }\n",423 " \n",424 " return model_name, mock_answer, status, metrics\n",425 "\n",426 " # Execute mock tasks in parallel\n",427 " tasks = [mock_api_call(config) for config in configs]\n",428 " results = await asyncio.gather(*tasks)\n",429 "\n",430 " # Prepare table headers\n",431 " header = f\"{'Model':<20} {'Status':<10} {'Start':<10} {'End':<10} {'Duration':<10} {'Size':<12}\"\n",432 " print(header)\n",433 " print(\"-\" * len(header))\n",434 "\n",435 " # Output rows and collect final success data\n",436 " for model_name, answer, status, row in results:\n",437 " print(f\"{row['model']:<20} {row['status']:<10} {row['start']:<10} {row['end']:<10} {row['duration']:<10} {row['size']:<12}\")\n",438 " if status == \"✅ Success\":\n",439 " competitors.append(model_name)\n",440 " answers.append(answer)\n",441 "\n",442 " print(f\"\\n✅ Completed. {len(competitors)}/{len(configs)} models simulated successfully.\")\n",443 " return competitors, answers\n"444 ]445 },446 {447 "cell_type": "code",448 "execution_count": null,449 "metadata": {},450 "outputs": [],451 "source": [452 "\n",453 "# --- Mock API Calls ---\n",454 "competitors, answers = await mock_execute_models_in_parallel(model_configs)\n"455 ]456 },457 {458 "cell_type": "code",459 "execution_count": null,460 "metadata": {},461 "outputs": [],462 "source": [463 "# Execute all models in parallel using the async functions\n",464 "competitors, answers = await execute_models_in_parallel(model_configs, messages)\n",465 " \n"466 ]467 },468 {469 "cell_type": "code",470 "execution_count": null,471 "metadata": {},472 "outputs": [],473 "source": [474 "# Optionally - Display the answers\n",475 "# for model_name, answer in zip(competitors, answers):\n",476 "# display(Markdown(f\"### {model_name}\\n\\n{answer}\"))\n"477 ]478 },479 {480 "cell_type": "markdown",481 "metadata": {},482 "source": [483 "## Step 4: Aggregator - Collect and Format Outputs\n",484 "\n",485 "The **Aggregator** collects all model outputs and formats them for the final evaluator."486 ]487 },488 {489 "cell_type": "markdown",490 "metadata": {},491 "source": []492 },493 {494 "cell_type": "code",495 "execution_count": null,496 "metadata": {},497 "outputs": [],498 "source": [499 "# ==========================================\n",500 "# Aggregator: Collect outputs and format for evaluation\n",501 "# ==========================================\n",502 "# The Aggregator collects all model outputs and prepares them\n",503 "# for the final Evaluator (judge) that will rank the responses\n",504 "\n",505 "def aggregator_format_outputs(competitors: List[str], answers: List[str]) -> str:\n",506 " \"\"\"\n",507 " Aggregator: Collects all model outputs and formats them for evaluation.\n",508 " Returns a formatted string ready for the judge/evaluator.\n",509 " \"\"\"\n",510 " together = \"\"\n",511 " for index, answer in enumerate(answers):\n",512 " together += f\"# Response from competitor {index+1}\\n\\n\"\n",513 " together += answer + \"\\n\\n\"\n",514 " return together\n",515 "\n",516 "# Use the aggregator to format all outputs\n",517 "together = aggregator_format_outputs(competitors, answers)\n",518 "print(f\"✅ Aggregator collected and formatted {len(competitors)} model responses\")"519 ]520 },521 {522 "cell_type": "markdown",523 "metadata": {},524 "source": [525 "## Step 5: Final Evaluator - Judge and Rank All Outputs\n",526 "\n",527 "The **Final Evaluator** (Judge) evaluates all aggregated responses and ranks them from best to worst."528 ]529 },530 {531 "cell_type": "code",532 "execution_count": null,533 "metadata": {},534 "outputs": [],535 "source": [536 "# ==========================================\n",537 "# Final Evaluator: Judge and Rank All Outputs\n",538 "# ==========================================\n",539 "# The final Evaluator (Judge) model evaluates all aggregated responses\n",540 "# and ranks them from best to worst\n",541 "\n",542 "judge = f\"\"\"You are judging a competition between {len(competitors)} competitors.\n",543 "Each model has been given this question:\n",544 "\n",545 "{question}\n",546 "\n",547 "Your job is to evaluate each response for clarity and strength of argument, and rank them in order of best to worst.\n",548 "Respond with JSON, and only JSON, with the following format:\n",549 "{{\"results\": [\"best competitor number\", \"second best competitor number\", \"third best competitor number\", ...]}}\n",550 "\n",551 "Here are the responses from each competitor:\n",552 "\n",553 "{together}\n",554 "\n",555 "Now respond with the JSON with the ranked order of the competitors, nothing else. Do not include markdown formatting or code blocks.\"\"\"\n",556 "\n",557 "judge_messages = [{\"role\": \"user\", \"content\": judge}]"558 ]559 },560 {561 "cell_type": "code",562 "execution_count": null,563 "metadata": {},564 "outputs": [],565 "source": [566 "# ==========================================\n",567 "# Final Evaluator Call: Judge Ranks All Outputs\n",568 "# ==========================================\n",569 "# The Evaluator (Judge) model evaluates and ranks all model responses\n",570 "\n",571 "openai = OpenAI()\n",572 "response = openai.chat.completions.create(\n",573 " model=\"gpt-5-mini\",\n",574 " messages=judge_messages,\n",575 ")\n",576 "results = response.choices[0].message.content\n",577 "print(\"✅ Final Evaluator (Judge) completed ranking:\")\n",578 "print(results)"579 ]580 },581 {582 "cell_type": "code",583 "execution_count": null,584 "metadata": {},585 "outputs": [],586 "source": [587 "# Parse and display the final rankings\n",588 "results_dict = json.loads(results)\n",589 "ranks = results_dict[\"results\"]\n",590 "print(\"\\n\" + \"=\"*50)\n",591 "print(\"FINAL RANKINGS\")\n",592 "print(\"=\"*50)\n",593 "for index, result in enumerate(ranks):\n",594 " competitor = competitors[int(result)-1]\n",595 " print(f\"Rank {index+1}: {competitor}\")"596 ]597 }598 ],599 "metadata": {600 "kernelspec": {601 "display_name": ".venv",602 "language": "python",603 "name": "python3"604 },605 "language_info": {606 "codemirror_mode": {607 "name": "ipython",608 "version": 3609 },610 "file_extension": ".py",611 "mimetype": "text/x-python",612 "name": "python",613 "nbconvert_exporter": "python",614 "pygments_lexer": "ipython3",615 "version": "3.12.4"616 }617 },618 "nbformat": 4,619 "nbformat_minor": 2620}621