Ramificación Condicional
Aprende cómo enrutar la ejecución dinámicamente usando aristas condicionales, funciones de enrutamiento y add_conditional_edges en LangGraph.
Ramificación Condicional
No todos los grafos son lineales. La ramificación condicional permite que tu grafo tome decisiones en tiempo de ejecución, enrutando la ejecución a diferentes caminos basándose en el estado actual.
¿Qué es la Ramificación Condicional?
Ramificación condicional significa que el siguiente nodo a ejecutar depende del estado actual, no de una topología fija.
add_conditional_edges
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
input_text: str
category: str
def classifier(state: State) -> dict:
text = state["input_text"].lower()
if "bill" in text or "payment" in text:
return {"category": "billing"}
elif "bug" in text or "error" in text:
return {"category": "technical"}
else:
return {"category": "general"}
def route_by_category(state: State) -> str:
return state["category"]
builder = StateGraph(State)
builder.add_node("classifier", classifier)
builder.add_node("tech_support", lambda s: {"tech_support": True})
builder.add_node("billing_support", lambda s: {"billing_support": True})
builder.add_node("general_support", lambda s: {"general_support": True})
builder.add_edge(START, "classifier")
builder.add_conditional_edges("classifier", route_by_category, {
"technical": "tech_support",
"billing": "billing_support",
"general": "general_support"
})
builder.add_edge("tech_support", END)
builder.add_edge("billing_support", END)
builder.add_edge("general_support", END)
app = builder.compile()Ruteo con LLM
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini")
def classify_route(state: RouteState) -> dict:
prompt = ChatPromptTemplate.from_messages([
("system", "Clasifica la consulta en: 'technical', 'billing' o 'general'."),
("human", "{query}")
])
chain = prompt | llm | StrOutputParser()
route = chain.invoke({"query": state["query"]}).strip().lower()
return {"route": route}Bucles Condicionales
class LoopState(TypedDict):
input: str
result: str
attempts: int
is_valid: bool
def process(state: LoopState) -> dict:
result = attempt_processing(state["input"])
is_valid = validate_result(result)
return {"result": result, "is_valid": is_valid, "attempts": state["attempts"] + 1}
def loop_router(state: LoopState) -> str:
if state["is_valid"]:
return "valid"
if state["attempts"] >= 3:
return "max_retries"
return "retry"
builder.add_conditional_edges("process", loop_router, {
"valid": "format_output",
"retry": "process",
"max_retries": "error_handler"
})Siempre incluye un número máximo de reintentos en bucles. Sin él, un fallo persistente causa un bucle infinito.
Preguntas de Práctica
¿Qué método se usa para la ramificación condicional en LangGraph?
¿Qué recibe y devuelve una función de enrutamiento?
¿Qué sucede si un enrutador devuelve un valor no presente en el mapeo?
¿Cuál es el uso más común de aristas condicionales en agentes?
¿Cómo crear una arista condicional binaria simple (sí/no)?
¿Qué deberías incluir siempre en un bucle con aristas condicionales?
¿Se puede usar un LLM como enrutador en LangGraph?
¿Qué constante puede usarse como destino en el mapeo de arista condicional?
Puntos Clave
add_conditional_edges(origen, enrutador, mapeo)permite enrutamiento dinámico- Las funciones de enrutamiento reciben el estado y devuelven una clave de string
- El mapeo traduce la salida del enrutador en nombres de nodos destino
- Los bucles se crean mapeando una ruta de vuelta a un nodo ejecutado anteriormente
- El enrutamiento con LLM usa un LLM para clasificar y escribir la ruta en el estado
- Cubre siempre todas las salidas posibles del enrutador en el mapeo
- Incluye condiciones de terminación en bucles para evitar ejecución infinita
- END puede ser un destino en mapeos de arista condicional