Outputs Module

Data classes for task outputs and reflections

Output Classes

PraisonAI provides structured output classes for handling task results and agent reflections. These classes ensure consistent data formats across all agent operations.

Import

from praisonaiagents import TaskOutput, ReflectionOutput

TaskOutput

Represents the output of a task execution, including the result, metadata, and execution details.

Class Definition

class TaskOutput:
    def __init__(
        self,
        description: str,
        result: Any,
        agent: str,
        task_id: Optional[str] = None,
        status: str = "completed",
        metadata: Optional[Dict[str, Any]] = None,
        execution_time: Optional[float] = None,
        error: Optional[str] = None
    )

Parameters

• description (str): Description of the task that was executed
• result (Any): The actual result of the task execution
• agent (str): Name of the agent that executed the task
• task_id (str, optional): Unique identifier for the task
• status (str, optional): Status of the task (“completed”, “failed”, “partial”). Defaults to “completed”
• metadata (Dict[str, Any], optional): Additional metadata about the task execution
• execution_time (float, optional): Time taken to execute the task in seconds
• error (str, optional): Error message if the task failed

Attributes

• description: Task description
• result: Task result
• agent: Agent name
• task_id: Task identifier
• status: Execution status
• metadata: Additional metadata
• execution_time: Execution duration
• error: Error message (if any)
• timestamp: Timestamp when the output was created

Methods

to_dict()

Converts the TaskOutput to a dictionary representation.

def to_dict() -> Dict[str, Any]

Returns:

• Dict[str, Any]: Dictionary containing all task output data

str()

Returns a string representation of the task output.

def __str__() -> str

Returns:

• str: Human-readable string representation

Example Usage

from praisonaiagents import TaskOutput

# Create a successful task output
output = TaskOutput(
    description="Analyze customer feedback data",
    result={
        "positive_feedback": 85,
        "negative_feedback": 15,
        "top_issues": ["delivery time", "product quality"]
    },
    agent="AnalysisAgent",
    task_id="task_001",
    status="completed",
    execution_time=3.45,
    metadata={"data_points": 1000}
)

# Access output data
print(f"Task: {output.description}")
print(f"Result: {output.result}")
print(f"Execution time: {output.execution_time}s")

# Convert to dictionary
output_dict = output.to_dict()

ReflectionOutput

Represents the output of an agent’s self-reflection process, including insights and confidence levels.

Class Definition

class ReflectionOutput:
    def __init__(
        self,
        content: str,
        agent: str,
        confidence: float = 1.0,
        insights: Optional[List[str]] = None,
        improvements: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None
    )

Parameters

• content (str): The main reflection content
• agent (str): Name of the agent performing the reflection
• confidence (float, optional): Confidence level in the reflection (0.0 to 1.0). Defaults to 1.0
• insights (List[str], optional): List of key insights from the reflection
• improvements (List[str], optional): List of suggested improvements
• metadata (Dict[str, Any], optional): Additional metadata about the reflection

Attributes

• content: Reflection content
• agent: Agent name
• confidence: Confidence level (0.0-1.0)
• insights: List of insights
• improvements: List of improvements
• metadata: Additional metadata
• timestamp: Timestamp when the reflection was created

Methods

to_dict()

Converts the ReflectionOutput to a dictionary representation.

def to_dict() -> Dict[str, Any]

Returns:

• Dict[str, Any]: Dictionary containing all reflection data

add_insight()

Adds a new insight to the reflection.

def add_insight(insight: str) -> None

Parameters:

• insight (str): The insight to add

add_improvement()

Adds a new improvement suggestion to the reflection.

def add_improvement(improvement: str) -> None

Parameters:

• improvement (str): The improvement suggestion to add

str()

Returns a string representation of the reflection output.

def __str__() -> str

Returns:

• str: Human-readable string representation

Example Usage

from praisonaiagents import ReflectionOutput

# Create a reflection output
reflection = ReflectionOutput(
    content="The analysis revealed patterns in customer behavior that were not initially apparent.",
    agent="AnalysisAgent",
    confidence=0.85,
    insights=[
        "Customers prefer morning deliveries",
        "Product bundling increases satisfaction"
    ],
    improvements=[
        "Include demographic data in future analyses",
        "Consider seasonal variations"
    ]
)

# Add additional insights
reflection.add_insight("Weekend orders have higher satisfaction rates")

# Access reflection data
print(f"Agent: {reflection.agent}")
print(f"Confidence: {reflection.confidence}")
print(f"Insights: {reflection.insights}")

# Convert to dictionary
reflection_dict = reflection.to_dict()

Integration with Agents

Both output classes integrate seamlessly with PraisonAI agents:

from praisonaiagents import Agent, Task

# Agent automatically returns TaskOutput
agent = Agent(
    name="DataAgent",
    instructions="Analyze and process data"
)

# Task execution returns TaskOutput
task = Task(
    description="Process sales data",
    agent=agent,
    expected_output="Processed data summary"
)

# Execute task and get TaskOutput
output = task.execute()
print(f"Task result: {output.result}")
print(f"Status: {output.status}")

# Agent with self-reflection returns ReflectionOutput
agent_with_reflection = Agent(
    name="ReflectiveAgent",
    instructions="Analyze data and reflect on findings",
    reflection=True
)

# Reflection outputs are automatically generated
result = agent_with_reflection.run("Analyze customer trends")
# Access reflection through agent's reflection history

Best Practices

1. Error Handling: Always check the status field in TaskOutput before using the result
2. Metadata Usage: Use metadata to store additional context that might be useful for debugging
3. Confidence Levels: Use confidence levels in ReflectionOutput to gauge reliability
4. Insights Management: Keep insights concise and actionable
5. Serialization: Use to_dict() methods for saving outputs to databases or files

Example: Complete Workflow

from praisonaiagents import Agent, Task, TaskOutput, ReflectionOutput
import json

# Create agent
agent = Agent(
    name="ResearchAgent",
    instructions="Research and analyze topics comprehensively"
)

# Execute task
task = Task(
    description="Research AI trends in 2024",
    agent=agent
)

try:
    # Execute and get output
    output = task.execute()
    
    if output.status == "completed":
        # Save successful output
        with open("task_results.json", "w") as f:
            json.dump(output.to_dict(), f, indent=2)
        
        # Create reflection on the task
        reflection = ReflectionOutput(
            content="Successfully gathered comprehensive data on AI trends",
            agent=agent.name,
            confidence=0.9,
            insights=[
                "Generative AI continues to dominate",
                "Focus shifting to practical applications"
            ],
            improvements=[
                "Include more international perspectives",
                "Add quantitative metrics"
            ]
        )
        
        # Save reflection
        with open("task_reflection.json", "w") as f:
            json.dump(reflection.to_dict(), f, indent=2)
    else:
        print(f"Task failed: {output.error}")
        
except Exception as e:
    # Create error output
    error_output = TaskOutput(
        description=task.description,
        result=None,
        agent=agent.name,
        status="failed",
        error=str(e)
    )
    print(f"Error: {error_output}")

See Also

• Task Module
• Agent Module
• Display Module
• Callbacks Documentation