Google ADK 2 Graph Workflows: A Complete Guide with Code Examples

28 min readAug 14, 2026

Recently, I sat down to learn ADK 2’s Graph-based agent workflows. The documentation was solid but I struggled because I learn things in a different way. I like no assumptions to be made on my knowledge and I usually get troubled if there is code syntax or method parameters/classes that I am not able to place in terms of how they work. So I took the assistance of Antigravity, ADK documentation and other material as my learning tutor to work through the basics and examples in a progressive fashion.

What you have now is a guide that helps you understand the ADK 2’s Graph Workflows via a scenario (a customer support ticket system), and which looks at various patterns from scratch:

Sequences
Parallel fan-outs
Conditional routing
Human-in-the-loop approvals
Dynamic runtime loops.

Each section introduces a pattern with the why before the how, provides a complete runnable example you can save to a file and execute immediately, and shows you exactly what to expect in the output. By the end, all the patterns converge into a single unified system.

Whether you’re evaluating ADK for your team or building your first production agent , I hope this saves you a lot of time.

Part 1: What Is a Graph Workflow?

Building production-ready AI applications often starts with a single prompt. Very quickly, you realize a single prompt isn’t enough. You start adding tools, giving the agent access to search, databases, and APIs. This leads to what we call “The Agentic Trap.” You wrap your entire business process in a massive, open-ended LLM agent loop, hoping it figures out the right sequence of tools to call. The result? Unpredictable behavior, high latency, massive token costs, and a system that is impossible to debug when it hallucinates a path you never intended.

The Graph Solution offers a way out. Instead of a monolithic agent, you break your application into a directed graph. Deterministic steps (like fetching user data, calling internal APIs, validating schemas) become standard function nodes that run instantly and cost nothing. Creative, reasoning-heavy steps (like summarizing an issue or writing a response) become targeted Agent nodes. By explicitly defining the edges (the pathways) between these nodes, you regain control over the application flow. The LLM is invoked only when necessary, exactly where you need its reasoning power.

In Google ADK 2, a Graph Workflow allows you to wire together agents, functions, and external services. Data flows from node to node, edges can branch conditionally, run in parallel, and even loop back for retries.

When NOT to use Graphs
If you are building a simple chat interface where the user talks back and forth with a single persona, and the agent just needs a few basic tools to answer questions (like a weather bot), you don’t need a graph. Just use a simple Agent with tools. Graphs are for multi-step processes, backend pipelines, and complex orchestrations.

Part 2: Environment Setup

All the code for this guide lives in a GitHub repository. Let’s clone it and set up our environment.

Step 1: Clone the Repository

# Clone the repo
git clone https://github.com/rominirani/adk-workflow-patterns.git
cd adk-workflow-patterns

Step 2: Create a Virtual Environment

# Requires Python 3.11+
python3 --version
# Create and activate virtual environment
python3 -m venv .venv
source .venv/bin/activate    # macOS / Linux
# .venv\Scriptsctivate     # Windows
# Install dependencies
pip install -r requirements.txt

⚠️ Always activate your venv
Every time you open a new terminal, you need to cd adk-workflow-patterns && source .venv/bin/activate before running any of the examples.

Step 3: Configure Your API Credentials

You have two options. Pick whichever fits your situation and the code thing is that your ADK code will be identical either way.

Option A: Google AI Studio (Fastest)

Get a free API key from Google AI Studio. This is the quickest way to start building.

# Set your API key (add to ~/.zshrc or ~/.bashrc to persist)
export GOOGLE_API_KEY="your-api-key-here"

Option B: Vertex AI (Enterprise / Google Cloud)

If you’re running on Google Cloud or need enterprise features, use Vertex AI instead. Be sure to have gcloud setup locally, a billing-enabled Google Cloud Project (keep the project id handy since you will need to set it in the below script).

# Authenticate with Google Cloud
gcloud auth application-default login

# Set environment variables
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"
export GOOGLE_GENAI_USE_VERTEXAI="True"

Step 4: Verify Everything Works

# Should print the message about ADK installed successfully
python -c "import google.adk; print('ADK installed successfully')"

In all the code examples that follow, we use a MODEL constant at the top of every file:

import os
MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

This lets you switch models by setting export ADK_MODEL="gemini-2.5-pro" without editing any code. If you’d like to use higher models, feel free to use that. Go ahead and try the recently released gemini-3.7-flash if you’d like.

Key Insight
Your ADK code is identical regardless of which backend you use. The only difference is environment variables. This means you can develop locally using the key obtained from Google AI Studio (Option A) and deploy to production with Vertex AI (Option B) without changing a single line of code.

Part 3: Data Handling

Data handling is consistently the #1 confusion point for new ADK users. When building a graph, you need to understand how data moves between nodes and how it reaches the user. The ADK uses three distinct channels: Output, Message, and State.

1. Output (The Conveyor Belt): Think of the graph as an assembly line. When Node A finishes its work, it places an object on the conveyor belt. Node B (the next node in the graph) picks up that exact object as its input. This is the output. It is strictly node-to-node communication. It does not go to the user.

2. Message (The Intercom): Sometimes, a node needs to talk to the human sitting outside the assembly line. It uses the message channel. Messages are typically text or UI components sent directly to the user interface. Messages do NOT travel to the next node.

3. State (The Shared Whiteboard): As the conveyor belt moves, you might want to keep track of a global variable, like a ticket_id or an accumulation of errors. State is a shared whiteboard attached to the session. Any node can read from it or write to it at any time.

Schema Definitions

It’s best practice to strictly type the data moving along your conveyor belt (the Output). Here is an example of standardizing inputs and outputs using schemas.

from pydantic import BaseModel

class TicketInput(BaseModel):
    user_id: str
    issue_text: str
class ResolutionOutput(BaseModel):
    status: str
    summary: str
    refund_amount: float = 0.0

Let’s break down what’s happening here:

TicketInput defines what goes into the support pipeline: a user ID, the issue text, and an optional priority level.
ResolutionOutput defines what comes out: the status of the resolution, a summary, and any refund amount.
By using Pydantic BaseModel, ADK can automatically validate that data flowing between nodes matches these shapes. If a node produces output that doesn't match the schema, you'll get a clear validation error instead of a silent data corruption downstream.

Here are key points to consider:

You’ll use schemas like these when setting input_schema and output_schema on Agent nodes.
For function nodes, the validation is implicit i.e. your Python function signature IS the schema.

4.1 Sequential Chain

The simplest workflow is a straight line: data flows from START through each node in order, one after another. No branching, no parallelism anywhere. It’s just a pipeline. This is the foundation everything else builds on.

In the example below, we build a 3-node sequence: extract_data (a function that pulls out the user's text), format_prompt (a function that wraps it into a prompt), and summarize_agent (an LLM agent that generates a summary). The first two nodes are free — they're pure Python, no LLM involved. Only the final agent incurs token costs.

Pay attention to the edges syntax: (START, extract_data, format_prompt, summarize_agent) is a tuple that means "START → extract_data → format_prompt → summarize_agent". Each element is connected to the next in sequence.

📄 graph-workflows/examples/01_sequential.py — Run: python graph-workflows/examples/01_sequential.py

import asyncio
import os
from google.adk import Workflow, Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

def extract_data(node_input):
    """Extract text from the upstream output (user message content)."""
    if isinstance(node_input, types.Content):
        return {"original": node_input.parts[0].text, "category": "inquiry"}
    return {"original": str(node_input), "category": "inquiry"}

def format_prompt(node_input):
    """Format the extracted data into a prompt string."""
    if isinstance(node_input, dict):
        return f"Please summarize this inquiry in one sentence: {node_input['original']}"
    return f"Please summarize this inquiry in one sentence: {node_input}"

summarize_agent = Agent(
    name="summarize_agent",
    model=MODEL,
    instruction="You are a helpful summarizer. Provide a one-sentence summary.",
)

workflow = Workflow(
    name="sequential_workflow",
    edges=[
        (START, extract_data, format_prompt, summarize_agent),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="How do I reset my password?")])
    async for event in runner.run_async(
        user_id="user_1", session_id="session_1", new_message=user_msg
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"[{event.author}]: {part.text}")
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

extract_data: A pure Python function. It takes the user's input, extracts the text, and returns it wrapped in an Event(output=...). This output becomes the input for the next node. Cost: 0 tokens.
format_prompt: Another pure Python function. It takes the extracted data and wraps it into a prompt string. Still 0 tokens.
summarize_agent: An LLM Agent node. This is where the model is invoked. It receives the formatted prompt and generates a natural language summary. This is the ONLY step that costs tokens.
Workflow: The edges list defines the graph. The tuple (START, extract_data, format_prompt, summarize_agent) creates a 3-step chain.
main: The Runner executes the workflow. We create an InMemorySessionService for local testing, send a user message, and iterate over the streamed events.

Expected Output

Output: {'original': 'How do I reset my password?', 'category': 'inquiry'}
Output: Please summarize this inquiry: How do I reset my password?
[summarize_agent] : The user is asking how to reset their password.

4.2 Conditional Branching

Real-world applications rarely flow in a single straight line. You often need to make decisions: if the user asks about billing, send them to the Billing Agent; if they ask about technical issues, send them to the Tech Support Agent. This is Conditional Branching or Routing.

In ADK, branching relies on the route field in the Event. A router node examines the input and returns an Event with a specific string set in route. The framework then matches this string against the keys in a dictionary (Python) attached to the workflow edge.

If the route value matches a key, execution flows to the target node for that key. If no key matches, the framework looks for a "default" fallback route. This ensures your graph doesn't break if the router generates an unexpected value.

📄 graph-workflows/examples/02_routing.py — Run: python graph-workflows/examples/02_routing.py

import asyncio
import os
from google.adk import Workflow, Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START
from google.adk.events.event_actions import EventActions
from google.adk import Event
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

def router_node(node_input):
    """Classify the user message and route to the correct specialist."""
    text = ""
    if isinstance(node_input, types.Content):
        text = node_input.parts[0].text.lower()
    else:
        text = str(node_input).lower()
    if "charge" in text or "refund" in text:
        route = "billing"
    elif "where" in text or "tracking" in text:
        route = "shipping"
    else:
        route = "default"
    return Event(
        output=text,
        actions=EventActions(route=route),
    )

billing_agent = Agent(name="billing_agent", model=MODEL, instruction="Handle billing queries. Be brief.")
shipping_agent = Agent(name="shipping_agent", model=MODEL, instruction="Handle shipping queries. Be brief.")
general_agent = Agent(name="general_agent", model=MODEL, instruction="Handle general queries. Be brief.")

workflow = Workflow(
    name="routing_workflow",
    edges=[
        (START, router_node),
        (router_node, {
            "billing": billing_agent,
            "shipping": shipping_agent,
            "default": general_agent,
        }),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    queries = ["Where is my tracking number?", "I want a refund!"]
    for i, q in enumerate(queries):
        print(f"\n--- Testing: {q} ---")
        user_msg = types.Content(role="user", parts=[types.Part(text=q)])
        async for event in runner.run_async(
            user_id="user_1", session_id=f"session_{i}", new_message=user_msg
        ):
            if event.content and event.content.parts:
                for part in event.content.parts:
                    if part.text:
                        print(f"[{event.author}]: {part.text}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

router_node: This is the decision maker. It examines the user's text for keywords ("charge", "refund" → billing; "where", "tracking" → shipping). The critical line is return Event(output=text, actions=EventActions(route=route)) — the actions.route field tells the framework WHICH downstream edge to follow.
The edges dict: (router_node, {"billing": billing_agent, "shipping": shipping_agent, "default": general_agent}) maps route strings to target nodes. If route="billing", the billing_agent fires. If the route doesn't match any key, the "default" key catches it.
The test loop: We test with two different queries to show how the same graph routes to different agents based on input.

Expected Output

--- Testing: Where is my tracking number? ---
Direct use of automatic function calling (AFC) in AsyncModels.generate_content is not recommended. Instead, we recommend to use AFC in AsyncChat.send_message. Similarly, direct use of AFC in AsyncModels.generate_content_stream is not recommended. Instead, we recommend to use AFC in AsyncChat.send_message_stream.
[shipping_agent]: Your tracking number is usually in your **shipping confirmation email** or accessible via your **order history** on our website.

--- Testing: I want a refund! ---
[billing_agent]: Please provide your order number or account email so I can assist you further.

Notice how the same workflow produces completely different responses depending on the input. The router is pure Python (0 tokens), and only the relevant specialist agent is invoked (1 LLM call per query, not 3).

4.3 Parallel Fan-Out + JoinNode

When you need to gather information from multiple independent sources, executing them sequentially wastes time. If fetching a user profile takes 1s, checking orders takes 1s, and verifying account status takes 1s, doing it sequentially takes 3 seconds. By running them in parallel (fan-out), it takes 1 second.

Fan-out is simple: you create multiple edges originating from the same node. The framework automatically executes the downstream nodes concurrently. However, when you fan out, you usually need to bring the data back together. This is where JoinNode comes in.

A JoinNode acts as a barrier. It pauses execution on that branch until ALL of its upstream dependencies have completed. Once they all finish, it bundles their outputs into a single dictionary, keyed by the name of the upstream node.

Be warned: if one upstream node crashes or is never executed due to routing logic, the JoinNode will wait forever (deadlock). Ensure all paths leading to a JoinNode are guaranteed to execute.

📄 graph-workflows/examples/03_parallel_join.py — Run: python graph-workflows/examples/03_parallel_join.py

import asyncio
import os
import time
from google.adk import Workflow, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, JoinNode
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

async def fetch_profile(node_input):
    await asyncio.sleep(1)
    return {"name": "Alice", "tier": "gold"}

async def fetch_orders(node_input):
    await asyncio.sleep(1)
    return {"recent_orders": 5}

async def fetch_account(node_input):
    await asyncio.sleep(1)
    return {"status": "active"}

def process_results(node_input):
    data = node_input
    summary = f"User {data['fetch_profile']['name']} is {data['fetch_profile']['tier']} tier with {data['fetch_orders']['recent_orders']} recent orders. Account: {data['fetch_account']['status']}."
    print(f"Summary: {summary}")
    return summary

join_node = JoinNode(name="data_joiner")

workflow = Workflow(
    name="parallel_workflow",
    edges=[
        (START, fetch_profile, join_node),
        (START, fetch_orders, join_node),
        (START, fetch_account, join_node),
        (join_node, process_results),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="start")])
    start_time = time.time()
    async for event in runner.run_async(
        user_id="user_1", session_id="session_1", new_message=user_msg
    ):
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")
    elapsed = time.time() - start_time
    print(f"Total time: {elapsed:.2f} seconds (3 tasks each taking 1s ran in parallel)")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

Three async functions: Each simulates a 1-second database lookup. In production, these would be real API calls to your CRM, order system, and billing database.
The edges pattern: Three edges from START — (START, fetch_profile, join_node), (START, fetch_orders, join_node), (START, fetch_account, join_node). Each edge fans out from START to a different function, then converges at the JoinNode.
JoinNode output: When all three functions complete, the JoinNode bundles their outputs into a dict keyed by function name: {"fetch_profile": {...}, "fetch_orders": {...}, "fetch_account": {...}}.
process_results: Receives the joined dict and can access each function's output by name.
Timing: We measure wall-clock time to prove parallelism. Three 1-second tasks complete in ~1 second, not 3.

Expected Output

[parallel_workflow] output: {'name': 'Alice', 'tier': 'gold'}
[parallel_workflow] output: {'recent_orders': 5}
[parallel_workflow] output: {'status': 'active'}
[parallel_workflow] output: {'fetch_account': {'status': 'active'}, 'fetch_profile': {'name': 'Alice', 'tier': 'gold'}, 'fetch_orders': {'recent_orders': 5}}
Summary: User Alice is gold tier with 5 recent orders. Account: active.
[parallel_workflow] output: User Alice is gold tier with 5 recent orders. Account: active.
Total time: 1.01 seconds (3 tasks each taking 1s ran in parallel)

If these three fetches ran sequentially, it would take ~3 seconds. Parallel fan-out cuts it to ~1 second i.e. the latency of the slowest branch.

4.4 Full Pipeline (Fan-Out → Join → Classify → Route → Specialist)

Now we can combine the previous concepts into a robust pattern. We fan out to gather data, join it back together, use a deterministic function to classify the intent based on the enriched data, and route to the correct specialized agent.

Get Romin Irani’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Cost Analysis: In this pattern, the 3 data fetching nodes consume 0 tokens. The join node consumes 0 tokens. The router function consumes 0 tokens. The only cost incurred is the final specialized agent. This is vastly more efficient than having a massive LLM agent evaluate tools and route internally. You’re effectively building a 5-step process where 4 steps are free and instant.

📄 graph-workflows/examples/04_full_pipeline.py — Run: python graph-workflows/examples/04_full_pipeline.py

import asyncio
import os
from google.adk import Workflow, Agent, Runner, Event
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, JoinNode
from google.adk.events.event_actions import EventActions
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

def fetch_db1(node_input):
    return {"tier": "premium"}

def fetch_db2(node_input):
    return {"last_order": "laptop"}

def classify_router(node_input):
    data = node_input
    enriched_prompt = f"User is {data['fetch_db1']['tier']}. Last order: {data['fetch_db2']['last_order']}. Help with their billing inquiry."
    return Event(output=enriched_prompt, actions=EventActions(route="billing"))

billing_agent = Agent(name="billing_agent", model=MODEL, instruction="You are billing support. Answer based on context provided. Be brief.")
tech_agent = Agent(name="tech_agent", model=MODEL, instruction="You are tech support. Be brief.")

join = JoinNode(name="joiner")

workflow = Workflow(
    name="full_pipeline",
    edges=[
        (START, fetch_db1, join),
        (START, fetch_db2, join),
        (join, classify_router),
        (classify_router, {
            "billing": billing_agent,
            "tech": tech_agent,
        }),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="I need help with my laptop charge")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=user_msg
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"[{event.author}]: {part.text}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

fetch_db1 and fetch_db2: Simulated database lookups. In production, these would call your actual APIs. Both are free (0 tokens).
classify_router: Receives the joined data from both databases. It reads the enriched context and decides the route. In this mock, it always routes to "billing" but in production, you'd inspect the actual user message (via session state) to classify.
Edge composition: The edges list combines fan-out, fan-in, sequential, and routing in one declaration. Read it top to bottom: fan-out → join → classify → route to specialist.

Expected Output

[billing_agent]: Okay, I can help with a billing inquiry for a premium user 
whose last order was a laptop. What information can I provide or what 
specific issue are you encountering?

Cost breakdown
fetch_db1 (0 tokens) + fetch_db2 (0 tokens) + join (0 tokens) + classify_router (0 tokens) + billing_agent (1 LLM call, ~200 tokens) = ~200 tokens total.

A monolithic agent doing the same work would cost ~1000+ tokens re-evaluating its context on every tool call.

4.5 Nested Workflows

Because a Workflow implements the same interface as a Node, you can nest graphs within graphs. You embed a child workflow directly into a parent workflow as if it were a single step.

When should you use this? When a sub-process has its own internal complexity. For example, if “Refund Processing” requires 5 distinct validation steps, routing, and database writes, you don’t want to clutter the main customer support graph with those details. You build a standalone refund_workflow and drop it into the main graph.

📄 graph-workflows/examples/05_nested.py — Run: python graph-workflows/examples/05_nested.py

import asyncio
import os
from google.adk import Workflow, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

def step_a(node_input):
    print("[sub] step_a: Sub process started")
    return "Sub process started"

def step_b(node_input):
    print(f"[sub] step_b: Received '{node_input}', finishing sub process")
    return "Sub process finished"

sub_workflow = Workflow(
    name="sub_process",
    edges=[(START, step_a, step_b)],
)

def prepare(node_input):
    print("[main] prepare: Getting ready")
    return "Ready"

def finalize(node_input):
    print(f"[main] finalize: Sub gave us '{node_input}'")
    return f"Done. Sub gave: {node_input}"

main_workflow = Workflow(
    name="main_process",
    edges=[(START, prepare, sub_workflow, finalize)],
)

async def main():
    runner = Runner(
        node=main_workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="go")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=user_msg
    ):
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

sub_workflow: A self-contained workflow with its own edges. It has two steps: step_a and step_b.
main_workflow: Uses sub_workflow directly in its edges tuple — (START, prepare, sub_workflow, finalize). The framework treats the entire sub_workflow as a single node. Data flows in, the sub-graph executes internally, and data flows out.

Why this matters
You can develop and test sub_workflow independently. When it's ready, drop it into any parent workflow. This is how you manage complexity in large agent systems.

Expected Output

[main] prepare: Getting ready
[main_process] output: Ready
[sub] step_a: Sub process started
[sub_process] output: Sub process started
[sub] step_b: Received 'Sub process started', finishing sub process
[sub_process] output: Sub process finished
[main] finalize: Sub gave us 'Sub process finished'
[main_process] output: Done. Sub gave: Sub process finished

4.6 Loops and Escalation

Graphs aren’t just one-directional. By wiring an edge from a downstream node back to an upstream node, you create a loop. This is incredibly useful for quality control: draft a response, evaluate it, and if it’s not good enough, send it back to the drafter.

The critical rule with loops is: always include a break condition. Without one, your graph will loop forever and which means burning LLM tokens with every iteration. In our example, we use a mock “critic” that rejects the first two attempts and approves on the third.

Notice the use of ctx.state to track the number of attempts. This is a clean way to maintain a counter across loop iterations without polluting the output channel.

📄 graph-workflows/examples/06_loop.py — Run: python graph-workflows/examples/06_loop.py

import asyncio
import os
from google.adk import Workflow, Agent, Runner, Event
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START
from google.adk.events.event_actions import EventActions
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

drafter = Agent(
    name="drafter",
    model=MODEL,
    instruction="Draft a short, 1-sentence customer apology for the issue described. Be specific and empathetic.",
)

def critic(node_input, ctx):
    """Evaluate the draft. Reject first 2 attempts, approve on 3rd."""
    attempts = ctx.state.get("attempts", 0) + 1
    ctx.state["attempts"] = attempts
    if attempts < 3:
        route = "fail"
        msg = f"Attempt {attempts} rejected. Too generic. Try again with more empathy."
    else:
        route = "pass"
        msg = f"Attempt {attempts} approved!"
    print(f"[critic] {msg}")
    return Event(output=msg, actions=EventActions(route=route))

finalize = Agent(
    name="finalize",
    model=MODEL,
    instruction="Output the approved message as the final customer response. Keep it brief.",
)

workflow = Workflow(
    name="loop_workflow",
    edges=[
        (START, drafter, critic),
        (critic, {
            "fail": drafter,   # Loop back!
            "pass": finalize,  # Break out
        }),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="I got the wrong item in my order.")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=user_msg
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"[{event.author}]: {part.text}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

drafter: An LLM agent that generates a 1-sentence apology. Each iteration produces a slightly different draft.
critic: A deterministic function (0 tokens) that uses ctx.state to track attempts. It rejects the first 2 attempts (actions=EventActions(route="fail")) and approves the 3rd (actions=EventActions(route="pass")).
The loop edge: (critic, {"fail": drafter, "pass": finalize}) — when the critic returns actions=EventActions(route="fail"), execution loops BACK to the drafter. When it returns actions=EventActions(route="pass"), execution breaks out to finalize.
State tracking: ctx.state["attempts"] persists across loop iterations because it's stored in the session, not in the output channel.

Expected Output

[drafter]: We are so sorry you received the wrong item in your order, 
and we understand how frustrating this mistake must be.

[critic] Attempt 1 rejected. Too generic. Try again with more empathy.

[drafter]: We sincerely apologize for the unexpected trouble and frustration 
this has caused you, and we deeply regret that your experience with us was negatively impacted.

[critic] Attempt 2 rejected. Too generic. Try again with more empathy.

[drafter]: We truly apologize that your second attempt was rejected, 
and we genuinely understand how frustrating and disheartening that must be.

[critic] Attempt 3 approved!

[finalize]: Attempt 3 approved!

Each loop iteration costs 1 LLM call (the drafter). The critic is free. So 3 iterations = 3 LLM calls. This is why bounding your loops matters and unbounded loops mean unbounded costs.

Part 5: Human-in-the-Loop (HITL)

Human-in-the-Loop (HITL) is a critical pattern for sensitive operations. It is a graph-level pause. This is fundamentally different from having an LLM ask a clarifying question. When an LLM asks a question, the conversation continues natively. When a graph triggers HITL, the workflow execution suspends, state is saved, and it waits for an external signal to resume.

How it works: A node decides it needs human approval (e.g., refund > $100). The node yields a RequestInput event. The framework intercepts this, pauses execution, and returns control to the application layer. The application layer prompts the user UI. When the user clicks "Approve", the application calls the runner again with the approval data. The framework loads the saved state and resumes execution from exactly where it paused.

Use this for:

Manager approvals
Confirming destructive actions (deleting data)
Requesting an explicit OTP code from the user.

📄 graph-workflows/examples/07_hitl.py — Run: python graph-workflows/examples/07_hitl.py

import asyncio
import os
from google.adk import Workflow, Agent, Runner, Event
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START
from google.adk.events.event_actions import EventActions
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

def check_refund(node_input, ctx):
    """Check refund amount. If > $100, pause for human approval."""
    already_approved = ctx.state.get("approval_status", None)
    if already_approved == "approved":
        print("[check_refund] Manager approved! Processing refund.")
        return Event(output="Refund Processed", actions=EventActions(route="approved"))
    elif already_approved == "rejected":
        print("[check_refund] Manager rejected the refund.")
        return Event(output="Refund Rejected", actions=EventActions(route="rejected"))
    # First time through: check the amount
    refund_amount = 500  # Simulated
    print(f"[check_refund] Refund amount is ${refund_amount}. Requires manager approval.")
    # Set state so the app layer can see it
    ctx.state["pending_refund"] = refund_amount
    ctx.state["needs_approval"] = True
    return Event(output=f"Pending approval for ${refund_amount} refund")

def process_approved(node_input):
    print(f"[process_approved] {node_input}")
    return f"Final Status: {node_input}"

def process_rejected(node_input):
    print(f"[process_rejected] {node_input}")
    return f"Final Status: {node_input}"

workflow = Workflow(
    name="hitl_workflow",
    edges=[
        (START, check_refund),
        (check_refund, {
            "approved": process_approved,
            "rejected": process_rejected,
        }),
    ],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    # --- First Run: Workflow pauses for approval ---
    print("=== RUN 1: Refund Request (Pauses) ===")
    user_msg = types.Content(role="user", parts=[types.Part(text="I want a refund for order #123")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=user_msg
    ):
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")
    print("\n--- Manager reviews... decides to approve ---\n")
    # --- Second Run: Manager approves, workflow resumes ---
    print("=== RUN 2: Manager Approves (Resumes) ===")
    approval_msg = types.Content(role="user", parts=[types.Part(text="Approved by manager")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=approval_msg,
        state_delta={"approval_status": "approved"},
    ):
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

check_refund: This function has two code paths. On the FIRST call, it checks the refund amount. If > $100, it returns RequestInput(...) which pauses the workflow. On the SECOND call (after the human responds), it checks node_input["approval_status"] and routes accordingly.
Two run_async calls: The first call starts the workflow and hits the pause. The second call resumes it with the manager's approval. In a real application, these two calls might be hours or days apart and the session state is persisted.
No LLM during pause: While the workflow is paused, no resources are consumed. The state sits in the session service waiting for the resume signal.

Expected Output

=== RUN 1: Refund Request (Pauses) ===
[check_refund] Refund amount is $500. Requires manager approval.
[hitl_workflow] output: Pending approval for $500 refund
Node 'check_refund' has conditional/DEFAULT edges but 
none were matched by the emitted route(s): None. The branch will end.

--- Manager reviews... decides to approve ---

=== RUN 2: Manager Approves (Resumes) ===
[check_refund] Manager approved! Processing refund.
[hitl_workflow] output: Refund Processed
[process_approved] Refund Processed
[hitl_workflow] output: Final Status: Refund Processed

Part 6: Dynamic Workflows

Up until now, every pattern we’ve explored has been a static graph: every node, branch, and edge was predetermined and wired before running the code.

However, real-world systems often face scenarios where the shape of the execution cannot be known in advance. For example:

Runtime-Sized Fan-Out: An intake agent decomposes a user request into N sub-tasks (where N depends on complexity), and you need to execute all N sub-tasks dynamically.
Dynamic Retry / Troubleshooting Loops: You want standard Python control flow (for, while, try/except) to govern when and how many times an agent or sub-workflow is invoked.
Conditional Sub-node Delegation: Deciding at runtime to execute a heavy multi-step agent only if certain criteria are met, without cluttering the top-level graph topology with bypass edges.

The Solution: `@node` and `ctx.run_node()`

In ADK, any Python function decorated with @node gains access to the execution context (ctx). This context provides a programmatic runner: await ctx.run_node(agent_or_node).

Why `@node(rerun_on_resume=True)` Matters

One of ADK’s most powerful capabilities is stateful checkpointing. When a workflow pauses (for human approval or waiting on an asynchronous event), ADK stores the state of every completed node.

Understanding rerun_on_resumeBy default, ADK memoizes node execution: once a node completes, ADK skips re-executing it upon session resumption. However, when a node acts as a dynamic orchestrator that contains loops or calls ctx.run_node(), ADK needs permission to re-enter the parent function so it can evaluate remaining loop conditions or child nodes. Setting @node(rerun_on_resume=True) instructs ADK to re-enter the function body while safely relying on checkpointed results for already-completed inner ctx.run_node() calls.

Example 1: Dynamic Troubleshooting Loop

Here we loop dynamically until a condition is met, without explicitly wiring an edge cycle.

📄 graph-workflows/examples/08_dynamic.py — Run: python graph-workflows/examples/08_dynamic.py

import asyncio
import os
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import Workflow, START, node
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

diagnose_agent = Agent(
    name="diagnose",
    model=MODEL,
    instruction="You are a tech support diagnostician. Suggest ONE specific troubleshooting step for the user's issue. Be brief (1-2 sentences).",
)

@node(rerun_on_resume=True)
async def troubleshoot_loop(ctx):
    """Dynamic node that runs the diagnose agent in a loop."""
    max_attempts = 2
    for i in range(1, max_attempts + 1):
        print(f"--- Iteration {i} ---")
        result = await ctx.run_node(diagnose_agent)
        print(f"[diagnose] result: {result}")
        # In a real system, you'd check if the issue is resolved
        if i == max_attempts:
            print(f"Max attempts reached. Escalating.")
            return f"Escalated after {max_attempts} iterations. Last suggestion: {result}"
    return "Issue resolved"

workflow = Workflow(
    name="dynamic_workflow",
    edges=[(START, troubleshoot_loop)],
)

async def main():
    runner = Runner(
        node=workflow,
        app_name="test_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    user_msg = types.Content(role="user", parts=[types.Part(text="My app keeps crashing when I upload photos")])
    async for event in runner.run_async(
        user_id="u1", session_id="s1", new_message=user_msg
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"[{event.author}]: {part.text}")
        if event.output is not None:
            print(f"[{event.author}] output: {event.output}")

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

@node(rerun_on_resume=True) decorator: Marks this asynchronous function as a dynamic node capable of invoking child nodes with full checkpointing support.
Standard Python Control Flow: Notice the graph has only one declared edge: (START, troubleshoot_loop). The looping logic, attempt counting, and breakout criteria are handled cleanly in standard Python code rather than complex cyclic graph schemas.
await ctx.run_node(diagnose_agent): Programmatically dispatches execution to the diagnose_agent. ADK handles streaming events, state propagation, and sub-node lifecycle automatically.
Strict Bounding: max_attempts = 2 acts as a hard safety brake. Always ensure dynamic loops have clear bounding limits to prevent runaway token expenditure.

Expected Output

--- Iteration 1 ---
[diagnose]: Clear the app's cache and data in your device's settings.
[diagnose] result: Clear the app's cache and data in your device's settings.

--- Iteration 2 ---
[diagnose]: Check for any available updates for the app in your device's app store.
[diagnose] result: Check for any available updates for the app in your 
device's app store.

Max attempts reached. Escalating.

[dynamic_workflow] output: Escalated after 2 iterations. 

Last suggestion: Check for any available updates for the app in your device's 
app store.

Part 7: Putting It All Together — The Complete Customer Support System

Every pattern we’ve learned feeds into one unified system. A production customer support pipeline doesn’t just do one thing. It needs to fetch data in parallel, classify the intent, route to specialized experts, trigger human review for sensitive actions, and dynamically troubleshoot complex issues.

Below is the architecture for our final system. We fan out to fetch Profile, Orders, and Account data. The JoinNode aggregates this context. A classification router decides if the user needs Billing, Shipping, Tech, or General help. If it's Billing, it goes to the Billing Agent, which might trigger a HITL refund check. If it's Tech, it enters the Dynamic Troubleshooting loop.

📄 graph-workflows/examples/09_complete_system.py — Run: python graph-workflows/examples/09_complete_system.py

"""Complete Customer Support System: combines all graph workflow patterns.

1. Parallel Data Fetchers (Profile, Orders, Billing) + JoinNode
2. Dynamic Classification & Routing based on user message
3. Tech Path -> Dynamic Troubleshooting Loop (@node with ctx.run_node)
4. Billing Path -> Billing Agent -> Human-in-the-Loop (HITL) Approval Check
5. Shipping & General Specialists
"""
import asyncio
import os
import time
import warnings
import logging

# Suppress internal SDK noise and warnings
warnings.filterwarnings("ignore", category=UserWarning, module="google.genai")
warnings.filterwarnings("ignore", category=ResourceWarning)
logging.getLogger("google.genai").setLevel(logging.ERROR)
logging.getLogger("aiohttp").setLevel(logging.ERROR)

from google.adk import Workflow, Agent, Runner, Event
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, JoinNode, node
from google.adk.events.event_actions import EventActions
from google.genai import types

MODEL = os.environ.get("ADK_MODEL", "gemini-2.5-flash")

# ===== Section 1: Parallel Data Fetchers =====
async def fetch_profile(node_input):
    await asyncio.sleep(0.3)
    return {"name": "Alice", "tier": "Gold", "account_age": "2 years"}

async def fetch_orders(node_input):
    await asyncio.sleep(0.3)
    return {"recent_orders": 3, "last_order": "Pro Laptop #8492"}

async def fetch_billing(node_input):
    await asyncio.sleep(0.3)
    return {"outstanding_balance": 0, "payment_method": "Visa-4242"}

# ===== Section 2: Classifier / Router =====
def classify_and_route(node_input, ctx):
    """Inspects user message in state and routes to the right specialist."""
    user_text = ctx.state.get("user_message", "").lower()
    profile = node_input.get("fetch_profile", {})
    context_summary = f"Customer: {profile.get('name', 'User')}, Tier: {profile.get('tier', 'Standard')}"
    
    if "approval_status" in ctx.state or "refund" in user_text or "charge" in user_text or "bill" in user_text:
        route = "billing"
    elif "crash" in user_text or "error" in user_text or "bug" in user_text:
        route = "tech"
    elif "track" in user_text or "package" in user_text or "delivery" in user_text:
        route = "shipping"
    else:
        route = "general"
        
    return Event(output=context_summary, actions=EventActions(route=route))

# ===== Section 3: Dynamic Tech Troubleshooting Node =====
diagnose_agent = Agent(
    name="tech_diagnostician",
    model=MODEL,
    instruction="Suggest ONE specific troubleshooting step in 1 sentence.",
)

@node(rerun_on_resume=True)
async def dynamic_troubleshoot(ctx):
    """Loops up to 2 attempts using ctx.run_node before escalating."""
    for attempt in (1, 2):
        print(f"  [Troubleshoot Loop] Attempt {attempt}...")
        result = await ctx.run_node(diagnose_agent)
        print(f"  [Diagnostician {attempt}]: {result}")
    return "Troubleshooting attempts completed. Escalated to Tier 2."

# ===== Section 4: Billing Agent & HITL Gate =====
billing_agent = Agent(
    name="billing_specialist",
    model=MODEL,
    instruction="Acknowledge the billing refund inquiry empathetically in 1 sentence.",
)

def check_refund_hitl(node_input, ctx):
    """HITL Gate: If refund amount > $100, pause for human approval."""
    already_approved = ctx.state.get("approval_status", None)
    if already_approved == "approved":
        return Event(output="Refund Processed: APPROVED by Manager", actions=EventActions(route="approved"))
    elif already_approved == "rejected":
        return Event(output="Refund Processed: REJECTED by Manager", actions=EventActions(route="rejected"))

    refund_amount = ctx.state.get("refund_amount", 150)
    ctx.state["pending_refund"] = refund_amount
    ctx.state["needs_approval"] = True
    print(f"  [HITL Gate] Refund of ${refund_amount} exceeds $100 auto-limit. Pausing for approval.")
    return Event(output=f"PAUSED: Requires manager approval for ${refund_amount} refund")

def process_approved(node_input):
    return f"Final Status: {node_input}"

# ===== Section 5: Shipping & General Agents =====
shipping_agent = Agent(
    name="shipping_specialist",
    model=MODEL,
    instruction="Provide a tracking status update for Order #8492 in 1 sentence.",
)

general_agent = Agent(
    name="general_specialist",
    model=MODEL,
    instruction="Answer general support inquiry helpfully in 1 sentence.",
)

# ===== Section 6: Build Unified Graph =====
join = JoinNode(name="data_joiner")

workflow = Workflow(
    name="complete_support_system",
    edges=[
        # 1. Parallel Fan-out from START to data fetchers
        (START, fetch_profile, join),
        (START, fetch_orders, join),
        (START, fetch_billing, join),
        
        # 2. Join -> Classify & Route
        (join, classify_and_route),
        
        # 3. Conditional branches
        (classify_and_route, {
            "tech": dynamic_troubleshoot,
            "billing": billing_agent,
            "shipping": shipping_agent,
            "general": general_agent,
        }),
        
        # 4. Billing flows into HITL Gate
        (billing_agent, check_refund_hitl),
        (check_refund_hitl, {
            "approved": process_approved,
        }),
    ],
)

# ===== Section 7: Demonstration of Multiple Paths =====
async def run_scenario(runner, label, user_text, session_id, initial_state=None, is_resume=False, resume_state=None):
    print(f"\n{'='*60}")
    print(f"=== {label} ===")
    print(f"Customer Input: \"{user_text}\"")
    print(f"{'='*60}")
    
    start_time = time.time()
    user_msg = types.Content(role="user", parts=[types.Part(text=user_text)])
    
    state_delta = initial_state or {}
    state_delta["user_message"] = user_text
    if is_resume and resume_state:
        state_delta.update(resume_state)

    async for event in runner.run_async(
        user_id="alice", session_id=session_id, new_message=user_msg, state_delta=state_delta
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"[{event.author}]: {part.text}")
        if event.output is not None and not isinstance(event.output, dict):
            print(f"[{event.author}] output: {event.output}")

    elapsed = time.time() - start_time
    print(f"Execution time: {elapsed:.2f}s")

async def main():
    runner = Runner(
        node=workflow,
        app_name="complete_support_app",
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    
    # Path 1: Tech Ticket -> Parallel Fetch -> Classify -> Dynamic Troubleshoot Loop
    await run_scenario(
        runner,
        label="SCENARIO 1: Tech Support (Dynamic Troubleshoot Loop)",
        user_text="The application crashes every time I upload photos.",
        session_id="session_tech_1",
    )

    # Path 2a: Billing Ticket -> Parallel Fetch -> Classify -> Billing Agent -> HITL Pause
    await run_scenario(
        runner,
        label="SCENARIO 2a: Billing Ticket (HITL Pause for Approval)",
        user_text="I was charged twice and need a refund of $150.",
        session_id="session_billing_1",
        initial_state={"refund_amount": 150},
    )

    # Path 2b: Manager Approves -> Workflow Resumes & Completes
    await run_scenario(
        runner,
        label="SCENARIO 2b: Manager Approves Refund (HITL Resume)",
        user_text="Approved by Manager",
        session_id="session_billing_1",
        is_resume=True,
        resume_state={"approval_status": "approved"},
    )

    # Path 3: Shipping Ticket -> Parallel Fetch -> Classify -> Shipping Specialist
    await run_scenario(
        runner,
        label="SCENARIO 3: Shipping Inquiry",
        user_text="Where is my package for order #8492?",
        session_id="session_shipping_1",
    )

    # Allow background aiohttp connections to finish closing
    await asyncio.sleep(0.25)

if __name__ == "__main__":
    asyncio.run(main())

Code walkthrough:

Layered Design: Notice how cleanly the graph is constructed. Parallel data fetchers (Profile, Orders, Billing) execute concurrently at the start of every request. The JoinNode aggregates the profile context for whatever downstream agent is chosen.
Dynamic Troubleshooting Branch (Tech Path): When a tech issue is classified, the flow enters dynamic_troubleshoot. The node uses a Python loop with await ctx.run_node(diagnose_agent) to try up to 2 distinct diagnostic attempts before escalating to Tier 2.
Human-in-the-Loop Branch (Billing Path): For high-value refund requests ($150), the billing agent first acknowledges the customer, and the downstream check_refund_hitl gate halts the workflow. When the manager later provides an approval payload via state_delta={"approval_status": "approved"}, the workflow resumes and finalizes the refund.
Specialized Fast Path (Shipping Path): Routine inquiries route directly to specialized agents (like shipping_specialist) with customer profile data pre-loaded without running unnecessary loops.

Expected Output

============================================================
=== SCENARIO 1: Tech Support (Dynamic Troubleshoot Loop) ===
Customer Input: "The application crashes every time I upload photos."
============================================================
[complete_support_system] output: Customer: Alice, Tier: Gold
  [Troubleshoot Loop] Attempt 1...
[tech_diagnostician]: Update the application to its latest version.
  [Diagnostician 1]: Update the application to its latest version.
  [Troubleshoot Loop] Attempt 2...
[tech_diagnostician]: I have provided one specific troubleshooting step for the application crashing issue. No further outputs are needed at this time.
  [Diagnostician 2]: I have provided one specific troubleshooting step for the application crashing issue. No further outputs are needed at this time.
[complete_support_system] output: Troubleshooting attempts completed. Escalated to Tier 2.
Execution time: 10.13s

============================================================
=== SCENARIO 2a: Billing Ticket (HITL Pause for Approval) ===
Customer Input: "I was charged twice and need a refund of $150."
============================================================
[complete_support_system] output: Customer: Alice, Tier: Gold
[billing_specialist]: Thank you for reaching out, Alice; I understand you're inquiring about a billing refund.
  [HITL Gate] Refund of $150 exceeds $100 auto-limit. Pausing for approval.
[complete_support_system] output: PAUSED: Requires manager approval for $150 refund
Node 'check_refund_hitl' has conditional/DEFAULT edges but none were matched by the emitted route(s): None. The branch will end.
Execution time: 2.49s

============================================================
=== SCENARIO 2b: Manager Approves Refund (HITL Resume) ===
Customer Input: "Approved by Manager"
============================================================
[complete_support_system] output: Customer: Alice, Tier: Gold

[billing_specialist]: We understand you're looking into a billing refund, Alice, and we're here to help.
[complete_support_system] output: Refund Processed: APPROVED by Manager
[complete_support_system] output: Final Status: Refund Processed: APPROVED by Manager
Execution time: 3.74s

============================================================
=== SCENARIO 3: Shipping Inquiry ===
Customer Input: "Where is my package for order #8492?"
============================================================
[complete_support_system] output: Customer: Alice, Tier: Gold
[shipping_specialist]: Alice, your Gold Tier Order #8492 has been shipped and is currently in transit.
Execution time: 9.16s

Conclusion: What You’ve Built

The Key Insight

The magic of ADK Graphs is that you pay only for reasoning, never for routing. By moving your orchestration, decision trees, and API calls into deterministic Python nodes, you eliminate LLM hallucinations, drop latency to milliseconds, and reduce token usage to nearly zero for those steps. You only invoke the LLM when you need its cognitive capabilities.

What’s Next?

In this guide, you drew every edge by hand and the graph decided what ran next. But what happens when you know the team of specialists, but the user’s request decides who gets involved?

That’s the world of Collaborative Workflows, where an LLM coordinator reads each agent’s description and routes the conversation to the right specialist automatically.

Google Cloud - Community

Google ADK 2 Graph Workflows: A Complete Guide with Code Examples

Part 1: What Is a Graph Workflow?

Part 2: Environment Setup

Step 1: Clone the Repository

Step 2: Create a Virtual Environment

Step 3: Configure Your API Credentials

Option A: Google AI Studio (Fastest)

Option B: Vertex AI (Enterprise / Google Cloud)

Step 4: Verify Everything Works

Part 3: Data Handling

Schema Definitions

4.1 Sequential Chain

4.2 Conditional Branching

4.3 Parallel Fan-Out + JoinNode

4.4 Full Pipeline (Fan-Out → Join → Classify → Route → Specialist)

Get Romin Irani’s stories in your inbox

4.5 Nested Workflows

4.6 Loops and Escalation

Part 5: Human-in-the-Loop (HITL)

Part 6: Dynamic Workflows

The Solution: @node and ctx.run_node()

Why @node(rerun_on_resume=True) Matters

Example 1: Dynamic Troubleshooting Loop

Part 7: Putting It All Together — The Complete Customer Support System

Conclusion: What You’ve Built

The Key Insight

What’s Next?

Written by Romin Irani

Published in Google Cloud - Community

Responses (2)

More from Romin Irani and Google Cloud - Community

Building an AI Document Reviewer with Google ADK, Gemini and the New Google Docs Comments &…

GitHub Repository: All complete source code, Dockerfile, and the one-command Cloud Run deployment script (deploy.sh) are available at…

The Vanishing Engineering Signal in AI Hackathons

Winning a hackathon used to be a reliable proxy for real technical chops. It meant you could drop into an ambiguous problem, write code…

DuckLess: serverless DuckDB on Google Cloud, in your own project

I like DuckDB a lot. On a laptop it chews through gigabytes of Parquet faster than most of the platforms I use at work, and it asks for…

Antigravity CLI Tutorial Series

Part 1 : Getting Started

Recommended from Medium

Google Cloud Skills Tutorial: The Complete Guide to AI-Powered Cloud Operations

Welcome to the definitive guide on Google Cloud Agent Skills. As the landscape of software engineering evolves, AI coding agents are…

Andrej Karpathy’s New suggestion- Ask AI output in ASD-STE100 format and not in Text format

A few days ago, Andrej Karpathy shared a perspective that immediately clicked with how my own daily workflow has been evolving. His premise…

5 Claude Skills That Fix Claude’s Ugly UI (AI Web Design)

If you must build your website with AI, at least make it look good

How to Become an AI Engineer in 2026: The Complete Learning Roadmap

Whether you’re starting from scratch or transitioning from software engineering, here’s a structured roadmap to learn the skills companies…

Structuring 40 Million Documents into an Agentic Knowledge Graph

929 million edges in 10.8 minutes, and 83.2% on 600 held-out questions

10 MCP Servers That Do Things I Didn’t Know MCP Could Do

I found some MCP servers that can give AI agents surprisingly powerful new abilities.

Responses (2)

More from Romin Irani and Google Cloud - Community

Building an AI Document Reviewer with Google ADK, Gemini and the New Google Docs Comments &…

GitHub Repository: All complete source code, Dockerfile, and the one-command Cloud Run deployment script (deploy.sh) are available at…

The Vanishing Engineering Signal in AI Hackathons

Winning a hackathon used to be a reliable proxy for real technical chops. It meant you could drop into an ambiguous problem, write code…

DuckLess: serverless DuckDB on Google Cloud, in your own project

I like DuckDB a lot. On a laptop it chews through gigabytes of Parquet faster than most of the platforms I use at work, and it asks for…

Antigravity CLI Tutorial Series

Part 1 : Getting Started

Recommended from Medium

Google Cloud Skills Tutorial: The Complete Guide to AI-Powered Cloud Operations

Welcome to the definitive guide on Google Cloud Agent Skills. As the landscape of software engineering evolves, AI coding agents are…

Andrej Karpathy’s New suggestion- Ask AI output in ASD-STE100 format and not in Text format

A few days ago, Andrej Karpathy shared a perspective that immediately clicked with how my own daily workflow has been evolving. His premise…

5 Claude Skills That Fix Claude’s Ugly UI (AI Web Design)

If you must build your website with AI, at least make it look good

How to Become an AI Engineer in 2026: The Complete Learning Roadmap

Whether you’re starting from scratch or transitioning from software engineering, here’s a structured roadmap to learn the skills companies…

Structuring 40 Million Documents into an Agentic Knowledge Graph

929 million edges in 10.8 minutes, and 83.2% on 600 held-out questions

10 MCP Servers That Do Things I Didn’t Know MCP Could Do

I found some MCP servers that can give AI agents surprisingly powerful new abilities.

The Solution: `@node` and `ctx.run_node()`

Why `@node(rerun_on_resume=True)` Matters