Task Validation & Feedback

from praisonaiagents import Agent, Task, AgentTeam

def validate_word_count(output):
    words = len(str(output.raw).split())
    if words >= 100:
        return True, output
    return False, f"Article has {words} words, expected at least 100"

writer = Agent(name="Writer", instructions="Write clear articles")

task = Task(
    description="Write a 500-word article about AI",
    agent=writer,
    guardrails=validate_word_count,
    max_retries=3,
)

AgentTeam(agents=[writer], tasks=[task]).start()

from typing import Tuple, Any

def validate_output(task_output) -> Tuple[bool, Any]:
    """
    Validate task output
    Returns: (is_valid, feedback_or_output)
    """
    content = task_output.raw
    
    # Your validation logic
    if len(content) > 100:
        return True, task_output
    else:
        return False, "Content too short, needs at least 100 characters"

task = Task(
    description="Generate detailed analysis",
    agent=agent,
    guardrails=validate_output,
    max_retries=3
)

writer = Agent(
    name="Writer",
    role="Content creator",
    goal="Write high-quality content",
    llm="gpt-4"  # Required for LLM guardrails
)

task = Task(
    description="Write a technical blog post about quantum computing",
    expected_output="Technical blog post",
    agent=writer,
    guardrails="Validate that the blog post: 1) Is technically accurate, 2) Contains at least 3 code examples, 3) Has proper introduction and conclusion sections",
    max_retries=2
)

from praisonaiagents import Agent, Task, AgentTeam

# Create agents
writer = Agent(
    name="Writer",
    role="Content creator",
    goal="Write content"
)

validator = Agent(
    name="Quality Checker",
    role="Content validator",
    goal="Ensure content meets requirements"
)

# Writing task
write_task = Task(
    name="write_article",
    description="Write a 500-word article about AI ethics",
    expected_output="500-word article",
    agent=writer,
    is_start=True,
    next_tasks=["validate_article"]
)

# Validation task (decision type)
validate_task = Task(
    name="validate_article",
    description="Check if article is exactly 500 words and covers key ethical points. Respond with 'valid' if correct, 'retry' with specific feedback if not.",
    expected_output="Validation decision with feedback",
    agent=validator,
    task_type="decision",
    condition={
        "valid": [],  # End workflow if valid
        "retry": ["write_article"]  # Retry writing if invalid
    }
)

# Create workflow with guardrails
from praisonaiagents import AgentFlow, Task

workflow = AgentFlow(
    steps=[
        Task(
            name="write",
            handler=write_handler,
            guardrails=validate_output,
            max_retries=3
        )
    ]
)

result = workflow.start("Write article")

from praisonaiagents import Agent, Task, AgentTeam

# After validation task runs
validation_task = workflow.get_task("validate_article")

# Check typed outcome (recommended)
if validation_task.validation_outcome.status == "invalid_output":
    print(f"Validation failed: {validation_task.validation_outcome.error}")
    retry_with_specific_fixes()
elif validation_task.validation_outcome.status == "success":
    proceed_to_next_step()

# Legacy dict access still works
if validation_task.validation_feedback["validation_response"] == "retry":
    handle_legacy_retry()

from praisonaiagents import Agent, Task, AgentTeam
import json

# Validation function for JSON data
def validate_json_schema(task_output) -> Tuple[bool, Any]:
    try:
        data = json.loads(task_output.raw)
        
        # Check required fields
        required_fields = ["name", "email", "age"]
        missing_fields = [f for f in required_fields if f not in data]
        
        if missing_fields:
            return False, f"Missing required fields: {', '.join(missing_fields)}"
        
        # Validate data types
        if not isinstance(data["age"], int) or data["age"] < 0:
            return False, "Age must be a positive integer"
        
        if "@" not in data["email"]:
            return False, "Invalid email format"
        
        return True, task_output
        
    except json.JSONDecodeError:
        return False, "Output is not valid JSON"

# Create data processor agent
processor = Agent(
    name="Data Processor",
    role="JSON data generator",
    goal="Generate valid user data in JSON format"
)

# Task with validation
generate_task = Task(
    description="Generate user data for John Doe, age 30, email john@example.com",
    expected_output="Valid JSON with name, email, and age fields",
    agent=processor,
    guardrails=validate_json_schema,
    max_retries=3
)

# Run pipeline
pipeline = AgentTeam(
    agents=[processor],
    tasks=[generate_task]
)

result = pipeline.start()

from praisonaiagents import Agent, Task, AgentTeam

# Create specialized agents
writer = Agent(
    name="Technical Writer",
    role="Documentation writer",
    goal="Write comprehensive technical documentation"
)

tech_reviewer = Agent(
    name="Technical Reviewer",
    role="Technical accuracy validator",
    goal="Ensure technical accuracy"
)

style_checker = Agent(
    name="Style Checker",
    role="Writing style validator",
    goal="Ensure documentation follows style guide"
)

# Writing task
write_docs = Task(
    name="write_documentation",
    description="Write API documentation for the new authentication endpoint",
    expected_output="Complete API documentation",
    agent=writer,
    is_start=True,
    next_tasks=["technical_review"]
)

# Technical validation
tech_review = Task(
    name="technical_review",
    description="Validate technical accuracy. Check: correct HTTP methods, proper authentication flow, accurate error codes. Respond 'pass' or 'rewrite' with specific issues.",
    expected_output="Technical validation result",
    agent=tech_reviewer,
    task_type="decision",
    condition={
        "pass": ["style_check"],
        "rewrite": ["write_documentation"]
    }
)

# Style validation
style_check = Task(
    name="style_check",
    description="Check documentation style. Verify: consistent formatting, clear examples, proper headings. Respond 'approved' or 'revise' with style issues.",
    expected_output="Style validation result",
    agent=style_checker,
    task_type="decision",
    condition={
        "approved": [],  # Complete
        "revise": ["write_documentation"]
    }
)

# Create validation pipeline with Workflow
from praisonaiagents import AgentFlow, Task

validation_pipeline = AgentFlow(
    steps=[
        Task(name="write", handler=write_docs_handler),
        Task(name="tech", handler=tech_review_handler, guardrails=tech_validator),
        Task(name="style", handler=style_check_handler, guardrails=style_validator)
    ]
)

result = validation_pipeline.start("Write documentation")

from praisonaiagents import Agent, Task, AgentTeam

# Validation that checks against requirements
class RequirementsValidator:
    def __init__(self, requirements):
        self.requirements = requirements
    
    def validate(self, task_output) -> Tuple[bool, Any]:
        content = task_output.raw
        missing_requirements = []
        
        for req in self.requirements:
            if req.lower() not in content.lower():
                missing_requirements.append(req)
        
        if missing_requirements:
            feedback = f"Missing requirements: {', '.join(missing_requirements)}"
            return False, feedback
        
        return True, task_output

# Create agent
analyst = Agent(
    name="Business Analyst",
    role="Requirements analyst",
    goal="Create comprehensive requirement documents"
)

# Define requirements
requirements = [
    "user authentication",
    "data encryption",
    "audit logging",
    "error handling",
    "performance metrics"
]

# Create validator instance
validator = RequirementsValidator(requirements)

# Task with custom validator
requirements_task = Task(
    description="Write security requirements for the new banking application",
    expected_output="Comprehensive security requirements document",
    agent=analyst,
    guardrails=validator.validate,
    max_retries=3
)

# Run task
agents = AgentTeam(
    agents=[analyst],
    tasks=[requirements_task]
)

result = agents.start()

# Access typed validation outcome
task.validation_outcome = AgentRunOutcome(
    status="invalid_output",    # Typed status for exhaustive matching  
    error="Article word count is 450, expected 500",
    elapsed_s=1.2,
    agent_name="quality_checker",
    context={
        "failed_criteria": ["word_count"],
        "suggestions": "Add 50 more words to meet requirement"
    }
)

# Use typed status and helper methods
if task.validation_outcome.is_retryable():
    schedule_retry(task.validation_outcome.error)

# Example of legacy validation feedback structure (still available)
task.validation_feedback = {
    "validation_response": "retry",
    "validation_details": {
        "reason": "Article word count is 450, expected 500",
        "suggestions": "Add 50 more words to meet requirement", 
        "failed_criteria": ["word_count"]
    },
    "rejected_output": "The original article content...",
    "validator_task": "validate_article",
    "validated_task": "write_article",
    "retry_count": 1
}

# Available from typed outcome's to_dict() method
typed_dict = task.validation_outcome.to_dict()

# Configure retry behavior
task = Task(
    description="Generate report",
    agent=agent,
    guardrails=validate_report,
    max_retries=3,
    retry_strategy="exponential",  # or "linear"
    retry_delay=2  # seconds between retries
)

def validate_with_detailed_feedback(task_output) -> Tuple[bool, Any]:
    # Perform validation
    issues = []
    
    if len(task_output.raw) < 100:
        issues.append("Content too short")
    
    if "conclusion" not in task_output.raw.lower():
        issues.append("Missing conclusion section")
    
    if issues:
        feedback = {
            "status": "failed",
            "issues": issues,
            "suggestions": "Please address the above issues",
            "example": "See template for reference"
        }
        return False, json.dumps(feedback)
    
    return True, task_output

def validate_length(min_words=100, max_words=500):
    def validator(task_output):
        word_count = len(task_output.raw.split())
        if min_words <= word_count <= max_words:
            return True, task_output
        else:
            return False, f"Word count {word_count} not in range {min_words}-{max_words}"
    return validator

task = Task(
    description="Write article",
    guardrails=validate_length(400, 600)
)

def validate_content_includes(required_sections):
    def validator(task_output):
        content = task_output.raw.lower()
        missing = [s for s in required_sections if s.lower() not in content]
        
        if missing:
            return False, f"Missing sections: {', '.join(missing)}"
        return True, task_output
    return validator

task = Task(
    description="Write report",
    guardrails=validate_content_includes([
        "Executive Summary",
        "Methodology",
        "Results",
        "Conclusion"
    ])
)