Skip to main content

​
Overview

Fast Agent provides powerful workflow patterns to orchestrate multiple agents for complex tasks. Workflows enable you to chain agents, run them in parallel, route requests intelligently, and implement advanced patterns like evaluator-optimizer loops.

​
Available Workflows

Chain

Execute agents sequentially, passing output from one to the next

Parallel

Run multiple agents simultaneously and aggregate results

Router

Intelligently route requests to the most appropriate agent

Orchestrator

Generate plans and coordinate multiple agents for complex tasks

Evaluator-Optimizer

Iteratively improve output through evaluation and refinement

MAKER

Achieve consensus through voting to reduce errors

Agents as Tools

Expose agents as tools for flexible orchestration

​
Chain Workflow

Chain workflows execute agents sequentially, where each agent processes the output of the previous agent.

​
Basic Chain

@fast.agent(
    name="url_fetcher",
    instruction="Fetch and summarize content from URLs",
    servers=["fetch"]
)
@fast.agent(
    name="social_media",
    instruction="Write 280 character social media posts. No hashtags."
)
@fast.chain(
    name="post_writer",
    sequence=["url_fetcher", "social_media"]
)
async def main():
    async with fast.run() as agent:
        await agent.post_writer("https://example.com")

​
Chain Configuration

@fast.chain(
    name="chain_name",
    sequence=["agent1", "agent2", "agent3"],
    instruction="Description for other workflows",
    cumulative=False,  # Pass only last output vs all outputs
    continue_with_final=True  # Open chat after execution
)

​
Chain Modes

Default mode: Each agent receives only the previous agent’s output.
@fast.chain(
    name="translator",
    sequence=["translate_fr", "translate_de"],
    cumulative=False  # Default
)
Flow: Input → Agent1 → Agent2 → Output
Chain agents can be used within other workflows by providing an instruction parameter that describes their capabilities.

​
Parallel Workflow

Parallel workflows send the same input to multiple agents simultaneously and optionally aggregate their responses.

​
Basic Parallel

@fast.agent("translate_fr", "Translate text to French")
@fast.agent("translate_de", "Translate text to German")
@fast.agent("translate_es", "Translate text to Spanish")

@fast.parallel(
    name="translate_all",
    fan_out=["translate_fr", "translate_de", "translate_es"]
)
async def main():
    async with fast.run() as agent:
        # Returns combined translations
        result = await agent.translate_all("Hello, world!")

​
With Aggregation

@fast.agent("search_google", "Search Google")
@fast.agent("search_bing", "Search Bing")
@fast.agent("aggregator", "Combine search results")

@fast.parallel(
    name="multi_search",
    fan_out=["search_google", "search_bing"],
    fan_in="aggregator",  # Aggregates results
    include_request=True  # Include original query
)

​
Configuration Options

@fast.parallel(
    name="parallel_name",
    fan_out=["agent1", "agent2", "agent3"],
    fan_in="aggregator_agent",  # Optional
    instruction="Description",
    include_request=True  # Send original input to fan_in
)
Parallel workflows are excellent for:
  • Multi-model ensembles
  • Parallel data processing
  • Gathering diverse perspectives

​
Router Workflow

Routers use an LLM to analyze requests and route them to the most appropriate agent. 
@fast.agent("math_expert", "Solve mathematical problems", servers=["calculator"])
@fast.agent("code_expert", "Write and debug code", servers=["python"])
@fast.agent("writer", "Creative writing and editing")

@fast.router(
    name="smart_router",
    agents=["math_expert", "code_expert", "writer"],
    model="o3-mini.high"  # Use capable model for routing
)
async def main():
    async with fast.run() as agent:
        # Automatically routes to math_expert
        await agent.smart_router("What is 15% of 240?")
        
        # Automatically routes to code_expert
        await agent.smart_router("Write a Python function to sort a list")

​
Router Configuration

@fast.router(
    name="router",
    agents=["agent1", "agent2", "agent3"],
    model="o3-mini.high",  # Routing model
    use_history=False,     # Stateless routing
    human_input=False      # No human intervention
)
The routing decision is based on agent instructions and available tools. Write clear, distinctive instructions for best routing accuracy.

​
Orchestrator Workflow

Orchestrators generate execution plans and coordinate multiple agents to accomplish complex tasks. 
@fast.agent("researcher", "Research topics online", servers=["fetch"])
@fast.agent("analyzer", "Analyze and synthesize information")
@fast.agent("writer", "Write clear documentation")

@fast.orchestrator(
    name="report_generator",
    instruction="Generate comprehensive reports on any topic",
    agents=["researcher", "analyzer", "writer"],
    plan_type="full",      # "full" or "iterative"
    plan_iterations=5      # Max planning attempts
)
async def main():
    async with fast.run() as agent:
        await agent.report_generator(
            "Create a report on renewable energy trends"
        )

​
Planning Strategies

Generate a complete plan upfront, then execute all steps.
@fast.orchestrator(
    name="planner",
    agents=["task1", "task2", "task3"],
    plan_type="full",
    plan_iterations=3
)
Best for: Well-defined tasks with clear requirements

​
Evaluator-Optimizer

Iteratively improve outputs through cycles of generation and evaluation. 
@fast.agent(
    "web_searcher",
    "Search and compile information",
    servers=["fetch"]
)
@fast.agent(
    "quality_assurance",
    """Evaluate information quality and completeness.
    Rate as: EXCELLENT, GOOD, FAIR, or POOR.
    Provide specific improvement suggestions."""
)

@fast.evaluator_optimizer(
    name="researcher",
    generator="web_searcher",
    evaluator="quality_assurance",
    min_rating="GOOD",      # Minimum acceptable rating
    max_refinements=3        # Maximum iterations
)
async def main():
    async with fast.run() as agent:
        result = await agent.researcher(
            "Research the benefits of renewable energy"
        )

​
How It Works

  1. Generate: Generator agent produces initial output
  2. Evaluate: Evaluator agent rates the output and provides feedback
  3. Refine: If rating < min_rating, generator improves based on feedback
  4. Repeat: Continue until min_rating reached or max_refinements exceeded
  5. Return: Final generator output is returned

​
Configuration

@fast.evaluator_optimizer(
    name="optimizer",
    generator="content_creator",
    evaluator="quality_checker",
    min_rating="EXCELLENT",  # EXCELLENT, GOOD, FAIR, POOR
    max_refinements=3         # Stop after N refinements
)
Set reasonable max_refinements to prevent excessive iterations and token usage.

​
MAKER Workflow

MAKER (“Massively decomposed Agentic processes with K-voting Error Reduction”) samples a worker agent multiple times until reaching consensus. 
@fast.agent(
    name="classifier",
    instruction="Reply with only: A, B, or C."
)

@fast.maker(
    name="reliable_classifier",
    worker="classifier",
    k=3,                          # Voting margin
    max_samples=25,               # Maximum samples
    match_strategy="normalized",  # exact, normalized, structured
    red_flag_max_length=16        # Flag long outputs
)
async def main():
    async with fast.run() as agent:
        result = await agent.reliable_classifier(
            "Classify this text..."
        )

​
Match Strategies

  • exact: Responses must match exactly
  • normalized: Ignore whitespace and case differences
  • structured: Parse and compare structured data (JSON, etc.)
MAKER is ideal for tasks where occasional errors in simple operations would compound in longer workflows. Based on research achieving zero errors over million-step tasks.

​
Agents as Tools

Expose child agents as tools that a parent orchestrator can call flexibly. 
@fast.agent(
    name="NY-Project-Manager",
    instruction="Return NY time and project status",
    servers=["time"]
)
@fast.agent(
    name="London-Project-Manager",
    instruction="Return London time and news",
    servers=["time"]
)

@fast.agent(
    name="PMO-orchestrator",
    instruction="""
    Get reports from project managers. Always use one tool call per project.
    Responsibilities:
    - NY projects: OpenAI, Fast-Agent, Anthropic
    - London news: Economics, Art, Culture
    Aggregate results and add PMO summary.
    """,
    default=True,
    agents=["NY-Project-Manager", "London-Project-Manager"],
    max_parallel=128,          # Max parallel child calls
    child_timeout_sec=600,     # Per-child timeout
    max_display_instances=20   # UI display limit
)
async def main():
    async with fast.run() as agent:
        await agent("Get PMO report. Projects: all. News: Art, Culture")

​
Key Features

  • Routing: Orchestrator selects appropriate child agents
  • Parallelization: Multiple child agents run concurrently
  • Decomposition: Break complex tasks into subtasks
The orchestrator agent uses its instruction to determine when and how to call child agents. Write detailed instructions for best results.

​
Combining Workflows

Workflows can be nested and combined: 
@fast.agent("fetch", "Fetch URLs", servers=["fetch"])
@fast.agent("summarize", "Summarize content")
@fast.agent("translate_fr", "Translate to French")
@fast.agent("translate_de", "Translate to German")
@fast.agent("combine", "Combine translations")

# Chain to fetch and summarize
@fast.chain(
    name="fetch_and_summarize",
    sequence=["fetch", "summarize"]
)

# Parallel translation
@fast.parallel(
    name="translate_both",
    fan_out=["translate_fr", "translate_de"],
    fan_in="combine"
)

# Chain the workflows together
@fast.chain(
    name="full_pipeline",
    sequence=["fetch_and_summarize", "translate_both"]
)

​
Workflow Best Practices

  • Chain: Sequential processing where order matters
  • Parallel: Independent operations that can run concurrently
  • Router: Multiple specialized agents, dynamic selection
  • Orchestrator: Complex tasks requiring planning
  • Evaluator-Optimizer: Quality-critical outputs
  • MAKER: High-reliability requirements
Use capable models (o3-mini.high, sonnet) for routing, orchestration, and evaluation. Worker agents can use faster/cheaper models.
Write specific instructions for each agent. Workflows rely on clear agent capabilities for routing and planning.
Set reasonable iteration limits (max_refinements, plan_iterations) to prevent runaway token usage.
Test individual agents before combining them in workflows. Use --agent flag to test specific agents.

​
Running Workflows

​
From Code

async with fast.run() as agent:
    # Call workflow by name
    result = await agent.workflow_name("input")
    
    # Or use send method
    result = await agent.workflow_name.send("input")
    
    # Interactive mode
    await agent.workflow_name.prompt()

​
From CLI

# Run specific workflow
uv run app.py --agent workflow_name --message "input"

# Quiet mode (no progress display)
uv run app.py --agent workflow_name --message "input" --quiet

# Interactive mode
uv run app.py --agent workflow_name

​
Next Steps

Agent Architecture

Learn about agent types and lifecycle

MCP Integration

Connect agents to external tools and data

Examples

See workflow examples in action

Configuration

Configure workflow behavior

Build docs developers (and LLMs) love

Get started for freeTalk to us