Save Parallel Workflow Steps

Save a Parallel research pipeline (HackerNews plus web) to Postgres and reload it by ID without a Registry, since all steps are agent-backed.

Demonstrates creating a workflow with parallel steps, saving it to the database, and loading it back.

save_parallel_steps.py

"""
Save Parallel Workflow Steps
============================

Demonstrates creating a workflow with parallel steps, saving it to the
database, and loading it back.
"""

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow.parallel import Parallel
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow, get_workflow_by_id

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
# Database
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)

# ---------------------------------------------------------------------------
# Create Agents
# ---------------------------------------------------------------------------
# Agents
hackernews_researcher = Agent(
    name="HackerNews Researcher",
    instructions="Research tech news and trends from Hacker News",
    tools=[HackerNewsTools()],
)

web_researcher = Agent(
    name="Web Researcher",
    instructions="Research general information from the web",
    tools=[WebSearchTools()],
)

writer = Agent(
    name="Content Writer",
    instructions="Write well-structured content from research findings",
)

reviewer = Agent(
    name="Content Reviewer",
    instructions="Review and improve the written content",
)

# ---------------------------------------------------------------------------
# Create Workflow Steps
# ---------------------------------------------------------------------------
# Steps
research_hn_step = Step(
    name="ResearchHackerNews",
    description="Research tech news from Hacker News",
    agent=hackernews_researcher,
)

research_web_step = Step(
    name="ResearchWeb",
    description="Research information from the web",
    agent=web_researcher,
)

write_step = Step(
    name="WriteArticle",
    description="Write article from research findings",
    agent=writer,
)

review_step = Step(
    name="ReviewArticle",
    description="Review and finalize the article",
    agent=reviewer,
)

# ---------------------------------------------------------------------------
# Create Workflow
# ---------------------------------------------------------------------------
# Workflow
workflow = Workflow(
    name="Parallel Research Pipeline",
    description="Research from multiple sources in parallel, then write and review",
    steps=[
        Parallel(
            research_hn_step,
            research_web_step,
            name="ParallelResearch",
            description="Run HackerNews and Web research in parallel",
        ),
        write_step,
        review_step,
    ],
    db=db,
)

# ---------------------------------------------------------------------------
# Run Workflow Example
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    # Save
    print("Saving workflow...")
    version = workflow.save(db=db)
    print(f"Saved workflow as version {version}")

    # Load
    print("\nLoading workflow...")
    loaded_workflow = get_workflow_by_id(db=db, id="parallel-research-pipeline")

    if loaded_workflow:
        print("Workflow loaded successfully!")
        print(f"  Name: {loaded_workflow.name}")
        print(f"  Steps: {len(loaded_workflow.steps) if loaded_workflow.steps else 0}")

        # Uncomment to run the loaded workflow
        # loaded_workflow.print_response(input="Latest developments in AI agents", stream=True)
    else:
        print("Workflow not found")

Run the Example

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   

   uv venv --python 3.12
   .venv\Scripts\activate
   
2. Install dependencies

   uv pip install -U agno "psycopg[binary]" ddgs fastapi openai sqlalchemy
   
3. Export your OpenAI API key

   Mac/Linux

   export OPENAI_API_KEY="your_openai_api_key_here"
   

   $Env:OPENAI_API_KEY="your_openai_api_key_here"
   
4. Run PgVector

   macOS / Linux

   docker run -d \
     -e POSTGRES_DB=ai \
     -e POSTGRES_USER=ai \
     -e POSTGRES_PASSWORD=ai \
     -e PGDATA=/var/lib/postgresql \
     -v pgvolume:/var/lib/postgresql \
     -p 5532:5432 \
     --name pgvector \
     agnohq/pgvector:18
   

   docker run -d `
     -e POSTGRES_DB=ai `
     -e POSTGRES_USER=ai `
     -e POSTGRES_PASSWORD=ai `
     -e PGDATA=/var/lib/postgresql `
     -v pgvolume:/var/lib/postgresql `
     -p 5532:5432 `
     --name pgvector `
     agnohq/pgvector:18
   
5. Run the example

   Save the code above as save_parallel_steps.py, then run:

   python save_parallel_steps.py
   

Full source: cookbook/93_components/workflows/save_parallel_steps.py