The only orchestrator built for AI agents

Orchestrate state machines and human-in-the-loop workflows. No precompiled DAGs—Prefect follows Python's control flow dynamically, handling runtime branching and unpredictable agent decisions.

Agents don't follow predetermined paths

While loops until success. Runtime branching on LLM output. Human approval at unpredictable points. Traditional orchestrators require precompiled DAGs—but agents are state machines that decide their next step at runtime.

Dynamic Control Flow

Orchestrate state machines, not static DAGs

Prefect follows Python's control flow—while loops, runtime branching, conditional logic. No precompiled graphs means agent state machines work natively. Your agent decides the next step, and Prefect executes it.

Explore dynamic workflows

agent_workflow.py

1from prefect import flow, task
2from pydantic_ai import Agent
3
4@task
5def call_agent(prompt: str, context: dict):
6    agent = Agent("openai:gpt-4")
7    result = agent.run_sync(prompt, deps=context)
8    return result.data
9
10@flow
11def agent_state_machine(user_request: str):
12    context = {"attempts": 0}
13
14    # Agent-driven while loop - no predetermined steps
15    while context["attempts"] < 5:
16        result = call_agent(user_request, context)
17
18        # Runtime branching based on agent output
19        if result.status == "success":
20            return result
21        elif result.status == "needs_human":
22            # Pause for human input (covered next)
23            break
24        else:
25            context["attempts"] += 1
26            user_request = f"Try again: {result.error}"
27
28    return result

human_approval.py

1from prefect import flow, pause_flow_run
2from prefect.input import RunInput
3
4class AgentApproval(RunInput):
5    approved: bool
6    feedback: str = ""
7
8@flow
9def agent_with_human_approval():
10    # Agent executes autonomously
11    result = research_agent.run()
12
13    # Pause for human review - generates type-safe UI
14    approval = pause_flow_run(
15        wait_for_input=AgentApproval.with_initial_data(
16            description=f"Review agent research: {result}",
17            approved=False
18        )
19    )
20
21    if not approval.approved:
22        # Incorporate human feedback into next attempt
23        result = research_agent.run(
24            feedback=approval.feedback
25        )
26
27    return result

Human-in-the-Loop

Pause for human input at any point

Agents pause and wait for type-safe human input through auto-generated UI forms. Approval workflows, feedback loops, and compliance gates work natively without custom infrastructure.

Learn about interactive workflows

Native Integration

Works natively with Pydantic AI

Prefect wraps Pydantic AI agents with durable execution—automatic retries, result caching, and task-level observability. Your agent framework handles reasoning. Prefect handles production operations.

View Pydantic AI integration

pydantic_ai_integration.py

1from prefect import flow, task
2from pydantic_ai import Agent
3
4# Pydantic AI handles agent reasoning
5agent = Agent(
6    'openai:gpt-4',
7    system_prompt='You are a helpful assistant.'
8)
9
10# Prefect wraps with production reliability
11@task(retries=3, cache_result_in_memory=True)
12def call_agent(prompt: str):
13    # Cached result on retry - no redundant LLM calls
14    result = agent.run_sync(prompt)
15    return result.data
16
17@flow(log_prints=True)
18def production_agent_workflow():
19    # Each agent call = independent task with retries
20    research = call_agent("Research topic X")
21    summary = call_agent(f"Summarize: {research}")
22
23    print(f"Result: {summary}")
24    return summary
25
26# Deploy to production with scheduling
27if __name__ == "__main__":
28    production_agent_workflow.serve(
29        name="agent-workflow",
30        cron="0 9 * * *"  # Daily at 9am
31    )

durable_execution.py

1from prefect import flow, task
2
3@task(
4    retries=3,
5    cache_result_in_memory=True,  # Cache LLM responses
6    persist_result=True            # Durable across restarts
7)
8def expensive_llm_call(prompt: str):
9    response = openai.ChatCompletion.create(
10        model="gpt-4",
11        messages=[{"role": "user", "content": prompt}]
12    )
13    return response.choices[0].message.content
14
15@flow
16def multi_step_agent():
17    # Step 1: Expensive research (cached)
18    research = expensive_llm_call("Research topic X")
19
20    # Step 2: If this fails, retry WITHOUT re-calling step 1
21    summary = expensive_llm_call(f"Summarize: {research}")
22
23    # Step 3: Generate output
24    output = expensive_llm_call(f"Format as report: {summary}")
25
26    return output
27
28# On retry: Steps 1 & 2 load from cache, only step 3 re-executes

Durable Execution

Retry without re-calling LLMs

Automatic result caching means retries and reruns load cached LLM responses instead of making redundant API calls. Idempotent workflows that save cost and preserve agent state across failures.

Learn about task caching

New

Serve context to AI agents with Horizon

Prefect orchestrates workflows. Horizon serves context via MCP with enterprise governance. Build production AI systems with both.

Explore Horizon

Trusted by AI and ML teams

From agent workflows to model training—production-ready orchestration

Actium Healthtestimonial

I used parallelized hyperparameter tuning with Prefect and Dask to run 350 experiments in 30 minutes—normally would have taken 2 days

Andrew Waterman

Machine Learning

Cash Appcase-study

Airflow was no longer viable for ML workflows. We needed security and ease of adoption—Prefect delivered both

Wendy Tang

Machine Learning Engineer

Start building production agent workflows

Free tier includes 2 users and 5 deployments. Works with Pydantic AI, LangGraph, and any Python agent framework.

Get Started View Documentation

1from prefect import flow, task

2from pydantic_ai import Agent

4@task

5def call_agent(prompt: str, context: dict):

6 agent = Agent("openai:gpt-4")

7 result = agent.run_sync(prompt, deps=context)

8 return result.data

10@flow

11def agent_state_machine(user_request: str):

12 context = {"attempts": 0}

14 # Agent-driven while loop - no predetermined steps

15 while context["attempts"] < 5:

16 result = call_agent(user_request, context)

18 # Runtime branching based on agent output

19 if result.status == "success":

20 return result

21 elif result.status == "needs_human":

22 # Pause for human input (covered next)

23 break

24 else:

25 context["attempts"] += 1

26 user_request = f"Try again: {result.error}"

28 return result

1from prefect import flow, pause_flow_run

2from prefect.input import RunInput

4class AgentApproval(RunInput):

5 approved: bool

6 feedback: str = ""

8@flow

9def agent_with_human_approval():

10 # Agent executes autonomously

11 result = research_agent.run()

13 # Pause for human review - generates type-safe UI

14 approval = pause_flow_run(

15 wait_for_input=AgentApproval.with_initial_data(

16 description=f"Review agent research: {result}",

17 approved=False

18 )

19 )

21 if not approval.approved:

22 # Incorporate human feedback into next attempt

23 result = research_agent.run(

24 feedback=approval.feedback

25 )

27 return result

1from prefect import flow, task

2from pydantic_ai import Agent

4# Pydantic AI handles agent reasoning

5agent = Agent(

6 'openai:gpt-4',

7 system_prompt='You are a helpful assistant.'

10# Prefect wraps with production reliability

11@task(retries=3, cache_result_in_memory=True)

12def call_agent(prompt: str):

13 # Cached result on retry - no redundant LLM calls

14 result = agent.run_sync(prompt)

15 return result.data

17@flow(log_prints=True)

18def production_agent_workflow():

19 # Each agent call = independent task with retries

20 research = call_agent("Research topic X")

21 summary = call_agent(f"Summarize: {research}")

23 print(f"Result: {summary}")

24 return summary

26# Deploy to production with scheduling

27if __name__ == "__main__":

28 production_agent_workflow.serve(

29 name="agent-workflow",

30 cron="0 9 * * *" # Daily at 9am

31 )

1from prefect import flow, task

3@task(

4 retries=3,

5 cache_result_in_memory=True, # Cache LLM responses

6 persist_result=True # Durable across restarts

8def expensive_llm_call(prompt: str):

9 response = openai.ChatCompletion.create(

10 model="gpt-4",

11 messages=[{"role": "user", "content": prompt}]

12 )

13 return response.choices[0].message.content

15@flow

16def multi_step_agent():

17 # Step 1: Expensive research (cached)

18 research = expensive_llm_call("Research topic X")

20 # Step 2: If this fails, retry WITHOUT re-calling step 1

21 summary = expensive_llm_call(f"Summarize: {research}")

23 # Step 3: Generate output

24 output = expensive_llm_call(f"Format as report: {summary}")

26 return output

28# On retry: Steps 1 & 2 load from cache, only step 3 re-executes