from praisonaiagents import Workflow, WorkflowContext, StepResult

def validate(ctx: WorkflowContext) -> StepResult:
    return StepResult(output=f"Valid: {ctx.input}")

def process(ctx: WorkflowContext) -> StepResult:
    return StepResult(output=f"Done: {ctx.previous_result}")

workflow = Workflow(steps=[validate, process])
result = workflow.start("Hello World")
print(result["output"])  # "Done: Valid: Hello World"

from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult
# Pipeline is an alias for Workflow (same thing)
from praisonaiagents import Pipeline
# Or from workflows module
from praisonaiagents.workflows import WorkflowManager, Workflow, WorkflowStep
# Pattern helpers
from praisonaiagents.workflows import route, parallel, loop, repeat

def on_start(workflow, input_text):
    print(f"Starting workflow with: {input_text}")

def on_complete(workflow, result):
    print(f"Workflow completed: {result['status']}")

def on_step_start(step_name, context):
    print(f"Starting step: {step_name}")

def on_step_complete(step_name, result):
    print(f"Step {step_name} completed: {result.output[:50]}...")

def on_step_error(step_name, error):
    print(f"Step {step_name} failed: {error}")

workflow = Workflow(
    steps=[step1, step2],
    on_workflow_start=on_start,
    on_workflow_complete=on_complete,
    on_step_start=on_step_start,
    on_step_complete=on_step_complete,
    on_step_error=on_step_error
)

def validate_output(result):
    if "error" in result.output.lower():
        return (False, "Output contains error, please fix")
    return (True, None)

workflow = Workflow(steps=[
    WorkflowStep(
        name="generator",
        handler=my_generator,
        guardrail=validate_output,
        max_retries=3
    )
])

workflow = Workflow(steps=[step1, step2])
print(workflow.status)  # "not_started"

result = workflow.start("input")
print(workflow.status)  # "completed"
print(workflow.step_statuses)  # {"step1": "completed", "step2": "completed"}

# Result includes status
print(result["status"])  # "completed"
print(result["steps"][0]["status"])  # "completed"
print(result["steps"][0]["retries"])  # 0

WorkflowContext(
    input: str = "",
    previous_result: Optional[str] = None,
    current_step: str = "",
    variables: Dict[str, Any] = {}
)

StepResult(
    output: str = "",
    stop_workflow: bool = False,
    variables: Dict[str, Any] = {}
)

def validate(ctx: WorkflowContext) -> StepResult:
    if "error" in ctx.input:
        return StepResult(output="Invalid", stop_workflow=True)
    return StepResult(output="Valid", variables={"validated": True})

Workflow(
    name: str = "Workflow",
    description: str = "",
    steps: List = [],
    variables: Dict[str, Any] = {},
    default_llm: Optional[str] = None,
    default_agent_config: Optional[Dict[str, Any]] = None
)

def start(
    input: str = "",
    llm: Optional[str] = None,
    verbose: bool = False
) -> Dict[str, Any]

async def astart(
    input: str = "",
    llm: Optional[str] = None,
    verbose: bool = False
) -> Dict[str, Any]

import asyncio

async def main():
    workflow = Workflow(steps=[step1, step2])
    result = await workflow.astart("Hello World")
    print(result["output"])

asyncio.run(main())

from praisonaiagents import Workflow, Agent, WorkflowStep

workflow = Workflow(
    steps=[
        my_function,                    # Function
        Agent(name="Writer", ...),      # Agent
        WorkflowStep(name="custom", handler=my_handler)  # WorkflowStep
    ]
)

WorkflowStep(
    name: str,
    description: str = "",
    action: str = "",
    handler: Optional[Callable] = None,
    should_run: Optional[Callable] = None,
    agent: Optional[Agent] = None,
    agent_config: Optional[Dict[str, Any]] = None,
    condition: Optional[str] = None,
    on_error: Literal["stop", "continue", "retry"] = "stop",
    max_retries: int = 1,
    context_from: Optional[List[str]] = None,
    retain_full_context: bool = True,
    output_variable: Optional[str] = None,
    tools: Optional[List[Any]] = None,
    next_steps: Optional[List[str]] = None,
    branch_condition: Optional[Dict[str, List[str]]] = None,
    loop_over: Optional[str] = None,
    loop_var: str = "item"
)

def my_handler(ctx: WorkflowContext) -> StepResult:
    # Access context
    print(f"Input: {ctx.input}")
    print(f"Previous: {ctx.previous_result}")
    print(f"Variables: {ctx.variables}")
    
    # Return result
    return StepResult(
        output="Step completed",
        stop_workflow=False,  # Set True to stop workflow
        variables={"key": "value"}  # Add/update variables
    )

def is_sensitive(ctx: WorkflowContext) -> bool:
    return "legal" in ctx.input.lower()

step = WorkflowStep(
    name="compliance",
    handler=check_compliance,
    should_run=is_sensitive  # Only runs for sensitive content
)

step = WorkflowStep(
    name="research",
    action="Research {{topic}}",
    agent_config={
        "role": "Researcher",
        "goal": "Find comprehensive information",
        "backstory": "Expert researcher"
    },
    tools=["tavily_search"],
    output_variable="research_data"
)

# Decision step with conditional branching
decision_step = WorkflowStep(
    name="evaluate",
    action="Evaluate if the task is complete. Reply with 'success' or 'failure'.",
    next_steps=["success_handler", "failure_handler"],
    branch_condition={
        "success": ["success_handler"],
        "failure": ["failure_handler"]
    }
)

# Loop step that iterates over a list
loop_step = WorkflowStep(
    name="process_items",
    action="Process item: {{current_item}}",
    loop_over="items",      # Variable containing the list
    loop_var="current_item" # Variable name for each item
)

# Execute with items
result = manager.execute(
    "my_workflow",
    variables={"items": ["item1", "item2", "item3"]}
)

Workflow(
    name: str,
    description: str = "",
    steps: List[WorkflowStep] = [],
    variables: Dict[str, Any] = {},
    file_path: Optional[str] = None,
    default_agent_config: Optional[Dict[str, Any]] = None,
    default_llm: Optional[str] = None,
    memory_config: Optional[Dict[str, Any]] = None,
    planning: bool = False,
    planning_llm: Optional[str] = None
)

workflow = Workflow(
    name="research_pipeline",
    description="Multi-agent research workflow",
    default_llm="gpt-4o-mini",
    planning=True,
    steps=[
        WorkflowStep(name="research", action="Research AI"),
        WorkflowStep(name="write", action="Write report")
    ],
    variables={"topic": "AI trends"}
)

WorkflowManager(
    workspace_path: Optional[str] = None,
    verbose: int = 0
)

def execute(
    workflow_name: str,
    executor: Optional[Callable[[str], str]] = None,
    variables: Optional[Dict[str, Any]] = None,
    on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
    on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
    default_agent: Optional[Any] = None,
    default_llm: Optional[str] = None,
    memory: Optional[Any] = None,
    planning: bool = False,
    stream: bool = False,
    verbose: int = 0,
    checkpoint: Optional[str] = None,
    resume: Optional[str] = None
) -> Dict[str, Any]

{
    "success": bool,
    "workflow": str,
    "results": [
        {
            "step": str,
            "status": "success" | "failed" | "skipped",
            "output": str | None,
            "error": str | None
        }
    ],
    "variables": Dict[str, Any]
}

from praisonaiagents import Agent
from praisonaiagents.memory import WorkflowManager

agent = Agent(name="Assistant", llm="gpt-4o-mini")
manager = WorkflowManager()

result = manager.execute(
    "deploy",
    default_agent=agent,
    variables={"environment": "production"},
    on_step=lambda step, i: print(f"Starting: {step.name}"),
    on_result=lambda step, output: print(f"Done: {step.name}")
)

if result["success"]:
    print("Workflow completed!")

async def aexecute(
    workflow_name: str,
    executor: Optional[Callable[[str], str]] = None,
    variables: Optional[Dict[str, Any]] = None,
    on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
    on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
    default_agent: Optional[Any] = None,
    default_llm: Optional[str] = None,
    memory: Optional[Any] = None,
    planning: bool = False,
    stream: bool = False,
    verbose: int = 0
) -> Dict[str, Any]

import asyncio
from praisonaiagents.memory import WorkflowManager

manager = WorkflowManager()

async def main():
    # Run multiple workflows concurrently
    results = await asyncio.gather(
        manager.aexecute("research", default_llm="gpt-4o-mini"),
        manager.aexecute("analysis", default_llm="gpt-4o-mini"),
    )
    return results

results = asyncio.run(main())

def list_workflows() -> List[Workflow]

manager = WorkflowManager()
workflows = manager.list_workflows()

for workflow in workflows:
    print(f"{workflow.name}: {len(workflow.steps)} steps")

def get_workflow(name: str) -> Optional[Workflow]

def create_workflow(
    name: str,
    description: str = "",
    steps: Optional[List[Dict[str, str]]] = None,
    variables: Optional[Dict[str, Any]] = None
) -> Workflow

manager = WorkflowManager()

workflow = manager.create_workflow(
    name="Code Review",
    description="Review code changes",
    steps=[
        {"name": "Lint", "action": "Run linting"},
        {"name": "Test", "action": "Run tests"},
        {"name": "Review", "action": "Review code"}
    ],
    variables={"branch": "main"}
)

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

{
    "total_workflows": int,
    "total_steps": int,
    "workflows_dir": str
}

def reload() -> None

workflow = Workflow(
    name="pipeline",
    variables={"topic": "AI"},
    steps=[
        WorkflowStep(
            name="research",
            action="Research {{topic}}",
            output_variable="research_data"
        ),
        WorkflowStep(
            name="analyze",
            action="Analyze: {{research_data}}"
        ),
        WorkflowStep(
            name="write",
            action="Write about {{previous_output}}"
        )
    ]
)

def list_checkpoints() -> List[Dict[str, Any]]

manager = WorkflowManager()
checkpoints = manager.list_checkpoints()

for cp in checkpoints:
    print(f"{cp['name']}: {cp['completed_steps']} steps completed")

def delete_checkpoint(name: str) -> bool

manager = WorkflowManager()

# Execute with checkpoint
result = manager.execute("deploy", checkpoint="deploy-v1")

# Resume if interrupted
result = manager.execute("deploy", resume="deploy-v1")

# Clean up
manager.delete_checkpoint("deploy-v1")

from praisonaiagents import Workflow, WorkflowContext, StepResult
from praisonaiagents.memory.workflows import route, parallel, loop, repeat

route(
    routes: Dict[str, List],  # Key: pattern to match, Value: steps to execute
    default: Optional[List] = None  # Fallback steps
) -> Route

workflow = Workflow(steps=[
    classify_request,  # Returns "approve" or "reject"
    route({
        "approve": [approve_handler, notify_user],
        "reject": [reject_handler],
        "default": [fallback_handler]
    })
])

parallel(steps: List) -> Parallel

workflow = Workflow(steps=[
    parallel([research_market, research_competitors, research_customers]),
    summarize_results  # Access via ctx.variables["parallel_outputs"]
])

loop(
    step: Any,                    # Step to execute for each item
    over: Optional[str] = None,   # Variable name containing list
    from_csv: Optional[str] = None,  # CSV file path
    from_file: Optional[str] = None, # Text file path
    var_name: str = "item"        # Variable name for current item
) -> Loop

# Loop over list variable
workflow = Workflow(
    steps=[loop(process_item, over="items")],
    variables={"items": ["a", "b", "c"]}
)

# Loop over CSV file
workflow = Workflow(steps=[
    loop(process_row, from_csv="data.csv")
])

def process_item(ctx: WorkflowContext) -> StepResult:
    item = ctx.variables["item"]  # Current item
    index = ctx.variables["loop_index"]  # Current index
    return StepResult(output=f"Processed: {item}")

repeat(
    step: Any,                                    # Step to repeat
    until: Optional[Callable[[WorkflowContext], bool]] = None,  # Stop condition
    max_iterations: int = 10                      # Maximum iterations
) -> Repeat

def is_complete(ctx: WorkflowContext) -> bool:
    return "done" in ctx.previous_result.lower()

workflow = Workflow(steps=[
    repeat(
        generator,
        until=is_complete,
        max_iterations=5
    )
])

workflow = Workflow(steps=[
    # Step 1: Parallel research
    parallel([research_a, research_b]),
    
    # Step 2: Route based on findings
    route({
        "positive": [expand_research],
        "negative": [summarize_and_stop]
    }),
    
    # Step 3: Iterate over results
    loop(process_finding, over="findings"),
    
    # Step 4: Repeat until quality threshold
    repeat(refine_output, until=is_high_quality, max_iterations=3)
])