Flujos de trabajo condicionales de LangGraph: ramificación sin desorden en las instrucciones
Codifique el enrutamiento if-else de LangGraph como aristas condicionales nombradas con contratos de estado claros y fallos recuperables, en lugar de incluir las ramificaciones dentro de los prompts.
Fuentes: Conditional Workflows in LangGraph: The If-Else of Agentic AI
Esta guía reconstruye un camino operativo para: Flujos de trabajo condicionales en LangGraph: El if-else de la IA agente. Se centra en contratos, verificaciones y código que se puede insertar en un repositorio sin tener que adivinar la intención. Para obtener una visión general, 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. 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.
¿Qué es exactamente un flujo de trabajo condicional?
¿Para qué sirve siquiera un flujo de trabajo condicional? Defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde 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 etapa falla, el error debe indicar una única responsabilidad y no un proceso complicado. Limite estrictamente los esquemas de las herramientas. Los argumentos de texto libre excesivos facilitan inyecciones maliciosas y hacen que las auditorías sean costosas.
Flujo de trabajo #1: Resolver una ecuación cuadrática
Para el Flujo de Trabajo n.º 1: Resolver una ecuación cuadrática, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta fase 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. Limite estrictamente los esquemas de las herramientas. Los argumentos de texto libre amplios invitan a inyecciones y hacen que las auditorías sean costosas.
La estructura del flujo de trabajo
En cuanto a la estructura del flujo de trabajo, 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. Registre los tiempos y costos junto con los resultados funcionales. Tener visibilidad desde el principio evita facturas inesperadas cuando el flujo pasa de entornos de demostración a entornos compartidos. Delimite estrictamente los esquemas de las herramientas; los argumentos de texto libre excesivos facilitan inyecciones maliciosas y encarecen las auditorías. En cuanto a la estructura del flujo de trabajo, 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. Documente tanto la ruta óptima como la ruta de recuperación. Las intentonas repetidas, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras aplicadas posteriormente.
Construcción del estado
Para construir el estado, 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. Realice puntos de control después de llamadas costosas al modelo para que una repetición no genere nuevos costos por el mismo trabajo.
class QuadState(TypedDict):
a: float
b: float
c: float
equation: str
discriminant: float
result: str
Los nodos
Para los nodos, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Considere esta fase 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. Realice un punto de control después de llamadas costosas al modelo para que una repetición no genere nuevos costos por el mismo trabajo.
def show_equation(state: QuadState):
equation = f"{state['a']}x² + {state['b']}x + {state['c']}"
return {"equation": equation}
def calculate_discriminant(state: QuadState):
discriminant = (state['b'] ** 2) - (4 * state['a'] * state['c'])
return {"discriminant": discriminant}
def real_roots(state: QuadState):
r1 = (-state['b'] + state['discriminant'] ** 0.5) / (2 * state['a'])
r2 = (-state['b'] - state['discriminant'] ** 0.5) / (2 * state['a'])
result = f"The roots are {r1} and {r2}"
return {"result": result}
def repeated_root(state: QuadState):
root = -state['b'] / (2 * state['a'])
result = f"Only repeating root is {root}"
return {"result": result}
def no_real_roots(state: QuadState):
return {"result": "No real roots"}
La función de enrutamiento: esta es la parte clave
Para la función de enrutamiento — esta es la parte clave: 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. Registre los tiempos y costos junto con los resultados funcionales. La visibilidad temprana evita facturas inesperadas cuando el camino pasa de un entorno de demostración a uno compartido. Coloque un punto de control después de llamadas a modelos costosas para que los intentos de repetición no generen nuevos costos por el mismo trabajo. Para la función de enrutamiento — esta es la parte clave: 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. Documente tanto el camino óptimo como el de recuperación. Los intentos de repetición, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.
def check_condition(state: QuadState):
if state['discriminant'] > 0:
return "real_roots"
elif state['discriminant'] == 0:
return "repeated_root"
else:
return "no_real_roots"
Vinculándolo todo
Para conectar todo, 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 desde 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 planificación de la ejecución de las herramientas. El planificador propone; el ejecutor realiza los cambios; el verificador comprueba los resultados en relación con el objetivo.
graph = StateGraph(QuadState)
graph.add_node("show_equation", show_equation)
graph.add_node("calculate_discriminant", calculate_discriminant)
graph.add_node("real_roots", real_roots)
graph.add_node("repeated_root", repeated_root)
graph.add_node("no_real_roots", no_real_roots)
graph.add_edge(START, "show_equation")
graph.add_edge("show_equation", "calculate_discriminant")
# This is the magic line
graph.add_conditional_edges("calculate_discriminant", check_condition)
graph.add_edge("real_roots", END)
graph.add_edge("repeated_root", END)
graph.add_edge("no_real_roots", END)
workflow = graph.compile()
Flujo de trabajo n.º 2: Gestor de revisiones de clientes (basado en LLM)
Para el Flujo de Trabajo n.º 2: Gestor de Revisiones del Cliente (basado en LLM), defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Separe la planificación de la ejecución de las herramientas. El planificador propone; el ejecutor realiza los cambios; el verificador comprueba los resultados en relación con el objetivo.
El escenario
Para el escenario, 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 desde un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos y costos junto con los resultados funcionales. La visibilidad temprana evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Separe la planificación de la ejecución de las herramientas. El planificador propone; el ejecutor realiza los cambios; el verificador comprueba los resultados en relación con el objetivo. Para el escenario, 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 desde un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Los intentos repetidos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son ajustes realizados posteriormente.
Estructuras de salida estructuradas
Para los esquemas de salida estructurada, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa 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 etapa falla, el error debe indicar una única responsabilidad y no un proceso complicado. Delimite estrictamente los esquemas de las herramientas. Los argumentos de texto libre amplios invitan a inyecciones y hacen que las auditorías sean costosas.
from pydantic import BaseModel
from typing import Literal
class SentimentSchema(BaseModel):
sentiment: Literal["positive", "negative"]
class DiagnosisSchema(BaseModel):
issue_type: Literal["UI", "performance", "bug", "support", "other"]
tone: Literal["frustrated", "angry", "disappointed", "neutral"]
urgency: Literal["low", "medium", "high"]
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
structured_model_sentiment = model.with_structured_output(SentimentSchema)
structured_model_diagnosis = model.with_structured_output(DiagnosisSchema)
El estado
Para el Estado, se deben definir las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Considere esta fase 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. Delimite estrictamente los esquemas de las herramientas. Los argumentos de texto libre amplios invitan a inyecciones y hacen que las auditorías sean costosas.
class ReviewState(TypedDict):
review: str
sentiment: Literal["positive", "negative"]
diagnosis: dict
response: str
Los nodos
En The Nodes, 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. Registre los tiempos y costos junto con los resultados funcionales. La visibilidad temprana evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Delimite estrictamente los esquemas de las herramientas. Los argumentos de texto libre excesivos facilitan inyecciones y encarecen las auditorías. En The Nodes, 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. 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 son mejoras posteriores.
def find_sentiment(state: ReviewState):
prompt = f"For the following review, find out the sentiment:\n{state['review']}"
result = structured_model_sentiment.invoke(prompt)
return {"sentiment": result.sentiment}
def positive_response(state: ReviewState):
prompt = f"""Write a warm thank you message in response to this review:
{state['review']}
Also kindly ask the user to leave feedback on our website."""
result = model.invoke(prompt)
return {"response": result.content}
def run_diagnosis(state: ReviewState):
prompt = f"""Diagnose this negative review:
{state['review']}
Return issue type, tone, and urgency."""
result = structured_model_diagnosis.invoke(prompt)
return {"diagnosis": result.model_dump()}
def negative_response(state: ReviewState):
diagnosis = state['diagnosis']
prompt = f"""You are a support assistant. The user had a {diagnosis['issue_type']} issue,
sounded {diagnosis['tone']}, and marked urgency as {diagnosis['urgency']}.
Write an empathetic, helpful resolution message."""
result = model.invoke(prompt)
return {"response": result.content}
def check_sentiment(state: ReviewState):
if state['sentiment'] == "positive":
return "positive_response"
else:
return "run_diagnosis"
El grafo
Para The Graph, defina las entradas, el propietario de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde 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 etapa falla, el error debe indicar una única responsabilidad y no un proceso complicado. Realice puntos de control después de llamadas costosas al modelo para que una repetición no genere nuevos costos por el mismo trabajo.
graph = StateGraph(ReviewState)
graph.add_node("find_sentiment", find_sentiment)
graph.add_node("positive_response", positive_response)
graph.add_node("run_diagnosis", run_diagnosis)
graph.add_node("negative_response", negative_response)
graph.add_edge(START, "find_sentiment")
graph.add_conditional_edges("find_sentiment", check_sentiment)
graph.add_edge("positive_response", END)
graph.add_edge("run_diagnosis", "negative_response")
graph.add_edge("negative_response", END)
workflow = graph.compile()
Pruebas
Para probarlo, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Considere esta fase 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. Realice un punto de control después de llamadas costosas al modelo para que una repetición no genere nuevos costos por el mismo trabajo.
El modelo mental fundamental
Para el Modelo Mental Central, 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. Registre los tiempos y costos junto con los resultados funcionales. Tener visibilidad desde el principio evita facturas inesperadas cuando la tarea pasa de un entorno de demostración a uno compartido. Establezca un punto de control después de llamadas al modelo costosas para que los intentos de repetición no generen nuevos costos por el mismo trabajo. Para el Modelo Mental Central, 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. Documente tanto la ruta óptima como la ruta de recuperación. Los intentos de repetición, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.
¿Qué sigue?
Para lo que viene a continuación, 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 planificación de la ejecución de las herramientas. El planificador propone; el ejecutor modifica; el verificador comprueba los resultados en relación con el objetivo.
Aprendizaje estructurado más allá de los blogs
Separe la planificación de la ejecución de las herramientas. El planificador propone; el ejecutor modifica; el verificador comprueba los resultados en relación con el objetivo.
Cursos
Separe la planificación de la ejecución de las herramientas. El planificador propone; el ejecutor modifica; el verificador comprueba los resultados en relación con el objetivo.
Lista de verificación operativa
Punto de verificación después de las llamadas a modelos costosos para que un intento de repetición no vuelva a facturar por el mismo trabajo.
Mantenga el estado del grafo simple y tipado. Los bloques anidados ocultan qué nodo escribió qué campo y dificultan la reanudación tras interrupciones.
Escriba un manual breve: cómo rotar claves, cómo vaciar la cola y cómo revertir el último cambio.
Lecturas relacionadas
- Por qué las equipos migran de LangChain chains a LangGraph workflows — Los agentes en producción necesitan un estado duradero, ramas y HITL. Mantenga las herramientas de LangChain dentro de los nodos; transfiera el flujo de control a un grafo explícito cuando lo exija la operabilidad.