Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.definable.ai/llms.txt

Use this file to discover all available pages before exploring further.

​
Sequential Pipeline

Steps execute in order, each receiving the previous step’s output. 
import asyncio
from definable.agent import Agent
from definable.agent.workflow import Workflow, Step

researcher = Agent(model="gpt-4o", instructions="Research the topic thoroughly.")
writer = Agent(model="gpt-4o", instructions="Write an article based on the research.")

workflow = Workflow(
    name="research-pipeline",
    steps=[
        Step(name="research", agent=researcher),
        Step(name="write", agent=writer),
    ],
)

async def main():
    result = await workflow.arun("The future of AI agents")
    print(result.content)
    print(f"Duration: {result.duration_ms:.0f}ms")

asyncio.run(main())

​
Parallel Execution

Run multiple steps concurrently. 
from definable.agent.workflow import Workflow, Parallel, Step

workflow = Workflow(
    name="multi-analysis",
    steps=[
        Parallel(name="analysis", steps=[
            Step(name="technical", agent=tech_analyst),
            Step(name="business", agent=biz_analyst),
            Step(name="legal", agent=legal_analyst),
        ]),
        Step(name="synthesizer", agent=synthesizer),
    ],
)

​
Conditional Branching

If/else branching based on step output. 
from definable.agent.workflow import Workflow, Step, Condition

workflow = Workflow(
    name="review-pipeline",
    steps=[
        Step(name="draft", agent=drafter),
        Step(name="review", agent=reviewer),
        Condition(
            name="quality-gate",
            condition=lambda ctx: "APPROVED" in (ctx.get_last_step_content() or ""),
            true_steps=Step(name="publish", agent=publisher),
            false_steps=Step(name="rewrite", agent=writer),
        ),
    ],
)

​
Iterative Loop

Repeat until an end condition is met. 
from definable.agent.workflow import Workflow, Loop, Step

workflow = Workflow(
    name="refinement",
    steps=[
        Loop(
            name="improve",
            steps=[
                Step(name="generate", agent=generator),
                Step(name="evaluate", agent=evaluator),
            ],
            end_condition=lambda outputs: any("APPROVED" in (o.content or "") for o in outputs),
            max_iterations=5,
        ),
    ],
)

​
Dynamic Router

Route to different steps based on input. 
from definable.agent.workflow import Workflow, Router, Step

workflow = Workflow(
    name="support",
    steps=[
        Router(
            name="classifier",
            selector=lambda ctx: "technical" if "bug" in (ctx.input or "").lower() else "general",
            routes={
                "technical": Step(name="tech", agent=tech_agent),
                "general": Step(name="general", agent=general_agent),
            },
        ),
    ],
)

​
Custom Function Steps

Use callables instead of agents. 
async def process(step_input):
    data = step_input.get_last_step_content()
    return f"Processed: {data[:100]}"

workflow = Workflow(
    name="mixed",
    steps=[
        Step(name="research", agent=researcher),
        Step(name="process", executor=process),
        Step(name="write", agent=writer),
    ],
)

​
Accessing Step Results

result = await workflow.arun("Research AI trends")

result.content            # Final step's content
result.success            # True if all steps succeeded
result.duration_ms        # Total execution time
result.step_outputs       # List of all step outputs

research = result.get_step_output("research")
print(research.content)
print(research.duration_ms)