mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-10 03:31:28 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f991f37cfe | ||
|
|
a43b3ac72c | ||
|
|
06f82249e1 | ||
|
|
d3aa78241a |
No files matched your search
@@ -112,6 +112,11 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
|
||||
| 🤖 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
|
||||
|
||||
@@ -456,6 +461,17 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
|
||||
- **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:
|
||||
@@ -506,33 +522,62 @@ loop_step = WorkflowStep(
|
||||
)
|
||||
```
|
||||
|
||||
### Task Callbacks & Variables (process="workflow")
|
||||
### Workflow Patterns (route, parallel, loop, repeat)
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent, Task, PraisonAIAgents
|
||||
from praisonaiagents import Workflow, WorkflowContext, StepResult
|
||||
# Or use Pipeline (alias for Workflow)
|
||||
from praisonaiagents import Pipeline
|
||||
from praisonaiagents.workflows import route, parallel, loop, repeat
|
||||
|
||||
# Per-task agent config (no need to create agent separately)
|
||||
task = Task(
|
||||
description="Research {{topic}}",
|
||||
agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
|
||||
variables={"topic": "AI trends"}
|
||||
# 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']}")
|
||||
)
|
||||
|
||||
# Callbacks for workflow execution
|
||||
def on_start(task, task_id):
|
||||
print(f"Starting: {task.name}")
|
||||
# 6. GUARDRAILS - Validate and retry
|
||||
def validate(result):
|
||||
return ("error" not in result.output, "Fix the error")
|
||||
|
||||
def on_complete(task, output):
|
||||
print(f"Completed: {task.name}")
|
||||
|
||||
agents = PraisonAIAgents(
|
||||
agents=[agent],
|
||||
tasks=[task],
|
||||
process="workflow",
|
||||
on_task_start=on_start,
|
||||
on_task_complete=on_complete,
|
||||
variables={"global_var": "shared_value"} # Global variables for all tasks
|
||||
)
|
||||
workflow = Workflow(steps=[
|
||||
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
|
||||
])
|
||||
```
|
||||
|
||||
### 9. Hooks
|
||||
@@ -678,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
|
||||
|
||||
@@ -689,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
@@ -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.19" \
|
||||
"praisonai>=2.3.23" \
|
||||
"praisonai[api]" \
|
||||
gunicorn \
|
||||
markdown
|
||||
|
||||
@@ -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.19" \
|
||||
"praisonai>=2.3.23" \
|
||||
"praisonai[chat]" \
|
||||
"embedchain[github,youtube]"
|
||||
|
||||
|
||||
@@ -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.19" \
|
||||
"praisonai>=2.3.23" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[chat]" \
|
||||
"praisonai[realtime]" \
|
||||
|
||||
@@ -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.19" \
|
||||
"praisonai>=2.3.23" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[crewai]"
|
||||
|
||||
|
||||
@@ -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,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']}")
|
||||
@@ -54,7 +54,13 @@ except ImportError:
|
||||
MCP = None
|
||||
from .session import Session
|
||||
from .memory.memory import Memory
|
||||
from .memory.workflows import Workflow, WorkflowStep, WorkflowContext, StepResult
|
||||
# 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,30 +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
|
||||
from ..workflows import WorkflowContext
|
||||
return WorkflowContext
|
||||
if name == "StepResult":
|
||||
from .workflows import 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
|
||||
from ..workflows import WorkflowContext
|
||||
return WorkflowContext
|
||||
if name == "StepOutput":
|
||||
from .workflows import StepResult
|
||||
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
|
||||
@@ -98,13 +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",
|
||||
|
||||
@@ -49,6 +49,137 @@ 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."""
|
||||
@@ -173,9 +304,6 @@ class Workflow:
|
||||
Returns:
|
||||
Dict with 'output' (final result) and 'steps' (all step results)
|
||||
"""
|
||||
# Normalize steps - convert Agents/functions to WorkflowStep
|
||||
normalized_steps = self._normalize_steps()
|
||||
|
||||
# Use default LLM if not specified
|
||||
model = llm or self.default_llm or "gpt-4o-mini"
|
||||
|
||||
@@ -185,7 +313,59 @@ class Workflow:
|
||||
results = []
|
||||
previous_output = None
|
||||
|
||||
for i, step in enumerate(normalized_steps):
|
||||
# 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,
|
||||
@@ -200,6 +380,7 @@ class Workflow:
|
||||
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}")
|
||||
@@ -270,6 +451,8 @@ class Workflow:
|
||||
# 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,
|
||||
@@ -306,11 +489,300 @@ class Workflow:
|
||||
|
||||
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:
|
||||
"""
|
||||
Manages workflow discovery, loading, and execution.
|
||||
|
||||
@@ -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
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "praisonaiagents"
|
||||
version = "0.0.197"
|
||||
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
Generated
+1
-1
@@ -2422,7 +2422,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "praisonaiagents"
|
||||
version = "0.0.197"
|
||||
version = "0.1.1"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "aiohttp" },
|
||||
|
||||
+87
-21
@@ -112,6 +112,11 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
|
||||
| 🤖 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
|
||||
|
||||
@@ -456,6 +461,17 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
|
||||
- **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:
|
||||
@@ -506,33 +522,62 @@ loop_step = WorkflowStep(
|
||||
)
|
||||
```
|
||||
|
||||
### Task Callbacks & Variables (process="workflow")
|
||||
### Workflow Patterns (route, parallel, loop, repeat)
|
||||
|
||||
```python
|
||||
from praisonaiagents import Agent, Task, PraisonAIAgents
|
||||
from praisonaiagents import Workflow, WorkflowContext, StepResult
|
||||
# Or use Pipeline (alias for Workflow)
|
||||
from praisonaiagents import Pipeline
|
||||
from praisonaiagents.workflows import route, parallel, loop, repeat
|
||||
|
||||
# Per-task agent config (no need to create agent separately)
|
||||
task = Task(
|
||||
description="Research {{topic}}",
|
||||
agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
|
||||
variables={"topic": "AI trends"}
|
||||
# 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']}")
|
||||
)
|
||||
|
||||
# Callbacks for workflow execution
|
||||
def on_start(task, task_id):
|
||||
print(f"Starting: {task.name}")
|
||||
# 6. GUARDRAILS - Validate and retry
|
||||
def validate(result):
|
||||
return ("error" not in result.output, "Fix the error")
|
||||
|
||||
def on_complete(task, output):
|
||||
print(f"Completed: {task.name}")
|
||||
|
||||
agents = PraisonAIAgents(
|
||||
agents=[agent],
|
||||
tasks=[task],
|
||||
process="workflow",
|
||||
on_task_start=on_start,
|
||||
on_task_complete=on_complete,
|
||||
variables={"global_var": "shared_value"} # Global variables for all tasks
|
||||
)
|
||||
workflow = Workflow(steps=[
|
||||
WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
|
||||
])
|
||||
```
|
||||
|
||||
### 9. Hooks
|
||||
@@ -678,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
|
||||
|
||||
@@ -689,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
|
||||
|
||||
@@ -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.19.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.19.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]"
|
||||
|
||||
@@ -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.19 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 @@
|
||||
__version__ = "2.3.19"
|
||||
__version__ = "2.3.23"
|
||||
@@ -12,7 +12,7 @@ dependencies = [
|
||||
"rich>=13.7",
|
||||
"markdown>=3.5",
|
||||
"pyparsing>=3.0.0",
|
||||
"praisonaiagents>=0.0.197",
|
||||
"praisonaiagents>=0.1.1",
|
||||
"python-dotenv>=0.19.0",
|
||||
"instructor>=1.3.3",
|
||||
"PyYAML>=6.0",
|
||||
|
||||
Generated
+4
-4
@@ -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.197" },
|
||||
{ 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.197"
|
||||
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/c6/00/342eae9c9b8bc25e812ad761ad91c0ad18adc533eaa2aa3c7c31a85cd7f8/praisonaiagents-0.0.197.tar.gz", hash = "sha256:9504dcfc861907201a10335ea7b74bfe2c6d189cc4f97951b782cd6ac46c8ee0", size = 520740, upload-time = "2025-12-16T23:26:31.078Z" }
|
||||
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/ee/5d/2aaa57e97d84b12dc75b66e083839eb3f9d3ef8d7721a9d8e4e7d2513ded/praisonaiagents-0.0.197-py3-none-any.whl", hash = "sha256:701b60a02d9e7147909a1311572e092846ef8247b7251b67b31dbcd581eec1d7", size = 497086, upload-time = "2025-12-16T23:26:29.208Z" },
|
||||
{ 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]]
|
||||
|
||||
Reference in new issue
Block a user