De un prompt largo a un grafo de nodos con decisión explícita.

01Por qué existen los workflows

El problema clásico: cuanto más crece el prompt de un agente, menos confiable es.

Hasta ahora, cuando querías que un LLM hiciera un proceso multi-paso (clasificar → buscar → responder → escalar), lo metías todo en un solo instruction largo y le dabas varias tools. Funciona con 3 pasos. Con 7, empieza a fallar: omite pasos, los invierte, se olvida de verificar, alucina transiciones.

El workflow agent en ADK existe para sacar el flujo del prompt y llevarlo a código. Cada paso se vuelve una unidad explícita, ejecutable, observable y testeable por separado.

Lo que un workflow te garantiza

02Modelo mental: un grafo de ejecución

Pensalo como un diagrama de flujo, pero donde cada caja es ejecutable.

Un grafo de workflow en ADK tiene tres ingredientes:

• Nodos — unidades de trabajo. Pueden ser agentes (LLM), tools, código, o incluso otros workflows.
• Edges — flechas que conectan nodos. Definen el orden y la condición de transición.
• START — keyword reservado que marca el punto de entrada del grafo.

Y eso es todo. La diferencia con un prompt largo es que acá cada caja corre en orden, y la flecha la definís vos en código. Si la flecha es condicional, se vuelve un edge de routing.

03Los 3 estilos de orquestación en ADK

ADK ofrece tres formas de componer trabajo multi-paso. No compiten — se complementan.

04Nodos: las unidades de trabajo

Un nodo es cualquier cosa que reciba un input, haga algo, y devuelva un output. Hay cuatro tipos.

1. Agent (LLM node)

Un agente estándar de ADK dentro del grafo. Recibe el output del nodo anterior, lo procesa con un LLM + sus tools, y devuelve su respuesta.

2. Function (código puro)

Una función Python cualquiera. No invoca LLM. Útil para transformaciones deterministas, validaciones, lookups en DB, llamadas a APIs internas, branching condicional.

from google.adk import Event

def my_function_node(node_input: str):
    # código determinista, sin LLM
    output_value = node_input.upper()
    return Event(output=output_value)  # "THE RESULT"

3. Tool (tool de ADK)

Cualquier tool registrada se puede usar como nodo. La diferencia con un Function es semántica: lo marcás como acción externa (típicamente con I/O, side effects, auth, etc.).

4. Sub-workflow (workflow anidado)

Un grafo entero puede ser un nodo dentro de otro grafo. Te permite encapsular sub-procesos reusables. Lo vemos en detalle en §14.

5. Human input (nodo humano)

Nodo especial que pausa la ejecución y espera input del usuario. Útil para aprobaciones, escalaciones, o cuando el LLM no puede decidir solo.

05Edges y routing: la lógica de transición

Si los nodos son "qué se ejecuta", los edges son "qué pasa después". Y acá es donde aparece la diferencia real con un prompt.

Edge secuencial

El caso más simple. El framework encadena los nodos en orden, y el output de uno se pasa al siguiente automáticamente.

# 3 nodos en orden, START implícito al primero
edges=[
    ("START", task_a_node, task_b_node, task_c_node)
]

Edge condicional (branching)

Un nodo devuelve un Event(route=...) con una clave. El edge-dict mapea cada clave a su nodo destino.

def router(node_input: str):
    if condition(node_input):
        return Event(route="RUN_TASK_C")
    return Event(route="RUN_TASK_B")

edges=[
    ("START", task_a_node, router),
    (router, {
        "RUN_TASK_B": task_b_node,
        "RUN_TASK_C": task_c_node,
    }),
]

El router es típicamente un LlmAgent que clasifica, o un Function que evalúa una condición. La clave del dict es el route key que el nodo emite en su Event.route.

06Data flow: cómo viaja la información entre nodos

Tres campos del Event dominan todo el flujo de datos. Entendé estos tres y entendés el 80% del modelo.

output

Dato que se pasa al siguiente nodo como su input tipado. Un nodo emite un solo output por ejecución.

message

Texto que se muestra al usuario. No se pasa al siguiente nodo — es para el canal de UI.

state

Parámetros clave-valor que se persisten en la sesión. Accesibles por cualquier nodo downstream y por tools/callbacks. Tiene scopings (lo vemos en §08).

Ejemplo: pasar un Pydantic model entre nodos

Un nodo puede devolver un objeto estructurado. El framework lo serializa a JSON y lo deserializa al tipo del input del siguiente nodo. No hay string parsing en el medio.

from google.adk import Agent, Workflow
from pydantic import BaseModel

class CityTime(BaseModel):
    time_info: str
    city: str

def lookup_time_function(city: str):
    return CityTime(time_info="10:10 AM", city=city)

city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-flash-latest",
    input_schema=CityTime,  # ← tipado fuerte
    instruction="""Return: It is {time_info} in {city} right now.""",
    output_schema=str,
)

Yield vs Return

Un nodo puede usar yield Event(...) para emitir varios eventos a lo largo de su ejecución (útil para streaming al usuario). Pero solo un Event.output por ejecución — si hacés dos yields con output, runtime error.

07Schemas tipados: hacer cumplir los contratos

Si los nodos se pasan datos, esos datos tienen que tener una forma. Los schemas son el contrato.

Cada agente en un grafo puede tener input_schema y output_schema (clases Pydantic). El framework:

1. Valida el input antes de pasarlo al agente.
2. Fuerza al LLM a responder con un JSON que matchee el output_schema.
3. Valida esa respuesta antes de pasarla al siguiente nodo.

Resultado: no más "el LLM devolvió algo raro". Si el output no matchea, falla rápido y de forma específica.

class FlightSearchInput(BaseModel):
    origin: str           # "SFO"
    destination: str      # "CDG"
    departure_date: str  # "2026-03-15"
    passengers: int = 1

class FlightSearchOutput(BaseModel):
    flights: list[dict]
    cheapest_price: float

flight_searcher = Agent(
    name="flight_searcher",
    instruction="Search for available flights.",
    input_schema=FlightSearchInput,
    output_schema=FlightSearchOutput,
    tools=[search_flights_api],
    mode="single_turn",
)

08State y scopings

A veces necesitás que un dato viva a lo largo de toda la sesión, no solo entre dos nodos. Para eso está state.

Scoping con prefijos

Las keys de state pueden tener prefijo que define su lifetime:

Acceso en instrucciones

Desde el instruction de un agente podés leer state con la sintaxis de template:

instruction = """
The user's name is {user_name}. Their current ticket is {ticket_id}.
Their previous interactions this session: {interaction_count}.
Respond accordingly.
"""

O leer del nombre del nodo fuente (más restrictivo):

instruction = """
The search result from <search_node> is: <search_node.results[0]>.
"""

09Patrón 1: Secuencia

El caso más simple. Tres nodos en orden, cada uno alimenta al siguiente. El ejemplo canónico de la doc oficial.

Escenario: generar nombre de ciudad → buscar la hora actual → reportar.

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

city_generator_agent = Agent(
    name="city_generator_agent",
    model="gemini-flash-latest",
    instruction="""Return the name of a random city. Return only the name, nothing else.""",
    output_schema=str,
)

class CityTime(BaseModel):
    time_info: str
    city: str

def lookup_time_function(node_input: str):
    return CityTime(time_info="10:10 AM", city=node_input)

city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-flash-latest",
    input_schema=CityTime,
    instruction="""Output: It is {time_info} in {city} right now.""",
    output_schema=str,
)

def completed_message_function(node_input: str):
    return Event(message=f"{node_input}\n WORKFLOW COMPLETED.")

root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", city_generator_agent, lookup_time_function,
         city_report_agent, completed_message_function)
    ],
)

Notá el patrón: alternás agentes LLM con funciones puras. Cada función es una decisión determinista o transformación. Cada agente es un paso que necesita razonar.

10Patrón 2: Branching (routing condicional)

Un nodo decide a qué otro nodo ir según su output. Es el patrón que rompe el "todo en línea recta".

process_message = Agent(
    name="process_message",
    model="gemini-flash-latest",
    instruction="""Classify into "BUG", "CUSTOMER_SUPPORT", or "LOGISTICS".
    If more than one applies, return comma-separated categories.""",
    output_schema=str,
)

def router(node_input: str):
    routes = [r.strip() for r in node_input.split(",")]
    return Event(route=routes)  # puede emitir múltiples rutas

def response_1_bug():           return Event(message="Handling bug...")
def response_2_support():       return Event(message="Handling customer support...")
def response_3_logistics():     return Event(message="Handling logistics...")

root_agent = Workflow(
    name="routing_workflow",
    edges=[
        ("START", process_message, router),
        (router, {
            "BUG": response_1_bug,
            "CUSTOMER_SUPPORT": response_2_support,
            "LOGISTICS": response_3_logistics,
        }),
    ],
)

Detalles que importan

• Un mismo nodo puede emitir múltiples rutas y el framework las ejecuta en paralelo (fan-out implícito). Útil cuando un ticket es BUG y LOGISTICS.
• Si la clasificación es determinista (ej: keyword matching), un FunctionNode alcanza — no necesitás un LLM gastando tokens en clasificar.
• El router también puede ser el LLM Agent mismo si devuelve un string con la categoría. Acá lo separamos para mantener el patrón limpio.

11Patrón 3: Fan-out / Join (paralelismo con barrera)

Varios nodos corren en paralelo, y un nodo de join espera a que todos terminen antes de continuar. Útil para enriquecer contexto desde múltiples fuentes.

Escenario típico: ante una consulta, correr en paralelo (1) búsqueda en knowledge base, (2) lookup de cliente en CRM, (3) historial de tickets previos, y luego sintetizar todo.

from google.adk.workflow import JoinNode

my_join_node = JoinNode(name="my_join_node")

edges=[
    ("START", parallel_task_A, my_join_node),
    ("START", parallel_task_B, my_join_node),
    ("START", parallel_task_C, my_join_node),
    (my_join_node, final_task_D),
]

12Patrón 4: Loop & escalación

Para procesos que necesitan iterar hasta cumplir una condición: refinar un output, validar calidad, esperar aprobación.

En un grafo, un loop es un back-edge: una arista que vuelve a un nodo anterior. Se rompe cuando el nodo emite un route distinto.

# Topología: writer → critic → router → (refiner → critic) loop OR done

edges=[
    ("START", writer_node, critic_node, router),
    (router, {
        "REFINE": refiner_node,
        "DONE":   done_node,
    }),
    (refiner_node, critic_node),  # back-edge = loop
]

El critic_node evalúa la salida. Si no cumple, emite route="REFINE" y el refiner la mejora. Si cumple, emite route="DONE" y sale.

13Patrón 5: Workflows anidados

Un workflow puede ser nodo de otro workflow. Encapsulás sub-procesos reusables y mantenés cada grafo legible.

# workflow_B es un sub-grafo completo, usado como nodo del padre

root_agent = Workflow(
    name="parent_workflow",
    edges=[
        ("START", task_a1, router),
        (router, {
            "RUN_WORKFLOW_B": workflow_b,
            "RUN_WORKFLOW_C": workflow_c,
        }),
    ],
)

Cómo fluye el output en workflows anidados

• Cada nodo del sub-workflow emite Event que se propagan al grafo padre (para trazabilidad).
• El output del sub-workflow hacia el padre es el output de su último nodo.
• El state se comparte — un nodo del sub-workflow puede leer state que setea un nodo del padre.

Cuándo anidar

• Cuando un sub-proceso se repite en varios workflows.
• Cuando un grafo se vuelve muy grande (>15 nodos) y querés particionar.
• Cuando distintos equipos mantienen distintas partes (cada uno su sub-workflow).

14Graph vs Template workflows

Ambos funcionan. Cuándo elegir uno, cuándo el otro.

15Graph vs Dynamic workflows

Dynamic workflows son para cuando el control flow no se puede declarar — se calcula en runtime.

Un dynamic workflow es una subclase de BaseAgent que implementa su propio _run_async_impl. Ahí adentro podés hacer lo que quieras: loops arbitrarios, recursión, branching que depende de respuestas de tools.

Cuándo dynamic es la respuesta correcta

• El número de pasos depende de datos externos (ej: "iterá hasta que la API devuelva status=done").
• Necesitás recursión (ej: descomponer un problema en sub-problemas, cada uno resuelto con el mismo workflow).
• Querés control imperativo total sobre la ejecución.

Para el caso de iPlan, graph es el sweet spot. Dynamic lo dejamos para cuando surjan casos que no entren en el modelo de nodos y aristas.

16Trampas conocidas y límites

Lo que la doc oficial te dice que no funciona con graph workflows.

❌ No compatible con graph workflows

• Live streaming. Si necesitás streaming de tokens en tiempo real, graph no lo soporta. Usá un LlmAgent directo fuera del grafo.
• Algunas integraciones third-party. Ciertas tools/MCPs asumen un contexto conversacional que un grafo no provee. Hay que testear caso por caso.
• LlmAgents conversacionales. No pueden ser multi-turn dentro del grafo. Deben ser single_turn o task mode.

⚠️ Errores comunes de diseño

✅ Buenas prácticas

• Dibujá el grafo en un diagrama antes de codear. Si no podés dibujarlo, no lo entendés.
• Cada nodo con un único propósito. Si un nodo hace 3 cosas, partilo.
• Preferí output tipado sobre state para datos entre nodos. State es para control.
• Testeá cada nodo aisladamente antes de integrarlos.
• Logging: el Event de cada nodo se loggea. Úsalo como observability built-in.

17Caso iPlan: el agente de soporte técnico

Cómo se aplica todo esto al problema real que tenemos entre manos.

El grafo propuesto

Por qué este grafo y no un agente único:

• La clasificación necesita entender la consulta entera — es un trabajo de LLM.
• Las 4 búsquedas (KB, docs API, CRM, monitoring) son independientes y se pueden paralelizar.
• La decisión final (responder vs escalar) depende de un score de confianza que se calcula sobre los 4 resultados — esto es un Function determinista, no un LLM.

18Flujo end-to-end, paso a paso

Qué hace cada nodo y qué datos se pasan.

1. Intake & validación (FunctionNode). Recibe el mensaje crudo del usuario. Valida formato, extrae user_id, channel, timestamp. Devuelve un IntakeData tipado. Si falla la validación, route a error_response.
2. Clasificar consulta (LlmAgent, single_turn). Recibe IntakeData. Clasifica en una o más categorías: BILLING, TECHNICAL, ACCOUNT, SERVICE_OUTAGE, OTHER. Devuelve un Classification con scores de confianza.
3. Fan-out: en función de la clasificación, se ejecutan en paralelo los nodos de búsqueda relevantes. SERVICE_OUTAGE salta la búsqueda en KB (no aporta) y prioriza monitoring.
4. JoinNode. Espera los 4 resultados. Cada nodo que alimenta emite un SourceResult tipado (o un fallback vacío si falló).
5. Evaluar confianza (FunctionNode). Calcula un confidence_score 0-1 combinando: relevancia de la búsqueda, completitud de los datos, claridad de la consulta. Determinista, sin LLM.
6. Router. Emite route="RESPOND" si score ≥ 0.75, route="ASK_CLARIFICATION" si 0.4-0.75, route="ESCALATE" si < 0.4 o si el usuario pidió humano.
7. Responder (LlmAgent). Sintetiza los resultados en una respuesta clara. Guarda la respuesta en artifacts y notifica al usuario.
8. Pedir aclaración (LlmAgent). Genera una pregunta de seguimiento. Mantiene el contexto en state.
9. Escalar a humano (FunctionNode). Crea un ticket en el sistema de soporte con todo el contexto (clasif + fuentes + score + historial). Notifica al agente humano disponible.

19Skeletons: cómo empezar a codear mañana

Tres bloques que muestran la forma del código real. No copy-paste literal: patrones para que el equipo los adapte.

El grafo completo, esquemático

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

# ── Schemas ──
class IntakeData(BaseModel):
    user_id: str
    channel: str
    raw_message: str
    timestamp: str

class Category(BaseModel):
    name: str          # "TECHNICAL" | "BILLING" | etc
    confidence: float   # 0-1

class Classification(BaseModel):
    categories: list[Category]
    intent: str          # "question" | "request" | "complaint"

class SourceResult(BaseModel):
    source: str          # "kb" | "api_docs" | "crm" | "monitoring"
    data: dict
    relevance: float    # 0-1

# ── Nodos ──
def intake_function(msg: str) -> IntakeData:
    # validar formato, extraer user_id, etc.
    ...
    return IntakeData(...)

classify_agent = Agent(
    name="classify", model="gemini-flash-latest",
    input_schema=IntakeData, output_schema=Classification,
    instruction="Clasificá la consulta y estimá confidence por categoría.",
    mode="single_turn",
)

# Tools externos (envueltos como FunctionTool o MCP)
kb_search         = build_kb_tool()
api_docs_search   = build_api_docs_tool()
crm_lookup        = build_crm_tool()
service_status    = build_monitoring_tool()

def score_confidence(results: dict[str, SourceResult]) -> float:
    # combinacion ponderada, sin LLM
    ...
    return score

def router(score: float):
    if score >= 0.75: return Event(route="RESPOND")
    if score >= 0.40: return Event(route="ASK_CLARIFICATION")
    return Event(route="ESCALATE")

respond_agent = Agent(
    name="respond", model="gemini-flash-latest",
    instruction="Sintetizá los resultados en una respuesta clara y accionable.",
    mode="single_turn",
)

clarify_agent = Agent(
    name="clarify", model="gemini-flash-latest",
    instruction="Generá UNA pregunta de seguimiento para destrabar la consulta.",
    mode="single_turn",
)

def escalate_to_human(payload: dict):
    # crear ticket, notificar, loggear
    ...
    return Event(message="Te derivo con un humano, ticket #XYZ creado.")

# ── El grafo ──
support_graph = Workflow(
    name="iplan_support",
    edges=[
        ("START", intake_function, classify_agent),
        (classify_agent, kb_search,         join_node),
        (classify_agent, api_docs_search,   join_node),
        (classify_agent, crm_lookup,        join_node),
        (classify_agent, service_status,    join_node),
        (join_node, score_confidence, router),
        (router, {
            "RESPOND":            respond_agent,
            "ASK_CLARIFICATION": clarify_agent,
            "ESCALATE":          escalate_to_human,
        }),
    ],
)

El scoring, aislado y testeable

Una de las grandes ventajas del grafo: podés testear score_confidence como una función pura, sin levantar el grafo entero ni gastar tokens.

# Test del FunctionNode de scoring

def test_score_high_when_all_sources_relevant():
    results = {
        "kb":         SourceResult(source="kb", data={}, relevance=0.9),
        "api_docs":   SourceResult(source="api_docs", data={}, relevance=0.85),
        "crm":        SourceResult(source="crm", data={}, relevance=0.7),
        "monitoring": SourceResult(source="monitoring", data={}, relevance=0.95),
    }
    score = score_confidence(results)
    assert score >= 0.75

Una tool real (ejemplo)

from google.adk import FunctionTool
from pydantic import BaseModel

class KbInput(BaseModel):
    query: str
    top_k: int = 5

class KbHit(BaseModel):
    doc_id: str
    title: str
    snippet: str
    score: float

async def kb_search(input: KbInput) -> list[KbHit]:
    # llamada al vector store de iPlan
    hits = await vector_store.search(input.query, limit=input.top_k)
    return [KbHit(...) for h in hits]

kb_search_tool = FunctionTool(
    func=kb_search,
    input_schema=KbInput,
    output_schema=list[KbHit],
)

20Próximos pasos para el equipo

Un roadmap realista para pasar de la teoría al agente funcionando.

Fase 1: Spike (1 semana)

• Setup del repo con ADK 2.0 + Python 3.11.
• Replicar el ejemplo canónico de la doc (sequential de 3 nodos) en local.
• Adaptar el ejemplo al caso más simple: intake + clasificador + respuesta hardcodeada.
• Setup de observability (logging de Events).

Fase 2: Núcleo (2 semanas)

• Implementar el clasificador con datos reales de consultas anonimizadas.
• Conectar la primera fuente de datos (KB de artículos).
• Implementar el score_confidence y el routing.
• Tests unitarios por nodo + test de integración del grafo.

Fase 3: Producción (3+ semanas)

• Conectar las 4 fuentes de datos reales.
• Integración con el canal (chat, email, lo que use iPlan).
• Integración con el sistema de tickets para escalación.
• Evaluación con dataset de conversaciones históricas.
• Deploy en Agent Runtime / Cloud Run.
• Observability en producción (traces, métricas de score de confianza, % de escalación).

Decisiones que tenemos que tomar juntos

1. Modelo: ¿Gemini Flash para clasificación, Gemini Pro para respuesta? ¿O un solo modelo? Costo vs calidad.
2. Knowledge base: ¿articles existentes, Confluence, Notion, otro? ¿Vector store propio o Vertex AI Search?
3. Umbrales de score: 0.75 y 0.4 son puntos de partida. Hay que calibrar con datos reales.
4. Loop de retroalimentación: ¿cómo capturamos feedback de los humanos sobre respuestas del agente? Necesario para mejorar.
5. Privacidad: ¿qué datos del cliente pueden ver los nodos? ¿PII enmascarada?

21Glosario rápido

Términos que vas a leer en la doc oficial y en este tutorial.

Nodo

Unidad ejecutable del grafo. Puede ser un Agent, Function, Tool, sub-workflow, o human input.

Edge

Conexión entre dos nodos. Define el flujo.

Workflow

El grafo completo. Tiene name y edges.

Event

El "paquete" que viaja entre nodos. Tiene output, message, state, route.

Route key

String que un nodo emite en Event.route y que el engine matchea contra el dict de edges para decidir el siguiente nodo.

JoinNode

Nodo barrier. Espera a que todos sus predecesores emitan output antes de avanzar.

Single-turn

Modo de un LlmAgent donde ejecuta una sola llamada al LLM y termina. Requerido para grafos.

Output schema

Pydantic class que define la forma del output de un nodo. El framework fuerza al LLM a cumplirlo.

State scope

Prefijo (app:, user:, temp:, o ninguno) que define cuánto vive un dato en state.

Artifact

Storage persistente de blobs asociado a una sesión. Para datos grandes (archivos, embeddings).

22Recursos para profundizar

Lo que está bueno leer después de este tutorial.

Documentación oficial

• Graph-based agent workflows — la página raíz.
• Build graph routes for agent workflows — patrones de routing avanzados.
• Data handling for agent workflows — cómo pasa la data entre nodos.
• Dynamic workflows — para cuando graph no alcanza.
• Template agent workflows — la alternativa prebuilt (legacy-style).

En el repo de ADK (GitHub)

• examples/workflow/complex/ — ejemplo completo con fan-out, join, loops.
• examples/workflow/routing/ — variantes de routing (string, int, LLM-driven).
• examples/workflow/nested/ — workflows anidados.

Convención de nombres para el equipo

Para mantener consistencia en el código, sugerimos:

• Nodos LLM: <verbo>_agent — classify_agent, respond_agent.
• Nodos Function: <verbo>_function o <sustantivo>_function — intake_function, score_confidence.
• Nodos Tool: <recurso>_tool — kb_search_tool, crm_lookup_tool.
• Schemas: sustantivos en PascalCase — IntakeData, Classification, SourceResult.