Ramificación Condicional

Aprende cómo enrutar la ejecución dinámicamente usando aristas condicionales, funciones de enrutamiento y add_conditional_edges en LangGraph.

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.

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

add_conditional_edges

python

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

python

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

python

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"
})

⚠️Advertencia

Siempre incluye un número máximo de reintentos en bucles. Sin él, un fallo persistente causa un bucle infinito.