Workflow Repeat (Evaluator-Optimizer)

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.

Steps inside this pattern inherit the same max_retries, guardrails, and output_file policies as top-level steps. See Nested workflows → Retry, guardrails, and output_file.

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

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

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 5

Repeating 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
) -> Repeat

Parameters

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 >= 10

Result 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 Discussion when you want a first-class N-agent conversation with round-robin/custom selectors, until, and per-turn logging.

Workflow Patterns

Overview of routing, parallel, loop, and repeat

Workflow Loop

Iterate over lists and files

Workflow Routing

Decision-based branching

Workflow Parallel

Run independent steps concurrently

Workflow Discussion

N-agent conversation loop

​Quick Start

​Repeating a list of steps

​Ordering guarantees

​API Reference

​repeat()

​Parameters

​Condition Function

​Result Variables

​Examples

​Quality-Based Stopping

​Counter-Based Stopping

​With Agents

​Evaluator-Optimizer Pattern

​With Early Stop

​Two-agent back-and-forth

​Use Cases

​How It Works

​Best Practices

​Comparison with Loop

​Repeat vs. Discussion

​Related

Workflow Patterns

Workflow Loop

Workflow Routing

Workflow Parallel

Workflow Discussion

Quick Start

Repeating a list of steps

Ordering guarantees

API Reference

repeat()

Parameters

Condition Function

Result Variables

Examples

Quality-Based Stopping

Counter-Based Stopping

With Agents

Evaluator-Optimizer Pattern

With Early Stop

Two-agent back-and-forth

Use Cases

How It Works

Best Practices

Comparison with Loop

Repeat vs. Discussion

Related