Applying 32 GenAI design patterns to a real project

   ┌──────────────────────────────────────────────────────────────────┐
   │                   Support Router (Workflow)                      │
   │                                                                  │
   │  load_context ──▶ classify ──▶ route ──▶ ... ──▶ save_context    │
   └───────────────────────────┬──────────────────────────────────────┘
                               │
         ┌───────────┬─────────┼─────────┬───────────┐
         │           │         │         │           │
         ▼           ▼         ▼         ▼           ▼
   ┌──────────┐ ┌─────────┐ ┌─────┐ ┌────────┐ ┌──────────┐
   │Knowledge │ │Diagnostc│ │ Ops │ │Billing │ │Escalation│
   │  Agent   │ │  Agent  │ │Agent│ │ Agent  │ │  Agent   │
   └────┬─────┘ └────┬────┘ └──┬──┘ └───┬────┘ └────┬─────┘
        │            │         │        │           │
        ▼            ▼         ▼        ▼           ▼
   ┌────────┐  ┌─────────┐ ┌───────┐ ┌───────┐ ┌─────────┐
   │  KB    │  │SSH Diag │ │  CW   │ │Billing│ │Intercom │
   │  RAG   │  │  Tool   │ │  API  │ │  API  │ │  API    │
   └────────┘  └─────────┘ └───────┘ └───────┘ └─────────┘
   ┌──────────────────────────────────────────────────────────────────┐
   │                    Shared Infrastructure                         │
   │  Customer Context │ Conv History │ Change Log │ Billing/CRM      │
   └──────────────────────────────────────────────────────────────────┘

from google.adk import Agent, Workflow, Event, Context
from google.adk.workflow import node, RetryConfig
from pydantic import BaseModel
from typing import Literal


# --- Context loading and persistence ---

@node(retry_config=RetryConfig(max_attempts=2))
async def load_context(ctx: Context, node_input: str):
    """Pull customer profile and recent history into state."""
    customer_id = ctx.state.get("customer_id")
    if customer_id:
        ctx.state["customer_profile"] = await fetch_customer_profile(customer_id)
        ctx.state["customer_history"] = await fetch_recent_conversations(customer_id)
    return node_input


async def save_context(ctx: Context, node_input: str):
    """Persist what was learned after the agent responds."""
    customer_id = ctx.state.get("customer_id")
    diagnosis = ctx.state.get("diagnosis")
    if customer_id and diagnosis:
        await persist_interaction(
            customer_id=customer_id,
            issue=diagnosis,
            resolution=node_input,
        )
    return node_input


# --- Tools (specifics come in later posts) ---

def search_knowledge_base(query: str):
    """Search the hosting knowledge base for relevant articles."""
    ...

def check_server_status(server_id: str):
    """CPU, memory, disk, and service health for a server."""
    ...

def get_server_logs(server_id: str, service: str, lines: int = 100):
    """Recent log entries for a service."""
    ...

def get_invoices(customer_id: str):
    """Recent invoices and payment status."""
    ...

def restart_service(server_id: str, service: str):
    """Restart a service. Confirmation gated by a callback."""
    ...

async def create_intercom_conversation(email: str, transcript: str):
    """Hand off to a human via Intercom with the full transcript attached."""
    ...


# --- Specialist agents ---

knowledge_agent = Agent(
    name="knowledge_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Answer the hosting question using the knowledge base.\n"
        "Customer: {customer_profile}\n"
        "Always cite the specific KB article. "
        "If you cannot find the answer, say so."
    ),
    tools=[search_knowledge_base],
)

diagnostics_agent = Agent(
    name="diagnostics_agent",
    model="gemini-2.5-pro",
    instruction=(
        "Diagnose this server issue step by step.\n"
        "Customer: {customer_profile}\n"
        "Check server status first, then relevant logs, then correlate "
        "findings before suggesting a fix."
    ),
    tools=[check_server_status, get_server_logs],
)

ops_agent = Agent(
    name="ops_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Execute approved server operations.\n"
        "Customer: {customer_profile}\n"
        "Never run destructive operations without confirmation."
    ),
    tools=[restart_service],
)

billing_agent = Agent(
    name="billing_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Handle billing and plan questions.\n"
        "Customer: {customer_profile}\n"
        "Look up invoices and plan details. "
        "Never process refunds or plan changes. Escalate those."
    ),
    tools=[get_invoices],
)

escalation_agent = Agent(
    name="escalation_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Hand the conversation off to a human support engineer.\n"
        "Customer: {customer_profile}\n"
        "Create an Intercom conversation and attach the full transcript."
    ),
    tools=[create_intercom_conversation],
)


# --- Intent classifier and router ---

class Intent(BaseModel):
    intent: Literal["KNOWLEDGE", "DIAGNOSTICS", "OPS", "BILLING", "ESCALATION"]
    urgency: Literal["LOW", "MEDIUM", "HIGH"]


classify = Agent(
    name="classify",
    model="gemini-2.5-flash",
    instruction=(
        "Classify this hosting support request.\n"
        "Customer: {customer_profile}\n"
        "KNOWLEDGE: how-to and configuration questions.\n"
        "DIAGNOSTICS: performance issues, errors, downtime.\n"
        "OPS: restart services, clear cache, apply approved changes.\n"
        "BILLING: invoices, plans, payments.\n"
        "ESCALATION: customer asks for a human."
    ),
    output_schema=Intent,
    output_key="classification",
)


def route(classification: Intent):
    yield Event(route=classification.intent)


# --- Workflow ---

root_agent = Workflow(
    name="support_router",
    edges=[
        ("START", load_context, classify, route),
        (route, {
            "KNOWLEDGE": knowledge_agent,
            "DIAGNOSTICS": diagnostics_agent,
            "OPS": ops_agent,
            "BILLING": billing_agent,
            "ESCALATION": escalation_agent,
        }),
        (knowledge_agent, save_context),
        (diagnostics_agent, save_context),
        (ops_agent, save_context),
        (billing_agent, save_context),
        (escalation_agent, save_context),
    ],
)