Multi-Component Decision Tree

Decision tree mixing Condition and Loop components with user confirmation at each pause point.

Demonstrates a decision tree mixing different HITL component types: Condition -> Loop -> Condition. Each pauses for user confirmation, enabling complex interactive workflows.

multi_component_decision_tree.py
"""
Multi-Component Decision Tree

Demonstrates a decision tree mixing different HITL component types:
Condition -> Loop -> Condition. Each pauses for user confirmation,
enabling complex interactive workflows.

Pattern:
1. Condition HITL: Choose analysis approach
2. Loop HITL: Confirm before starting iterative refinement
3. Condition HITL: Choose output format
"""

from agno.db.sqlite import SqliteDb
from agno.workflow import OnReject
from agno.workflow.condition import Condition
from agno.workflow.loop import Loop
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput
from agno.workflow.workflow import Workflow


# ============================================================
# Step functions
# ============================================================
def detailed_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(
        content="Detailed analysis: all metrics computed, edge cases covered."
    )


def quick_summary(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Quick summary: key highlights identified.")


def refine_results(step_input: StepInput) -> StepOutput:
    prev = step_input.previous_step_content or ""
    return StepOutput(
        content=f"Refinement pass complete. Quality improved.\nPrevious: {prev[:80]}..."
    )


def formal_report(step_input: StepInput) -> StepOutput:
    prev = step_input.previous_step_content or "No analysis"
    return StepOutput(content=f"=== FORMAL REPORT ===\n{prev}")


def internal_notes(step_input: StepInput) -> StepOutput:
    prev = step_input.previous_step_content or "No analysis"
    return StepOutput(content=f"--- Internal Notes ---\n{prev}")


# ============================================================
# Build workflow: Condition -> Loop -> Condition
# ============================================================
workflow = Workflow(
    name="multi_component_tree",
    db=SqliteDb(db_file="tmp/multi_component_tree.db"),
    steps=[
        # Decision 1: Analysis depth
        Condition(
            name="analysis_depth",
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Run detailed analysis? (No = quick summary)",
                on_reject=OnReject.else_branch,
            ),
            steps=[Step(name="detailed", executor=detailed_analysis)],
            else_steps=[Step(name="quick", executor=quick_summary)],
        ),
        # Decision 2: Optional refinement loop
        Loop(
            name="refinement",
            steps=[Step(name="refine", executor=refine_results)],
            max_iterations=3,
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Start iterative refinement? (up to 3 passes)",
                on_reject=OnReject.skip,
            ),
        ),
        # Decision 3: Output format
        Condition(
            name="output_format",
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Generate formal report? (No = internal notes)",
                on_reject=OnReject.else_branch,
            ),
            steps=[Step(name="formal", executor=formal_report)],
            else_steps=[Step(name="notes", executor=internal_notes)],
        ),
    ],
)

if __name__ == "__main__":
    print("Multi-Component Decision Tree")
    print("=" * 50)

    run_output = workflow.run("Quarterly performance review")

    while run_output.is_paused:
        for req in run_output.steps_requiring_confirmation:
            print(f"\n[Decision] {req.step_name}")
            print(f"  {req.confirmation_message}")

            choice = input("\n  Your choice (yes/no): ").strip().lower()

            if choice in ("yes", "y"):
                req.confirm()
                print("  -> Confirmed")
            else:
                req.reject()
                print("  -> Rejected")

        run_output = workflow.continue_run(run_output)

    print("\n" + "=" * 50)
    print(f"Status: {run_output.status}")
    print("=" * 50)
    print(run_output.content)

Carry refinement results forward

The source’s refinement loop repeats the same incoming analysis on each pass. To build on the previous pass, add forward_iteration_output=True to the Loop(name="refinement", ...) constructor in your saved copy. The loop still runs at most three passes; the start confirmation approves the loop as a whole.

Run the Example

Set up your virtual environment

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

Install dependencies

uv pip install -U agno fastapi sqlalchemy

Run the example

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

python multi_component_decision_tree.py

Full source: cookbook/04_workflows/08_human_in_the_loop/decision_tree/02_multi_component_decision_tree.py