Guardrails

Output validation and quality assurance for tasks

Guardrails System

Guardrails provide output validation and quality assurance for agent tasks, ensuring results meet specified criteria before being accepted.

Overview

Guardrails ensure task outputs meet quality and safety criteria through:

• Function-based validation for structured checks
• LLM-based validation for natural language criteria
• Automatic retry mechanisms for failed validations
• Custom validation logic for specific requirements

Sync vs Async Parity: Guardrails now work uniformly in both sync (.start() / .run()) and async execution paths. Previously, async execution bypassed guardrail validation.

Quick Start

Function Guardrail

from praisonaiagents import Agent, Task

# Define validation function
def validate_length(task_output):
    """Ensure output is at least 500 words"""
    word_count = len(task_output.raw.split())
    
    if word_count < 500:
        return False, f"Output too short: {word_count} words"
    
    return True, task_output

# Create task with guardrail
task = Task(
    description="Write a detailed article about AI",
    agent=writer_agent,
    guardrails=validate_length,
    expected_output="Article of at least 500 words"
)

Guardrail Types

Function-Based Guardrails

Function guardrails provide programmatic validation:

Basic Validation

def validate_json(task_output):
    """Ensure output is valid JSON"""
    import json
    
    try:
        data = json.loads(task_output.raw)
        return True, task_output
    except json.JSONDecodeError as e:
        return False, f"Invalid JSON: {str(e)}"

LLM-Based Guardrails

LLM guardrails use natural language for validation:

Simple String Guardrail

task = Task(
    description="Write a children's story",
    agent=writer_agent,
    guardrails="The story must be appropriate for ages 5-8, use simple language, and have a positive message"
)

Advanced Features

Retry Configuration

Configure retry behaviour for failed validations (works in both sync and async execution paths):

task = Task(
    description="Generate validated content",
    agent=agent,
    guardrails=validation_function,
    max_retries=3,  # Maximum retry attempts
    retry_delay=2,  # Delay between retries (seconds)
    retry_with_feedback=True  # Pass failure reason to agent
)

Execution Order: Guardrail validation → retry (if failed) or pass → memory/user callbacks → task marked completed.

How Retries Work

When a guardrail validation fails:

1. Before PR #1514: Async execution bypassed retry logic
2. After PR #1514: Both sync and async execution paths properly retry failed validations

The executor increments task.retry_count, sets task.status = "in progress", logs the retry, and continues the execution loop. On final failure after max_retries, it raises an exception. When a guardrail returns a modified TaskOutput or string, the downstream task.result and any memory callbacks receive the modified value, not the original.

Cached Field Clearing: When a guardrail modifies string output, cached fields like json_dict and pydantic are automatically cleared on TaskOutput to prevent stale cached values from being used by structured output consumers.

from praisonaiagents import Agent, Task, PraisonAIAgents

def must_mention_price(output):
    ok = "$" in output.raw
    return (ok, output if ok else "Rewrite and include a price in USD.")

agent = Agent(name="Writer", instructions="Write a one-line product blurb.")

task = Task(
    description="Write a blurb for a coffee mug.",
    agent=agent,
    guardrail=must_mention_price,
    max_retries=3,
)

PraisonAIAgents(agents=[agent], tasks=[task]).start()

Composite Guardrails

Combine multiple validation criteria:

def composite_guardrail(task_output):
    """Multiple validation checks"""
    # Check 1: Length
    if len(task_output.raw) < 100:
        return False, "Output too short"
    
    # Check 2: Format
    if not task_output.raw.strip().endswith('.'):
        return False, "Output must end with period"
    
    # Check 3: Content quality (using LLM)
    quality_check = LLMGuardrail(
        "Is this content professional and error-free?"
    )
    
    result = quality_check(task_output)
    if not result.success:
        return False, result.error
    
    return True, task_output

Guardrail Results

Access detailed validation results:

from praisonaiagents.guardrails import GuardrailResult

class CustomGuardrail:
    def __call__(self, task_output):
        # Perform validation
        if self.validate(task_output):
            return GuardrailResult(
                success=True,
                result=task_output
            )
        else:
            return GuardrailResult(
                success=False,
                error="Validation failed: specific reason",
                details={
                    "score": 0.3,
                    "issues": ["issue1", "issue2"]
                }
            )

Use Cases

Content Safety

Ensure generated content is safe and appropriate

guardrails="No offensive, harmful, or inappropriate content"

Data Validation

Validate analysis results and reports

def validate_analysis(output):
    # Check data accuracy
    # Verify calculations
    # Ensure completeness

Code Quality

Ensure generated code is safe and functional

def validate_code(output):
    # Syntax checking
    # Security scanning
    # Best practices

Compliance

Meet regulatory and policy requirements

guardrails="Must comply with GDPR, include privacy notice"

Integration Patterns

With Agents

from praisonaiagents import Agent, Task, AgentTeam

# Define guardrails
def technical_accuracy(output):
    """Validate technical content"""
    # Implementation
    pass

# Create specialized agents
writer = Agent(
    name="Technical Writer",
    description="Write accurate technical content"
)

reviewer = Agent(
    name="Technical Reviewer",
    description="Review and improve technical content"
)

# Create tasks with guardrails
tasks = [
    Task(
        description="Write Python tutorial",
        agent=writer,
        guardrails=technical_accuracy,
        expected_output="Accurate Python tutorial"
    ),
    Task(
        description="Review and enhance the tutorial",
        agent=reviewer,
        guardrails="Ensure tutorial is beginner-friendly and error-free",
        expected_output="Polished tutorial"
    )
]

# Run with validation
agents = AgentTeam(agents=[writer, reviewer], tasks=tasks)
result = agents.start()

With Dynamic Guardrails

class DynamicGuardrail:
    def __init__(self, config):
        self.min_length = config.get('min_length', 100)
        self.required_sections = config.get('sections', [])
        self.quality_threshold = config.get('quality', 0.8)
    
    def __call__(self, task_output):
        content = task_output.raw
        
        # Length check
        if len(content.split()) < self.min_length:
            return False, f"Minimum {self.min_length} words required"
        
        # Section check
        for section in self.required_sections:
            if section not in content:
                return False, f"Missing section: {section}"
        
        # Quality check (using LLM)
        quality_prompt = f"""
        Rate this content quality from 0-1:
        {content[:500]}...
        
        Consider: clarity, accuracy, completeness
        """
        
        # Implement quality scoring logic
        
        return True, task_output

# Use dynamic guardrail
guardrail = DynamicGuardrail({
    'min_length': 500,
    'sections': ['Introduction', 'Main Content', 'Conclusion'],
    'quality': 0.85
})

task = Task(
    description="Write comprehensive guide",
    agent=agent,
    guardrails=guardrail
)

Best Practices

Clear Criteria

• Define specific, measurable criteria
• Document validation requirements
• Provide helpful error messages
• Include examples of valid output

Balanced Approach

• Use function guardrails for simple checks
• Reserve LLM guardrails for complex validation
• Implement caching for repeated validations

Retry Strategy

• Set appropriate retry limits
• Provide detailed failure reasons
• Suggest corrections when possible

Testing

• Log validation failures
• Handle edge cases gracefully
• Test guardrails independently
• Verify both pass and fail cases
• Check retry behaviour
• Monitor validation performance

Complete Example

Blog Post Generator

from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents.guardrails import LLMGuardrail

# Define agents
researcher = Agent(
    name="Researcher",
    description="Research topics thoroughly and provide accurate information"
)

writer = Agent(
    name="Blog Writer",
    description="Write engaging, SEO-friendly blog posts"
)

editor = Agent(
    name="Editor",
    description="Polish and perfect blog posts"
)

# Define guardrails
def seo_validation(task_output):
    """Validate SEO requirements"""
    content = task_output.raw
    
    # Check title
    lines = content.split('\n')
    if not lines[0].startswith('#'):
        return False, "Missing H1 title"
    
    # Check meta description
    if 'Meta description:' not in content:
        return False, "Missing meta description"
    
    # Check keyword density (example: "AI" keyword)
    keyword_count = content.lower().count('ai')
    word_count = len(content.split())
    density = (keyword_count / word_count) * 100
    
    if density < 1 or density > 3:
        return False, f"Keyword density {density:.1f}% (target: 1-3%)"
    
    return True, task_output

# LLM-based quality guardrail
quality_guardrail = LLMGuardrail(
    description="""
    Evaluate if the blog post:
    1. Has a compelling introduction that hooks readers
    2. Provides valuable, actionable information
    3. Uses clear, concise language
    4. Includes relevant examples or case studies
    5. Has a strong conclusion with call-to-action
    6. Maintains consistent tone throughout
    7. Is free of factual errors
    8. Flows logically from point to point
    """,
    llm="gpt-4"
)

# Create tasks with guardrails
tasks = [
    Task(
        description="Research the topic: 'Future of AI in Healthcare'",
        agent=researcher,
        expected_output="Comprehensive research notes"
    ),
    Task(
        description="Write a 1000-word blog post based on the research",
        agent=writer,
        guardrails=seo_validation,
        expected_output="SEO-optimized blog post"
    ),
    Task(
        description="Edit and polish the blog post",
        agent=editor,
        guardrails=quality_guardrail,
        expected_output="Publication-ready blog post",
        context=[tasks[1]]  # Use previous task output
    )
]

# Run the blog generation pipeline
agents = AgentTeam(
    agents=[researcher, writer, editor],
    tasks=tasks,
    
)

# Execute with guardrails
result = agents.start()

# The system will:
# 1. Research the topic
# 2. Write the blog post (retry if SEO validation fails)
# 3. Edit the post (retry if quality standards not met)
# 4. Return the final, validated blog post

Next Steps

Learn about human-in-the-loop approvals

Explore advanced task configurations