Compare commits

...
6 Commits
Author SHA1 Message Date
Mervin Praison f991f37cfe Release v2.3.23 2025-12-17 00:41:48 +00:00
Mervin Praison a43b3ac72c Release v2.3.22 2025-12-17 00:25:14 +00:00
Mervin Praison 06f82249e1 Release v2.3.21 2025-12-17 00:05:40 +00:00
Mervin Praison d3aa78241a Release v2.3.20 2025-12-16 23:46:47 +00:00
Mervin Praison dfb77c39f9 Release v2.3.19 2025-12-16 23:26:53 +00:00
Mervin Praison 7e8f9b0574 Release v2.3.18 2025-12-16 23:18:36 +00:00
35 changed files with 6116 additions and 38 deletions

No files matched your search

+156 -1
View File
@@ -103,7 +103,20 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](#workflow-checkpoints) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |
## Supported Providers
@@ -445,6 +458,127 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
- **Async Execution**: Use `aexecute()` for async workflows
- **Planning Mode**: Enable at workflow level with `planning=True`
- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
- **Loops**: Use `loop_over` and `loop_var` to iterate over data
### Choosing the Right Workflow System
| Use Case | Recommended |
|----------|-------------|
| Simple function pipelines | `Workflow` class ⭐ |
| Agent-only pipelines | `Workflow` class |
| CSV batch processing | `process="workflow"` |
| Complex task routing | `process="workflow"` |
| Markdown templates | `WorkflowManager` |
| Early stop / conditional | `Workflow` class |
### Simple Workflow (Recommended)
The easiest way to create workflows - just pass functions as steps:
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
if not ctx.input:
return StepResult(output="No input", stop_workflow=True)
return StepResult(output=f"Valid: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.previous_result}")
# Create and run workflow
workflow = Workflow(steps=[validate, process])
result = workflow.start("Hello World", verbose=True)
print(result["output"]) # "Processed: Valid: Hello World"
```
**Key Features:**
- **Just pass functions** - No complex configuration needed
- **Early stop** - Return `stop_workflow=True` to stop the workflow
- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
- **Verbose mode** - See step-by-step progress
### Workflow Branching & Loops
```python
from praisonaiagents import WorkflowStep
# Branching step
decision_step = WorkflowStep(
name="decide",
action="Evaluate if task is complete",
next_steps=["success_step", "retry_step"],
branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
)
# Loop step
loop_step = WorkflowStep(
name="process_items",
action="Process {{item}}",
loop_over="items", # Variable containing list
loop_var="item" # Current item variable name
)
```
### Workflow Patterns (route, parallel, loop, repeat)
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Or use Pipeline (alias for Workflow)
from praisonaiagents import Pipeline
from praisonaiagents.workflows import route, parallel, loop, repeat
# 1. ROUTING - Decision-based branching
workflow = Workflow(steps=[
classify_request, # Returns "approve" or "reject"
route({
"approve": [approve_handler],
"reject": [reject_handler],
"default": [fallback_handler]
})
])
# 2. PARALLEL - Concurrent execution
workflow = Workflow(steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_results # Gets all parallel outputs
])
# 3. LOOP - Iterate over list or CSV
workflow = Workflow(
steps=[loop(process_item, over="items")],
variables={"items": ["a", "b", "c"]}
)
# Or from CSV file:
workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
# 4. REPEAT - Evaluator-Optimizer pattern
workflow = Workflow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result,
max_iterations=5
)
])
# 5. CALLBACKS - Monitor workflow execution
workflow = Workflow(
steps=[step1, step2],
on_workflow_start=lambda w, i: print(f"Starting: {i}"),
on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
)
# 6. GUARDRAILS - Validate and retry
def validate(result):
return ("error" not in result.output, "Fix the error")
workflow = Workflow(steps=[
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
])
```
### 9. Hooks
@@ -589,6 +723,12 @@ praisonai workflow run deploy --workflow-var environment=staging --workflow-var
# Execute with planning mode (AI creates sub-steps for each workflow step)
praisonai workflow run "Research Blog" --planning --verbose
# Execute with reasoning mode (chain-of-thought)
praisonai workflow run "Analysis" --reasoning --verbose
# Execute with memory enabled
praisonai workflow run "Research" --memory
# Show workflow details
praisonai workflow show deploy
@@ -600,8 +740,23 @@ praisonai "What is AI?" --workflow "Research,Summarize" --save
# Inline workflow with step actions
praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
# Workflow CLI help
praisonai workflow help
```
**Workflow CLI Options:**
| Flag | Description |
|------|-------------|
| `--workflow-var key=value` | Set workflow variable (can be repeated) |
| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
| `--planning` | Enable planning mode |
| `--reasoning` | Enable reasoning mode |
| `--memory` | Enable memory |
| `--verbose` | Enable verbose output |
| `--save` | Save output to file |
### Hooks CLI:
```bash
# List configured hooks
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
flask \
"praisonai>=2.3.16" \
"praisonai>=2.3.23" \
"praisonai[api]" \
gunicorn \
markdown
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.16" \
"praisonai>=2.3.23" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.16" \
"praisonai>=2.3.23" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.16" \
"praisonai>=2.3.23" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -29,7 +29,7 @@
},
{
"cell_type": "code",
"execution_count": 1,
"execution_count": null,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
@@ -71,7 +71,6 @@
"import pandas as pd\n",
"\n",
"print(\"📦 Dependencies installed!\")\n",
"print(\"🔄 Converting from AGNO → PraisonAI\")\n",
"print(\"🎯 Building Job Hunting Agent System\")"
]
},
@@ -0,0 +1,36 @@
"""
Simple Workflow Example
The easiest way to create and run workflows in PraisonAI.
Just pass functions or agents as steps - no complex configuration needed!
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
"""Validate the input data."""
if not ctx.input:
return StepResult(output="No input provided", stop_workflow=True)
return StepResult(output=f"Validated: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
"""Process the validated data."""
return StepResult(output=f"Processed: {ctx.previous_result}")
def format_output(ctx: WorkflowContext) -> StepResult:
"""Format the final output."""
return StepResult(output=f"✅ Final: {ctx.previous_result}")
# Create workflow - just list your functions!
workflow = Workflow(
name="Simple Pipeline",
steps=[validate, process, format_output]
)
if __name__ == "__main__":
# Run the workflow - PraisonAI style uses .start()
result = workflow.start("Hello, World!", verbose=True)
print(f"\nFinal output: {result['output']}")
print(f"Steps completed: {len(result['steps'])}")
@@ -0,0 +1,57 @@
"""
Task Callbacks Example
Demonstrates using on_task_start and on_task_complete callbacks
with PraisonAIAgents for workflow monitoring and logging.
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
# Create an agent
researcher = Agent(
name="Researcher",
role="Research Assistant",
goal="Find and summarize information",
llm="gpt-4o-mini"
)
# Create tasks with variables
task1 = Task(
name="research",
description="Research the topic: {{topic}}",
expected_output="A brief summary of the topic",
agent=researcher,
variables={"topic": "artificial intelligence"}
)
task2 = Task(
name="summarize",
description="Create a one-paragraph summary of the research.",
expected_output="A concise summary",
agent=researcher
)
# Define callbacks
def on_start(task, task_id):
"""Called before each task starts."""
print(f"🚀 Starting task: {task.name} (ID: {task_id})")
def on_complete(task, output):
"""Called after each task completes."""
print(f"✅ Completed task: {task.name}")
print(f" Output preview: {str(output)[:100]}...")
if __name__ == "__main__":
# Create agents with callbacks
agents = PraisonAIAgents(
agents=[researcher],
tasks=[task1, task2],
process="workflow",
on_task_start=on_start,
on_task_complete=on_complete,
variables={"global_var": "shared_value"} # Global variables
)
# Run the workflow
result = agents.start()
print("\n🎉 Workflow completed!")
@@ -0,0 +1,48 @@
"""
Workflow Branching Example
Demonstrates conditional branching in workflows where steps can route
to different next steps based on output content.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a workflow with branching
workflow = Workflow(
name="Decision Workflow",
description="A workflow that branches based on validation result",
steps=[
WorkflowStep(
name="validate",
action="Check if the number 42 is positive. Reply with 'valid' or 'invalid'.",
next_steps=["success_handler", "error_handler"],
branch_condition={
"valid": ["success_handler"],
"invalid": ["error_handler"]
}
),
WorkflowStep(
name="success_handler",
action="The validation passed! Generate a success message."
),
WorkflowStep(
name="error_handler",
action="The validation failed. Generate an error message."
)
]
)
if __name__ == "__main__":
# Create manager and register workflow
manager = WorkflowManager()
manager.workflows["Decision Workflow"] = workflow
# Execute - will branch to success_handler since 42 is positive
result = manager.execute(
"Decision Workflow",
default_llm="gpt-4o-mini"
)
print("Workflow completed!")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
@@ -0,0 +1,63 @@
"""
Workflow Checkpoints Example
Demonstrates saving and resuming workflow execution using checkpoints.
Useful for long-running workflows that may be interrupted.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a multi-step workflow
workflow = Workflow(
name="Long Process",
description="A workflow with checkpoints for resumability",
steps=[
WorkflowStep(
name="step1",
action="Initialize the process and prepare data."
),
WorkflowStep(
name="step2",
action="Process the first batch of data."
),
WorkflowStep(
name="step3",
action="Process the second batch of data."
),
WorkflowStep(
name="step4",
action="Finalize and generate report."
)
]
)
if __name__ == "__main__":
manager = WorkflowManager()
manager.workflows["Long Process"] = workflow
# Execute with checkpoint - saves after each step
print("=== Starting workflow with checkpoint ===")
result = manager.execute(
"Long Process",
default_llm="gpt-4o-mini",
checkpoint="my-checkpoint" # Saves progress after each step
)
print(f"Completed {len(result['results'])} steps")
# List available checkpoints
print("\n=== Available checkpoints ===")
checkpoints = manager.list_checkpoints()
for cp in checkpoints:
print(f" {cp['name']}: {cp['completed_steps']} steps completed")
# Resume from checkpoint (if workflow was interrupted)
# result = manager.execute(
# "Long Process",
# default_llm="gpt-4o-mini",
# resume="my-checkpoint" # Resumes from last saved state
# )
# Clean up checkpoint
manager.delete_checkpoint("my-checkpoint")
print("\n✅ Checkpoint deleted")
@@ -0,0 +1,68 @@
"""
Workflow Conditional Execution Example
Demonstrates using should_run to conditionally execute steps
based on input or previous results.
"""
from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult
# Condition functions - return True to run the step, False to skip
def is_sensitive_content(ctx: WorkflowContext) -> bool:
"""Only run compliance check for sensitive topics."""
sensitive_keywords = ["legal", "medical", "financial", "security"]
return any(keyword in ctx.input.lower() for keyword in sensitive_keywords)
def needs_translation(ctx: WorkflowContext) -> bool:
"""Only translate if content is not in English."""
return ctx.variables.get("language", "en") != "en"
# Handler functions
def process_content(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.input}")
def compliance_check(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"✅ Compliance verified for: {ctx.previous_result}")
def translate(ctx: WorkflowContext) -> StepResult:
lang = ctx.variables.get("language", "unknown")
return StepResult(output=f"🌐 Translated to {lang}: {ctx.previous_result}")
def finalize(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"📄 Final: {ctx.previous_result}")
# Create workflow with conditional steps
workflow = Workflow(
name="Conditional Pipeline",
steps=[
WorkflowStep(name="process", handler=process_content),
WorkflowStep(
name="compliance",
handler=compliance_check,
should_run=is_sensitive_content # Only runs for sensitive content
),
WorkflowStep(
name="translate",
handler=translate,
should_run=needs_translation # Only runs if language != "en"
),
WorkflowStep(name="finalize", handler=finalize)
]
)
if __name__ == "__main__":
# Test 1: Normal content (skips compliance)
print("=== Test 1: Normal content ===")
result = workflow.start("Hello world", verbose=True)
print(f"Output: {result['output']}\n")
# Test 2: Sensitive content (runs compliance)
print("=== Test 2: Legal content ===")
result = workflow.start("Review this legal document", verbose=True)
print(f"Output: {result['output']}\n")
# Test 3: Non-English content (runs translation)
print("=== Test 3: Non-English content ===")
workflow.variables = {"language": "es"}
result = workflow.start("Hola mundo", verbose=True)
print(f"Output: {result['output']}")
@@ -0,0 +1,69 @@
"""
Workflow Early Stop Example
Demonstrates how to stop a workflow early using custom handler
functions that return StepResult with stop_workflow=True.
"""
from praisonaiagents.memory.workflows import (
WorkflowManager, WorkflowStep, Workflow,
WorkflowContext, StepResult
)
# Custom validator that can stop the workflow
def validate_data(context: WorkflowContext) -> StepResult:
"""Check if data is valid and stop workflow if not."""
data = context.variables.get("data", {})
if data.get("value", 0) < 0:
return StepResult(
output="❌ Validation failed: negative value detected. Stopping workflow.",
stop_workflow=True # This stops the entire workflow
)
return StepResult(
output="✅ Validation passed: data is valid.",
stop_workflow=False # Continue to next step
)
# Create workflow with early stop capability
workflow = Workflow(
name="Data Processing",
description="Process data with validation gate",
variables={
"data": {"value": -5} # Invalid data - will trigger early stop
},
steps=[
WorkflowStep(
name="validate",
handler=validate_data # Custom function
),
WorkflowStep(
name="process",
action="Process the validated data." # Won't run if validation fails
),
WorkflowStep(
name="report",
action="Generate final report." # Won't run if validation fails
)
]
)
if __name__ == "__main__":
manager = WorkflowManager()
manager.workflows["Data Processing"] = workflow
# Test with invalid data (will stop early)
print("=== Testing with invalid data ===")
result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['output']}")
# Test with valid data (will complete)
print("\n=== Testing with valid data ===")
workflow.variables["data"] = {"value": 42}
result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
@@ -0,0 +1,55 @@
"""
Workflow Loop with CSV Example
Demonstrates iterating over a CSV file, processing each row.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import loop
import tempfile
import os
# Create sample CSV file
csv_content = """name,email,task
Alice,[email protected],Write documentation
Bob,[email protected],Review pull requests
Charlie,[email protected],Fix critical bug
Diana,[email protected],Deploy to production"""
# Processor for each CSV row
def process_task(ctx: WorkflowContext) -> StepResult:
"""Process a single task from CSV."""
item = ctx.variables.get("item", {})
name = item.get("name", "Unknown")
task = item.get("task", "No task")
index = ctx.variables.get("loop_index", 0)
return StepResult(
output=f"[{index+1}] {name}: {task} ✅",
variables={"last_processed": name}
)
if __name__ == "__main__":
# Create temp CSV file
with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
f.write(csv_content)
csv_path = f.name
try:
print("=== Testing Loop with CSV ===\n")
print(f"CSV file: {csv_path}\n")
# Create workflow
workflow = Workflow(
name="CSV Task Processor",
steps=[loop(process_task, from_csv=csv_path)]
)
result = workflow.start("Process all tasks", verbose=True)
print(f"\nAll outputs:\n{result['output']}")
print(f"\nProcessed {len(result['variables']['loop_outputs'])} items")
print(f"Last processed: {result['variables'].get('last_processed')}")
finally:
os.unlink(csv_path)
@@ -0,0 +1,52 @@
"""
Workflow Loop with List Example
Demonstrates iterating over a list of items, processing each one.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import loop
# Sample data
fruits = ["apple", "banana", "cherry", "date", "elderberry"]
# Processor for each item
def process_fruit(ctx: WorkflowContext) -> StepResult:
"""Process a single fruit from the list."""
fruit = ctx.variables.get("item", "unknown")
index = ctx.variables.get("loop_index", 0)
# Simulate processing
result = f"🍎 [{index + 1}] {fruit.upper()} - processed successfully"
return StepResult(
output=result,
variables={"last_fruit": fruit}
)
# Aggregator to summarize results
def summarize(ctx: WorkflowContext) -> StepResult:
"""Summarize all processed fruits."""
outputs = ctx.variables.get("loop_outputs", [])
summary = f"📊 Processed {len(outputs)} fruits:\n"
summary += "\n".join(f" {o}" for o in outputs)
return StepResult(output=summary)
# Create workflow
workflow = Workflow(
name="Fruit Processor",
steps=[
loop(process_fruit, over="fruits"),
summarize
],
variables={"fruits": fruits}
)
if __name__ == "__main__":
print("=== Testing Loop with List ===\n")
result = workflow.start("Process all fruits", verbose=True)
print(f"\nFinal Output:\n{result['output']}")
print(f"\nLoop outputs count: {len(result['variables']['loop_outputs'])}")
print(f"Last processed: {result['variables'].get('last_fruit')}")
@@ -0,0 +1,46 @@
"""
Workflow Loops Example
Demonstrates iterating over a list of items in a workflow step.
Each item is processed individually with the loop variable.
"""
from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
# Create a workflow with loop
workflow = Workflow(
name="Process Items",
description="Process a list of items using loop",
variables={
"items": ["apple", "banana", "cherry"] # List to iterate over
},
steps=[
WorkflowStep(
name="process_each",
action="Describe the fruit: {{item}}",
loop_over="items", # Variable name containing the list
loop_var="item" # Variable name for current item
),
WorkflowStep(
name="summarize",
action="Summarize all the fruits that were processed."
)
]
)
if __name__ == "__main__":
# Create manager and register workflow
manager = WorkflowManager()
manager.workflows["Process Items"] = workflow
# Execute - will process each fruit in the list
result = manager.execute(
"Process Items",
default_llm="gpt-4o-mini"
)
print("Workflow completed!")
for step_result in result["results"]:
print(f" {step_result['step']}: {step_result['status']}")
if isinstance(step_result.get("output"), list):
print(f" Loop processed {len(step_result['output'])} items")
@@ -0,0 +1,68 @@
"""
Mixed Steps Workflow Example
Demonstrates combining different step types in a single workflow:
- Agent instances
- Handler functions
- WorkflowStep with action strings
"""
from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult, Agent
# Create an agent
analyzer = Agent(
name="Analyzer",
role="Data Analyst",
goal="Analyze and summarize data",
llm="gpt-4o-mini"
)
# Custom handler functions
def validate_input(ctx: WorkflowContext) -> StepResult:
"""Validate the input before processing."""
if len(ctx.input) < 5:
return StepResult(
output="Input too short. Minimum 5 characters required.",
stop_workflow=True
)
return StepResult(output=f"✅ Valid input: {ctx.input}")
def format_output(ctx: WorkflowContext) -> StepResult:
"""Format the final output."""
return StepResult(
output=f"📊 Report:\n{ctx.previous_result}",
variables={"report_generated": True}
)
# Create workflow with mixed step types
workflow = Workflow(
name="Mixed Pipeline",
steps=[
# Step 1: Function - validates input
validate_input,
# Step 2: Agent - analyzes content
analyzer,
# Step 3: WorkflowStep with action string
WorkflowStep(
name="enhance",
action="Add a conclusion to this analysis: {{previous_output}}"
),
# Step 4: Function - formats output
format_output
],
default_llm="gpt-4o-mini" # Used for action-based steps
)
if __name__ == "__main__":
# Test with valid input
print("=== Running mixed workflow ===")
result = workflow.start(
"Analyze the trends in AI adoption for 2024",
verbose=True
)
print(f"\nFinal output:\n{result['output']}")
print(f"\nVariables: {result['variables']}")
@@ -0,0 +1,53 @@
"""
Workflow Parallel Execution Example
Demonstrates running multiple steps concurrently and
combining their results.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import parallel
import time
# Parallel workers - each does independent work
def research_market(ctx: WorkflowContext) -> StepResult:
"""Research market trends."""
time.sleep(0.1) # Simulate work
return StepResult(output="📊 Market: Growth expected 15% YoY")
def research_competitors(ctx: WorkflowContext) -> StepResult:
"""Research competitor analysis."""
time.sleep(0.1) # Simulate work
return StepResult(output="🏢 Competitors: 3 major players identified")
def research_customers(ctx: WorkflowContext) -> StepResult:
"""Research customer feedback."""
time.sleep(0.1) # Simulate work
return StepResult(output="👥 Customers: 85% satisfaction rate")
# Aggregator - combines parallel results
def summarize_research(ctx: WorkflowContext) -> StepResult:
"""Summarize all research findings."""
outputs = ctx.variables.get("parallel_outputs", [])
summary = "📋 RESEARCH SUMMARY:\n" + "\n".join(f" • {o}" for o in outputs)
return StepResult(output=summary)
# Create workflow with parallel execution
workflow = Workflow(
name="Parallel Research",
steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_research
]
)
if __name__ == "__main__":
print("=== Testing Parallel Workflow ===\n")
start_time = time.time()
result = workflow.start("Analyze the business landscape", verbose=True)
elapsed = time.time() - start_time
print(f"\nFinal Output:\n{result['output']}")
print(f"\nCompleted in {elapsed:.2f}s (parallel execution)")
print(f"Individual outputs: {len(result['variables']['parallel_outputs'])}")
@@ -0,0 +1,69 @@
"""
Workflow Repeat (Evaluator-Optimizer) Example
Demonstrates repeating a step until a condition is met.
This is useful for iterative improvement patterns.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import repeat
# Simulated content generator that improves each iteration
class ContentGenerator:
def __init__(self):
self.iteration = 0
self.content = []
def generate(self, ctx: WorkflowContext) -> StepResult:
"""Generate content, adding more each iteration."""
self.iteration += 1
# Add new content each iteration
new_items = [
f"Point {self.iteration}.1: Key insight about the topic",
f"Point {self.iteration}.2: Supporting evidence"
]
self.content.extend(new_items)
output = f"Iteration {self.iteration}: Generated {len(self.content)} points total"
return StepResult(
output=output,
variables={
"total_points": len(self.content),
"content": self.content.copy()
}
)
# Condition: stop when we have enough content
def has_enough_content(ctx: WorkflowContext) -> bool:
"""Check if we have generated enough content."""
total = ctx.variables.get("total_points", 0)
return total >= 6 # Stop when we have 6+ points
# Create generator instance
generator = ContentGenerator()
# Create workflow with repeat pattern
workflow = Workflow(
name="Content Generator",
steps=[
repeat(
generator.generate,
until=has_enough_content,
max_iterations=10
)
]
)
if __name__ == "__main__":
print("=== Testing Repeat (Evaluator-Optimizer) Pattern ===\n")
result = workflow.start("Generate comprehensive content", verbose=True)
print(f"\nFinal output: {result['output']}")
print(f"Total iterations: {result['variables']['repeat_iterations']}")
print(f"Total points generated: {result['variables']['total_points']}")
print("\nGenerated content:")
for point in result['variables']['content']:
print(f" • {point}")
@@ -0,0 +1,61 @@
"""
Workflow Routing Example
Demonstrates decision-based routing where the workflow
takes different paths based on the output of a decision step.
"""
from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.workflows import route
# Decision maker - determines which route to take
def classify_request(ctx: WorkflowContext) -> StepResult:
"""Classify the input and decide the route."""
input_lower = ctx.input.lower()
if "urgent" in input_lower or "emergency" in input_lower:
return StepResult(output="priority: high")
elif "question" in input_lower or "help" in input_lower:
return StepResult(output="priority: support")
else:
return StepResult(output="priority: normal")
# Route handlers
def handle_high_priority(ctx: WorkflowContext) -> StepResult:
return StepResult(output="🚨 HIGH PRIORITY: Escalating to senior team immediately!")
def handle_support(ctx: WorkflowContext) -> StepResult:
return StepResult(output="💬 SUPPORT: Routing to help desk for assistance.")
def handle_normal(ctx: WorkflowContext) -> StepResult:
return StepResult(output="📋 NORMAL: Added to standard processing queue.")
# Create workflow with routing
workflow = Workflow(
name="Request Router",
steps=[
classify_request,
route({
"high": [handle_high_priority],
"support": [handle_support],
"normal": [handle_normal],
"default": [handle_normal]
})
]
)
if __name__ == "__main__":
# Test different inputs
print("=== Testing Routing Workflow ===\n")
# Test 1: Urgent request
result = workflow.start("This is an URGENT matter!", verbose=True)
print(f"Result: {result['output']}\n")
# Test 2: Support request
result = workflow.start("I have a question about my account", verbose=True)
print(f"Result: {result['output']}\n")
# Test 3: Normal request
result = workflow.start("Please process my order", verbose=True)
print(f"Result: {result['output']}")
@@ -0,0 +1,45 @@
"""
Workflow with Agents Example
Demonstrates using Agent instances directly as workflow steps.
Agents are automatically wrapped and executed in sequence.
"""
from praisonaiagents import Workflow, Agent
# Create agents
researcher = Agent(
name="Researcher",
role="Research Specialist",
goal="Find accurate information",
llm="gpt-4o-mini"
)
writer = Agent(
name="Writer",
role="Content Writer",
goal="Write engaging content",
llm="gpt-4o-mini"
)
editor = Agent(
name="Editor",
role="Content Editor",
goal="Polish and improve content",
llm="gpt-4o-mini"
)
# Create workflow with agents as steps
workflow = Workflow(
name="Content Pipeline",
steps=[researcher, writer, editor]
)
if __name__ == "__main__":
# Run the workflow
result = workflow.start(
"Write a short paragraph about artificial intelligence",
verbose=True
)
print(f"\nFinal output:\n{result['output']}")
@@ -54,6 +54,13 @@ except ImportError:
MCP = None
from .session import Session
from .memory.memory import Memory
# Workflows - import from dedicated workflows module
from .workflows import (
Workflow, WorkflowStep, WorkflowContext, StepResult,
Route, Parallel, Loop, Repeat,
route, parallel, loop, repeat,
Pipeline # Alias for Workflow
)
from .guardrails import GuardrailResult, LLMGuardrail
# Fast Context support (lazy loaded to avoid performance impact)
@@ -48,17 +48,58 @@ def __getattr__(name):
if name == "create_auto_memory":
from .auto_memory import create_auto_memory
return create_auto_memory
# Workflows - now in dedicated module, re-export for backward compatibility
if name == "WorkflowManager":
from .workflows import WorkflowManager
from ..workflows import WorkflowManager
return WorkflowManager
if name == "Workflow":
from .workflows import Workflow
from ..workflows import Workflow
return Workflow
if name == "WorkflowStep":
from .workflows import WorkflowStep
from ..workflows import WorkflowStep
return WorkflowStep
if name == "WorkflowContext":
from ..workflows import WorkflowContext
return WorkflowContext
if name == "StepResult":
from ..workflows import StepResult
return StepResult
if name == "Pipeline":
from ..workflows import Pipeline
return Pipeline
if name == "Route":
from ..workflows import Route
return Route
if name == "Parallel":
from ..workflows import Parallel
return Parallel
if name == "Loop":
from ..workflows import Loop
return Loop
if name == "Repeat":
from ..workflows import Repeat
return Repeat
if name == "route":
from ..workflows import route
return route
if name == "parallel":
from ..workflows import parallel
return parallel
if name == "loop":
from ..workflows import loop
return loop
if name == "repeat":
from ..workflows import repeat
return repeat
# Backward compatibility aliases
if name == "StepInput":
from ..workflows import WorkflowContext
return WorkflowContext
if name == "StepOutput":
from ..workflows import StepResult
return StepResult
if name == "create_workflow_manager":
from .workflows import create_workflow_manager
from ..workflows.workflows import create_workflow_manager
return create_workflow_manager
if name == "HooksManager":
from .hooks import HooksManager
@@ -85,11 +126,23 @@ __all__ = [
"AutoMemory",
"AutoMemoryExtractor",
"create_auto_memory",
# Workflows
# Workflows (re-exported from praisonaiagents.workflows for backward compatibility)
"WorkflowManager",
"Workflow",
"Pipeline",
"WorkflowStep",
"WorkflowContext",
"StepResult",
"create_workflow_manager",
# Workflow patterns
"Route",
"Parallel",
"Loop",
"Repeat",
"route",
"parallel",
"loop",
"repeat",
# Hooks
"HooksManager",
"HookResult",
@@ -29,13 +29,166 @@ from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class WorkflowContext:
"""Context passed to step handlers. Contains all information about the current workflow state."""
input: str = "" # Original workflow input
previous_result: Optional[str] = None # Output from previous step
current_step: str = "" # Current step name
variables: Dict[str, Any] = field(default_factory=dict) # All workflow variables
@dataclass
class StepResult:
"""Result returned from step handlers."""
output: str = "" # Step output content
stop_workflow: bool = False # If True, stop the entire workflow early
variables: Dict[str, Any] = field(default_factory=dict) # Variables to add/update
# Aliases for backward compatibility
StepInput = WorkflowContext
StepOutput = StepResult
# =============================================================================
# Workflow Pattern Helpers
# =============================================================================
@dataclass
class Route:
"""
Decision-based branching. Routes to different steps based on previous output.
Usage:
workflow = Workflow(steps=[
decision_maker,
route({
"approve": [approve_step],
"reject": [reject_step],
"default": [fallback_step]
})
])
"""
routes: Dict[str, List] = field(default_factory=dict)
default: Optional[List] = None
def __init__(self, routes: Dict[str, List], default: Optional[List] = None):
self.routes = routes
self.default = default or routes.get("default", [])
@dataclass
class Parallel:
"""
Execute multiple steps concurrently and combine results.
Usage:
workflow = Workflow(steps=[
parallel([agent1, agent2, agent3]),
aggregator
])
"""
steps: List = field(default_factory=list)
def __init__(self, steps: List):
self.steps = steps
@dataclass
class Loop:
"""
Iterate over a list or CSV file, executing step for each item.
Usage:
# Loop over list variable
workflow = Workflow(
steps=[loop(processor, over="items")],
variables={"items": ["a", "b", "c"]}
)
# Loop over CSV file
workflow = Workflow(steps=[
loop(processor, from_csv="data.csv")
])
"""
step: Any = None
over: Optional[str] = None # Variable name containing list
from_csv: Optional[str] = None # CSV file path
from_file: Optional[str] = None # Text file path (one item per line)
var_name: str = "item" # Variable name for current item
def __init__(
self,
step: Any,
over: Optional[str] = None,
from_csv: Optional[str] = None,
from_file: Optional[str] = None,
var_name: str = "item"
):
self.step = step
self.over = over
self.from_csv = from_csv
self.from_file = from_file
self.var_name = var_name
@dataclass
class Repeat:
"""
Repeat a step until a condition is met (evaluator-optimizer pattern).
Usage:
workflow = Workflow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result.lower(),
max_iterations=5
)
])
"""
step: Any = None
until: Optional[Callable[[WorkflowContext], bool]] = None
max_iterations: int = 10
def __init__(
self,
step: Any,
until: Optional[Callable[[WorkflowContext], bool]] = None,
max_iterations: int = 10
):
self.step = step
self.until = until
self.max_iterations = max_iterations
# Convenience functions for cleaner API
def route(routes: Dict[str, List], default: Optional[List] = None) -> Route:
"""Create a routing decision point."""
return Route(routes=routes, default=default)
def parallel(steps: List) -> Parallel:
"""Execute steps in parallel."""
return Parallel(steps=steps)
def loop(step: Any, over: Optional[str] = None, from_csv: Optional[str] = None,
from_file: Optional[str] = None, var_name: str = "item") -> Loop:
"""Loop over items executing step for each."""
return Loop(step=step, over=over, from_csv=from_csv, from_file=from_file, var_name=var_name)
def repeat(step: Any, until: Optional[Callable[[WorkflowContext], bool]] = None,
max_iterations: int = 10) -> Repeat:
"""Repeat step until condition is met."""
return Repeat(step=step, until=until, max_iterations=max_iterations)
@dataclass
class WorkflowStep:
"""A single step in a workflow."""
name: str
description: str = ""
action: str = "" # The action/prompt to execute
condition: Optional[str] = None # Optional condition for execution
condition: Optional[str] = None # Optional condition for execution (string)
should_run: Optional[Callable[['WorkflowContext'], bool]] = None # Function to check if step should run
handler: Optional[Callable[['WorkflowContext'], 'StepResult']] = None # Custom function instead of agent
on_error: Literal["stop", "continue", "retry"] = "stop"
max_retries: int = 1
@@ -46,6 +199,7 @@ class WorkflowStep:
# Agent configuration fields
agent_config: Optional[Dict[str, Any]] = None # Per-step agent config {role, goal, backstory, llm}
agent: Optional[Any] = None # Direct agent instance for this step
tools: Optional[List[Any]] = None # Tools for this step
# Branching fields
@@ -56,12 +210,24 @@ class WorkflowStep:
loop_over: Optional[str] = None # Variable name to iterate over (e.g., "items")
loop_var: str = "item" # Variable name for current item in loop
# Backward compatibility aliases
@property
def evaluator(self):
return self.should_run
@property
def executor(self):
return self.handler
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"action": self.action,
"condition": self.condition,
"should_run": self.should_run is not None,
"handler": self.handler is not None,
"agent": self.agent is not None,
"on_error": self.on_error,
"max_retries": self.max_retries,
"context_from": self.context_from,
@@ -79,9 +245,9 @@ class WorkflowStep:
@dataclass
class Workflow:
"""A complete workflow with multiple steps."""
name: str
name: str = "Workflow"
description: str = ""
steps: List[WorkflowStep] = field(default_factory=list)
steps: List = field(default_factory=list) # Can be WorkflowStep, Agent, or function
variables: Dict[str, Any] = field(default_factory=dict)
file_path: Optional[str] = None
@@ -105,6 +271,516 @@ class Workflow:
"planning": self.planning,
"planning_llm": self.planning_llm
}
def run(
self,
input: str = "",
llm: Optional[str] = None,
verbose: bool = False
) -> Dict[str, Any]:
"""
Run the workflow with the given input.
This is the simplest way to execute a workflow:
```python
from praisonaiagents import Workflow, Agent
workflow = Workflow(
steps=[
Agent(name="Writer", role="Write content"),
my_validator_function,
Agent(name="Editor", role="Edit content"),
]
)
result = workflow.run("Write about AI")
```
Args:
input: The input text/prompt for the workflow
llm: LLM model to use (default: gpt-4o-mini)
verbose: Print step outputs
Returns:
Dict with 'output' (final result) and 'steps' (all step results)
"""
# Use default LLM if not specified
model = llm or self.default_llm or "gpt-4o-mini"
# Add input to variables
all_variables = {**self.variables, "input": input}
results = []
previous_output = None
# Process steps (may include Route, Parallel, Loop, Repeat)
i = 0
while i < len(self.steps):
step = self.steps[i]
# Handle special pattern types
if isinstance(step, Route):
# Decision-based routing
route_result = self._execute_route(
step, previous_output, input, all_variables, model, verbose
)
results.extend(route_result["steps"])
previous_output = route_result["output"]
all_variables.update(route_result.get("variables", {}))
i += 1
continue
elif isinstance(step, Parallel):
# Parallel execution
parallel_result = self._execute_parallel(
step, previous_output, input, all_variables, model, verbose
)
results.extend(parallel_result["steps"])
previous_output = parallel_result["output"]
all_variables.update(parallel_result.get("variables", {}))
i += 1
continue
elif isinstance(step, Loop):
# Loop over items
loop_result = self._execute_loop(
step, previous_output, input, all_variables, model, verbose
)
results.extend(loop_result["steps"])
previous_output = loop_result["output"]
all_variables.update(loop_result.get("variables", {}))
i += 1
continue
elif isinstance(step, Repeat):
# Repeat until condition
repeat_result = self._execute_repeat(
step, previous_output, input, all_variables, model, verbose
)
results.extend(repeat_result["steps"])
previous_output = repeat_result["output"]
all_variables.update(repeat_result.get("variables", {}))
i += 1
continue
# Normalize single step
step = self._normalize_single_step(step, i)
# Create context for handlers
context = WorkflowContext(
input=input,
previous_result=str(previous_output) if previous_output else None,
current_step=step.name,
variables=all_variables.copy()
)
# Check should_run condition
if step.should_run:
try:
if not step.should_run(context):
if verbose:
print(f"⏭️ Skipped: {step.name}")
i += 1
continue
except Exception as e:
logger.error(f"should_run failed for {step.name}: {e}")
# Execute step
output = None
stop = False
if step.handler:
# Custom handler function
try:
result = step.handler(context)
if isinstance(result, StepResult):
output = result.output
stop = result.stop_workflow
if result.variables:
all_variables.update(result.variables)
else:
output = str(result)
except Exception as e:
output = f"Error: {e}"
elif step.agent:
# Direct agent
try:
output = step.agent.chat(step.action or input)
except Exception as e:
output = f"Error: {e}"
elif step.action:
# Action with agent_config - create temporary agent
try:
from ..agent.agent import Agent
config = step.agent_config or self.default_agent_config or {}
temp_agent = Agent(
name=config.get("name", step.name),
role=config.get("role", "Assistant"),
goal=config.get("goal", "Complete the task"),
llm=config.get("llm", model),
verbose=verbose
)
# Substitute variables in action
action = step.action
for key, value in all_variables.items():
action = action.replace(f"{{{{{key}}}}}", str(value))
if previous_output:
action = action.replace("{{previous_output}}", str(previous_output))
output = temp_agent.chat(action)
except Exception as e:
output = f"Error: {e}"
# Store result
results.append({
"step": step.name,
"output": output
})
previous_output = output
if verbose:
print(f"✅ {step.name}: {str(output)[:100]}...")
# Handle early stop
if stop:
if verbose:
print(f"🛑 Workflow stopped at: {step.name}")
break
# Store output in variables
var_name = step.output_variable or f"{step.name}_output"
all_variables[var_name] = output
i += 1
return {
"output": previous_output,
"steps": results,
"variables": all_variables
}
def _normalize_steps(self) -> List['WorkflowStep']:
"""Convert mixed steps (Agent, function, WorkflowStep) to WorkflowStep list."""
normalized = []
for i, step in enumerate(self.steps):
if isinstance(step, WorkflowStep):
normalized.append(step)
elif callable(step):
# It's a function - wrap as handler
normalized.append(WorkflowStep(
name=getattr(step, '__name__', f'step_{i+1}'),
handler=step
))
elif hasattr(step, 'chat'):
# It's an Agent - wrap with agent reference
normalized.append(WorkflowStep(
name=getattr(step, 'name', f'agent_{i+1}'),
agent=step,
action="{{input}}"
))
else:
# Unknown type - try to use as string action
normalized.append(WorkflowStep(
name=f'step_{i+1}',
action=str(step)
))
return normalized
def _normalize_single_step(self, step: Any, index: int) -> 'WorkflowStep':
"""Normalize a single step to WorkflowStep."""
if isinstance(step, WorkflowStep):
return step
elif callable(step):
return WorkflowStep(
name=getattr(step, '__name__', f'step_{index+1}'),
handler=step
)
elif hasattr(step, 'chat'):
return WorkflowStep(
name=getattr(step, 'name', f'agent_{index+1}'),
agent=step,
action="{{input}}"
)
else:
return WorkflowStep(
name=f'step_{index+1}',
action=str(step)
)
def _execute_single_step_internal(
self,
step: Any,
previous_output: Optional[str],
input: str,
all_variables: Dict[str, Any],
model: str,
verbose: bool,
index: int = 0
) -> Dict[str, Any]:
"""Execute a single step and return result."""
normalized = self._normalize_single_step(step, index)
context = WorkflowContext(
input=input,
previous_result=str(previous_output) if previous_output else None,
current_step=normalized.name,
variables=all_variables.copy()
)
output = None
stop = False
if normalized.handler:
try:
result = normalized.handler(context)
if isinstance(result, StepResult):
output = result.output
stop = result.stop_workflow
if result.variables:
all_variables.update(result.variables)
else:
output = str(result)
except Exception as e:
output = f"Error: {e}"
elif normalized.agent:
try:
output = normalized.agent.chat(normalized.action or input)
except Exception as e:
output = f"Error: {e}"
elif normalized.action:
try:
from ..agent.agent import Agent
config = normalized.agent_config or self.default_agent_config or {}
temp_agent = Agent(
name=config.get("name", normalized.name),
role=config.get("role", "Assistant"),
goal=config.get("goal", "Complete the task"),
llm=config.get("llm", model),
verbose=verbose
)
action = normalized.action
for key, value in all_variables.items():
action = action.replace(f"{{{{{key}}}}}", str(value))
if previous_output:
action = action.replace("{{previous_output}}", str(previous_output))
output = temp_agent.chat(action)
except Exception as e:
output = f"Error: {e}"
if verbose:
print(f"✅ {normalized.name}: {str(output)[:100]}...")
return {
"step": normalized.name,
"output": output,
"stop": stop,
"variables": all_variables
}
def _execute_route(
self,
route_step: Route,
previous_output: Optional[str],
input: str,
all_variables: Dict[str, Any],
model: str,
verbose: bool
) -> Dict[str, Any]:
"""Execute routing based on previous output."""
results = []
output = previous_output
# Find matching route
matched_route = None
prev_lower = str(previous_output).lower() if previous_output else ""
for key in route_step.routes:
if key.lower() in prev_lower or key == "default":
if key != "default":
matched_route = route_step.routes[key]
break
if matched_route is None:
matched_route = route_step.default or []
if verbose:
route_name = next((k for k in route_step.routes if route_step.routes[k] == matched_route), "default")
print(f"🔀 Routing to: {route_name}")
# Execute matched route steps
for idx, step in enumerate(matched_route):
step_result = self._execute_single_step_internal(
step, output, input, all_variables, model, verbose, idx
)
results.append({"step": step_result["step"], "output": step_result["output"]})
output = step_result["output"]
all_variables.update(step_result.get("variables", {}))
if step_result.get("stop"):
break
return {"steps": results, "output": output, "variables": all_variables}
def _execute_parallel(
self,
parallel_step: Parallel,
previous_output: Optional[str],
input: str,
all_variables: Dict[str, Any],
model: str,
verbose: bool
) -> Dict[str, Any]:
"""Execute steps in parallel (simulated with sequential for now)."""
import concurrent.futures
results = []
outputs = []
if verbose:
print(f"⚡ Running {len(parallel_step.steps)} steps in parallel...")
# Use ThreadPoolExecutor for parallel execution
with concurrent.futures.ThreadPoolExecutor(max_workers=len(parallel_step.steps)) as executor:
futures = []
for idx, step in enumerate(parallel_step.steps):
future = executor.submit(
self._execute_single_step_internal,
step, previous_output, input, all_variables.copy(), model, False, idx
)
futures.append((idx, future))
for idx, future in futures:
try:
step_result = future.result()
results.append({"step": step_result["step"], "output": step_result["output"]})
outputs.append(step_result["output"])
except Exception as e:
results.append({"step": f"parallel_{idx}", "output": f"Error: {e}"})
outputs.append(f"Error: {e}")
# Combine outputs
combined_output = "\n---\n".join(str(o) for o in outputs)
all_variables["parallel_outputs"] = outputs
if verbose:
print(f"✅ Parallel complete: {len(outputs)} results")
return {"steps": results, "output": combined_output, "variables": all_variables}
def _execute_loop(
self,
loop_step: Loop,
previous_output: Optional[str],
input: str,
all_variables: Dict[str, Any],
model: str,
verbose: bool
) -> Dict[str, Any]:
"""Execute step for each item in loop."""
import csv
results = []
outputs = []
items = []
# Get items from variable, CSV, or file
if loop_step.over:
items = all_variables.get(loop_step.over, [])
if isinstance(items, str):
items = [items]
elif loop_step.from_csv:
try:
with open(loop_step.from_csv, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
items = list(reader)
except Exception as e:
logger.error(f"Failed to read CSV {loop_step.from_csv}: {e}")
items = []
elif loop_step.from_file:
try:
with open(loop_step.from_file, 'r', encoding='utf-8') as f:
items = [line.strip() for line in f if line.strip()]
except Exception as e:
logger.error(f"Failed to read file {loop_step.from_file}: {e}")
items = []
if verbose:
print(f"🔁 Looping over {len(items)} items...")
for idx, item in enumerate(items):
# Add current item to variables
loop_vars = all_variables.copy()
loop_vars[loop_step.var_name] = item
loop_vars["loop_index"] = idx
step_result = self._execute_single_step_internal(
loop_step.step, previous_output, input, loop_vars, model, verbose, idx
)
results.append({"step": f"{step_result['step']}_{idx}", "output": step_result["output"]})
outputs.append(step_result["output"])
previous_output = step_result["output"]
all_variables["loop_outputs"] = outputs
combined_output = "\n".join(str(o) for o in outputs) if outputs else ""
return {"steps": results, "output": combined_output, "variables": all_variables}
def _execute_repeat(
self,
repeat_step: Repeat,
previous_output: Optional[str],
input: str,
all_variables: Dict[str, Any],
model: str,
verbose: bool
) -> Dict[str, Any]:
"""Repeat step until condition is met."""
results = []
output = previous_output
if verbose:
print(f"🔄 Repeating up to {repeat_step.max_iterations} times...")
for iteration in range(repeat_step.max_iterations):
step_result = self._execute_single_step_internal(
repeat_step.step, output, input, all_variables, model, verbose, iteration
)
results.append({"step": f"{step_result['step']}_{iteration}", "output": step_result["output"]})
output = step_result["output"]
all_variables.update(step_result.get("variables", {}))
# Check until condition
if repeat_step.until:
context = WorkflowContext(
input=input,
previous_result=str(output) if output else None,
current_step=f"repeat_{iteration}",
variables=all_variables.copy()
)
try:
if repeat_step.until(context):
if verbose:
print(f"✅ Repeat condition met at iteration {iteration + 1}")
break
except Exception as e:
logger.error(f"Repeat until condition failed: {e}")
if step_result.get("stop"):
break
all_variables["repeat_iterations"] = iteration + 1
return {"steps": results, "output": output, "variables": all_variables}
def start(self, input: str = "", **kwargs) -> Dict[str, Any]:
"""Alias for run() for consistency with PraisonAIAgents."""
return self.run(input, **kwargs)
# Alias: Pipeline = Workflow (they are the same concept)
Pipeline = Workflow
class WorkflowManager:
@@ -717,6 +1393,16 @@ class WorkflowManager:
on_result=on_result
)
# Handle skipped steps
if step_result.get("skipped"):
results.append({
"step": step.name,
"status": "skipped",
"output": None
})
current_step_idx += 1
continue
results.append({
"step": step.name,
"status": "success" if step_result["success"] else "failed",
@@ -724,6 +1410,11 @@ class WorkflowManager:
"error": step_result.get("error")
})
# Handle early stop
if step_result.get("stop"):
self._log(f"Workflow stopped early at step '{step.name}'")
break
# Save checkpoint after each step if enabled
if checkpoint:
self._save_checkpoint(
@@ -784,9 +1475,64 @@ class WorkflowManager:
planning: bool = False,
verbose: int = 0,
on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
on_result: Optional[Callable[[WorkflowStep, str], None]] = None
on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
original_input: str = ""
) -> Dict[str, Any]:
"""Execute a single workflow step."""
# Get previous step output
previous_output = results[-1].get("output") if results else None
# Create context for step handlers
context = WorkflowContext(
input=original_input,
previous_result=str(previous_output) if previous_output else None,
current_step=step.name,
variables=all_variables.copy()
)
# Check should_run condition if provided
if step.should_run:
try:
if not step.should_run(context):
return {
"success": True,
"output": None,
"skipped": True,
"stop": False
}
except Exception as e:
self._log(f"should_run check for step '{step.name}' failed: {e}")
# If step has a custom handler function
if step.handler:
try:
result = step.handler(context)
# Handle StepResult
if isinstance(result, StepResult):
# Update variables from result
if result.variables:
all_variables.update(result.variables)
return {
"success": True,
"output": result.output,
"stop": result.stop_workflow, # Early termination flag
"error": None
}
else:
return {
"success": True,
"output": str(result),
"stop": False,
"error": None
}
except Exception as e:
return {
"success": False,
"output": None,
"stop": False,
"error": str(e)
}
# Build context from previous steps
context = self._build_step_context(step, step_idx, results, all_variables)
@@ -800,15 +1546,20 @@ class WorkflowManager:
# Get or create executor for this step
step_executor = executor
if step_executor is None:
# Create agent for this step
step_agent = self._create_step_agent(
step=step,
default_agent=default_agent,
default_llm=default_llm,
memory=memory,
planning=planning,
verbose=verbose
)
# Use step's direct agent if provided
step_agent = step.agent
# Otherwise create agent from config
if step_agent is None:
step_agent = self._create_step_agent(
step=step,
default_agent=default_agent,
default_llm=default_llm,
memory=memory,
planning=planning,
verbose=verbose
)
if step_agent:
if planning and hasattr(step_agent, 'start'):
def agent_executor(prompt, agent=step_agent):
@@ -823,6 +1574,7 @@ class WorkflowManager:
return {
"success": False,
"output": None,
"stop": False,
"error": f"No executor available for step '{step.name}'"
}
@@ -853,6 +1605,7 @@ class WorkflowManager:
return {
"success": step_success,
"output": output,
"stop": False, # Normal execution doesn't request stop
"error": error
}
@@ -826,7 +826,34 @@ Provide a JSON with the structure:
logging.info("Hierarchical task execution finished")
def workflow(self):
"""Synchronous version of workflow method"""
"""Synchronous version of workflow method.
DEPRECATED: This method is deprecated. Use the Workflow class instead:
```python
from praisonaiagents import Workflow, StepResult
from praisonaiagents.workflows import route, parallel, loop, repeat
workflow = Workflow(steps=[step1, step2])
result = workflow.start("input")
```
The Workflow class provides a simpler API with more features:
- route() for decision-based branching
- parallel() for concurrent execution
- loop() for iteration over lists/CSV
- repeat() for evaluator-optimizer patterns
- Callbacks (on_step_start, on_step_complete, etc.)
- Guardrails with validation feedback
- Status tracking
"""
import warnings
warnings.warn(
"process='workflow' is deprecated. Use the Workflow class instead: "
"from praisonaiagents import Workflow; workflow = Workflow(steps=[...]); workflow.start()",
DeprecationWarning,
stacklevel=2
)
current_iter = 0 # Track how many times we've looped
# Build workflow relationships first
for task in self.tasks.values():
@@ -0,0 +1,55 @@
"""
Workflows module for PraisonAI Agents.
Provides workflow/pipeline patterns for orchestrating agents and functions.
"""
from .workflows import (
# Core classes
Workflow,
Pipeline, # Alias for Workflow
WorkflowStep,
WorkflowContext,
StepResult,
WorkflowManager,
# Pattern helpers
Route,
Parallel,
Loop,
Repeat,
# Convenience functions
route,
parallel,
loop,
repeat,
# Backward compatibility aliases
StepInput,
StepOutput,
)
__all__ = [
# Core
"Workflow",
"Pipeline",
"WorkflowStep",
"WorkflowContext",
"StepResult",
"WorkflowManager",
# Patterns
"Route",
"Parallel",
"Loop",
"Repeat",
"route",
"parallel",
"loop",
"repeat",
# Aliases
"StepInput",
"StepOutput",
]
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.0.194"
version = "0.1.1"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
requires-python = ">=3.10"
authors = [
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.194"
version = "0.1.1"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+156 -1
View File
@@ -103,7 +103,20 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
| @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
| 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](#workflow-checkpoints) | [📖](https://docs.praison.ai/cli/session) |
| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |
## Supported Providers
@@ -445,6 +458,127 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
- **Async Execution**: Use `aexecute()` for async workflows
- **Planning Mode**: Enable at workflow level with `planning=True`
- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
- **Loops**: Use `loop_over` and `loop_var` to iterate over data
### Choosing the Right Workflow System
| Use Case | Recommended |
|----------|-------------|
| Simple function pipelines | `Workflow` class ⭐ |
| Agent-only pipelines | `Workflow` class |
| CSV batch processing | `process="workflow"` |
| Complex task routing | `process="workflow"` |
| Markdown templates | `WorkflowManager` |
| Early stop / conditional | `Workflow` class |
### Simple Workflow (Recommended)
The easiest way to create workflows - just pass functions as steps:
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Define simple handler functions
def validate(ctx: WorkflowContext) -> StepResult:
if not ctx.input:
return StepResult(output="No input", stop_workflow=True)
return StepResult(output=f"Valid: {ctx.input}")
def process(ctx: WorkflowContext) -> StepResult:
return StepResult(output=f"Processed: {ctx.previous_result}")
# Create and run workflow
workflow = Workflow(steps=[validate, process])
result = workflow.start("Hello World", verbose=True)
print(result["output"]) # "Processed: Valid: Hello World"
```
**Key Features:**
- **Just pass functions** - No complex configuration needed
- **Early stop** - Return `stop_workflow=True` to stop the workflow
- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
- **Verbose mode** - See step-by-step progress
### Workflow Branching & Loops
```python
from praisonaiagents import WorkflowStep
# Branching step
decision_step = WorkflowStep(
name="decide",
action="Evaluate if task is complete",
next_steps=["success_step", "retry_step"],
branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
)
# Loop step
loop_step = WorkflowStep(
name="process_items",
action="Process {{item}}",
loop_over="items", # Variable containing list
loop_var="item" # Current item variable name
)
```
### Workflow Patterns (route, parallel, loop, repeat)
```python
from praisonaiagents import Workflow, WorkflowContext, StepResult
# Or use Pipeline (alias for Workflow)
from praisonaiagents import Pipeline
from praisonaiagents.workflows import route, parallel, loop, repeat
# 1. ROUTING - Decision-based branching
workflow = Workflow(steps=[
classify_request, # Returns "approve" or "reject"
route({
"approve": [approve_handler],
"reject": [reject_handler],
"default": [fallback_handler]
})
])
# 2. PARALLEL - Concurrent execution
workflow = Workflow(steps=[
parallel([research_market, research_competitors, research_customers]),
summarize_results # Gets all parallel outputs
])
# 3. LOOP - Iterate over list or CSV
workflow = Workflow(
steps=[loop(process_item, over="items")],
variables={"items": ["a", "b", "c"]}
)
# Or from CSV file:
workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
# 4. REPEAT - Evaluator-Optimizer pattern
workflow = Workflow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result,
max_iterations=5
)
])
# 5. CALLBACKS - Monitor workflow execution
workflow = Workflow(
steps=[step1, step2],
on_workflow_start=lambda w, i: print(f"Starting: {i}"),
on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
)
# 6. GUARDRAILS - Validate and retry
def validate(result):
return ("error" not in result.output, "Fix the error")
workflow = Workflow(steps=[
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
])
```
### 9. Hooks
@@ -589,6 +723,12 @@ praisonai workflow run deploy --workflow-var environment=staging --workflow-var
# Execute with planning mode (AI creates sub-steps for each workflow step)
praisonai workflow run "Research Blog" --planning --verbose
# Execute with reasoning mode (chain-of-thought)
praisonai workflow run "Analysis" --reasoning --verbose
# Execute with memory enabled
praisonai workflow run "Research" --memory
# Show workflow details
praisonai workflow show deploy
@@ -600,8 +740,23 @@ praisonai "What is AI?" --workflow "Research,Summarize" --save
# Inline workflow with step actions
praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
# Workflow CLI help
praisonai workflow help
```
**Workflow CLI Options:**
| Flag | Description |
|------|-------------|
| `--workflow-var key=value` | Set workflow variable (can be repeated) |
| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
| `--planning` | Enable planning mode |
| `--reasoning` | Enable reasoning mode |
| `--memory` | Enable memory |
| `--verbose` | Enable verbose output |
| `--save` | Save output to file |
### Hooks CLI:
```bash
# List configured hooks
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.16.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.16.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.23.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.23.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.16 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.23 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.16"
__version__ = "2.3.23"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.0.194",
"praisonaiagents>=0.1.1",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.0.194" },
{ name = "praisonaiagents", specifier = ">=0.1.1" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.194"
version = "0.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e9/5f/f572e88d4148e61da3d114f71357c9103a822099103b0b57adc7d88bedc2/praisonaiagents-0.0.194.tar.gz", hash = "sha256:e70c85203c9cf3792697535844a8bba4da158ecb9959abc5025967199eb86d36", size = 518340, upload-time = "2025-12-16T22:58:43.725Z" }
sdist = { url = "https://files.pythonhosted.org/packages/b0/41/14a8e49c644c26ee565511776c67cc978427f36aec68a1442ac666c35401/praisonaiagents-0.1.1.tar.gz", hash = "sha256:76f69188e4893b0e868f33ea02d26cccfa90b99d679c2db4185d8cf624fa740e", size = 550745, upload-time = "2025-12-17T00:37:35.974Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/01/18/d0d0d99ec828d0cc749eb642e0246abda45575c18646893786cdad01fd9c/praisonaiagents-0.0.194-py3-none-any.whl", hash = "sha256:a1f89d3169a4df8c0c96706afcc7c71f20560c4fd2a37aaa0b8193ee6445fdef", size = 494859, upload-time = "2025-12-16T22:58:41.237Z" },
{ url = "https://files.pythonhosted.org/packages/f7/93/2cef070fae77bbef32a986ad7e1787f799442047f5dac831b9c137839353/praisonaiagents-0.1.1-py3-none-any.whl", hash = "sha256:c116551a35df0f8766496ba5102c54633f071735eae2a71b33eb4df5e2728e86", size = 518796, upload-time = "2025-12-17T00:37:33.824Z" },
]
[[package]]