Workflow

A Workflow coordinates multiple agents through an explicitly ordered pipeline that you implement in Python. Unlike Graph and Swarm, there is no framework orchestrator class — you control every step, branch, and retry directly.

Not a framework class

Workflow is a pattern, not an import from elsai.multiagent. You call agents in sequence (or in parallel) inside your own functions. For framework-managed dependency graphs, use Graph instead.

Workflow pipeline

1. Define agents — Create specialist Agent instances with focused system prompts.
2. Write pipeline — Call agents in order inside a function, passing each output as the next input.
3. Control flow — Add conditionals, retries, or parallel execution with standard Python.
4. Return — Collect and return the final result from your function.

Usage

Sequential pipeline — each agent's output feeds the next:

from elsai import Agent
from elsai.agent import AgentConfig

researcher = Agent(
    system_prompt="You are a research specialist. Find accurate facts about the given topic.",
    config=AgentConfig(name="researcher"),
)
analyst = Agent(
    system_prompt="You analyse research data and extract key insights.",
    config=AgentConfig(name="analyst"),
)
writer = Agent(
    system_prompt="You write polished, reader-friendly reports.",
    config=AgentConfig(name="writer"),
)

def run_pipeline(topic: str) -> str:
    research = researcher(f"Research: {topic}")
    analysis = analyst(f"Analyse this research:\n\n{research}")
    report = writer(f"Write a report based on this analysis:\n\n{analysis}")
    return str(report)

print(run_pipeline("the impact of AI on healthcare"))

Parallel execution — run independent steps concurrently, then merge:

import concurrent.futures
from elsai import Agent
from elsai.agent import AgentConfig

researcher = Agent(system_prompt="Research the given topic thoroughly.", config=AgentConfig(name="researcher"))
data_agent = Agent(system_prompt="Find statistics and figures.", config=AgentConfig(name="data"))
analyst = Agent(system_prompt="Synthesise research and data into insights.", config=AgentConfig(name="analyst"))
writer = Agent(system_prompt="Write a polished report.", config=AgentConfig(name="writer"))

def run_parallel_pipeline(topic: str) -> str:
    with concurrent.futures.ThreadPoolExecutor() as pool:
        fut_research = pool.submit(researcher, f"Research: {topic}")
        fut_data = pool.submit(data_agent, f"Find statistics on: {topic}")
        research = str(fut_research.result())
        data = str(fut_data.result())

    insights = analyst(f"Synthesise:\nResearch: {research}\nData: {data}")
    return str(writer(f"Write a report:\n{insights}"))

Error recovery — retry individual steps because you own the execution loop:

import time

def run_with_retry(agent: Agent, prompt: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            return str(agent(prompt))
        except Exception:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)
    return ""

research = run_with_retry(researcher, f"Research: {topic}")