LangGraph Orchestrator: StateGraph and Routing Logic

The orchestrator is the central nervous system of Mesa de Ayuda IA, implemented in backend/app/core/orchestrator.py as a LangGraph StateGraph. LangGraph was chosen over LangChain’s classic AgentExecutor for a concrete set of architectural reasons: the execution path is modeled as an explicit directed graph with named nodes and edges, making the flow fully auditable and testable; routing decisions are deterministic (the classifier runs at temperature 0 and the conditional edge function is pure Python); parallel fan-out for mixed-intent queries is a native feature, not a workaround; and the graph topology itself prevents infinite loops since there are no back-edges — every path leads to consolidate → END. The result is a system that is predictable enough to confidently deploy in a corporate sales context where wrong answers have real commercial consequences.

GraphState

The entire execution context of a single user request is stored in a TypedDict called GraphState. Every node in the graph reads from and writes to this shared state object — nodes never call each other directly.

class GraphState(TypedDict):
    question: str
    category: str
    history: List[Dict[str, str]]        # Historial de mensajes previos
    image_data: Optional[str]            # Base64 de la imagen (si existe)
    confirmation: Optional[bool]         # Confirmación del usuario para acción
    agent_results: Annotated[List[AgentResult], operator.add]
    final_answer: str
    sources: List[Dict[str, Any]]
    agents_invoked: Annotated[List[str], operator.add]
    warnings: List[str]
    start_time: float
    tokens_input: Annotated[int, operator.add]
    tokens_output: Annotated[int, operator.add]

Field	Type	Set By	Purpose
`question`	`str`	API request	The user’s raw question text
`category`	`str`	`classify` node	One of the 6 intent categories
`history`	`List[Dict]`	API request	Previous conversation turns for context
`image_data`	`Optional[str]`	API request	Base64-encoded image for multimodal queries
`confirmation`	`Optional[bool]`	API request	Explicit frontend confirmation flag for CRM registration
`agent_results`	`Annotated[List, operator.add]`	Agent nodes	Accumulated `AgentResult` objects from all invoked agents
`final_answer`	`str`	`consolidate` node	The synthesized answer returned to the user
`sources`	`List[Dict]`	`consolidate` node	Merged source list from all agents
`agents_invoked`	`Annotated[List, operator.add]`	Agent nodes	Names of every agent that participated
`warnings`	`List[str]`	`classify` / `consolidate`	Non-fatal advisory messages
`start_time`	`float`	`classify` node	`time.time()` at classification start, used for latency calculation
`tokens_input`	`Annotated[int, operator.add]`	Agent nodes + `consolidate`	Accumulated input token count across all Gemini calls
`tokens_output`	`Annotated[int, operator.add]`	Agent nodes + `consolidate`	Accumulated output token count across all Gemini calls

The `Annotated[List, operator.add]` pattern

Four fields — agent_results, agents_invoked, tokens_input, and tokens_output — use Annotated[..., operator.add] as their type. This is LangGraph’s reducer mechanism: when multiple parallel branches write to the same field, LangGraph uses the reducer function (operator.add) to merge the writes rather than letting one overwrite the other. For lists this means concatenation; for integers it means addition. Without this annotation, a mixta query with three parallel agent nodes would produce a race condition where the last-finishing agent’s result would silently discard the other two.

Graph Nodes

The graph contains seven named nodes, each corresponding to a Python function:

classify

Entry point. Calls classify_intent(), detects image data, checks the confirmation flag, and inspects conversation history to detect in-progress registration flows. Sets state["category"] and state["start_time"].

catalogo

Calls agent_registry.get_agent("agente_catalogo").process_query(state["question"]) and writes the AgentResult plus token counts into state.

politicas

Calls agent_registry.get_agent("agente_politicas").process_query(state["question"]) and writes the AgentResult into state.

proceso_ventas

Calls agent_registry.get_agent("agente_proceso_ventas").process_query(state["question"]) and writes the AgentResult into state.

multimodal

Retrieves agente_multimodal from registry (asserts it is a MultimodalAgent instance) and calls process_query(question, image_data=state.get("image_data")).

accion

Retrieves agente_accion from registry (asserts it is an AccionAgent instance) and calls process_query(question, confirmation=..., history=...), passing both the confirmation flag and the full conversation history.

consolidate — the final convergence node:

If exactly one AgentResult is present, its answer field becomes final_answer directly. No additional LLM call is made.
If two or more AgentResult objects are present (possible for mixta), the node loads prompts/orchestrator_prompt.md, builds a consolidation prompt with all partial answers labeled by agent name, and makes one final Gemini call to synthesize a single coherent response. If Gemini fails, a safe fallback concatenates the individual answers separated by --- dividers.

# Consolidation prompt structure (simplified)
context = "\n\n".join(
    [f"### [{r.agent_name}]:\n{r.answer}" for r in results]
)

prompt = f"""{orchestrator_prompt}

---

### Respuestas de los agentes participantes:
{context}

### Pregunta original del usuario:
{state['question']}

Genera una respuesta consolidada, única y bien estructurada."""

Conditional Routing

The route_agents() function is registered as the decision function for the conditional edge that follows the classify node. It reads state["category"] and returns a list of node names to execute next:

def route_agents(state: GraphState):
    """
    Decide qué agente(s) invocar según la categoría clasificada.
    Retorna una lista de nombres de nodo.
    """
    cat = state["category"]

    if cat == "catalogo_precios":
        return ["catalogo"]
    elif cat == "politicas_comerciales":
        return ["politicas"]
    elif cat == "proceso_ventas":
        return ["proceso_ventas"]
    elif cat == "multimodal":
        return ["multimodal"]
    elif cat == "accion_registro":
        return ["accion"]
    else:
        # mixta: invocar los 3 agentes RAG principales en paralelo
        return ["catalogo", "politicas", "proceso_ventas"]

Returning a list with a single element routes execution to exactly one agent node. Returning ["catalogo", "politicas", "proceso_ventas"] for mixta triggers LangGraph’s parallel fan-out — all three nodes start executing concurrently, writing their results into the shared GraphState via operator.add accumulation.

Graph Construction

The complete graph topology is declared using LangGraph’s StateGraph builder API:

workflow = StateGraph(GraphState)

# Register all nodes
workflow.add_node("classify", node_classify)
workflow.add_node("catalogo", node_catalogo)
workflow.add_node("politicas", node_politicas)
workflow.add_node("proceso_ventas", node_proceso_ventas)
workflow.add_node("multimodal", node_multimodal)
workflow.add_node("accion", node_accion)
workflow.add_node("consolidate", node_consolidate)

# Entry point
workflow.set_entry_point("classify")

# Conditional routing from classify → agent node(s)
workflow.add_conditional_edges(
    "classify",
    route_agents,
    {
        "catalogo": "catalogo",
        "politicas": "politicas",
        "proceso_ventas": "proceso_ventas",
        "multimodal": "multimodal",
        "accion": "accion",
    },
)

# All agent nodes converge to consolidate
workflow.add_edge("catalogo", "consolidate")
workflow.add_edge("politicas", "consolidate")
workflow.add_edge("proceso_ventas", "consolidate")
workflow.add_edge("multimodal", "consolidate")
workflow.add_edge("accion", "consolidate")
workflow.add_edge("consolidate", END)

# Compile to a runnable
orchestrator_app = workflow.compile()

The add_conditional_edges() call takes three arguments: the source node ("classify"), the decision function (route_agents), and a mapping dictionary that translates the string values returned by the decision function into actual node names. Every agent node is then wired to consolidate with add_edge(), and consolidate leads unconditionally to END. This topology guarantees that:

Every execution path passes through classify exactly once.
Every execution path terminates at consolidate → END — there are no cycles.
Parallel branches from mixta all converge at consolidate before the final answer is produced.

Confirmation Flow

The accion_registro category is special because the CRM registration agent uses a multi-step confirmation flow. The classify node detects confirmation in three distinct ways before the LLM classifier even runs:

def node_classify(state: GraphState):
    q_lower = state.get("question", "").strip().lower()
    is_confirmation_text = q_lower in [
        "sí", "si", "sí, registrar", "si, registrar",
        "confirmar", "confirmo", "sí, registrar."
    ]

    if state.get("confirmation") or is_confirmation_text:
        return {
            "category": "accion_registro",
            "start_time": time.time(),
            "warnings": [],
        }

    # Check history for in-progress registration flow
    history = state.get("history", [])
    if history:
        last_assistant_msg = None
        for msg in reversed(history):
            if msg.get("role") == "assistant":
                last_assistant_msg = msg
                break

        if last_assistant_msg:
            last_content = last_assistant_msg.get("content", "").lower()
            if "registrar la oportunidad" in last_content or "para continuar" in last_content:
                logger.info("Orquestador: Detectado flujo de registro en curso desde el historial")
                return {
                    "category": "accion_registro",
                    "start_time": time.time(),
                    "warnings": [],
                }

The three detection mechanisms, in priority order:

Explicit confirmation flag — the frontend sends "confirmation": true in the JSON body when the user clicks a confirmation button in the UI. This is the most reliable signal and short-circuits everything else.
Short confirmation text — if the user types any of the recognized short phrases ("sí", "confirmar", "confirmo", etc.), it is treated as a confirmation even without the explicit flag.
History inspection — if neither of the above applies, the node walks backwards through the conversation history to find the last assistant message. If that message contains the phrases "registrar la oportunidad" or "para continuar", the system infers that the previous turn was an agente_accion summary pending confirmation, and routes the new message to accion to complete the registration.

The compiled orchestrator_app object is a LangGraph CompiledStateGraph. It is invoked by ChatService (in backend/app/services/chat_service.py) via orchestrator_app.invoke(initial_state), where initial_state is a dictionary populated from the incoming API request. The return value is the final GraphState dictionary, from which ChatService extracts final_answer, sources, agents_invoked, tokens_input, tokens_output, and warnings to build the HTTP response. The compiled graph is module-level and stateless between requests — each invoke() call creates a fresh, isolated GraphState from the provided initial values.

Get Started

Core Concepts

Agents

Configuration & Deployment

Observability & Evaluation

LangGraph Orchestrator: StateGraph and Routing Logic

GraphState

The `Annotated[List, operator.add]` pattern

Graph Nodes

classify

catalogo

politicas

proceso_ventas

multimodal

accion

Conditional Routing

Graph Construction

Confirmation Flow

Build docs developers (and LLMs) love

Get Started

Core Concepts

Agents

Configuration & Deployment

Observability & Evaluation

Documentation Index

​GraphState

​The Annotated[List, operator.add] pattern

​Graph Nodes

classify

catalogo

politicas

proceso_ventas

multimodal

accion

​Conditional Routing

​Graph Construction

​Confirmation Flow

Build docs developers (and LLMs) love

GraphState

The `Annotated[List, operator.add]` pattern

Graph Nodes

Conditional Routing

Graph Construction

Confirmation Flow