Output Review HITL Example (AgentOS)

Pausing after a step runs so a human can review the step output before the workflow continues in AgentOS.

output_review.py

"""
Output Review HITL Example (AgentOS)

This example demonstrates pausing after a step runs so a human can review the
step output before the workflow continues in AgentOS.
"""

from agno.os import AgentOS
from agno.workflow import HumanReview, OnReject
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
from workflow_db import db


def draft_email(step_input: StepInput) -> StepOutput:
    topic = step_input.input or "the schedule change"
    return StepOutput(
        content=f"Subject: Update: {topic}\n\n"
        "Hi team,\n\n"
        f"Please note that {topic}. Let me know if this creates any conflicts.\n\n"
        "Thanks"
    )


def send_email(step_input: StepInput) -> StepOutput:
    approved_draft = step_input.previous_step_content or "No approved draft"
    return StepOutput(content=f"Email queued for sending:\n\n{approved_draft}")


workflow = Workflow(
    name="output_review_workflow",
    description="Review a drafted email before it moves to the send step.",
    db=db,
    steps=[
        Step(
            name="draft_email",
            executor=draft_email,
            human_review=HumanReview(
                requires_output_review=True,
                output_review_message="Review this email draft before sending.",
                on_reject=OnReject.cancel,
            ),
        ),
        Step(name="send_email", executor=send_email),
    ],
)
workflow.id = "workflow-hitl-output-review"

agent_os = AgentOS(
    name="Workflow Output Review HITL",
    description="AgentOS workflow example for post-step output review.",
    workflows=[workflow],
    db=db,
)
app = agent_os.get_app()


if __name__ == "__main__":
    agent_os.serve(app="output_review:app", reload=True)

The example imports this helper module from the same directory:

workflow_db.py

from agno.db.postgres import PostgresDb

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

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[os]" "psycopg[binary]"
   
3. 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
   
4. Run the example

   Save the code blocks above as output_review.py and workflow_db.py in the same directory, then run:

   python output_review.py
   

Full source: cookbook/05_agent_os/human_in_the_loop/workflow/output_review.py