graph-workflow-designer

graph-workflow-designer

Design graph-based workflows introduced in ADK 2.0. Each node is an agent (or function), each edge is a deterministic transition predicate. Use when you need predictable routing without LLM-driven delegation.

When to use

• Compliance / safety-critical flows where routing must be auditable
• Replace fragile coordinator-LLM routing with explicit logic
• Mix LLM steps and pure-function steps in one declarative graph
• Need cycles, conditional branches, and merge points

When NOT to use

• Routing decisions need natural-language understanding → coordinator-subagent-pattern
• Pure linear flow → SequentialAgent (workflow-agent-scaffold)
• Iterative refinement loop → LoopAgent (workflow-agent-scaffold)

Template

from google.adk.agents import LlmAgent, GraphAgent
from google.adk.agents.graph import Node, Edge

# Nodes
classify = LlmAgent(
    name="classify",
    model="gemini-2.5-flash",
    instruction="Classify the request as 'urgent' or 'standard'. Write to state['priority'].",
    output_key="priority",
)
urgent_handler = LlmAgent(
    name="urgent",
    model="gemini-2.5-pro",
    instruction="Handle urgent requests with immediate response.",
)
standard_handler = LlmAgent(
    name="standard",
    model="gemini-2.5-flash",
    instruction="Handle standard requests at normal pace.",
)
audit = LlmAgent(
    name="audit",
    model="gemini-2.5-flash",
    instruction="Append a one-line audit log entry.",
)

# Graph
root_agent = GraphAgent(
    name="priority_router",
    nodes=[classify, urgent_handler, standard_handler, audit],
    edges=[
        Edge(from_="classify", to="urgent",
             when=lambda ctx: ctx.session.state.get("priority") == "urgent"),
        Edge(from_="classify", to="standard",
             when=lambda ctx: ctx.session.state.get("priority") == "standard"),
        Edge(from_="urgent", to="audit"),
        Edge(from_="standard", to="audit"),
    ],
    entry="classify",
    exit_nodes=["audit"],
)

Edge predicates

when is a callable receiving InvocationContext. Return True to take the edge. If multiple edges match from a node, the first listed wins (or use priority).

Cycles

Edge(from_="reviewer", to="drafter",
     when=lambda ctx: ctx.session.state.get("approved") is False,
     max_traversals=5),  # safety limit

Validation

• Every node referenced in edges must be in nodes=[]
• Exactly one entry node
• At least one exit_nodes entry
• No unreachable nodes (run graph.validate() if available)
• Cycles must have max_traversals to avoid infinite loops

See also

• dynamic-workflow-builder for code-driven branching when graph topology itself depends on runtime state