File: //opt/PraisonAI/examples/general/workflow_example_detailed.py
from praisonaiagents import Agent, Task, PraisonAIAgents
import random
from typing import List, Dict, Union
import json
from pydantic import BaseModel
# Add Pydantic models for data validation
class Person(BaseModel):
name: str
age: int
job: str
city: str
salary: int
class ProcessedPerson(Person):
salary_category: str
age_group: str
processing_status: str
class DataList(BaseModel):
items: List[Dict]
class ValidationResult(BaseModel):
validation_result: str
details: str = ""
def random_data_of_individuals() -> List[Dict]:
"""Generate random individual data"""
names = ["John", "Jane", "Mike", "Sarah", "David", "Emma"]
jobs = ["Engineer", "Doctor", "Teacher", "Artist", "Developer"]
cities = ["New York", "London", "Tokyo", "Paris", "Berlin"]
data = []
for _ in range(random.randint(3, 7)):
person = {
"name": random.choice(names),
"age": random.randint(25, 65),
"job": random.choice(jobs),
"city": random.choice(cities),
"salary": random.randint(30000, 120000)
}
data.append(person)
return data
def process_data_of_individuals(data: Union[List[Dict], Dict, str]) -> Dict:
"""Process individual data by adding categories and analysis"""
try:
print("\n[DEBUG] process_data_of_individuals input:", data)
print("[DEBUG] Current workflow state:")
print("- total_records:", workflow.get_state("total_records"))
print("- current_index:", workflow.get_state("current_index"))
print("- remaining:", workflow.get_state("remaining"))
# Get the items list from the collect_task result
collect_result = None
for task in workflow.tasks.values():
if task.name == "collect_data" and task.result:
try:
collect_data = json.loads(task.result.raw)
collect_result = collect_data.get("items", [])
print("[DEBUG] Found collect_data items:", len(collect_result))
except:
print("[DEBUG] Failed to parse collect_data result")
# Handle string input by trying to parse it as JSON
if isinstance(data, str):
if ":" in data and not data.strip().startswith("{"):
# Convert string format to dictionary
pairs = [pair.strip() for pair in data.split(",")]
data_dict = {}
for pair in pairs:
key, value = pair.split(":")
key = key.strip().lower()
value = value.strip()
if key == "age" or key == "salary":
value = int(value)
data_dict[key] = value
data = data_dict
else:
data = json.loads(data)
print("[DEBUG] Parsed data:", data)
# Handle single record
if isinstance(data, dict):
person = data
# Initialize total records if not set
total_records = workflow.get_state("total_records")
if total_records is None and collect_result:
total_records = len(collect_result)
workflow.set_state("total_records", total_records)
print(f"[DEBUG] Initialized total_records to {total_records}")
current_index = workflow.get_state("current_index", 0)
total_records = total_records or 1
remaining = total_records - (current_index + 1)
workflow.set_state("remaining", remaining)
print(f"[DEBUG] Processing record {current_index + 1}/{total_records}")
elif isinstance(data, list):
if len(data) == 0:
raise ValueError("Empty data list")
person = data[0]
workflow.set_state("total_records", len(data))
workflow.set_state("current_index", 0)
workflow.set_state("remaining", len(data) - 1)
print(f"[DEBUG] First record from list of {len(data)} items")
else:
raise ValueError("Input must be a dictionary or list with at least one record")
processed_person = person.copy()
# Add salary category
salary = person.get("salary", 0)
if salary < 50000:
processed_person["salary_category"] = "entry"
elif salary < 90000:
processed_person["salary_category"] = "mid"
else:
processed_person["salary_category"] = "senior"
# Add age group
age = person.get("age", 0)
if age < 35:
processed_person["age_group"] = "young"
elif age < 50:
processed_person["age_group"] = "mid"
else:
processed_person["age_group"] = "senior"
# Add processing status using workflow state
remaining = workflow.get_state("remaining", 0)
current_index = workflow.get_state("current_index", 0)
total_records = workflow.get_state("total_records", 1)
# Update current index for next iteration
workflow.set_state("current_index", current_index + 1)
print(f"[DEBUG] Status check - remaining: {remaining}, current_index: {current_index}, total_records: {total_records}")
if remaining <= 0 and current_index >= total_records - 1:
print("[DEBUG] Setting status to 'all records processed'")
processed_person["processing_status"] = "all records processed"
else:
print(f"[DEBUG] More records to process. Remaining: {remaining}")
processed_person["processing_status"] = f"more records to process ({remaining} remaining)"
print("[DEBUG] Final processed person:", processed_person)
return processed_person
except Exception as e:
print(f"[DEBUG] Error processing data: {str(e)}")
return {"error": str(e), "processing_status": "error occurred"}
# Create agents
data_agent = Agent(
name="DataCollector",
role="Data collection specialist",
goal="Collect and validate data about individuals",
backstory="Expert in gathering and validating demographic data",
tools=[random_data_of_individuals],
self_reflect=False
)
process_agent = Agent(
name="DataProcessor",
role="Data processor",
goal="Process and categorize individual data",
backstory="Expert in data analysis and categorization",
tools=[process_data_of_individuals],
self_reflect=False
)
# Modify tasks to use Pydantic models
collect_task = Task(
description="Collect random individual data using the random_data_of_individuals tool. Return as a JSON object with 'items' array.",
expected_output="List of individual records with basic information",
agent=data_agent,
name="collect_data",
tools=[random_data_of_individuals],
is_start=True,
next_tasks=["validate_data"],
output_json=DataList
)
validate_task = Task(
description="""Validate the collected data. Check if:
1. All required fields are present (name, age, job, city, salary)
2. Age is between 25 and 65
3. Salary is between 30000 and 120000
Return validation_result as 'valid' or 'invalid' with optional details.""",
expected_output="Validation result indicating if data is valid or invalid",
agent=data_agent,
name="validate_data",
task_type="decision",
condition={
"valid": ["process_data"],
"invalid": ["collect_data"]
},
output_json=ValidationResult
)
process_task = Task(
description="""Process one record at a time from the input data.
Current progress will be shown in Loop Status.
For the current record:
1. Use process_data_of_individuals tool to add categories
2. Return the processed record with remaining count
Current remaining: {remaining}
Current item: {current_item}
Process this record and indicate if more records remain.""",
expected_output="Processed record with categories and status",
agent=process_agent,
name="process_data",
tools=[process_data_of_individuals],
task_type="loop",
condition={
"more records to process": ["process_data"],
"all records processed": []
},
context=[collect_task],
output_json=ProcessedPerson
)
# Create PraisonAIAgents instance with workflow process
workflow = PraisonAIAgents(
agents=[data_agent, process_agent],
tasks=[collect_task, validate_task, process_task],
verbose=1,
process="workflow"
)
# Run the workflow
result = workflow.start()
# Print results
print("\nWorkflow Results:")
print("----------------")
for task_id, task in workflow.tasks.items():
print(f"\nTask: {task.name}")
print(f"Status: {task.status}")
if task.result:
print("Output:")
try:
# Try to format as pretty JSON
import json
output = json.loads(task.result.raw)
print(json.dumps(output, indent=2))
except:
# If not JSON, print raw output
print(task.result.raw[:500] + "..." if len(task.result.raw) > 500 else task.result.raw)