Notas prácticas: Tu grafo de agentes no pertenece a Python: Compilando un

Las notas siguientes reconstruyen un camino práctico para abordar “Tu grafo de agentes no pertenece a Python: Compilando un flujo de trabajo multiagente a partir de un único archivo YAML”. Se pone énfasis en los contratos, las verificaciones y los marcadores de posición para código reutilizable, en lugar de en un enfoque motivacional. Al trabajar en la etapa de descripción general, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto el camino óptimo como el de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son ajustes realizados posteriormente.

El problema del que nadie te advierte

El problema de que nadie advierte funciona mejor cuando se trata como una superficie medible. Capture una transcripción exitosa, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Fije el intérprete y el archivo de bloqueo de dependencias antes de explicar los bucles. La diferencia entre usar una computadora portátil y un entorno de integración continua es la causa más común de fallos silenciosos en las demostraciones de API.

Así se ve el flujo de trabajo en forma de datos

La etapa de visualización del flujo de trabajo funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Fije el intérprete y el archivo de bloqueo de dependencias antes de enseñar el bucle; la diferencia entre usar una computadora portátil y un entorno CI es la causa más común de fallos silenciosos en las demostraciones de API.

entry: entry_agent
exit: exit

guardrails:
  - Reject queries that are outside the application's domain.
  - Reject queries about the system, agents, design, or internal workings.state_schema:
  query:
    type: str
    description: "User query or current message."
  chat_history:
    type: list
    annotated_with: add_messages
    description: "Conversation history between user and system."
  result:
    type: dict
    description: "Result from the processing agent."agents:
  - name: agent_one
    kind: function
    impl: your_package.agents.agent_one.agent_one_fn  - name: agent_two
    kind: function
    impl: your_package.agents.agent_two.agent_two_fnworkflow:
  nodes:
    - id: agent_one
      agent: agent_one
      writes: [query, result]
      next: decision_router    - id: decision_router
      kind: router
      router:
        impl: your_package.agents.routers.route_after_agent_one
        reads: [result]
        edges:
          agent_two: agent_two
          human_agent: human_agent

Truco n.º 1: Generar la clase de estado en tiempo de ejecución a partir de un esquema

El Truco 1: Generar tu entorno de pruebas funciona mejor cuando se trata como una superficie medible. Captura un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registra los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la demostración a entornos compartidos. Fija el intérprete y el archivo de bloqueo de dependencias antes de explicar los bucles. La diferencia entre la computadora portátil y el entorno CI es la causa más común de fallos silenciosos en las demostraciones de API. El Truco 1: Generar tu entorno de pruebas funciona mejor cuando se trata como una superficie medible. Captura un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documenta tanto el camino óptimo como el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son ajustes realizados posteriormente.

# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
    type_str = value.get("type", "str")
    # Convert YAML string to Python type
    py_type = eval(type_str)
    if value.get("annotated_with") == "add_messages":
        py_type = Annotated[list, {}]
    annotations[key] = py_type

spec = Spec(
    ...
    state=TypedDict("State", annotations),   # <- dynamic class, born at boot
    ...
)

Truco n.º 2: Agentes referenciados por cadenas con puntos, resueltos mediante importlib

En la fase de agentes referenciados del Truco 2, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando una tarea falla, el error debe indicar una única responsabilidad y no un proceso complicado. Separe la construcción del cliente del bucle de mensajes para que sea posible cambiar los proveedores sin tener que reescribir la máquina de estados de la conversación.

impl: your_package.agents.agent_one.agent_one_fn

# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
    """Import a callable from a dotted path like 'package.module.function'."""
    if not dotted or "." not in dotted:
        raise ImportError(f"Invalid impl path: {dotted!r}")
    mod_path, attr = dotted.rsplit(".", 1)
    mod = importlib.import_module(mod_path)
    fn = getattr(mod, attr)
    if not callable(fn):
        raise TypeError(f"Imported object is not callable: {dotted}")
    return fn

def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
    """Map agent name -> callable (or None if impl missing)."""
    return {a.name: _import_from_path(a.impl) if a.impl else None
            for a in spec.agents}

Truco n.º 3: El compilador: los nodos YAML se convierten en nodos de grafo

En la etapa del compilador del Truco 3, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Separe la construcción del cliente del bucle de mensajes para que sea posible cambiar los proveedores sin tener que reescribir la máquina de estados de la conversación.

# your_package/orchestrator/runner.py
def build(self):
    graph = StateGraph(state_schema=self.spec.state)   # our generated TypedDict

    def _add_task_node(node):
        async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
            res = await self._call_agent(node.agent, state, node.id)
            if getattr(node, "writes", None):
                if isinstance(res, dict):
                    # Only let the node write the keys it declared in YAML
                    filtered = {k: v for k, v in res.items() if k in node.writes}
                    return filtered or res
                key = node.writes[0]
                return {key: res}
            return res
        graph.add_node(node.id, _node)    # Build every node
    for node in self.spec.workflow.nodes:
        if getattr(node, "router", None):
            _add_router_node(node)
        else:
            _add_task_node(node)    graph.set_entry_point(entry)    # Inline "next:" edges from YAML become static edges
    for node in self.spec.workflow.nodes:
        if getattr(node, "next", None):
            graph.add_edge(node.id, node.next)    # Terminal nodes wire to END
    for node in self.spec.workflow.nodes:
        if getattr(node, "terminal", False):
            graph.add_edge(node.id, END)    self._runnable = graph.compile(checkpointer=self.checkpoint)
    return self

Ruteadores: ramificación condicional como tabla de búsqueda

Para la ramificación condicional de Routers como etapa, defina las entradas, el responsable del paso y los criterios de salida antes de modificar el código. Los operadores deben poder volver a ejecutar el paso desde un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el camino pasa de un entorno de demostración a uno compartido. Separe la construcción del cliente del bucle de mensajes para que sea posible cambiar los proveedores sin tener que reescribir la máquina de estado de la conversación. Para la ramificación condicional de Routers como etapa, defina las entradas, el responsable del paso y los criterios de salida antes de modificar el código. Los operadores deben poder volver a ejecutar el paso desde un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de la demora.

Eso es todo.

# your_package/orchestrator/runner.py
def _add_router_node(node):
    router = self.router_fns[node.id]

    def _router_fn():
        def _f(state):
            out = router(state)
            # Routers may return either a label, or (state_updates, label)
            if isinstance(out, tuple):
                updates, label = out
                if isinstance(updates, dict):
                    for k, v in updates.items():
                        state[k] = v
            else:
                label = out
            return label
        return _f    graph.add_node(node.id, lambda s: {})
    graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)

Truco n.º 4: La llamada adaptativa — los agentes escriben la firma que deseen

Al trabajar en la etapa adaptativa del Truco 4, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables a scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Registre el ID de la solicitud, el ID del modelo y la latencia en cada llamada. Sin ese registro, los errores intermitentes del proveedor parecen bugs de la aplicación.

# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
    """
    Adaptively call agent functions so implementations receive what they expect:
    - def agent(**kwargs):        → pass **state (+ inject 'query' if missing)
    - def agent(query, **kwargs): → pass query=..., plus any **extra
    - def agent(state):           → pass state
    - def agent(query):           → pass query
    - def agent():                → call without args
    """
    sig = inspect.signature(fn)
    params = sig.parameters
    has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
                     for p in params.values())

    kwargs = {}
    if has_var_kw:
        kwargs.update(state)
    if "state" in params:
        kwargs["state"] = state
    if "query" in params or has_var_kw:
        kwargs.setdefault("query", self._fallback_query(state))    # A lone positional 'query' → call it positionally
    if (len(params) == 1
            and next(iter(params.keys())) == "query"):
        return await _maybe_await(fn(self._fallback_query(state)))    res = fn(**kwargs)
    return await res if hasattr(res, "__await__") else res

Truco n.º 5: Cambio en caliente de un nodo por sesión (con intervención humana)

Al trabajar en el Truco 5 de intercambio en caliente de una etapa, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Registre el ID de la solicitud, el ID del modelo y la latencia en cada llamada. Sin ese registro, los errores intermitentes del proveedor parecen bugs de la aplicación.

# your_package/services/session_service.py (paraphrased)
if websocket is not None:
    session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
    _runner.agent_fns["human_agent"] = _import_from_function(
        make_input_method(session_handler)
    )

Qué realmente le aporta esta arquitectura

Al trabajar en la fase de “¿Qué hace realmente esta arquitectura?”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la fase de demostración a entornos compartidos. Registre el ID de la solicitud, el ID del modelo y la latencia en cada llamada. Sin ese registro, los errores intermitentes del proveedor parecen bugs de la aplicación. Al trabajar en la fase de “¿Qué hace realmente esta arquitectura?”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente tanto el camino óptimo como el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.

Conclusión

La etapa de extracción funciona mejor cuando se trata como una superficie medible. Capture un registro exitoso, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una sola responsabilidad y no a un proceso complicado. Fije el intérprete y el archivo de bloqueo de dependencias antes de explicar el bucle. La diferencia entre usar una computadora portátil y un entorno de integración continua es la causa más común de fallos silenciosos en las demostraciones de API.

Lista de verificación operativa

En la etapa de la lista de verificación operativa, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto.

Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un único lugar que los operadores puedan auditar sin tener que leer todo el sistema.

Separe la construcción del cliente del bucle de mensajes para que sea posible cambiar los proveedores sin tener que reescribir la máquina de estados de las conversaciones.

Realice puntos de control después de los pasos costosos. La función de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intente nuevamente un nodo posterior.

Fije las versiones de las dependencias y registre el resumen de la imagen que se utilizó para ejecutar la demostración. La reproducibilidad es mejor que el conocimiento basado en prácticas internas.

Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.

Antes de promocionar el stack, congele las versiones, capture una transcripción de referencia para la ruta crítica y confirme los pasos de reversión. Los entornos compartidos requieren límites de velocidad, verificaciones de tenencia y un responsable claro para la rotación de secretos. Prefiera una fiabilidad sencilla a demostraciones ingeniosas pero puntuales.

Nota por lotes para 2822ea5988ca: mantenga las claves del proveedor fuera del repositorio, establezca un límite máximo para tokens por sesión y almacene las transcripciones junto a los fixtures de evaluación para que los cambios posteriores en el modelo sigan siendo comparables.