from praisonaiagents.memory import WorkflowManager, Workflow, WorkflowStep

WorkflowStep(
    name: str,
    description: str = "",
    action: str = "",
    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,
    agent_config: Optional[Dict[str, Any]] = None,
    tools: Optional[List[Any]] = None
)

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"
)

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
) -> 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}}"
        )
    ]
)