Agno Workflow

상위: Agno

요약

Agno의 Workflow는 예측 가능한 다단계 파이프라인이다. Agent/Team이 동적 의사결정이라면, Workflow는 개발자가 단계를 미리 정의하는 구조적 오케스트레이션이다. Step, Condition, Parallel, Loop 4가지 구성 요소로 복잡한 실행 흐름을 표현한다.

Workflow vs Team

Step — 실행 단위

Step은 Workflow의 기본 빌딩 블록이다. 3가지 실행자 유형을 지원한다:

from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput

# Agent 스텝
research_step = Step(
    name="ResearchStep",
    description="Research the topic",
    agent=research_agent,
)

# Team 스텝
team_step = Step(
    name="TeamResearch",
    team=research_team,
)

# 커스텀 함수 스텝
def transform_content(step_input: StepInput) -> StepOutput:
    previous = step_input.previous_step_content or ""
    return StepOutput(
        step_name="TransformContent",
        content=f"[TRANSFORMED] {previous}",
        success=True,
    )

transform_step = Step(
    name="TransformContent",
    description="Transform the content",
    executor=transform_content,
)

StepInput / StepOutput

스텝 간 데이터 전달 구조:

Condition — 조건부 실행

evaluator 함수의 반환값(bool)에 따라 내부 steps를 실행하거나 건너뛴다.

from agno.workflow.condition import Condition
from agno.workflow.types import StepInput

def should_research(step_input: StepInput) -> bool:
    topic = step_input.input or ""
    keywords = ["ai", "tech", "programming"]
    return any(kw in topic.lower() for kw in keywords)

workflow = Workflow(
    steps=[
        Condition(
            name="ResearchCondition",
            description="Check if research is needed",
            evaluator=should_research,
            steps=[research_step, analysis_step],
        ),
        write_step,  # 조건과 무관하게 항상 실행
    ],
)

session_state 접근

evaluator에서 session_state 파라미터를 추가하면 워크플로우 상태에 접근 가능:

def check_user_context(step_input: StepInput, session_state: dict) -> bool:
    return session_state.get("has_been_greeted", False)

Parallel — 병렬 실행

여러 스텝을 동시에 실행한다. Condition과 조합 가능.

from agno.workflow.parallel import Parallel

workflow = Workflow(
    steps=[
        Parallel(
            research_hackernews_step,
            research_finance_step,
            name="ParallelResearch",
            description="Run research in parallel",
        ),
        write_step,
    ],
)

Condition + Parallel 조합

조건부로 병렬 실행하는 패턴. 실무에서 가장 많이 쓰이는 구조:

workflow = Workflow(
    steps=[
        Parallel(
            Condition(
                name="HNCondition",
                evaluator=check_tech,
                steps=[research_hn_step],
            ),
            Condition(
                name="FinanceCondition",
                evaluator=check_finance,
                steps=[research_finance_step],
            ),
            name="ConditionalResearch",
        ),
        prepare_step,
        write_step,
    ],
)

Loop — 반복 실행

end_condition 함수가 True를 반환하거나 max_iterations에 도달할 때까지 반복한다.

from agno.workflow.loop import Loop
from agno.workflow.types import StepOutput
from typing import List

def check_complete(outputs: List[StepOutput]) -> bool:
    """True를 반환하면 루프 종료."""
    if not outputs:
        return False
    for output in outputs:
        if output.content and len(output.content) > 500:
            return True
    return False

workflow = Workflow(
    steps=[
        Loop(
            name="ResearchLoop",
            steps=[research_step],
            end_condition=check_complete,
            max_iterations=3,
        ),
        summarize_step,
    ],
)

Session State

→ Agno Session State 참조. Workflow의 session_state 딕셔너리로 스텝 간 상태를 공유. Condition evaluator와 executor에서 접근.

HITL in Workflow

→ Agno HITL 참조. 스텝 단위 Confirmation, User Input (Schema 기반), Error Handling (retry/skip).

실행 모드

4가지 실행 모드를 지원한다:

스트리밍 이벤트 모니터링

from agno.run.workflow import StepPausedEvent

for event in workflow.run("input", stream=True, stream_events=True):
    if isinstance(event, StepPausedEvent):
        print(f"Paused at: {event.step_name}")

Save/Load + Registry

워크플로우를 DB에 영속화하고 복원할 수 있다. 커스텀 executor 함수는 Registry를 통해 이름 기반으로 복원한다.

from agno.registry import Registry
from agno.workflow.workflow import get_workflow_by_id

# 커스텀 함수를 Registry에 등록
registry = Registry(name="My Registry", functions=[transform_content])

# 저장
workflow.save(db=db)

# 로드 (registry로 커스텀 함수 복원)
loaded = get_workflow_by_id(db=db, id="my-workflow", registry=registry)

전체 예시 (종합)

from agno.agent import Agent
from agno.team import Team
from agno.workflow.workflow import Workflow
from agno.workflow.step import Step
from agno.workflow.condition import Condition
from agno.workflow.parallel import Parallel
from agno.db.sqlite import SqliteDb
from agno.tools.hackernews import HackerNewsTools
from agno.tools.yfinance import YFinanceTools

# 에이전트 정의
hn_agent = Agent(name="HN Researcher", tools=[HackerNewsTools()])
finance_agent = Agent(name="Finance Researcher", tools=[YFinanceTools()])
writer_agent = Agent(name="Writer", instructions="Write a report.")

# 워크플로우 구성
workflow = Workflow(
    name="Content Creation Workflow",
    description="Automated content creation pipeline",
    db=SqliteDb(session_table="workflow_session", db_file="tmp/workflow.db"),
    steps=[
        Parallel(
            Condition(
                name="TechCheck",
                evaluator=lambda si: "tech" in (si.input or "").lower(),
                steps=[Step(name="HN", agent=hn_agent)],
            ),
            Condition(
                name="FinanceCheck",
                evaluator=lambda si: "stock" in (si.input or "").lower(),
                steps=[Step(name="Finance", agent=finance_agent)],
            ),
            name="Research",
        ),
        Step(name="Write", agent=writer_agent),
    ],
)

workflow.print_response(input="AI tech and stock trends", stream=True)

관련 개념

• Agno: Agno 프레임워크 MOC
• Agno Agent: 단일 에이전트 구성
• Agno Team: 멀티 에이전트 협업 (Workflow와 비교)
• Agno Tools & Guardrails: HITL 패턴
• Agno AgentOS: 프로덕션 배포, Registry