Caso: orquestar un flujo con pydantic-graph

En este caso práctico construimos un flujo de trabajo como un grafo de estados usando pydantic-graph. Es una librería del ecosistema de Pydantic para definir procesos paso a paso —donde cada paso decide cuál viene después— con la seguridad de tipos de Python. Es la base sobre la que se arman los agentes modernos.

La idea: un proceso como grafo

Muchos problemas son en realidad una máquina de estados: recibir → validar → procesar → terminar, con bifurcaciones por el camino. pydantic-graph te deja escribir cada paso como un nodo; cada nodo, al ejecutarse, devuelve el siguiente nodo o una señal de fin (End). El grafo se encarga de encadenarlos.

Paso 1: instalar

pip install pydantic-graph

Paso 2: definir los nodos

Cada nodo es una clase que hereda de BaseNode y define un método run. Lo que ese método retorna determina el flujo: otro nodo (seguir) o End (terminar).

from __future__ import annotations
from dataclasses import dataclass
from pydantic_graph import BaseNode, End, Graph, GraphRunContext


@dataclass
class Recibir(BaseNode):
    texto: str

    async def run(self, ctx: GraphRunContext) -> Validar:
        return Validar(self.texto.strip())      # pasa al siguiente nodo


@dataclass
class Validar(BaseNode[None, None, str]):
    texto: str

    async def run(self, ctx: GraphRunContext) -> Procesar | End[str]:
        if not self.texto:
            return End("rechazado: texto vacío")  # termina el flujo
        return Procesar(self.texto)


@dataclass
class Procesar(BaseNode[None, None, str]):
    texto: str

    async def run(self, ctx: GraphRunContext) -> End[str]:
        return End(f"procesado: {self.texto.upper()}")

Fíjate en Validar: bifurca. Si el texto está vacío termina con un rechazo; si no, continúa a Procesar.

Paso 3: armar y ejecutar el grafo

Se declara el grafo con la lista de nodos y se corre desde un nodo inicial:

flujo = Graph(nodes=[Recibir, Validar, Procesar])

print(flujo.run_sync(Recibir("  hola mundo  ")).output)
# procesado: HOLA MUNDO

print(flujo.run_sync(Recibir("   ")).output)
# rechazado: texto vacío

El grafo siguió el camino que cada nodo fue eligiendo, hasta llegar a un End.

¿Por qué un grafo y no solo funciones?

• Claridad: cada paso y cada transición quedan explícitos y tipados.
• Bifurcaciones y ciclos: un nodo puede volver a otro anterior (reintentos, bucles) con naturalidad.
• Estado compartido: mediante el GraphRunContext los nodos comparten un estado común.
• Diagramas: el grafo puede exportar su estructura como diagrama mermaid para documentarlo.

Esta forma de pensar —pasos que deciden el siguiente paso— es exactamente la que usan los agentes de IA para encadenar razonamiento, llamadas a herramientas y decisiones.

Para llevar

pydantic-graph modela un proceso como un grafo de nodos tipados: cada nodo hace su trabajo y devuelve el siguiente, o End para terminar. Ganas claridad, bifurcaciones, ciclos y estado compartido sin enredos. Es una forma robusta de orquestar flujos complejos y el cimiento de los agentes modernos.