Workflow 001: Self-Evolving Loop
Overview
| Property | Value |
|---|---|
| Workflow ID | 001 |
| Title | Self-Evolving Loop |
| Complexity | Advanced |
| Duration | Variable (depends on dataset size) |
Purpose
Execute the complete self-evolving agent loop that autonomously improves prompts through iterative evaluation and optimization.
Architecture
┌─────────────────────────────────────────────────────────────────────┐
│ SELF-EVOLVING LOOP │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Dataset │ │
│ │ (sections) │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ FOR EACH SECTION │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │Summarization│───►│ Run Eval │───►│ Check │ │ │
│ │ │ Agent │ │ (4 graders)│ │ Pass? │ │ │
│ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │
│ │ │ │ │
│ │ ┌──────────┬──────────┘ │ │
│ │ │ │ │ │
│ │ ▼ ▼ │ │
│ │ [PASS] [FAIL] │ │
│ │ │ │ │ │
│ │ │ ▼ │ │
│ │ │ ┌─────────────┐ │ │
│ │ │ │ Metaprompt │ │ │
│ │ │ │ Agent │ │ │
│ │ │ └──────┬──────┘ │ │
│ │ │ │ │ │
│ │ │ ▼ │ │
│ │ │ ┌─────────────┐ │ │
│ │ │ │ Update │ │ │
│ │ │ │ Prompt │ │ │
│ │ │ └──────┬──────┘ │ │
│ │ │ │ │ │
│ │ │ ▼ │ │
│ │ │ ┌─────────────┐ │ │
│ │ │ │ Retry? │──► MAX_RETRIES │ │
│ │ │ └──────┬──────┘ exceeded │ │
│ │ │ │ │ │
│ │ └────┬─────┘ │ │
│ │ ▼ │ │
│ │ [NEXT SECTION] │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ FINAL OUTPUT │ │
│ │ • Best prompt (highest scoring) │ │
│ │ • Version history │ │
│ │ • Aggregate statistics │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘Prerequisites
Complete these SOPs before starting:
- [x] SOP-001: Environment Setup
- [x] SOP-002: Dataset Preparation
- [x] SOP-003: Eval Creation
- [x] SOP-004: Baseline Agent Setup
- [x] SOP-005: Run Evaluation Loop
Complete Implementation
Step 1: Import Dependencies and Initialize
python
import asyncio
import json
import time
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field, ConfigDict, field_validator
from openai import OpenAI
from agents import Agent, Runner
# Initialize OpenAI client
client = OpenAI()
# Configuration
EVAL_ID = "eval_..." # Your eval ID from SOP-003
MAX_OPTIMIZATION_RETRIES = 3
LENIENT_PASS_RATIO = 0.75
LENIENT_AVERAGE_THRESHOLD = 0.85Step 2: Define Data Models
python
class PromptVersionEntry(BaseModel):
"""Track prompt versions with metadata."""
version: int = Field(..., ge=0)
model: str = Field("gpt-5", min_length=1)
prompt: str = Field(..., min_length=1)
timestamp: datetime = Field(default_factory=datetime.utcnow)
eval_id: Optional[str] = None
run_id: Optional[str] = None
metadata: Optional[dict[str, Any]] = None
model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True)
@field_validator("prompt")
@classmethod
def prompt_not_blank(cls, v: str) -> str:
if not v.strip():
raise ValueError("prompt must not be blank")
return v
class VersionedPrompt:
"""Manage prompt version history."""
def __init__(self, initial_prompt: str, model: str = "gpt-5", **kwargs):
self._versions = [PromptVersionEntry(version=0, prompt=initial_prompt, model=model, **kwargs)]
def update(self, new_prompt: str, model: str = "gpt-5", **kwargs) -> PromptVersionEntry:
entry = PromptVersionEntry(
version=self.current().version + 1,
prompt=new_prompt,
model=model,
**kwargs
)
self._versions.append(entry)
return entry
def current(self) -> PromptVersionEntry:
return self._versions[-1]
def history(self) -> list[PromptVersionEntry]:
return self._versions.copy()Step 3: Define Evaluation Functions
python
def run_eval(eval_id: str, section: str, summary: str):
"""Create an eval run for a section-summary pair."""
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}}],
},
},
)
def poll_eval_run(eval_id: str, run_id: str, max_polls: int = 10):
"""Poll until eval run completes."""
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)
return client.evals.runs.output_items.list(eval_id=eval_id, run_id=run_id)
def parse_eval_run_output(items) -> list[dict]:
"""Extract grader scores from eval output."""
results = []
for item in items.data:
for result in item.results:
reasoning = None
try:
content = result.sample["output"][0]["content"]
content_json = json.loads(content)
reasoning = " ".join([step["conclusion"] for step in content_json["steps"]])
except Exception:
pass
results.append({
"grader_name": result.name,
"score": result.score,
"passed": result.passed,
"reasoning": reasoning,
})
return results
def calculate_grader_score(grader_scores: list[dict]) -> float:
"""Calculate average score across all graders."""
if not grader_scores:
return 0.0
return sum(entry.get("score", 0.0) for entry in grader_scores) / len(grader_scores)
def is_lenient_pass(grader_scores: list[dict], average_score: float) -> bool:
"""Check if evaluation passes lenient criteria."""
if not grader_scores:
return False
passed_count = sum(1 for entry in grader_scores if entry.get("passed"))
total_graders = len(grader_scores)
if total_graders and (passed_count / total_graders) >= LENIENT_PASS_RATIO:
return True
return average_score >= LENIENT_AVERAGE_THRESHOLD
def collect_grader_feedback(grader_scores: list[dict]) -> str:
"""Consolidate grader feedback for metaprompt."""
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("Chemical names missing from summary.")
elif grader.startswith("word_length_deviation_grader"):
feedback_lines.append("Summary length deviates from target.")
elif grader.startswith("cosine_similarity"):
feedback_lines.append("Summary not similar enough to source.")
elif grader.startswith("llm_as_judge") and reasoning:
feedback_lines.append(reasoning)
if not feedback_lines:
feedback_lines.append("All graders passed; tighten coverage and precision.")
return " ".join(feedback_lines)Step 4: Define Metaprompt Template
python
METAPROMPT_TEMPLATE = """
# Context:
## Original prompt:
{original_prompt}
## Section:
{section}
## Summary:
{summary}
## Reason to improve the prompt:
{reasoning}
# Task:
Write a new summarization prompt that is significantly improved and more specific.
The new prompt should instruct the model to produce concise yet comprehensive
technical summaries that precisely preserve all explicit information from the
source text. It should emphasize the inclusion of all named entities, quantities,
compounds, and technical terminology without paraphrasing or omission.
The resulting prompt should read like a clear, directive system message for a
technical summarization assistant—structured, unambiguous, and generalizable
across scientific or regulatory document sections.
"""Step 5: Create Agents
python
# Metaprompt agent (static)
metaprompt_agent = Agent(
name="MetapromptAgent",
instructions="You are a prompt optimizer."
)
# Initialize versioned prompt
summarization_prompt = VersionedPrompt(
initial_prompt="You are a summarization assistant. Given a section of text, produce a summary."
)
def make_summarization_agent(prompt_entry: PromptVersionEntry) -> Agent:
"""Factory to create summarization agent from prompt entry."""
return Agent(
name="SummarizationAgent",
instructions=prompt_entry.prompt,
model=prompt_entry.model,
)
# Create initial agent
summarization_agent = make_summarization_agent(summarization_prompt.current())
# Tracking state
eval_cache: dict[tuple[str, str], list[dict]] = {}
best_candidate: dict[str, Any] = {
"score": float("-inf"),
"prompt": summarization_prompt.current().prompt,
"model": summarization_prompt.current().model,
"summary": None,
"version": summarization_prompt.current().version,
"passed_lenient": False,
}
aggregate_prompt_stats: dict[int, dict[str, Any]] = {}Step 6: Main Orchestration Loop
python
async def process_section(
section: str,
summarization_agent: Agent,
eval_id: str,
current_prompt: str,
retry_count: int = 0
) -> tuple[str, list[dict], bool]:
"""Process a single section through the loop."""
global best_candidate
# Generate summary
result = await Runner.run(summarization_agent, section)
summary = result.final_output
# Check cache
cache_key = (section, summary)
if cache_key in eval_cache:
grader_scores = eval_cache[cache_key]
else:
# Run evaluation
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)
grader_scores = parse_eval_run_output(run_output)
eval_cache[cache_key] = grader_scores
# Calculate scores
average_score = calculate_grader_score(grader_scores)
passed = is_lenient_pass(grader_scores, average_score)
# Update best candidate
if average_score > best_candidate["score"]:
best_candidate = {
"score": average_score,
"prompt": current_prompt,
"model": summarization_agent.model,
"summary": summary,
"version": summarization_prompt.current().version,
"passed_lenient": passed,
}
return summary, grader_scores, passed
async def optimize_prompt(
section: str,
summary: str,
grader_scores: list[dict],
current_prompt: str
) -> str:
"""Use metaprompt agent to generate improved prompt."""
feedback = collect_grader_feedback(grader_scores)
metaprompt_input = METAPROMPT_TEMPLATE.format(
original_prompt=current_prompt,
section=section,
summary=summary,
reasoning=feedback
)
result = await Runner.run(metaprompt_agent, metaprompt_input)
return result.final_output
async def run_self_evolving_loop(sections: list[str], eval_id: str):
"""Main orchestration loop."""
global summarization_agent
print(f"Starting self-evolving loop with {len(sections)} sections")
print(f"Initial prompt version: v{summarization_prompt.current().version}")
print("-" * 60)
for idx, section in enumerate(sections):
print(f"\n[Section {idx + 1}/{len(sections)}]")
retry_count = 0
passed = False
while not passed and retry_count < MAX_OPTIMIZATION_RETRIES:
current_prompt = summarization_prompt.current().prompt
# Process section
summary, grader_scores, passed = await process_section(
section=section,
summarization_agent=summarization_agent,
eval_id=eval_id,
current_prompt=current_prompt,
retry_count=retry_count
)
average_score = calculate_grader_score(grader_scores)
print(f" Attempt {retry_count + 1}: score={average_score:.3f}, passed={passed}")
if not passed and retry_count < MAX_OPTIMIZATION_RETRIES - 1:
# Optimize prompt
print(" Optimizing prompt...")
new_prompt = await optimize_prompt(
section=section,
summary=summary,
grader_scores=grader_scores,
current_prompt=current_prompt
)
# Update versioned prompt
summarization_prompt.update(
new_prompt=new_prompt,
model="gpt-5",
eval_id=eval_id,
metadata={"retry": retry_count + 1}
)
# Recreate agent with new prompt
summarization_agent = make_summarization_agent(summarization_prompt.current())
print(f" Updated to prompt v{summarization_prompt.current().version}")
retry_count += 1
# Update aggregate stats
version = summarization_prompt.current().version
if version not in aggregate_prompt_stats:
aggregate_prompt_stats[version] = {"sections": 0, "total_score": 0.0, "passed": 0}
aggregate_prompt_stats[version]["sections"] += 1
aggregate_prompt_stats[version]["total_score"] += average_score
if passed:
aggregate_prompt_stats[version]["passed"] += 1
print("\n" + "=" * 60)
print("SELF-EVOLVING LOOP COMPLETE")
print("=" * 60)
print(f"\nBest Candidate:")
print(f" Version: v{best_candidate['version']}")
print(f" Score: {best_candidate['score']:.3f}")
print(f" Passed: {best_candidate['passed_lenient']}")
print(f"\nPrompt History: {len(summarization_prompt.history())} versions")
return best_candidate, summarization_prompt.history()Step 7: Run the Loop
python
async def main():
"""Main entry point."""
# Load your dataset
import pandas as pd
df = pd.read_csv("data/sections.csv")
sections = df["content"].tolist()
# Run the loop
best, history = await run_self_evolving_loop(
sections=sections,
eval_id=EVAL_ID
)
# Save results
with open("results/best_prompt.txt", "w") as f:
f.write(best["prompt"])
print(f"\nBest prompt saved to results/best_prompt.txt")
return best, history
# Execute
if __name__ == "__main__":
asyncio.run(main())Expected Output
Starting self-evolving loop with 10 sections
Initial prompt version: v0
------------------------------------------------------------
[Section 1/10]
Attempt 1: score=0.650, passed=False
Optimizing prompt...
Updated to prompt v1
Attempt 2: score=0.820, passed=True
[Section 2/10]
Attempt 1: score=0.780, passed=True
[Section 3/10]
Attempt 1: score=0.710, passed=False
Optimizing prompt...
Updated to prompt v2
Attempt 2: score=0.890, passed=True
...
============================================================
SELF-EVOLVING LOOP COMPLETE
============================================================
Best Candidate:
Version: v2
Score: 0.890
Passed: True
Prompt History: 3 versionsVerification Checklist
- [ ] All dependencies installed
- [ ] EVAL_ID configured
- [ ] Dataset loaded
- [ ] Loop executes without errors
- [ ] Prompt versions incrementing
- [ ] Best candidate tracking working
- [ ] Results saved