Tutorial Google ADK 2.0: Graph Workflow dan Multi-Agent

Versi pertama asisten pengadaan kami membiarkan model memutuskan siapa yang menyetujui pembelian. Model membaca permintaan, menebak rentang nominal, lalu transfer ke agent manajer atau direktur. Hasilnya benar hampir setiap kali, dan justru itu standar yang salah untuk aturan yang oleh tim finance sudah ditulis sebagai dua angka: di bawah 5 juta rupiah otomatis, di bawah 50 juta perlu manajer, di atas itu perlu direktur.

Tutorial Google ADK 2.0 ini membangun ulang alur tersebut di atas graph engine yang baru, di mana LLM hanya mengekstrak permintaan dan Python biasa yang mengurus routing. Semua snippet saya jalankan di google-adk 2.10.0 dengan Python 3.11, dengan intake agent diganti function stub supaya jalur deterministiknya bisa berjalan tanpa model. Bagian yang perilakunya mengejutkan saya tulis apa adanya; sisanya merujuk ke dokumentasi resmi ADK 2.0.

Apa yang berubah di ADK 2.0: agent menjadi node

ADK 2.0 mengganti eksekusi agent yang hierarkis dengan graph engine. Agent, tool, dan function biasa semuanya adalah node, dan BaseAgent sekarang meng-extend BaseNode. Python mencapai GA pada 19 Mei 2026, Go pada 30 Juni 2026, dan TypeScript pada 21 Agustus 2026, dan sejak itu Python merilis versi minor kira-kira tiap dua minggu hingga 2.10.0 di akhir September. Rilis ini memberi tiga cara menyusun node.

Pertanyaan untuk memilih barisnya sederhana: apakah analis junior bisa menjalankan aturannya dari kebijakan tertulis? Kalau bisa, tempatnya di graph atau dynamic workflow, dan bukan model yang memilih cabangnya. Batas nominal approval lolos tes ini, jadi masuk ke kode. Mengubah pesan yang samar menjadi vendor, nominal, dan cost center tidak lolos, jadi langkah itu tetap di model.

Tutorial Google ADK 2.0: graph approval pengadaan

Seluruh kebijakan muat dalam satu Workflow. Daftar edges dibaca dari atas ke bawah: START masuk ke intake agent, output terstrukturnya masuk ke router nominal, dan setiap nilai route dipetakan ke satu cabang. Semua cabang approval bertemu di router kedua yang memisahkan disetujui dan ditolak.

# agent.py — google-adk 2.10.0, Python 3.11
from pydantic import BaseModel, Field
from google.adk import Agent, Workflow, Event, Context
from google.adk.events import RequestInput
from google.adk.workflow import DEFAULT_ROUTE

class PurchaseRequest(BaseModel):
    vendor: str
    amount_idr: int = Field(ge=0)
    cost_center: str
    justification: str

class Decision(BaseModel):
    approved: bool
    approver: str
    note: str | None = None   # Optional, not str = "" — see below

# The one LLM node: free text in, a validated PurchaseRequest out.
intake_agent = Agent(
    name="intake_agent",
    model="gemini-flash-latest",
    instruction="Extract the purchase request from the user's message.",
    output_schema=PurchaseRequest,
)

def route_by_amount(node_input: PurchaseRequest, ctx: Context):
    ctx.state["request"] = node_input.model_dump()   # later nodes read it back
    if node_input.amount_idr < 5_000_000:
        return Event(route="AUTO", output=node_input)
    if node_input.amount_idr < 50_000_000:
        return Event(route="MANAGER", output=node_input)
    return Event(route="DIRECTOR", output=node_input)

def auto_approve(node_input: PurchaseRequest):
    return Event(output=Decision(approved=True, approver="policy:auto"))

def manager_approval(node_input: PurchaseRequest):
    # Pauses the run. The human's reply becomes the NEXT node's input.
    yield RequestInput(message="Manager approval",
                       payload=node_input.model_dump(),
                       response_schema=Decision)

def director_approval(node_input: PurchaseRequest):
    yield RequestInput(message="Director approval",
                       payload=node_input.model_dump(),
                       response_schema=Decision)

# ADK coerces node_input to the annotation, so the human's JSON reply
# arrives here as a validated Decision.
def decision_router(node_input: Decision):
    route = "APPROVED" if node_input.approved else "REJECTED"
    return Event(route=route, output=node_input)

def create_po(node_input: Decision, ctx: Context):
    return Event(message=f"PO drafted for {ctx.state['request']['vendor']}")

def notify_rejection(node_input: Decision):
    return Event(message=f"Rejected by {node_input.approver}")

def needs_review():
    return Event(message="Unrecognised route; sent to procurement inbox")

root_agent = Workflow(
    name="procurement_approval",
    edges=[
        ("START", intake_agent, route_by_amount),
        (route_by_amount, {
            "AUTO": auto_approve,
            "MANAGER": manager_approval,
            "DIRECTOR": director_approval,
            DEFAULT_ROUTE: needs_review,     # never ship a router without it
        }),
        (auto_approve, decision_router),
        (manager_approval, decision_router),
        (director_approval, decision_router),
        (decision_router, {"APPROVED": create_po, "REJECTED": notify_rejection}),
    ],
)

Tiga detail menopang desain ini. Intake agent memakai output_schema, sehingga router menerima PurchaseRequest yang sudah tervalidasi, bukan prosa. ADK meng-coerce setiap node_input ke type annotation parameternya, itu sebabnya sebuah node boleh mengembalikan dict biasa dan node berikutnya tetap menerima model. Lalu route_by_amount menulis permintaan ke ctx.state, sehingga create_po bisa membaca vendor dua langkah kemudian tanpa harus mengopernya lewat setiap return value.

Menunggu manusia dengan RequestInput

Manajer tidak menjawab dalam hitungan milidetik, jadi node approval tidak boleh blocking. Yield RequestInput mengakhiri run saat itu dan mengirim event berisi function call bernama adk_request_input, membawa pesan, payload Anda, dan interrupt id yang juga tercantum di long_running_tool_ids. Untuk melanjutkan, kirim function response dengan id tersebut ke session yang sama.

import asyncio
from google.adk import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from agent import root_agent

async def main():
    sessions = InMemorySessionService()
    runner = Runner(app_name="proc", node=root_agent, session_service=sessions)
    session = await sessions.create_session(app_name="proc", user_id="budi")

    ask = types.Content(role="user", parts=[types.Part(
        text="Need 12 juta for printer paper from PT Kertas, cost centre OPS")])

    pending = None
    async for ev in runner.run_async(user_id="budi", session_id=session.id,
                                     new_message=ask):
        if ev.long_running_tool_ids:            # the RequestInput event
            pending = next(iter(ev.long_running_tool_ids))
            print("paused at", ev.node_info.path)
            # -> procurement_approval@1/manager_approval@1

    # Hours later, from the approval UI. Same session id, same interrupt id.
    answer = types.Content(role="user", parts=[types.Part(
        function_response=types.FunctionResponse(
            id=pending, name="adk_request_input",
            response={"approved": False, "approver": "siti"}))])

    async for ev in runner.run_async(user_id="budi", session_id=session.id,
                                     new_message=answer):
        if ev.content and ev.content.parts[0].text:
            print(ev.content.parts[0].text)     # -> Rejected by siti

asyncio.run(main())

• Balasan diteruskan ke node penerus, bukan ke node yang berhenti, jadi manager_approval tidak pernah berjalan dua kali. Itulah perilaku default function node, yang rerun_on_resume-nya false.
• response_schema ditegakkan saat balasan masuk. Balasan dengan approved bernilai maybe membuat run_async melempar WorkflowDataError berisi error Pydantic, jadi bungkus pemanggilan resume dan kirim error-nya kembali ke UI approval Anda.
• Nilai default tidak bertahan dalam perjalanan pulang-pergi. Model Decision pertama saya punya note bertipe str dengan default string kosong; nilai yang di-resume membawa note sebagai None, dan decision_router yang ber-annotation gagal dengan event ValidationError. Mendeklarasikan note sebagai optional dengan default None memperbaikinya.

Route yang tidak dipetakan menghentikan run diam-diam

Inilah kegagalan yang paling saya harap ada orang yang memperingatkan sebelumnya. Ketika router mengembalikan nilai route yang bukan key di map-nya, ADK tidak melempar exception dan tidak mengirim event error. Event output router muncul, lalu run langsung berakhir. Dalam pengujian saya mengembalikan nilai yang tidak dipetakan dan hanya mendapat satu event.

# Wrong: a router whose value is not a key. No exception, no error event —
# the run emits router's output and simply stops.
(router, {"AUTO": auto_approve, "MANAGER": manager_approval})
# router returns Event(route="Manager")   <- casing drift from an LLM, or a new tier

# Right: catch everything you did not name.
from google.adk.workflow import DEFAULT_ROUTE
(router, {"AUTO": auto_approve, "MANAGER": manager_approval,
          DEFAULT_ROUTE: needs_review})

Session yang bertahan setelah restart

Jeda hanya berguna kalau jedanya lebih awet daripada prosesnya. InMemorySessionService kehilangan semuanya saat restart, jadi production butuh DatabaseSessionService. Saya membuat session, menjalankannya sampai interrupt manajer, lalu membuat service dan Runner baru di file SQLite yang sama dan mengirim approval. Workflow melanjutkan di record_decision dengan data permintaan masih ada di state.

# pip install "google-adk[db]==2.10.0" aiosqlite greenlet
#                                               ^^^^^^^^ not pulled in by [db];
# without it you get: ValueError: Failed to create database engine for URL ...
from google.adk.sessions import DatabaseSessionService

DB_URL = "sqlite+aiosqlite:///./proc.db"   # the service needs an async driver, not plain sqlite://

# Process 1: run until the manager_approval interrupt, then exit.
sessions = DatabaseSessionService(db_url=DB_URL)
runner = Runner(app_name="proc", node=root_agent, session_service=sessions)

# Process 2, after a restart: a brand-new service and runner on the same file.
sessions = DatabaseSessionService(db_url=DB_URL)
runner = Runner(app_name="proc", node=root_agent, session_service=sessions)
# run_async(..., new_message=<function_response for the stored interrupt id>)
# -> record_decision runs; nothing before the interrupt is executed again.

# What lands on disk (tables: sessions, events, app_states, user_states,
# adk_internal_metadata). Each events.event_data row is JSON like:
# {"author": ..., "node_info": {"path": "procurement_approval@1/route_by_amount@1",
#   "output_for": [...]}, "output": {...}, "actions": {...}}

Sampai ke sana saya terjebak satu error yang menyesatkan. Extra db meng-install SQLAlchemy, yang ter-resolve ke 2.1.1, dan layer asyncio di rilis itu butuh greenlet, yang tidak ikut ter-install. ADK hanya melaporkan gagal membuat database engine; penyebab sebenarnya adalah ImportError di bawahnya. Meng-install greenlet menyelesaikannya. Baris event juga menunjukkan kenapa custom session store rusak di 2.0: node_info dan output sekarang ada di setiap event yang disimpan, dan validator yang menolak property tak dikenal akan menolaknya.

Saat graph kurang cocok: dynamic dan collaborative workflow

Matriks approval sungguhan jarang hanya tiga kotak tetap. Di atas 500 juta rupiah rantai kami menambah CFO, dan menggambar graph untuk setiap panjang rantai cepat jadi berantakan. Dynamic workflow menuliskannya sebagai loop, dengan setiap approval didelegasikan ke child node lewat ctx.run_node.

# Same module as before: reuses PurchaseRequest and intake_agent
from google.adk import Context, Workflow
from google.adk.events import RequestInput
from google.adk.workflow import node

def approval_chain(amount_idr: int) -> list[str]:
    chain = ["manager"]
    if amount_idr >= 50_000_000:
        chain.append("director")
    if amount_idr >= 500_000_000:
        chain.append("cfo")
    return chain

@node(name="ask_approver")
def ask_approver(node_input: dict):
    yield RequestInput(message=f"{node_input['role']} approval", payload=node_input)

@node(rerun_on_resume=True)          # required for an orchestrator using run_node
async def approval_loop(ctx: Context, node_input: PurchaseRequest):
    for role in approval_chain(node_input.amount_idr):
        # Each resume re-enters this loop from the top. Completed ask_approver
        # calls return their stored answer instantly instead of pausing again.
        decision = await ctx.run_node(
            ask_approver, node_input={"role": role, **node_input.model_dump()})
        if not decision["approved"]:
            return f"rejected by {role}"
    return "approved"

root_agent = Workflow(name="approval_chain",
                      edges=[("START", intake_agent, approval_loop)])

# Trace for 600,000,000 IDR, three resumes:
#   body: manager                      -> pause
#   body: manager, director            -> pause
#   body: manager, director, cfo       -> pause
#   body: manager, director, cfo       -> "approved"
# Anything with side effects belongs in a child node, never in this body.

Trace di komentar itu berasal dari run sungguhan, dan itulah yang perlu dipahami: setiap kali resume, body orchestrator mulai lagi dari atas, dan child yang sudah selesai mengembalikan jawaban tersimpan alih-alih berhenti lagi. Checkpoint-nya ada di child. Apa pun yang punya side effect, seperti email, penulisan ke ERP, atau counter, harus menjadi node tersendiri, kalau tidak ia akan terpicu sekali per resume. Ketika pekerjaannya sendiri terbuka, seperti mencari vendor dan mengumpulkan penawaran, koordinator dengan subagent lebih cocok.

vendor_lookup = Agent(
    name="vendor_lookup",
    mode="single_turn",                 # no user interaction, can run in parallel
    tools=[search_vendor_master],
)

quote_collector = Agent(
    name="quote_collector",
    mode="task",                        # may ask clarifying questions, auto-returns
    input_schema=QuoteRequest,
    output_schema=QuoteSummary,
    tools=[request_quote, read_quote_inbox],
)

root_agent = Agent(                     # no mode on the root
    name="sourcing_coordinator",
    sub_agents=[vendor_lookup, quote_collector],
)

• chat adalah mode default: subagent berbicara dengan user dan harus mengembalikan kontrol secara eksplisit.
• task boleh mengajukan pertanyaan klarifikasi dan kembali otomatis saat selesai. Mode ini tidak boleh punya subagent sendiri dan dinonaktifkan di dalam graph workflow pada Python 2.0.
• single_turn tidak mengizinkan interaksi dengan user, langsung kembali, dan merupakan satu-satunya mode yang bisa berjalan paralel.

Koordinator mendapat satu tool yang dibuat otomatis per subagent, dinamai sesuai subagent-nya, dan agent dengan mode task dan single_turn berjalan di cabang session yang terisolasi, sehingga sibling paralel tidak bisa melihat pekerjaan satu sama lain. Jangan pasang mode pada root agent. Model yang memilih subagent mana yang dipanggil, dan justru karena itu approval tetap di graph sementara sourcing ada di sini.

Menguji graph dan mengevaluasi agent

Bagi pengujian di garis yang sama dengan desainnya. Node deterministik adalah function Python biasa, jadi uji dengan pytest tanpa API key. Satu-satunya node yang memakai model adalah tempat adk eval berguna: tool_trajectory_avg_score membandingkan tool call dengan daftar yang diharapkan, dan response_match_score membandingkan jawaban akhir.

// tests/test_config.json — for the intake_agent eval set
{
  "criteria": {
    "tool_trajectory_avg_score": { "threshold": 1.0, "match_type": "IN_ORDER" },
    "response_match_score": 0.8
  }
}

# Run it
adk eval procurement/ tests/intake.evalset.json \
  --config_file_path=tests/test_config.json --print_detailed_results

# The routing nodes need no model at all: plain pytest
def test_director_threshold():
    assert approval_chain(50_000_000) == ["manager", "director"]
    assert approval_chain(49_999_999) == ["manager"]

Kriteria trajectory secara default memakai threshold 1.0 dengan exact match, artinya tidak boleh ada call yang berlebih atau hilang. IN_ORDER menerima call tambahan di antara call yang diharapkan, dan ANY_ORDER mengabaikan urutan, jadi pilih mode paling longgar yang masih menangkap regresi yang Anda pedulikan. response_match_score default-nya 0.8. Untuk intake agent, yang penting adalah agent tidak pernah melewatkan pencarian vendor, jadi IN_ORDER adalah setelan yang jujur.

Migrasi dari ADK 1.x tanpa merusak session

Sebagian besar agent 1.x tetap jalan, tetapi empat perubahan ini yang sering menggigit dalam praktik.

1. Perbarui custom session store dan validator JSON yang ketat untuk field Event baru: node_info, output, route, dan isolation scope di semua bahasa, ditambah RequestedInput di Go.
2. Di Python, override _run_async_impl dilewati. Pindahkan logikanya ke callback before dan after agent, yield event alih-alih menambahkan langsung ke context.session.events, dan berhenti menangkap BaseException supaya retry bisa bekerja.
3. Di Go, ganti module path menjadi google.golang.org/adk/v2 dan berikan context sebagai argumen pertama session.NewEvent.
4. Di TypeScript, SequentialAgent, ParallelAgent, dan LoopAgent sudah deprecated dan digantikan graph workflow, dan InvocationContext.agent sekarang bisa undefined.

// Go 1.x
import "google.golang.org/adk/session"
ev := session.NewEvent(ctx.InvocationID())

// Go 2.0 — new module path, and NewEvent takes the context first
import "google.golang.org/adk/v2/session"
ev := session.NewEvent(ctx, ctx.InvocationID())

# Python: pin 1.x while you migrate
pip install "google-adk~=1.0"
# TypeScript
npm install @google/adk@^1.6.0

Aturan yang saya bawa dari pembangunan ulang ini: biarkan model membaca, biarkan kode memutuskan. Di ADK 2.0 bentuknya adalah Workflow yang satu-satunya node LLM-nya mengubah teks menjadi schema, dengan setiap batas nominal berada di route map yang punya DEFAULT_ROUTE. Simpan session di database sebelum jeda manusia pertama, dan taruh side effect di child node supaya resume tidak mengulanginya.