SOP-005: Run Evaluation Loop
Document Control
| Property | Value |
|---|---|
| SOP ID | 005 |
| Title | Run Evaluation Loop |
| Version | 1.0 |
| Status | Active |
| Complexity | Medium |
Purpose
Execute evaluations on agent outputs, parse grader scores, and manage the feedback loop.
Prerequisites
- Completed SOP-003: Eval Creation
- Completed SOP-004: Baseline Agent Setup
- Valid Eval ID
Evaluation Flow
┌─────────────────────────────────────────────────────────────────────┐
│ EVALUATION LOOP FLOW │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Agent │────►│ Run Eval │────►│ Poll for │ │
│ │ Generates │ │ (API) │ │ Completion │ │
│ │ Summary │ │ │ │ │ │
│ └─────────────┘ └─────────────┘ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Calculate │◄────│ Parse │◄────│ Retrieve │ │
│ │ Scores │ │ Results │ │ Output │ │
│ └──────┬──────┘ └─────────────┘ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ DECISION POINT │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Lenient Pass? │ │ Max Retries? │ │ │
│ │ │ (≥75% graders │ │ (default: 3) │ │ │
│ │ │ OR ≥85% avg) │ │ │ │ │
│ │ └────────┬────────┘ └────────┬────────┘ │ │
│ │ │ │ │ │
│ │ YES ──┴── NO YES ──┴── NO │ │
│ │ │ │ │ │ │ │
│ │ ▼ ▼ ▼ ▼ │ │
│ │ [PASS] [OPTIMIZE] [ALERT] [RETRY] │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘Step-by-Step Procedure
Step 1: Define Eval Runner Function
python
import time
import json
from openai import OpenAI
client = OpenAI()
def run_eval(eval_id: str, section: str, summary: str):
"""Creates a run of the eval with the input section and output summary."""
return client.evals.runs.create(
eval_id=eval_id,
name="self-evolving-eval",
data_source={
"type": "jsonl",
"source": {
"type": "file_content",
"content": [
{
"item": {
"section": section,
"summary": summary,
}
}
],
},
},
)Step 2: Create Polling Function
python
def poll_eval_run(eval_id: str, run_id: str, max_polls: int = 10):
"""
Polls the evaluation run until completion or timeout.
Handles asynchronous behavior by periodically checking run status.
Balances responsiveness and resource use with fixed intervals.
"""
run = None
for attempt in range(1, max_polls + 1):
run = client.evals.runs.retrieve(eval_id=eval_id, run_id=run_id)
if run.status == "completed":
break
if attempt == max_polls:
print("❌ Exceeded retries, aborting")
break
time.sleep(5) # Wait 5 seconds between polls
run_output_items = client.evals.runs.output_items.list(
eval_id=eval_id, run_id=run_id
)
return run_output_itemsStep 3: Parse Eval Results
python
def parse_eval_run_output(items):
"""Extract all grader scores and any available conclusion outputs."""
all_results = []
for item in items.data:
for result in item.results:
grader_name_full = result.name
score = result.score
passed = result.passed
reasoning = None
# Try to extract reasoning from LLM-as-judge
try:
sample = result.sample
if sample:
content = result.sample["output"][0]["content"]
content_json = json.loads(content)
steps = content_json["steps"]
reasoning = " ".join([step["conclusion"] for step in steps])
except Exception:
pass
all_results.append({
"grader_name": grader_name_full,
"score": score,
"passed": passed,
"reasoning": reasoning,
})
return all_resultsStep 4: Calculate Scores
python
def calculate_grader_score(grader_scores):
"""Simple average score of all graders from the eval."""
if not grader_scores:
return 0.0
score_sum = sum(entry.get("score", 0.0) for entry in grader_scores)
return score_sum / len(grader_scores)
def calculate_total_grader_score(grader_scores):
"""Sum of all grader scores for aggregate tracking."""
if not grader_scores:
return 0.0
return sum(entry.get("score", 0.0) for entry in grader_scores)Step 5: Check Lenient Pass
python
LENIENT_PASS_RATIO = 0.75 # 75% of graders must pass (binary)
LENIENT_AVERAGE_THRESHOLD = 0.85 # 85% average score across graders
def is_lenient_pass(grader_scores, average_score: float) -> bool:
"""
Check if evaluation passes lenient criteria.
Returns True if:
- At least 75% of graders pass (binary), OR
- Average score is at least 85%
"""
if not grader_scores:
return False
passed_count = sum(1 for entry in grader_scores if entry.get("passed"))
total_graders = len(grader_scores)
# Check binary pass ratio
if total_graders and (passed_count / total_graders) >= LENIENT_PASS_RATIO:
return True
# Check average score threshold
return average_score >= LENIENT_AVERAGE_THRESHOLDStep 6: Collect Grader Feedback
python
DEFAULT_PASSING_FEEDBACK = (
"All graders passed; tighten factual coverage, chemical completeness, and conciseness."
)
def collect_grader_feedback(grader_scores):
"""Consolidate grader reasoning into actionable feedback for the metaprompt agent."""
feedback_lines = []
for entry in grader_scores:
grader = entry.get("grader_name", "")
passed = entry.get("passed", False)
reasoning = entry.get("reasoning")
if not passed:
if grader.startswith("chemical_name_grader"):
feedback_lines.append(
"Not all chemical names in the input section were included in the summary."
)
elif grader.startswith("word_length_deviation_grader"):
feedback_lines.append(
"The summary length deviates too much from the expected length."
)
elif grader.startswith("cosine_similarity"):
feedback_lines.append(
"The summary is not sufficiently similar to the source section."
)
elif grader.startswith("llm_as_judge") and reasoning:
feedback_lines.append(reasoning)
if not feedback_lines:
feedback_lines.append(DEFAULT_PASSING_FEEDBACK)
return " ".join(feedback_lines)Step 7: Async Eval with Caching
python
eval_cache: dict[tuple[str, str], list[dict]] = {}
async def get_eval_grader_score(eval_id: str, section: str, summary: str):
"""Retrieve grader scores for a section-summary pair with caching."""
cache_key = (section, summary)
if cache_key in eval_cache:
return eval_cache[cache_key]
eval_run = run_eval(eval_id=eval_id, section=section, summary=summary)
run_output = poll_eval_run(eval_id=eval_id, run_id=eval_run.id)
results = parse_eval_run_output(run_output)
eval_cache[cache_key] = results
return resultsStep 8: Test the Evaluation Loop
python
async def test_evaluation():
"""Test the complete evaluation flow."""
EVAL_ID = "eval_..." # Your eval ID
# Test section and summary
SECTION = """3.2.S.1 General Information ([1-13C]pyruvic acid)
The active ingredient in Hyperpolarized Pyruvate (13C) Injection is
hyperpolarized [1-13C]pyruvate."""
SUMMARY = """The active ingredient in Hyperpolarized Pyruvate (13C) Injection
is hyperpolarized [1-13C]pyruvate, derived from [1-13C]pyruvic acid."""
# Run evaluation
grader_scores = await get_eval_grader_score(
eval_id=EVAL_ID,
section=SECTION,
summary=SUMMARY
)
# Calculate scores
average_score = calculate_grader_score(grader_scores)
total_score = calculate_total_grader_score(grader_scores)
lenient_passed = is_lenient_pass(grader_scores, average_score)
# Print results
print(f"Grader Scores: {grader_scores}")
print(f"Average Score: {average_score:.3f}")
print(f"Total Score: {total_score:.3f}")
print(f"Lenient Pass: {lenient_passed}")
if not lenient_passed:
feedback = collect_grader_feedback(grader_scores)
print(f"Feedback: {feedback}")
# Run test
import asyncio
asyncio.run(test_evaluation())Expected Output
Grader Scores: [
{'grader_name': 'chemical_name_grader-xxx', 'score': 0.5, 'passed': False},
{'grader_name': 'word_length_deviation_grader-xxx', 'score': 0.8, 'passed': True},
{'grader_name': 'cosine_similarity-xxx', 'score': 0.91, 'passed': True},
{'grader_name': 'llm_as_judge-xxx', 'score': 0.8, 'passed': True}
]
Average Score: 0.753
Total Score: 3.010
Lenient Pass: TrueVerification Checklist
- [ ]
run_eval()function creates eval runs - [ ]
poll_eval_run()waits for completion - [ ]
parse_eval_run_output()extracts all grader scores - [ ] Score calculation functions working
- [ ] Lenient pass logic implemented
- [ ] Feedback collection working
- [ ] Caching prevents redundant API calls
- [ ] Test evaluation completes successfully
Troubleshooting
Issue: Eval run never completes
Solution: Increase max_polls or check for API errors:
python
run = client.evals.runs.retrieve(eval_id=eval_id, run_id=run_id)
print(f"Status: {run.status}, Error: {getattr(run, 'error', None)}")Issue: Empty results list
Solution: Verify data source format matches eval schema.
Issue: Score always 0
Solution: Check grader pass_threshold settings and template variables.