Workflow Parallel Execution

from praisonaiagents import Agent, AgentFlow, parallel, WorkflowContext, StepResult

coordinator = Agent(name="Coordinator", instructions="Merge parallel results")

def task_a(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Result A")

def task_b(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Result B")

workflow = AgentFlow(agents=[coordinator], steps=[parallel([task_a, task_b])])
workflow.start("Run independent tasks in parallel")

from praisonaiagents import Agent, AgentFlow, parallel

planner  = Agent(name="Planner",  instructions="Produce a detailed plan.")
reviewer = Agent(name="Reviewer", instructions="Review the plan critically.")
writer   = Agent(name="Writer",   instructions="Rewrite the plan clearly.")
scorer   = Agent(name="Scorer",   instructions="Score the plan 0–100 with reasons.")

flow = AgentFlow(steps=[
    planner,
    parallel([reviewer, writer, scorer]),   # context="full" by default
])

flow.start("Draft a GTM plan for a developer tool launch.")

# Token-saving opt-in — WARNS whenever it drops the middle:
parallel([reviewer, writer, scorer], context="truncate")

⚡ Running 3 steps in parallel...
✅ Parallel complete: 3 results
✅ summarize: 📋 SUMMARY:...

📋 SUMMARY:
  • 📊 Market: Growth 15% YoY
  • 🏢 Competitors: 3 major players
  • 👥 Customers: 85% satisfaction

parallel(
    steps: List,
    max_workers: Optional[int] = None,
    on_failure: str = "partial_ok",
    timeout: Optional[float] = None,
    context: str = "full",
) -> Parallel

def aggregator(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables["parallel_outputs"]
    # outputs = ["Result A", "Result B", "Result C"]
    return StepResult(output=f"Combined: {len(outputs)} results")

from praisonaiagents import AgentFlow, Task, WorkflowContext, StepResult, parallel

def branch_a(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="A")

def branch_b(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="B")

def after(ctx: WorkflowContext) -> StepResult:
    a = ctx.variables["branch_a_output"]
    b = ctx.variables["branch_b_output"]
    return StepResult(output=f"{a} + {b}")

workflow = AgentFlow(steps=[
    parallel([
        Task(name="branch_a", handler=branch_a, output_variable="branch_a_output"),
        Task(name="branch_b", handler=branch_b, output_variable="branch_b_output"),
    ]),
    Task(name="after", handler=after),
])
result = workflow.start("go")
print(result["variables"]["branch_a_output"])  # "A"
print(result["variables"]["branch_b_output"])  # "B"

workflow = AgentFlow(steps=[
    parallel([
        Task(name="first",  handler=lambda ctx: StepResult(output="FROM-FIRST"),
             output_variable="shared"),
        Task(name="second", handler=lambda ctx: StepResult(output="FROM-SECOND"),
             output_variable="shared"),
    ]),
])
result = workflow.start("go")
print(result["variables"]["shared"])  # "FROM-SECOND"
# WARNING logged: Parallel branches 0 and 1 both wrote variable 'shared'...

import threading
from praisonaiagents import AgentFlow, Task, WorkflowContext, StepResult, parallel

client = threading.Lock()   # stands in for an HTTP/DB/LLM client

def append_row(label):
    def run(ctx: WorkflowContext) -> StepResult:
        # The surrounding dict and its 'rows' list ARE branch-local.
        ctx.variables["state"]["rows"].append(label)
        # The 'client' leaf is SHARED by reference across branches.
        with ctx.variables["state"]["client"]:
            pass
        return StepResult(
            output=label,
            variables={f"rows_{label}": list(ctx.variables["state"]["rows"])},
        )
    return run

wf = AgentFlow(
    steps=[parallel([
        Task(name="a", handler=append_row("A")),
        Task(name="b", handler=append_row("B")),
    ])],
    variables={"state": {"client": client, "rows": []}},
)
result = wf.start("go")
# Each branch saw only its own append — rows_A == ["A"], rows_B == ["B"].
# The lock in state["client"] is the one real object, usable for coordination.

from praisonaiagents import Agent

researcher = Agent(name="Researcher", role="Research topics")
analyst = Agent(name="Analyst", role="Analyze data")
writer = Agent(name="Writer", role="Write content")

workflow = AgentFlow(steps=[
    parallel([researcher, analyst, writer]),
    final_aggregator
])

workflow = AgentFlow(steps=[
    parallel([
        my_function,           # Function
        Agent(name="Bot"),     # Agent
        Task(...)      # Task
    ]),
    aggregator
])

workflow = AgentFlow(steps=[
    parallel([
        parallel([task_a1, task_a2]),  # Group A
        parallel([task_b1, task_b2])   # Group B
    ]),
    final_aggregator
])

# Performance comparison
# Sequential: 3 steps × 1s each = 3s total
# Parallel:   3 steps × 1s each = ~1s total (concurrent)

# Default — lossless (new default as of #5719)
parallel([worker_a, worker_b, worker_c])
# equivalent to:
parallel([worker_a, worker_b, worker_c], context="full")

# Opt in to the head+tail excerpt (and warn on drop)
parallel([worker_a, worker_b, worker_c], context="truncate")

# Opt in to the LLM summariser (large fan-outs); falls back to truncate
parallel([worker_a, worker_b, worker_c], context="summarize")

from praisonaiagents import Agent, AgentFlow, parallel, WorkflowStepError

# partial_ok (default) — continue even if one branch fails
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c]),
    aggregator,
])

# fail_fast — abort on first failure
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c], on_failure="fail_fast"),
    aggregator,
])

# fail_all — gather all errors then fail
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c], on_failure="fail_all"),
    aggregator,
])

def aggregator(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables["parallel_outputs"]

    # A failed branch contributes None at its index
    successful = [o for o in outputs if o is not None]
    failures = [o for o in outputs if o is None]

    if len(successful) >= 2:  # Minimum threshold
        return StepResult(output=f"Processed {len(successful)} sources")
    else:
        return StepResult(output=f"Insufficient data: {len(failures)} failures")

from praisonaiagents import Workflow, StepResult, parallel

def ok(ctx):  return StepResult(output="a")
def bad(ctx): raise KeyError("missing")

r = Workflow(steps=[parallel([ok, bad])]).start("x")

print(r["status"])   # "failed"
print(r["error"])    # "parallel_1: 'missing'"
for s in r["steps"]:
    if s.get("status") == "failed":
        print(s["step"], s["error"])   # parallel_1 'missing'

from praisonaiagents import Task, parallel, AgentFlow

risky = Task(name="charge_customer", handler=charge, on_error="stop")   # halts on failure
safe  = Task(name="log_event", handler=log_event)

wf = AgentFlow(steps=[
    parallel([risky, safe], on_failure="partial_ok"),
    Task(name="ship_order", handler=ship),   # must NOT run if charge_customer failed
])
wf.run(input=order)
# charge_customer fails → workflow halts with status="failed"; ship_order never runs.
# Printed: 🛑 Workflow stopped by nested parallel step

try:
    result = workflow.start("Process all branches")
except WorkflowStepError as e:
    print(f"Workflow failed: {e}")
    print(f"Root cause: {e.cause}")
    for error in e.errors:
        print(f"  Branch {error['step']}: {error['error']}")

from praisonaiagents import Agent, AgentFlow, parallel

# Bound the whole block at 10 seconds — a stuck branch no longer blocks the run.
flow = AgentFlow(steps=[
    parallel([researcher, analyst, writer], timeout=10),
    summarizer,
])
flow.start("Analyse Q3 numbers")

# Same Workflow instance used across many inputs
wf = AgentFlow(steps=[
    parallel([slow_branch, quick_fail_branch], on_failure="fail_fast"),
    report,
])

wf.start("job 1")   # fail_fast raises; slow_branch is abandoned
wf.start("job 2")   # run 2 begins; the abandoned branch finally fails afterwards
# Result: run 2 completes with status="completed".
# The late failure from run 1's abandoned branch cannot flip run 2's status.