Google ADK 2.0: bikin agent workflow berbasis graph di Python

Agent support dengan satu prompt panjang dan dua belas tool kelihatan bagus di demo. Terus ada customer kirim pesan yang butuh urutan tetap: klasifikasi, cek order, putuskan, balas. Agent-nya jawab, jawabannya kedengeran masuk akal, dan kamu nggak punya cara buat tahu dia benar-benar cek order atau cuma mengarang statusnya. Control flow-nya ada di dalam kepala model, jadi nggak bisa di-test.

Agent Development Kit (ADK) 2.0 dari Google nambahin graph runtime buat kasus ini. Kamu deklarasikan node dan edge, pakai model di tempat yang butuh penalaran, dan pakai Python biasa di tempat yang nggak butuh. Artikel ini bikin satu workflow triage support, jalanin lokal, dan pasang eval set biar perubahan prompt nggak diam-diam merusak routing.

Yang perlu disiapkan

• Python 3.10 atau lebih baru
• pip
• API key Gemini dari Google AI Studio
• Nyaman dengan type hint dan Pydantic

Langkah 1: install dan scaffold

python -m venv .venv
source .venv/bin/activate
pip install google-adk
adk create my_agent

adk create bikin tiga file:

my_agent/
    agent.py      # root agent kamu
    .env          # API key
    __init__.py

Taruh key-nya di my_agent/.env:

echo 'GOOGLE_API_KEY="YOUR_API_KEY"' > my_agent/.env

Satu-satunya objek yang wajib ada di agent.py adalah root_agent. Sisanya cuma ngedit satu file itu.

Langkah 2: bikin tool yang dipanggil model dengan benar

ADK nyusun schema tool dari signature fungsi dan docstring-nya. Parameter yang punya type hint tanpa default itu wajib. Kasih default dan dia jadi opsional. Deskripsi parameter diambil dari docstring, jadi docstring di sini berfungsi sebagai prompt, bukan sekadar dokumentasi.

from google.adk.tools import ToolContext

ORDERS = {
    "A-1042": {"status": "delivered", "total_idr": 149000, "days_since_delivery": 3},
    "A-2098": {"status": "in_transit", "total_idr": 89000, "days_since_delivery": None},
}

def lookup_order(order_id: str, tool_context: ToolContext) -> dict:
    """Looks up one order in the local order table.

    Args:
        order_id (str): Order id in the form A-1042.

    Returns:
        dict: status, total_idr, and days_since_delivery, or status "not_found".
    """
    key = (order_id or "").strip().upper()
    order = ORDERS.get(key)
    if order is None:
        return {"status": "not_found", "order_id": key}
    tool_context.state["temp:last_order_id"] = key
    return {"status": "success", "order_id": key, **order}

Ada dua hal yang layak ditiru dari snippet itu. Balikin {"status": "not_found"} daripada melempar exception, biar model punya bahan buat memperbaiki diri. Dan parameter bertipe ToolContext di-inject framework dan disembunyikan dari model, itu cara sebuah tool baca session state atau memicu transfer antar agent.

Parameter variadic (*args, **kwargs) diabaikan waktu ADK bikin schema, jadi semua data yang harus diisi model mesti jadi parameter eksplisit.

Langkah 3: pahami di mana session state tinggal

session.state itu dictionary, dan prefix key-nya nentuin scope:

• tanpa prefix: cuma untuk percakapan ini
• user:, misal user:preferred_language, dipakai bareng di semua session milik user itu
• app: dipakai bareng semua user di aplikasi itu
• temp: dibuang begitu invocation sekarang selesai

Apakah isinya bertahan setelah restart tergantung session service yang kamu pilih. InMemorySessionService kehilangan semuanya, DatabaseSessionService dan VertexAiSessionService nggak.

Penulisan state di dalam tool atau callback lewat ToolContext.state atau CallbackContext.state dan dicatat sebagai event. Nulis ke objek session yang kamu ambil langsung dari session service kelihatan jalan, tapi diam-diam gagal disimpan kalau backend-nya database. Baca di situ, tulis di sini.

Kamu juga bisa nyuntik state ke instruction pakai kurung kurawal, misalnya "Answer in {user:preferred_language}." Key yang nggak ada bikin error, jadi pakai {key?} untuk nilai yang belum tentu ada.

Langkah 4: definisikan workflow sebagai graph

Workflow ini punya empat bagian: satu agent klasifikasi, satu fungsi router, dua agent cabang, dan satu jalur tanpa model buat pesan yang butuh manusia.

from typing import Literal
from pydantic import BaseModel
from google.adk import Agent, Event, Workflow

class TicketIntent(BaseModel):
    """Routing decision for one incoming support message."""
    route: Literal["REFUND", "TRACKING", "OTHER"]
    order_id: str
    reason: str

intent_agent = Agent(
    name="intent_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Read the customer message and classify it as REFUND, TRACKING, or OTHER. "
        "Extract the order id if it appears, otherwise use an empty string. "
        "Keep reason under 20 words."
    ),
    output_schema=TicketIntent,
)

def route(intent: TicketIntent) -> str:
    return intent.route

output_schema bikin model balikin TicketIntent yang sudah divalidasi, bukan prosa, jadi node berikutnya terima objek Pydantic, bukan string yang harus kamu parse manual. input_schema ngelakuin hal yang sama untuk arah masuknya.

Agent cabangnya Agent biasa dengan tool lookup terpasang:

refund_agent = Agent(
    name="refund_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Call lookup_order with the order id before writing anything. "
        "If the order is delivered and fewer than 7 days have passed, approve the refund. "
        "Otherwise explain the policy and offer a manual review. Reply in the customer's language."
    ),
    tools=[lookup_order],
    input_schema=TicketIntent,
)

tracking_agent = Agent(
    name="tracking_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Call lookup_order and report the delivery status in one short paragraph. "
        "If the order is not found, ask for the order id again."
    ),
    tools=[lookup_order],
    input_schema=TicketIntent,
)

def escalate(intent: TicketIntent) -> Event:
    return Event(message=f"Needs a human: {intent.reason}")

Terus graph-nya. START itu titik masuk, tuple ngerangkai node, dan tuple yang elemen keduanya dictionary ngubah sebuah node jadi router:

root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", intent_agent, route),
        (route, {
            "REFUND": refund_agent,
            "TRACKING": tracking_agent,
            "OTHER": escalate,
        }),
    ],
)

route balikin salah satu dari tiga string dan dictionary-nya memetakan tiap string ke node berikutnya. Nggak ada model yang dilibatkan buat keputusan routing di bagian ini, jadi pesan yang jelas-jelas soal status pengiriman nggak mungkin nyasar ke jalur refund.

Satu detail yang perlu diingat: output tiap node jadi input node berikutnya. Agent klasifikasi nyerahin TicketIntent ke route, route nyerahin key string ke cabangnya, dan agent cabang menutup run. Kalau mau ada langkah tetap setelah semua cabang, tambahkan node itu ke edge tiap cabang.

Langkah 5: jalanin dan baca event-nya

adk run my_agent          # chat di terminal
adk web --port 8000       # UI browser buat development

UI web itu buat testing dan debugging, dan dokumentasinya tegas bilang itu bukan deployment production.

Buat akses programatik, nyalain API server lalu tembak pakai curl:

adk api_server

curl -X POST http://localhost:8000/apps/my_agent/users/u_123/sessions/s_123 \
  -H "Content-Type: application/json" \
  -d '{"channel": "whatsapp"}'

curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "appName": "my_agent",
    "userId": "u_123",
    "sessionId": "s_123",
    "newMessage": {
      "role": "user",
      "parts": [{"text": "Order A-1042 arrived damaged, I want my money back"}]
    }
  }'

/run balikin semua event sekaligus. /run_sse streaming event yang sama lewat Server-Sent Events dan nerima "streaming": true buat output per token. Baca event-nya, bukan teks akhirnya: event function call untuk lookup_order itu bukti lookup-nya benar-benar kejadian.

Langkah 6: pasang eval set

Bug routing bakal balik lagi tiap kali ada yang ngedit instruction. Eval set nangkep itu.

adk eval my_agent tests/triage.test.json --print_detailed_results

{
  "eval_set_id": "triage_refund_path",
  "name": "Refund path",
  "eval_cases": [
    {
      "eval_id": "damaged_order_refund",
      "conversation": [
        {
          "user_content": {
            "parts": [{"text": "Order A-1042 arrived damaged"}],
            "role": "user"
          },
          "final_response": {
            "parts": [{"text": "We have approved the refund for order A-1042."}],
            "role": "model"
          },
          "intermediate_data": {
            "tool_uses": [
              {"name": "lookup_order", "args": {"order_id": "A-1042"}}
            ],
            "intermediate_responses": []
          }
        }
      ],
      "session_input": {
        "app_name": "my_agent",
        "user_id": "test_user",
        "state": {}
      }
    }
  ]
}

List tool_uses itu trajectory yang kamu harapkan, dan ADK membandingkannya dengan yang benar-benar dilakukan agent. Itu pengecekan yang nggak bisa dilakukan assertion biasa ke teks akhir. Schema test file-nya berbasis Pydantic, dan dokumentasinya juga bahas AgentEvaluator.evaluate buat jalanin set yang sama di dalam pytest atau job CI.

Langkah 7: deploy

adk deploy docker --with_ui my_agent
adk deploy cloud_run --with_ui my_agent

Jalur Docker ngasih kamu container yang bisa ditaruh di mana saja, dan Cloud Run itu opsi terkelola. Lepas --with_ui buat apa pun yang publik.

Kapan graph jadi bentuk yang tepat

Satu agent dengan tool tetap pilihan yang benar kalau urutan langkahnya nggak penting dan model bebas memilih. Template workflow (SequentialAgent, LoopAgent, ParallelAgent) ngurus fan-out dan loop antar sub-agent. Di ADK 2.0, graph runtime menangani kasus-kasus itu dengan kontrol lebih banyak, dan dokumentasinya menandai template workflow sebagai superseded untuk Python dan Go. Kalau kamu di TypeScript, Java, atau Kotlin, template workflow masih jalur yang didukung.

Pakai graph kalau rute antar langkah harus bisa diprediksi, kalau kamu butuh kode di antara panggilan model, atau kalau ada node yang harus berhenti nunggu manusia. Jauhkan model dari keputusan yang jawabannya sudah kamu tahu.

Langkah selanjutnya

• Bungkus intent_agent jadi AgentTool kalau agent lain butuh, biar logika routing-nya cuma ada satu salinan.
• Ganti InMemorySessionService dengan service berbasis database sebelum ada orang lain yang pakai workflow-nya.
• Tambahkan callback buat kirim daftar event ke stack tracing kamu, karena event-nya sudah ada di situ.

Referensi

• Graph-based agent workflows, ADK docs
• State: the session's scratchpad
• Why evaluate agents
• google/adk-python on GitHub