Workflow Repeat (Evaluator-Optimizer)
Repeat steps until a condition is met using the repeat() helper
Repeat a step — or a fixed list of steps — until a condition is met. This pattern is ideal for iterative improvement, quality checking, and optimization workflows.
from praisonaiagents import Agent, AgentFlow, repeat, WorkflowContext, StepResult
editor = Agent(name="Editor", instructions="Improve the draft each iteration")
def refine(ctx: WorkflowContext) -> StepResult:
n = ctx.variables.get("repeat_index", 0)
return StepResult(output=f"Draft revision {n}")
workflow = AgentFlow(agents=[editor], steps=[repeat(refine, times=3)])
workflow.start("Iterate until quality is acceptable")The user requests repeated refinement; the repeat helper runs the step N times.
Quick Start
- Define generator and stop condition
python from praisonaiagents import AgentFlow, WorkflowContext, StepResult from praisonaiagents import repeat class ContentGenerator: def __init__(self): self.points = [] def generate(self, ctx: WorkflowContext) -> StepResult: self.points.append(f"Point {len(self.points) + 1}") return StepResult( output=f"Generated {len(self.points)} points", variables={"point_count": len(self.points)} ) # Condition: stop when we have enough def has_enough(ctx: WorkflowContext) -> bool: return ctx.variables.get("point_count", 0) >= 5 generator = ContentGenerator() - Run repeat workflow
python workflow = AgentFlow(steps=[ repeat(generator.generate, until=has_enough, max_iterations=10) ]) result = workflow.start("Generate content")
Output:
🔄 Repeating up to 10 times...
✅ generate: Generated 1 points...
✅ generate: Generated 2 points...
✅ generate: Generated 3 points...
✅ generate: Generated 4 points...
✅ generate: Generated 5 points...
✅ Repeat condition met at iteration 5Repeating a list of steps
Pass a list to run several steps in order, once per iteration. Each step sees the previous step’s output.
from praisonaiagents import Agent, AgentFlow, repeat
writer = Agent(name="Writer", instructions="Draft a 1-sentence answer.")
critic = Agent(name="Critic", instructions="Rewrite the previous answer more clearly.")
workflow = AgentFlow(steps=[repeat([writer, critic], max_iterations=2)])
workflow.start("Explain vector databases in one sentence.")
# Runs: writer → critic → writer → critic
# Each step receives the previous step's output as ctx.previous_result.Ordering guarantees
| Behaviour | Guarantee |
|---|---|
| Order of members | Runs in the order given, once per iteration |
| Data flow | Each member’s ctx.previous_result is the previous member’s output |
max_iterations | Counts full iterations of the list, not individual member calls |
until | Checked after a full iteration completes |
stop_workflow=True | Halts the current iteration immediately and ends the repeat |
Empty list repeat([]) | No-op — the flow continues |
API Reference
repeat()
repeat(
step: Any | list, # Step or list of steps to repeat
until: Optional[Callable[[WorkflowContext], bool]] = None, # Stop condition
max_iterations: int = 10 # Maximum iterations
) -> RepeatParameters
| Parameter | Type | Default | Description | |
|---|---|---|---|---|
step | `Any \ | list` | required | A single step OR a list of steps to run in order each iteration (function, Agent, Task) |
until | Optional[Callable] | None | Function returning True to stop (evaluated after each full iteration) | |
max_iterations | int | 10 | Maximum number of iterations |
Condition Function
The until function receives WorkflowContext and returns bool:
def my_condition(ctx: WorkflowContext) -> bool:
# Access previous output
output = ctx.previous_result
# Access variables
count = ctx.variables.get("count", 0)
# Return True to stop, False to continue
return "done" in output.lower() or count >= 10Result Variables
After repeat completion:
| Variable | Type | Description |
|---|---|---|
repeat_iterations | int | Number of iterations executed |
Examples
Quality-Based Stopping
def evaluate_quality(ctx: WorkflowContext) -> bool:
output = ctx.previous_result or ""
# Stop when output contains "excellent"
return "excellent" in output.lower()
workflow = AgentFlow(steps=[
repeat(improve_content, until=evaluate_quality, max_iterations=5)
])Counter-Based Stopping
def generate_item(ctx: WorkflowContext) -> StepResult:
count = ctx.variables.get("item_count", 0) + 1
return StepResult(
output=f"Item {count}",
variables={"item_count": count}
)
def has_enough_items(ctx: WorkflowContext) -> bool:
return ctx.variables.get("item_count", 0) >= 10
workflow = AgentFlow(steps=[
repeat(generate_item, until=has_enough_items, max_iterations=20)
])With Agents
from praisonaiagents import Agent
improver = Agent(
name="Improver",
role="Content improver",
instructions="Improve this content. Say 'DONE' when perfect."
)
def is_done(ctx: WorkflowContext) -> bool:
return "DONE" in (ctx.previous_result or "")
workflow = AgentFlow(steps=[
repeat(improver, until=is_done, max_iterations=5)
])Evaluator-Optimizer Pattern
Classic pattern with separate generator and evaluator:
def generator(ctx: WorkflowContext) -> StepResult:
# Generate or improve content
current = ctx.previous_result or ""
improved = f"{current}\n- New point added"
return StepResult(output=improved)
def evaluator(ctx: WorkflowContext) -> bool:
output = ctx.previous_result or ""
# Check if we have enough points
point_count = output.count("- ")
return point_count >= 5
workflow = AgentFlow(steps=[
repeat(generator, until=evaluator, max_iterations=10)
])With Early Stop
def process_with_stop(ctx: WorkflowContext) -> StepResult:
if "error" in ctx.input:
return StepResult(output="Error detected", stop_workflow=True)
return StepResult(output="Processed")
workflow = AgentFlow(steps=[
repeat(process_with_stop, max_iterations=5)
])Two-agent back-and-forth
The simplest N-agent turn-taking pattern: alternate two agents for a fixed number of rounds.
from praisonaiagents import Agent, AgentFlow, repeat
proposer = Agent(name="Proposer", instructions="Propose one improvement to the previous message.")
reviewer = Agent(name="Reviewer", instructions="Critique the last proposal in one sentence.")
workflow = AgentFlow(steps=[repeat([proposer, reviewer], max_iterations=3)])
workflow.start("Improve this sentence: 'The product is good.'")For a fixed rotation of more than two speakers, prefer Discussion — it exists for exactly this.
Use Cases
| Use Case | Description |
|---|---|
| Content Generation | Generate until quality threshold met |
| Optimization | Iterate until optimal solution found |
| Validation | Retry until validation passes |
| Data Collection | Collect until enough data gathered |
| Self-Correction | Agent corrects itself until correct |
How It Works
| Phase | What happens |
|---|---|
| 1. Start | repeat() records the initial repeat_index = 0 in context variables |
| 2. Generate | The wrapped step runs and returns a StepResult |
| 3. Evaluate | The until callable receives the updated WorkflowContext; returns True to stop |
| 4. Repeat | If until returns False and max_iterations not reached, the step runs again |
| 5. Return | Final StepResult is passed to the next workflow step |
Best Practices
Set a sensible max_iterations
Cap repeat loops to prevent runaway cost when exit conditions never trigger.
Write clear until conditions
Evaluator functions should return explicit booleans — avoid ambiguous string matching.
Use repeat for refinement, loop for batches
Repeat improves a single artefact; loop processes many items — do not confuse the two.
Log each iteration in staging
Inspect intermediate outputs when tuning evaluator-optimiser workflows.
Comparison with Loop
| Feature | loop() | repeat() |
|---|---|---|
| Purpose | Iterate over data | Iterate until condition |
| Data source | List, CSV, file | None (generates) |
| Stopping | When data exhausted | When condition met |
| Use case | Batch processing | Iterative improvement, or a small fixed list of steps run in order each iteration |
Repeat vs. Discussion
repeat() now handles both a single step and a fixed list of steps — pick the right tool:
- Use
repeat(step, until=...)when the same step should iterate to improve. - Use
repeat([a, b, ...], max_iterations=N)for a light back-and-forth of a small, fixed list. - Use
Discussionwhen you want a first-class N-agent conversation with round-robin/custom selectors,until, and per-turn logging.