Skip to main content
Repeat a step 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

1

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()
2

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

API Reference

repeat()

repeat(
    step: Any,                                              # Step to repeat
    until: Optional[Callable[[WorkflowContext], bool]] = None,  # Stop condition
    max_iterations: int = 10                                # Maximum iterations
) -> Repeat

Parameters

ParameterTypeDefaultDescription
stepAnyrequiredStep to repeat (function, Agent, Task)
untilOptional[Callable]NoneFunction returning True to stop
max_iterationsint10Maximum 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:
VariableTypeDescription
repeat_iterationsintNumber 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)
])

Use Cases

Use CaseDescription
Content GenerationGenerate until quality threshold met
OptimizationIterate until optimal solution found
ValidationRetry until validation passes
Data CollectionCollect until enough data gathered
Self-CorrectionAgent corrects itself until correct

How It Works

PhaseWhat happens
1. Startrepeat() records the initial repeat_index = 0 in context variables
2. GenerateThe wrapped step runs and returns a StepResult
3. EvaluateThe until callable receives the updated WorkflowContext; returns True to stop
4. RepeatIf until returns False and max_iterations not reached, the step runs again
5. ReturnFinal StepResult is passed to the next workflow step

Best Practices

Cap repeat loops to prevent runaway cost when exit conditions never trigger.
Evaluator functions should return explicit booleans — avoid ambiguous string matching.
Repeat improves a single artefact; loop processes many items — do not confuse the two.
Inspect intermediate outputs when tuning evaluator-optimiser workflows.

Comparison with Loop

Featureloop()repeat()
PurposeIterate over dataIterate until condition
Data sourceList, CSV, fileNone (generates)
StoppingWhen data exhaustedWhen condition met
Use caseBatch processingIterative improvement

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