GraphFlow: 워크플로우와 스케줄링

GraphFlow: 워크플로우와 스케줄링

에이전트가 정해진 순서대로, 때로는 조건에 따라 실행되도록 제어하고 싶을 때가 있어요. autogen_agentchat.teams.GraphFlow(줄여서 "플로우")는 바로 이걸 위한 멀티에이전트 워크플로우를 만들어요. 구조화된 실행을 사용해 에이전트가 작업을 수행하는 방식을 정밀하게 제어하죠. 먼저 플로우를 만들고 실행하는 방법을 보여주고, 이어서 플로우 동작을 관찰·디버깅하는 방법과 실행을 관리하는 중요한 연산들을 설명할게요.

출처: 공식문서

AutoGen AgentChat은 방향 그래프(directed graph) 실행을 위한 팀을 제공해요.

• autogen_agentchat.teams.GraphFlow: autogen_agentchat.teams.DiGraph를 따라 에이전트 간 실행 흐름을 제어하는 팀. 순차·병렬·조건 분기·루프 동작을 지원해요.

**언제 `GraphFlow`를 써야 할까?**

에이전트가 동작하는 순서를 엄격히 제어해야 하거나, 서로 다른 결과가 서로 다른 다음 단계로 이어져야 할 때 Graph를 쓰세요. 애드혹 대화 흐름으로 충분하다면 `RoundRobinGroupChat`이나 `SelectorGroupChat` 같은 단순한 팀으로 시작하세요. 결정적 제어, 조건 분기, 사이클이 있는 복잡한 다단계 프로세스가 필요할 때 구조화된 워크플로우로 전환하면 됩니다.

Warning: GraphFlow는 실험적 기능이에요. API·동작·기능은 향후 릴리스에서 변경될 수 있습니다.

플로우 만들고 실행하기

DiGraphBuilder는 워크플로우를 위한 실행 그래프를 쉽게 구성하게 해주는 fluent 유틸리티예요. 다음을 지원해요.

• 순차 체인
• 병렬 팬아웃
• 조건 분기
• 안전한 종료 조건이 있는 루프

그래프의 각 노드는 에이전트를 나타내고, 엣지는 허용된 실행 경로를 정의해요. 엣지는 선택적으로 에이전트 메시지에 기반한 조건을 가질 수 있어요.

순차 플로우

**작가(writer)**가 문단을 작성하고 **리뷰어(reviewer)**가 피드백을 주는 간단한 워크플로우로 시작할게요. 이 그래프는 리뷰어가 작가의 글에 댓글을 단 뒤 종료돼요. 참고로 플로우는 그래프의 모든 소스·리프 노드를 자동으로 계산하고, 그래프의 모든 소스 노드에서 실행을 시작해 실행할 노드가 없을 때 완료됩니다.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create an OpenAI model client
client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

# Create the writer agent
writer = AssistantAgent("writer", model_client=client, system_message="Draft a short paragraph on climate change.")

# Create the reviewer agent
reviewer = AssistantAgent("reviewer", model_client=client, system_message="Review the draft and suggest improvements.")

# Build the graph
builder = DiGraphBuilder()
builder.add_node(writer).add_node(reviewer)
builder.add_edge(writer, reviewer)

# Build and validate the graph
graph = builder.build()

# Create the flow
flow = GraphFlow([writer, reviewer], graph=graph)

# Use `asyncio.run(...)` and wrap the below in a async function when running in a script.
stream = flow.run_stream(task="Write a short paragraph about climate change.")
async for event in stream:  # type: ignore
    print(event)
# Use Console(flow.run_stream(...)) for better formatting in console.

Join이 있는 병렬 플로우

이번엔 조금 더 복잡한 플로우를 만들게요.

• 작가가 문단을 작성해요.
• 두 명의 편집자가 문법과 스타일에 대해 각각 독립적으로 수정해요(병렬 팬아웃).
• 최종 리뷰어가 그 수정들을 통합해요(join).

실행은 작가에서 시작해 editor1과 editor2로 동시에 팬아웃되고, 둘 다 최종 리뷰어로 합쳐집니다.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create an OpenAI model client
client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

# Create the writer agent
writer = AssistantAgent("writer", model_client=client, system_message="Draft a short paragraph on climate change.")

# Create two editor agents
editor1 = AssistantAgent("editor1", model_client=client, system_message="Edit the paragraph for grammar.")

editor2 = AssistantAgent("editor2", model_client=client, system_message="Edit the paragraph for style.")

# Create the final reviewer agent
final_reviewer = AssistantAgent(
    "final_reviewer",
    model_client=client,
    system_message="Consolidate the grammar and style edits into a final version.",
)

# Build the workflow graph
builder = DiGraphBuilder()
builder.add_node(writer).add_node(editor1).add_node(editor2).add_node(final_reviewer)

# Fan-out from writer to editor1 and editor2
builder.add_edge(writer, editor1)
builder.add_edge(writer, editor2)

# Fan-in both editors into final reviewer
builder.add_edge(editor1, final_reviewer)
builder.add_edge(editor2, final_reviewer)

# Build and validate the graph
graph = builder.build()

# Create the flow
flow = GraphFlow(
    participants=builder.get_participants(),
    graph=graph,
)

# Run the workflow
await Console(flow.run_stream(task="Write a short paragraph about climate change."))

메시지 필터링

실행 그래프 vs 메시지 그래프

GraphFlow에서 **실행 그래프(execution graph)**는 DiGraph로 정의되며 에이전트가 실행되는 순서를 제어해요. 하지만 실행 그래프는 에이전트가 다른 에이전트로부터 어떤 메시지를 받을지는 제어하지 않아요. 기본적으로 모든 메시지는 그래프의 모든 에이전트에게 전송됩니다.

메시지 필터링은 각 에이전트가 받는 메시지를 필터링해 모델 컨텍스트를 관련 정보로만 제한하는 별도의 기능이에요. 메시지 필터의 집합이 플로우의 **메시지 그래프(message graph)**를 정의합니다. 메시지 그래프를 지정하면 다음에 도움이 돼요.

• 환각(hallucination) 줄이기
• 메모리 부하 제어
• 에이전트가 관련 정보에만 집중하도록 하기

이 규칙을 정의하려면 MessageFilterAgent를 MessageFilterConfig 및 PerSourceFilter와 함께 사용할 수 있어요.

from autogen_agentchat.agents import AssistantAgent, MessageFilterAgent, MessageFilterConfig, PerSourceFilter
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Model client
client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

# Create agents
researcher = AssistantAgent(
    "researcher", model_client=client, system_message="Summarize key facts about climate change."
)
analyst = AssistantAgent("analyst", model_client=client, system_message="Review the summary and suggest improvements.")
presenter = AssistantAgent(
    "presenter", model_client=client, system_message="Prepare a presentation slide based on the final summary."
)

# Apply message filtering
filtered_analyst = MessageFilterAgent(
    name="analyst",
    wrapped_agent=analyst,
    filter=MessageFilterConfig(per_source=[PerSourceFilter(source="researcher", position="last", count=1)]),
)

filtered_presenter = MessageFilterAgent(
    name="presenter",
    wrapped_agent=presenter,
    filter=MessageFilterConfig(per_source=[PerSourceFilter(source="analyst", position="last", count=1)]),
)

# Build the flow
builder = DiGraphBuilder()
builder.add_node(researcher).add_node(filtered_analyst).add_node(filtered_presenter)
builder.add_edge(researcher, filtered_analyst).add_edge(filtered_analyst, filtered_presenter)

# Create the flow
flow = GraphFlow(
    participants=builder.get_participants(),
    graph=builder.build(),
)

# Run the flow
await Console(flow.run_stream(task="Summarize key facts about climate change."))

🔁 고급 예제: 조건 루프 + 필터링된 요약

이 예제는 다음을 보여줘요.

• 생성기(generator)와 리뷰어(reviewer) 사이의 루프(리뷰어가 "APPROVE"라고 하면 종료)
• 첫 사용자 입력과 마지막 리뷰어 메시지만 보는 요약(summarizer) 에이전트

from autogen_agentchat.agents import AssistantAgent, MessageFilterAgent, MessageFilterConfig, PerSourceFilter
from autogen_agentchat.teams import (
    DiGraphBuilder,
    GraphFlow,
)
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

# Agents
generator = AssistantAgent("generator", model_client=model_client, system_message="Generate a list of creative ideas.")
reviewer = AssistantAgent(
    "reviewer",
    model_client=model_client,
    system_message="Review ideas and provide feedbacks, or just 'APPROVE' for final approval.",
)
summarizer_core = AssistantAgent(
    "summary", model_client=model_client, system_message="Summarize the user request and the final feedback."
)

# Filtered summarizer
filtered_summarizer = MessageFilterAgent(
    name="summary",
    wrapped_agent=summarizer_core,
    filter=MessageFilterConfig(
        per_source=[
            PerSourceFilter(source="user", position="first", count=1),
            PerSourceFilter(source="reviewer", position="last", count=1),
        ]
    ),
)

# Build graph with conditional loop
builder = DiGraphBuilder()
builder.add_node(generator).add_node(reviewer).add_node(filtered_summarizer)
builder.add_edge(generator, reviewer)
builder.add_edge(reviewer, filtered_summarizer, condition=lambda msg: "APPROVE" in msg.to_model_text())
builder.add_edge(reviewer, generator, condition=lambda msg: "APPROVE" not in msg.to_model_text())
builder.set_entry_point(generator)  # Set entry point to generator. Required if there are no source nodes.
graph = builder.build()

termination_condition = MaxMessageTermination(10)

# Create the flow
flow = GraphFlow(
    participants=builder.get_participants(),
    graph=graph,
    termination_condition=termination_condition
)

# Run the flow and pretty print the output in the console
await Console(flow.run_stream(task="Brainstorm ways to reduce plastic waste."))

🔁 고급 예제: Activation Group이 있는 사이클

다음 예제들은 activation_group과 activation_condition을 사용해 사이클 그래프의 복잡한 의존성 패턴을 처리하는 방법을 보여줘요. 특히 여러 경로가 같은 대상 노드로 이어질 때 유용합니다.

예제 1: 여러 경로가 있는 루프 — "All" Activation (A→B→C→B)

이 시나리오는 A → B → C → B 형태로, B는 두 개의 들어오는 엣지(A와 C에서)를 가져요. 기본적으로 B는 실행 전에 모든 의존성이 충족되기를 요구합니다. 이 예제는 초기 입력(A)과 피드백(C)이 모두 처리된 뒤에야 B가 다시 실행될 수 있는 리뷰 루프를 보여줘요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Model client
client = OpenAIChatCompletionClient(model="gpt-4o-mini")

# Create agents for A→B→C→B→E scenario
agent_a = AssistantAgent("A", model_client=client, system_message="Start the process and provide initial input.")
agent_b = AssistantAgent(
    "B",
    model_client=client,
    system_message="Process input from A or feedback from C. Say 'CONTINUE' if it's from A or 'STOP' if it's from C.",
)
agent_c = AssistantAgent("C", model_client=client, system_message="Review B's output and provide feedback.")
agent_e = AssistantAgent("E", model_client=client, system_message="Finalize the process.")

# Build the graph with activation groups
builder = DiGraphBuilder()
builder.add_node(agent_a).add_node(agent_b).add_node(agent_c).add_node(agent_e)

# A → B (initial path)
builder.add_edge(agent_a, agent_b, activation_group="initial")

# B → C
builder.add_edge(agent_b, agent_c, condition="CONTINUE")

# C → B (loop back - different activation group)
builder.add_edge(agent_c, agent_b, activation_group="feedback")

# B → E (exit condition)
builder.add_edge(agent_b, agent_e, condition="STOP")

termination_condition = MaxMessageTermination(10)
# Build and create flow
graph = builder.build()
flow = GraphFlow(participants=[agent_a, agent_b, agent_c, agent_e], graph=graph, termination_condition=termination_condition)

print("=== Example 1: A→B→C→B with 'All' Activation ===")
print("B will exit when it receives a message from C")
# await Console(flow.run_stream(task="Start a review process for a document."))

예제 2: 여러 경로가 있는 루프 — "Any" Activation (A→B→(C1,C2)→B)

이번엔 더 복잡한 시나리오인 A → B → (C1, C2) → B 를 볼게요.

• B는 C1과 C2로 병렬 팬아웃
• C1과 C2 모두 B로 피드백
• B는 "any" 활성화를 사용해 C1이나 C2 중 하나라도 완료되면 바로 실행

이건 가장 빠른 응답이 다음 단계를 트리거하게 만들고 싶은 시나리오에 유용해요.

# Create agents for A→B→(C1,C2)→B scenario
agent_a2 = AssistantAgent("A", model_client=client, system_message="Initiate a task that needs parallel processing.")
agent_b2 = AssistantAgent(
    "B",
    model_client=client,
    system_message="Coordinate parallel tasks. Say 'PROCESS' to start parallel work or 'DONE' to finish.",
)
agent_c1 = AssistantAgent("C1", model_client=client, system_message="Handle task type 1. Say 'C1_COMPLETE' when done.")
agent_c2 = AssistantAgent("C2", model_client=client, system_message="Handle task type 2. Say 'C2_COMPLETE' when done.")
agent_e = AssistantAgent("E", model_client=client, system_message="Finalize the process.")

# Build the graph with "any" activation
builder2 = DiGraphBuilder()
builder2.add_node(agent_a2).add_node(agent_b2).add_node(agent_c1).add_node(agent_c2).add_node(agent_e)

# A → B (initial)
builder2.add_edge(agent_a2, agent_b2)

# B → C1 and B → C2 (parallel fan-out)
builder2.add_edge(agent_b2, agent_c1, condition="PROCESS")
builder2.add_edge(agent_b2, agent_c2, condition="PROCESS")

# B → E (exit condition)
builder2.add_edge(agent_b2, agent_e, condition=lambda msg: "DONE" in msg.to_model_text())

# C1 → B and C2 → B (both in same activation group with "any" condition)
builder2.add_edge(
    agent_c1, agent_b2, activation_group="loop_back_group", activation_condition="any", condition="C1_COMPLETE"
)

builder2.add_edge(
    agent_c2, agent_b2, activation_group="loop_back_group", activation_condition="any", condition="C2_COMPLETE"
)

# Build and create flow
graph2 = builder2.build()
flow2 = GraphFlow(participants=[agent_a2, agent_b2, agent_c1, agent_c2, agent_e], graph=graph2)

print("=== Example 2: A→B→(C1,C2)→B with 'Any' Activation ===")
print("B will execute as soon as EITHER C1 OR C2 completes (whichever finishes first)")
# await Console(flow2.run_stream(task="Start a parallel processing task."))

예제 3: 혼합 Activation Group

이 예제는 서로 다른 activation group이 같은 그래프에 공존할 수 있음을 보여줘요. 다음 시나리오를 다룹니다.

• 노드 D는 서로 다른 활성화 요구사항을 가진 여러 소스에서 입력을 받아요.
• 일부 의존성은 "all" 활성화(모든 입력을 기다려야 함)
• 다른 의존성은 "any" 활성화(첫 입력이 오면 진행)

이 패턴은 서로 다른 타입의 의존성이 다른 긴급도를 가진 복잡한 워크플로우에 유용해요.

# Create agents for mixed activation scenario
agent_a3 = AssistantAgent("A", model_client=client, system_message="Provide critical input that must be processed.")
agent_b3 = AssistantAgent("B", model_client=client, system_message="Provide secondary critical input.")
agent_c3 = AssistantAgent("C", model_client=client, system_message="Provide optional quick input.")
agent_d3 = AssistantAgent("D", model_client=client, system_message="Process inputs based on different priority levels.")

# Build graph with mixed activation groups
builder3 = DiGraphBuilder()
builder3.add_node(agent_a3).add_node(agent_b3).add_node(agent_c3).add_node(agent_d3)

# Critical inputs that must ALL be present (activation_group="critical", activation_condition="all")
builder3.add_edge(agent_a3, agent_d3, activation_group="critical", activation_condition="all")
builder3.add_edge(agent_b3, agent_d3, activation_group="critical", activation_condition="all")

# Optional input that can trigger execution on its own (activation_group="optional", activation_condition="any")
builder3.add_edge(agent_c3, agent_d3, activation_group="optional", activation_condition="any")

# Build and create flow
graph3 = builder3.build()
flow3 = GraphFlow(participants=[agent_a3, agent_b3, agent_c3, agent_d3], graph=graph3)

print("=== Example 3: Mixed Activation Groups ===")
print("D will execute when:")
print("- BOTH A AND B complete (critical group with 'all' activation), OR")
print("- C completes (optional group with 'any' activation)")
print("This allows for both required dependencies and fast-path triggers.")
# await Console(flow3.run_stream(task="Process inputs with mixed priority levels."))

Activation Group 핵심 요약

1. activation_group: 같은 대상 노드를 가리키는 엣지들을 그룹화해, 서로 다른 의존성 패턴을 정의할 수 있게 해요.
2. activation_condition:
   • "all"(기본): 대상 노드는 그룹의 모든 엣지가 충족될 때까지 기다려요
   • "any": 대상 노드는 그룹의 어떤 엣지 하나라도 충족되면 바로 실행해요
3. 사용 사례:
   • 여러 진입점이 있는 사이클: 서로 다른 activation group이 충돌을 막아요
   • 우선순위 기반 실행: "all"과 "any" 조건을 혼합해 다른 긴급도를 처리
   • 조기 종료가 있는 병렬 처리: "any"를 사용해 가장 빠른 결과로 진행
4. 모범 사례:
   • 설명적인 그룹 이름("critical", "optional", "feedback" 등) 사용
   • 같은 그룹 안에서 activation condition을 일관되게 유지
   • 다른 실행 경로로 그래프 로직을 테스트

이런 패턴들은 명확하고 이해하기 쉬운 실행 시맨틱을 유지하면서 정교한 워크플로우 제어를 가능하게 해요.

더 알아보기 (Learn more)

• 워크플로우 대신 자유로운 대화 흐름을 원한다면 Selector Group Chat을 확인하세요.
• 종료 조건을 더 세밀하게 제어하려면 종료 조건 튜토리얼을 참고하세요.
• 병렬·동시 실행 패턴은 동시 에이전트를 보세요.