Workflow Validation Loop

The Workflow Validation Loop feature allows you to create workflows with built-in quality checks and automatic retry mechanisms. When a task fails validation, it can automatically retry with feedback from the validator, creating a continuous improvement loop until the output meets quality standards.

from praisonaiagents import Agent, AgentFlow

validator = Agent(name="Validator", instructions="Check quality and request fixes")
workflow = AgentFlow(agents=[validator], steps=["validate_input", "run", "validate_output"])
workflow.start("Run with schema checks")

The user runs a workflow; validation steps guard inputs and outputs.

How It Works

Quick Start

Define guardrail and handlers

from praisonaiagents import AgentFlow, Task, WorkflowContext, StepResult

def validate_quality(result: StepResult) -> tuple[bool, str]:
    """Check if content meets quality standards."""
    content = result.output.lower()
    if len(content) < 100:
        return (False, "Content too short. Add more detail.")
    if "ai" not in content:
        return (False, "Content must discuss AI topics.")
    return (True, None)

# Step handlers
def write_content(ctx: WorkflowContext) -> StepResult:
    feedback = ctx.variables.get("validation_feedback", "")
    prompt = f"Write a blog post about AI trends. {feedback}"
    # In real usage, call an agent here
    return StepResult(output="AI is transforming industries...")

def publish_content(ctx: WorkflowContext) -> StepResult:
    return StepResult(output=f"Published: {ctx.previous_result}")

Run workflow with guardrail retry

workflow = AgentFlow(
    steps=[
        Task(
            name="write_content",
            handler=write_content,
            guardrails=validate_quality,
            max_retries=3
        ),
        Task(
            name="publish",
            handler=publish_content
        )
    ]
)

result = workflow.start("Write about AI")
print(result["output"])

Key Features

1. Guardrails with Automatic Retry

Guardrails validate step outputs and automatically retry with feedback:

from praisonaiagents import AgentFlow, Task, StepResult
from praisonaiagents import route

def check_quality(result: StepResult) -> tuple[bool, str]:
    if "error" in result.output.lower():
        return (False, "Output contains errors. Please fix.")
    return (True, None)

workflow = AgentFlow(steps=[
    Task(
        name="generate",
        action="Generate content about {{topic}}",
        guardrails=check_quality,
        max_retries=3  # Retry up to 3 times
    )
])

2. Routing for Decision Logic

Use route() for conditional branching based on output:

from praisonaiagents import AgentFlow, WorkflowContext, StepResult
from praisonaiagents import route

def review_content(ctx: WorkflowContext) -> StepResult:
    # Analyze content and decide
    if "good" in ctx.previous_result:
        return StepResult(output="decision: approved")
    return StepResult(output="decision: needs_revision")

def edit_content(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Edited content")

def publish_content(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Published!")

workflow = AgentFlow(steps=[
    write_step,
    review_content,
    route({
        "approved": [publish_content],
        "needs_revision": [edit_content, review_content]
    })
])

3. Multi-Stage Validation

Create complex validation workflows with multiple checkpoints:

from praisonaiagents import Agent, Task

# Define validation agents
tech_reviewer = Agent(
    name="TechnicalReviewer",
    role="Technical accuracy validator",
    goal="Ensure technical correctness and accuracy"
)

style_reviewer = Agent(
    name="StyleReviewer",
    role="Writing style validator",
    goal="Ensure clarity and proper writing style"
)

senior_reviewer = Agent(
    name="SeniorReviewer",
    role="Senior quality reviewer",
    goal="Perform final quality assurance"
)

# Multiple validation stages
technical_review = Task(
    name="technical_review",
    description="Check technical accuracy",
    expected_output="Technical validation result",
    agent=tech_reviewer,
    task_type="decision",
    condition={
        "approved": ["style_review"],
        "rejected": ["write_content"]
    }
)

style_review = Task(
    name="style_review",
    description="Check writing style and clarity",
    expected_output="Style validation result",
    agent=style_reviewer,
    task_type="decision",
    condition={
        "approved": ["final_review"],
        "rejected": ["edit_content"]
    }
)

final_review = Task(
    name="final_review",
    description="Final quality check",
    expected_output="Final approval decision",
    agent=senior_reviewer,
    task_type="decision",
    condition={
        "approved": ["publish"],
        "rejected": ["write_content"]
    }
)

Advanced Usage

Custom Validation Logic

Implement custom validation criteria:

from praisonaiagents import Agent, Task, AgentTeam

# Validator with specific criteria
validator = Agent(
    name="QualityValidator",
    role="Content quality assurance",
    goal="Ensure content meets all quality criteria",
    backstory="""You are a meticulous reviewer who checks for:
    1. Accuracy and factual correctness
    2. Clarity and readability
    3. Proper structure and flow
    4. Grammar and spelling
    5. Engagement and value to readers
    
    Provide specific, actionable feedback for improvements."""
)

# Task with detailed validation instructions
validate_task = Task(
    name="validate_with_criteria",
    description="""Review the content and evaluate:
    - Is the information accurate?
    - Is it well-structured?
    - Does it provide value?
    - Are there any errors?
    
    If any criteria fails, provide specific feedback for improvement.""",
    expected_output="Detailed validation report with pass/fail decision",
    agent=validator,
    task_type="decision",
    condition={
        "passed": ["publish"],
        "failed": ["revise_content"]
    }
)

Validation with External Tools

Integrate external validation tools:

from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents import tool

@tool("grammar_checker")
def check_grammar(text: str) -> dict:
    """Check grammar and return issues found"""
    # Integration with grammar checking service
    return {"errors": [], "suggestions": []}

@tool("plagiarism_checker")
def check_plagiarism(text: str) -> dict:
    """Check for plagiarism"""
    # Integration with plagiarism detection
    return {"similarity_score": 0.05, "sources": []}

validator = Agent(
    name="AutomatedValidator",
    role="Automated quality checker",
    tools=["grammar_checker", "plagiarism_checker"],
    goal="Ensure content quality using automated tools"
)

Retry Limits and Escalation

Implement retry limits to prevent infinite loops:

from praisonaiagents import Agent, Task

# Define required agents
validator = Agent(
    name="Validator",
    role="Content validator",
    goal="Validate content quality"
)

senior_reviewer = Agent(
    name="SeniorReviewer",
    role="Senior escalation reviewer",
    goal="Handle escalated reviews"
)

class ValidationWorkflow:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
        self.retry_count = 0
        self.validator = validator
        self.senior_reviewer = senior_reviewer
        
    def create_validation_task(self):
        return Task(
            name="validate_with_limit",
            description=f"Validate content (attempt {self.retry_count + 1}/{self.max_retries})",
            expected_output="Validation result",
            agent=self.validator,
            task_type="decision",
            condition={
                "approved": ["publish"],
                "rejected": ["retry_or_escalate"]
            }
        )
    
    def retry_or_escalate_task(self):
        self.retry_count += 1
        if self.retry_count >= self.max_retries:
            return Task(
                name="escalate_to_senior",
                description="Escalate to senior reviewer",
                expected_output="Senior review decision",
                agent=self.senior_reviewer
            )
        else:
            return self.create_validation_task()

Validation Patterns

1. Clear Validation Criteria

Define specific, measurable validation criteria:

validation_criteria = """
Evaluate the content based on:
1. Accuracy: All facts must be verifiable
2. Completeness: All required sections present
3. Clarity: Flesch reading score > 60
4. Length: Between 800-1200 words
5. Tone: Professional but engaging

Provide a score for each criterion and overall pass/fail.
"""

2. Structured Feedback

Ensure validators provide actionable feedback:

validator = Agent(
    name="StructuredValidator",
    role="Quality reviewer",
    goal="Provide structured, actionable feedback",
    backstory="""When rejecting content, always provide:
    1. Specific issues found
    2. Examples of problems
    3. Clear suggestions for improvement
    4. Priority of fixes needed"""
)

3. Progress Tracking

Monitor validation cycles:

from praisonaiagents.callbacks import Callback

class ValidationTracker(Callback):
    def __init__(self):
        self.validation_attempts = {}
        
    def on_task_start(self, task, **kwargs):
        if task.task_type == "decision":
            task_id = task.name
            self.validation_attempts[task_id] = \
                self.validation_attempts.get(task_id, 0) + 1
            print(f"Validation attempt #{self.validation_attempts[task_id]} for {task_id}")
    
    def on_task_complete(self, task, result, **kwargs):
        if task.task_type == "decision" and "rejected" in str(result).lower():
            print(f"Validation failed, feedback: {result}")

# Use callbacks with Workflow
from praisonaiagents import AgentFlow
from praisonaiagents import AgentFlowHooksConfig

workflow = AgentFlow(
    steps=[write_step, validate_step],
    hooks=WorkflowHooksConfig(
        on_step_complete=lambda name, result: print(f"Step {name}: {result.output[:50]}..."),
        on_step_error=lambda name, error: print(f"Step {name} failed: {error}")
    )
)

Common Patterns

1. Content Creation Pipeline

# Define agents for content pipeline
writer = Agent(
    name="Writer",
    role="Content writer",
    goal="Create high-quality written content"
)

editor = Agent(
    name="Editor",
    role="Content editor",
    goal="Review and improve content quality"
)

qa_lead = Agent(
    name="QALead",
    role="Quality assurance lead",
    goal="Ensure final content quality"
)

# Full content creation pipeline with validation
draft_task = Task(
    name="create_draft",
    description="Create initial draft",
    expected_output="First draft",
    agent=writer
)

review_task = Task(
    name="review_draft",
    description="Review and provide feedback",
    expected_output="Review decision",
    agent=editor,
    task_type="decision",
    condition={
        "approved": ["polish_content"],
        "needs_work": ["create_draft"]
    }
)

polish_task = Task(
    name="polish_content",
    description="Polish the approved draft",
    expected_output="Polished content",
    agent=writer
)

final_check = Task(
    name="final_quality_check",
    description="Final quality assurance",
    expected_output="Final decision",
    agent=qa_lead,
    task_type="decision",
    condition={
        "approved": ["publish"],
        "rejected": ["polish_content"]
    }
)

2. Code Review Workflow

# Define agents for code review workflow
developer = Agent(
    name="Developer",
    role="Software developer",
    goal="Implement features with high code quality"
)

senior_developer = Agent(
    name="SeniorDeveloper",
    role="Senior code reviewer",
    goal="Ensure code quality and standards"
)

qa_engineer = Agent(
    name="QAEngineer",
    role="Quality assurance engineer",
    goal="Test functionality and ensure quality"
)

# Code review with validation loop
code_task = Task(
    name="implement_feature",
    description="Implement the new feature",
    expected_output="Implemented code",
    agent=developer
)

code_review = Task(
    name="review_code",
    description="Review code for quality and standards",
    expected_output="Code review decision",
    agent=senior_developer,
    task_type="decision",
    condition={
        "approved": ["test_code"],
        "changes_requested": ["implement_feature"]
    }
)

test_task = Task(
    name="test_code",
    description="Run tests and verify functionality",
    expected_output="Test results",
    agent=qa_engineer,
    task_type="decision",
    condition={
        "passed": ["deploy"],
        "failed": ["implement_feature"]
    }
)

Troubleshooting

Common Issues

Infinite Loops: Always implement retry limits or escalation paths
Context Loss: Ensure feedback is properly propagated between retries
Unclear Criteria: Define specific, measurable validation criteria

Debugging Tips

Enable verbose logging to track validation flows:

from praisonaiagents import AgentFlow

workflow = AgentFlow(
    steps=[write_step, validate_step],
      # Enable detailed logging
)

result = workflow.start("Write content", )

Best Practices

Define explicit expected outputs

Validation steps need concrete criteria — vague prompts produce flaky pass/fail loops.

Cap validation cycles

Pair validation with repeat max iterations so agents do not loop indefinitely.

Use callbacks for visibility

Hook into validation events to log scores and failures for operators.

Separate generate and validate agents

A dedicated validator reduces bias compared to self-grading in the same agent.

Workflow Repeat

Repeat steps until a condition is met

Self Reflection

Agent self-improvement loops

Callbacks

Monitor validation cycles

Autonomous Workflows

Environment-based feedback loops

​How It Works

​Quick Start

​Key Features

​1. Guardrails with Automatic Retry

​2. Routing for Decision Logic

​3. Multi-Stage Validation

​Advanced Usage

​Custom Validation Logic

​Validation with External Tools

​Retry Limits and Escalation

​Validation Patterns

​1. Clear Validation Criteria

​2. Structured Feedback

​3. Progress Tracking

​Common Patterns

​1. Content Creation Pipeline

​2. Code Review Workflow

​Troubleshooting

​Common Issues

​Debugging Tips

​Best Practices

​Related