Si hay algo que he aprendido al construir sistemas de IA en los últimos años, es esto:los patrones importan Ya sea que diseñemos software tradicional o experimentemos con agentes de modelos de lenguaje grandes (LLM), la forma en que estructuramos los flujos de trabajo determina cuán robustos, flexibles y escalables serán.

En lugar de conectar agentes con una lógica personalizada interminable, LangGraph nos brinda un marco basado en gráficos para definir, visualizar y depurar flujos de trabajo.

En esta publicación, te guiaré por algunos de los patrones de diseño de agencia más comunes que puedes implementar en LangGraph. No entraré en explicaciones detalladas (ya los cubrí en el artículo anterior), pero sí destacaré la importancia de cada uno, mi perspectiva sobre ellos en sistemas reales y cuándo deberían usarse. También he incluido ejemplos de código para que puedas experimentarlos en un notebook de Google Colab.

En resumen:
Desarrollar agentes de IA es complejo porque el campo es aún joven y las arquitecturas evolucionan rápidamente. En esta publicación, muestro cómo implementar patrones de diseño de agentes probados con LangGraph, haciéndolos más robustos y listos para producción, incluyendo:

  • Encadenamiento de indicaciones : dividir tareas complejas en pasos manejables
  • Enrutamiento y paralelización : dirigir consultas de manera eficiente y ejecutar tareas simultáneamente
  • Reflexión : permitir que los agentes critiquen y mejoren sus propios resultados
  • Uso de herramientas : integración de sistemas externos y API para una funcionalidad del mundo real
  • Planificación : estructurar objetivos en secuencias claras y ejecutables
  • Colaboración entre múltiples agentes : coordinación de agentes especializados para resolver problemas complejos

Antes de profundizar en patrones de diseño específicos, primero debemos configurar nuestro entorno e importaciones. Dado que LangGraph se basa en LangChain, incorporaremos las utilidades principales para definir el estado, gestionar mensajes y conectar herramientas.
En los siguientes ejemplos, también usaré el modelo de IA generativa de Google a través de LangChain, así que asegúrate de tener tu GOOGLE_API_KEY (disponible gratuitamente) almacenada en los datos de usuario de Colab (o como una variable de entorno si se ejecuta localmente).

import os
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from pydantic import BaseModel

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import add_messages
from langgraph.types import Command

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage

from google.colab import userdata

GOOGLE_API_KEY = userdata.get('GOOGLE_API_KEY')

Encadenamiento de indicaciones

En esencia, el encadenamiento de indicaciones es el patrón más simple: la salida de un modelo se convierte en la entrada de otro. Piénselo como una carrera de relevos donde cada nodo refina, amplía o transforma la información antes de transmitirla.

En el siguiente ejemplo, creamos un flujo de trabajo que realiza las siguientes tareas mediante encadenamiento de indicaciones:

  1. Extraer temas clave del texto de entrada.
  2. Genere títulos de blogs basados ​​en esos temas.

Inicializamos el cliente LLM, definimos una Stateclase para el seguimiento de datos e implementamos cada paso como una función de nodo.
Finalmente, conectamos los nodos secuencialmente y compilamos el grafo, creando un flujo de trabajo LangGraph ejecutable.

# Define LLM client
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
    system_instruction="""You are an expert technical writer. Always give clear,
     concise, and straight-to-the-point answers."""
)

# Define the graph state
class State(dict):
    text: str
    topics: str
    title: str

# Define nodes
def extract_topics(state: State) -> State:
    prompt = f"Extract 1-3 key topics from the following text:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    state["topics"] = resp.content.strip()
    return state

def generate_title(state: State) -> State:
    prompt = f"Generate two catchy blog titles for each one these topics:\n\n{state['topics']}"
    resp = llm.invoke(prompt)
    state["title"] = resp.content.strip()
    return state

# Build the graph
workflow = StateGraph(State)
workflow.add_node("extract_topics", extract_topics)
workflow.add_node("generate_title", generate_title)

# Define the node connections
workflow.set_entry_point("extract_topics")
workflow.add_edge("extract_topics", "generate_title")
workflow.add_edge("generate_title", END)

# Compile runnable graph
graph = workflow.compile()

Así es como se ve el gráfico como un diagrama de flujo de sirena:

Con el flujo de trabajo definido, podemos ejecutar el gráfico con una entrada de muestra. Aquí, proporcionamos un texto breve sobre LangGraph, y el gráfico extrae temas secuencialmente y genera sugerencias de títulos para blogs.

# Run the graph
input_text = (
    "LangGraph introduces a graph-based paradigm for building LLM-powered agents. "
    "It allows developers to create modular, debuggable, and reliable agent workflows "
    "using nodes, edges, and state passing."
)

result = graph.invoke({"text": input_text})
print("Topics:", result["topics"])
print("\n"+"="*50+"\n")
print("Suggested Blog Title:", result["title"])

Producción:

Temas: Aquí hay 3 temas clave extraídos del texto: 

1.   **Agentes LLM basados ​​en gráficos:** El concepto central de usar una estructura gráfica para construir agentes LLM. 
2.   **Modularidad, depuración y confiabilidad:** Los beneficios de usar LangGraph para el desarrollo de agentes. 
3.   **Nodos, bordes y paso de estado:** Los componentes fundamentales del marco LangGraph. 

=================================================== 

Título de blog sugerido: Bien, aquí hay dos títulos de blog atractivos para cada uno de los tres temas, que apuntan a una mezcla de claridad e intriga: 

**1. Agentes LLM basados ​​en grafos:** 

*    **Título 1:** Liberando el poder de las mentes gráficas: Construyendo agentes LLM más inteligentes 
*    **Título 2:** Más allá de las cadenas: Cómo los agentes basados ​​en grafos están revolucionando los LLM 

** 2. Modularidad, depurabilidad y confiabilidad:** 

*    **Título 1:** LangGraph: El secreto para construir agentes LLM confiables y depurables 
*    **Título 2:** Deje de perseguir errores: El enfoque modular de LangGraph para el desarrollo de agentes LLM 

** 3. Nodos, bordes y paso de estado:** 

*    **Título 1:** LangGraph desmitificado: Nodos, bordes y el flujo de conversaciones inteligentes 
*    **Título 2:** Bloques de construcción de brillantez: Dominando nodos, bordes y estado en LangGraph

Enrutamiento:

No todas las entradas merecen el mismo tratamiento. Aquí es donde entra en juego el enrutamiento: clasificamos la entrada y la enviamos por la rama derecha del grafo.

En la práctica, esto podría verse así:

  • Sentimiento positivo → respuesta alentadora
  • Sentimiento negativo → respuesta de apoyo

En este flujo de trabajo de ejemplo, el classificationnodo primero clasifica el sentimiento de un texto y actualiza su estado. A continuación, una función de borde condicional router_funcdetermina qué rama seguir: positiveo negative. Cada rama genera una respuesta adaptada al sentimiento.

Al utilizar bordes condicionales, LangGraph facilita el enrutamiento dinámico de datos y al mismo tiempo mantiene el gráfico legible y mantenible.

# LLM client
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
    system_instruction="""You are a helpful assistant that can classify text 
    sentiment and respond accordingly."""
)

# State definition
class State(dict):
    text: str
    sentiment: str
    response: str

# Nodes
def calssification(state: State) -> str:
    """Classify sentiment."""
    prompt = f"Is the following text positive or negative? Answer with one word only: Positive or Negative.\n\n{state['text']}"
    resp = llm.invoke(prompt)
    sentiment = resp.content.strip().lower()
    state["sentiment"] = sentiment
    return state

def positive_node(state: State) -> State:
    prompt = f"Generate an encouraging reply to this positive text:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    state["response"] = resp.content.strip()
    return state

def negative_node(state: State) -> State:
    prompt = f"Generate a supportive reply to this negative text:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    state["response"] = resp.content.strip()
    return state

def router_func(state: State) -> Literal["positive", "negative"]:
    """Return next node name."""
    return "positive" if "positive" in state["sentiment"] else "negative"

# Build the graph
workflow = StateGraph(State)

workflow.add_node("calssification", calssification)
workflow.add_node("positive", positive_node)
workflow.add_node("negative", negative_node)

# classify node decides the next step
workflow.set_entry_point("calssification")
workflow.add_conditional_edges("calssification", router_func, {
    "positive": "positive",
    "negative": "negative",
})

# Both branches lead to END
workflow.add_edge("positive", END)
workflow.add_edge("negative", END)

graph = workflow.compile()

Diagrama de flujo de trabajo del agente:

Flujo de trabajo de enrutamiento con LangGraph

Una vez definido el flujo de trabajo de enrutamiento, podemos probarlo con una entrada de muestra. Aquí, el agente evalúa la percepción de un mensaje positivo y genera una respuesta adecuada.

# Run example
input_text = "I'm so happy with how my project turned out!"
result = graph.invoke({"text": input_text})
print("Sentiment:", result["sentiment"])
print("Response:", result["response"])

Producción:

Sentiment: positive
Response: That's fantastic! I'm so happy to hear that your hard work paid off and you're pleased with the results. Celebrate that success! You deserve it! 🎉

Paralelización

Una de mis características favoritas: ¿por qué procesar todo secuencialmente cuando puedes ejecutar algunas tareas en paralelo?

Resumen y crítica, extracción de sentimientos y temas, investigación en múltiples fuentes, todos estos son trabajos que pueden realizarse en paralelo y converger más adelante.

En este ejemplo, el flujo de trabajo ejecuta dos nodos en paralelo: uno resume el texto de entrada y el otro lo critica. Una vez completados ambos nodos, sus resultados se combinan en un único párrafo coherente.

Al utilizar bordes paralelos y un estado de solo anexión para las salidas, LangGraph permite que los nodos operen simultáneamente mientras aún fusionan los resultados de manera limpia, lo que demuestra cómo los agentes pueden realizar múltiples análisis a la vez.

# LLM client
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
)

# State definition
class State(TypedDict):
    text: str
    # Reducer makes these append-only so multiple nodes can update in parallel
    outputs: Annotated[list, operator.add]

# Nodes
def summarize(state: State):
    prompt = f"Summarize in one sentence:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    return {"outputs": [f"Summary: {resp.content.strip()}"]}

def critique(state: State):
    prompt = f"Critique briefly:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    return {"outputs": [f"Critique: {resp.content.strip()}"]}

def combine(state: State):
    prompt = f"Combine the following Critique and Summarization in one \
    paragraph:\n\n{state['text']}"
    resp = llm.invoke(prompt)
    return {"outputs": [f"Combined:\n{resp.content.strip()}"]}

# Build the graph
builder = StateGraph(State)
builder.add_node("summarize", summarize)
builder.add_node("critique", critique)
builder.add_node("combine", combine)

# Parallel edges: summarize and critique run side by side
builder.add_edge(START, "summarize")
builder.add_edge(START, "critique")

# Both join into combine
builder.add_edge("summarize", "combine")
builder.add_edge("critique", "combine")

builder.add_edge("combine", END)

graph = builder.compile()
Flujo de trabajo de paralelización con LangGraph

Con el flujo de trabajo paralelo implementado, podemos ejecutarlo con una entrada de muestra. El gráfico ejecuta los nodos summarizey critiquesimultáneamente y luego combina sus resultados en una única salida.

# --- Run example ---
input_text = "LangGraph helps developers design and run agent workflows with LLMs."
result = graph.invoke({"text": input_text, "outputs": []})

print("\nFinal outputs:")
for out in result["outputs"]:
    print(out)
    print("="*100 + "\n")

Producción:

Final outputs:
Critique: The statement is accurate and concise, but lacks detail and context. Here's a brief critique, highlighting its strengths and weaknesses:

**Strengths:**

*   **Accurate:** It correctly identifies LangGraph's primary function.
*   **Concise:** It's a short and to-the-point description.

**Weaknesses:**

*   **Lacks Specificity:** Doesn't explain *how* LangGraph helps, what kind of workflows, or its advantages.
*   **Missing Context:** Assumes the reader knows what "agent workflows" and "LLMs" are.
*   **Doesn't Highlight Value Proposition:** Doesn't mention benefits like improved reliability, observability, or scalability.

**In essence, it's a functional description, not a compelling one.** It tells you *what* LangGraph does, but not *why* it's useful or *how* it's different.
====================================================================================================

Summary: LangGraph is a framework for building multi-actor workflows powered by large language models.
====================================================================================================

Combined:
LangGraph provides a powerful framework for developers to construct and execute complex agent workflows powered by Large Language Models (LLMs). In essence, it simplifies the process of building sophisticated AI agents that can interact with various tools and data sources to achieve specific goals. While the provided information is concise, it lacks detail regarding the specific functionalities and advantages offered by LangGraph, such as its capabilities for managing state, handling errors, or optimizing performance, which would be crucial for a comprehensive understanding and evaluation of the tool.
====================================================================================================

Reflexión

Si hay un patrón que creo que todo agente debería tener, es la reflexión.

Sé que esto puede ser polémico, pero creo que la reflexión es la única manera de ir más allá de las respuestas puntuales de LLM. Los modelos alucinan. Pierden el contexto. Producen borradores que parecen buenos, pero no resisten el escrutinio. La reflexión cierra ese círculo.

En el siguiente ejemplo, el flujo de trabajo realiza estos pasos:

  1. El generatornodo produce un borrador inicial (o perfecciona uno existente).
  2. El evaluatornodo revisa el borrador y proporciona comentarios o aprobación.
  3. La decidefunción determina si se debe retroceder para refinar o proceder a finalizar.
  4. El finalizenodo guarda el borrador aprobado.

Al combinar bordes condicionales con evaluación iterativa, LangGraph permite al agente mejorar continuamente sus resultados hasta que cumplan con los estándares deseados.

# LLM client
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
)

# State
class State(TypedDict):
    task: str
    draft: str
    feedback: str
    final: str

# Nodes
def generator(state: State):
    """Generate an initial or refined draft."""
    prompt = f"""
You are an assistant helping to complete the following task:

Task:
{state['task']}

Current Draft:
{state.get('draft', 'None')}

Feedback:
{state.get('feedback', 'None')}

Instructions:
- If there is no draft and no feedback, generate a clear and complete response to the task.
- If there is a draft but no feedback, improve the draft as needed for clarity and quality.
- If there is both a draft and feedback, revise the draft by incorporating the feedback directly.
- Always produce a single, improved draft as your output.
"""
    resp = llm.invoke(prompt)
    return {"draft": resp.content.strip()}

def evaluator(state: State):
    """Evaluate the draft and give feedback or approval."""
    prompt = f"""Evaluate the following draft, based on the given task. 
If it meets the requirements, reply exactly 'APPROVED'. 
Otherwise, provide constructive feedback for improvement.
Task:
{state['task']}
Draft:
{state['draft']}"""
    resp = llm.invoke(prompt)
    print(f"""
================= DRAFT =================
{state['draft']}

================ FEEDBACK ===============
{resp.content.strip()}
========================================
""")
    return {"feedback": resp.content.strip()}

def decide(state: State) -> str:
    """Decide next step: either approve and finish, or refine again."""
    if "APPROVED" in state["feedback"].upper():
        return "approved"
    return "refine"

def finalize(state: State):
    """Save the final approved draft."""
    return {"final": state["draft"]}

# Build the graph
builder = StateGraph(State)

builder.add_node("generator", generator)
builder.add_node("evaluator", evaluator)
builder.add_node("finalize", finalize)

builder.add_edge(START, "generator")
builder.add_edge("generator", "evaluator")
builder.add_edge("evaluator", "finalize")

# Conditional edges from decide
builder.add_conditional_edges(
    "evaluator",
    decide,
    {
        "approved": "finalize",   # stop loop
        "refine": "generator",    # go back for improvement
    },
)

builder.add_edge("finalize", END)

graph = builder.compile()

Diagrama de flujo de trabajo de reflexión:

Patrón de diseño de reflexión con LangGraph

Con el flujo de trabajo de reflexión definido, ejecutémoslo en una tarea de ejemplo. El agente genera un borrador de solución, lo evalúa y lo itera según sea necesario hasta que se aprueba el resultado.

# Run example
input_task = "You have six horses and want to race them to see which is fastest. What is the best way to do this?"
result = graph.invoke({"task": input_task})

print("\nFinal Answer:\n", result["final"])

Producción:

================= DRAFT =================
Here's a strategy to determine the fastest horse out of six, minimizing the number of races required:
1. **Initial Races:** Divide the six horses into two groups of three. Race each group. Label the horses A1, A2, A3 (with A1 being the fastest in the first race, A2 second, and A3 third) and B1, B2, B3 (similarly ranked from the second race).
2. **Race of the Winners:** Race the winners of the two initial races (A1 and B1). The winner of this race is the fastest horse overall. Let's say A1 wins.
3. **Determining Second Fastest:** The second fastest horse could be either B1 (the horse that lost to A1 in the winners' race) or A2 (the second-place horse from the race that A1 won). Race B1 and A2. The winner of *this* race is the second-fastest horse overall.
Therefore, this method requires a total of three races.

================ FEEDBACK ===============
The draft is incorrect. While it identifies the fastest horse correctly, it fails to reliably identify the second-fastest horse.
Here's why and how to improve it:
The logic in step 3 is flawed. The second-fastest horse could be B1, A2, *or* possibly even A3 if A1 and A2 were significantly faster in their initial race than B1, B2, and B3.  Since we only have ordinal data (rankings within a race), not interval data (relative speed differences), we can't rule out A3.
Here's a corrected strategy requiring more races:
1. **Initial Races:** Divide the six horses into two groups of three. Race each group. Label the horses A1, A2, A3 (with A1 being the fastest in the first race, A2 second, and A3 third) and B1, B2, B3 (similarly ranked from the second race).
2. **Race of the Winners:** Race the winners of the two initial races (A1 and B1). The winner of this race is the fastest horse overall. Let's say A1 wins.
3. **Determining Second Fastest:** The second fastest horse could be either B1 (the horse that lost to A1 in the winners' race), A2 (the second-place horse from the race that A1 won), or B2 (the second place horse from the race that B1 won). Race B1, A2, and B2. The winner of *this* race is the second-fastest horse overall.

Therefore, this revised method requires a total of three races. But this isn't the *best* method. A better method that guarantees finding the fastest horse is:
1. **Initial Races:** Divide the six horses into two groups of three. Race each group. Label the horses A1, A2, A3 (with A1 being the fastest in the first race, A2 second, and A3 third) and B1, B2, B3 (similarly ranked from the second race).
2. **Race of the Winners:** Race the winners of the two initial races (A1 and B1). The winner of this race is the fastest horse overall. Let's say A1 wins.
3. **Race for 2nd Place**: The second fastest horse could be B1 (the horse that lost to A1 in the winners' race), A2 (the second-place horse from the race that A1 won) or B2 (the second-place horse from the race that B1 won). Race these three horses. The winner of this race is the second-fastest horse overall.

This also requires three races, but is a more robust solution than the original. However, a better solution exists:
1. **Race all horses together.**
This requires only one race, and will identify the fastest horse.
Revised Answer:
The provided strategy is flawed. It correctly identifies the fastest horse but fails to guarantee finding the second-fastest. A better approach is simply to race all six horses together in a single race. The horse that wins is the fastest.

IMPROVEMENT NEEDED
========================================


================= DRAFT =================
Here's a strategy to determine the fastest horse out of six. The most efficient approach is:
1. **Race all six horses together.**
The horse that wins this single race is definitively the fastest.
While other strategies exist that involve multiple preliminary races, they are unnecessary. For instance, one could divide the horses into groups, race those groups, and then race the winners. However, this requires more races than simply racing all horses together. Similarly, attempting to determine the *second* fastest horse complicates the process and requires additional races. To strictly determine the *fastest* horse, a single race is the most efficient solution.

================ FEEDBACK ===============
APPROVED
========================================

Final Answer:
Here's a strategy to determine the fastest horse out of six. The most efficient approach is:
1. **Race all six horses together.**
The horse that wins this single race is definitively the fastest.
While other strategies exist that involve multiple preliminary races, they are unnecessary. For instance, one could divide the horses into groups, race those groups, and then race the winners. However, this requires more races than simply racing all horses together. Similarly, attempting to determine the *second* fastest horse complicates the process and requires additional races. To strictly determine the *fastest* horse, a single race is the most efficient solution.

Uso de herramientas

En LangGraph, puede integrar fácilmente herramientas externas, desde funciones matemáticas sencillas hasta API complejas, en el flujo de trabajo de su agente.
LLM no solo genera, sino que delega. Esto transforma su modelo de un generador de texto a un sistema de resolución de problemas.

En este ejemplo, definimos una herramienta de cálculo sencilla y la vinculamos a nuestro LLM. El flujo de trabajo:

  1. Llama al modelo con mensajes de usuario.
  2. Comprueba si el modelo solicitó una herramienta.
  3. Si se necesita una herramienta, la ejecuta y envía el resultado al modelo.

Al alternar entre los nodos del modelo y de la herramienta, LangGraph permite a los agentes interactuar con las herramientas de forma dinámica, ampliando sus capacidades más allá de la mera generación de texto.

# State definition
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# Define a simple tool
def calculator(expression: str):
    """Evaluate a math expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

# Initialize Gemini model with tools
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
)

model_with_tools = llm.bind_tools([calculator])
tool_node = ToolNode([calculator])

# Nodes
def call_model(state: State):
    """Call the model; it may request a tool."""
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: State):
    """Decide whether to go to tools or finish."""
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:  # If the model requested a tool
        return "tools"
    return END

# Build the graph
builder = StateGraph(State)

builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue, ["tools", END])
builder.add_edge("tools", "call_model")

graph = builder.compile()

Diagrama de flujo de uso de herramientas:

Flujo de trabajo de uso de herramientas con LangGraph

Con el flujo de trabajo habilitado por la herramienta, podemos ejecutar una consulta de ejemplo. El modelo reconoce la necesidad de cálculo, invoca la calculatorherramienta y devuelve el resultado dentro de la conversación.

# Run example
query = {"role": "user", "content": "What is 12 * 7?"}
result = graph.invoke({"messages": [query]})

print("\nConversation:")
for m in result["messages"]:
    print(m)

Producción:

Conversation:
content='What is 12 * 7?' additional_kwargs={} response_metadata={} id='5ceeda02-510d-4bad-849a-759212097013'
content='' additional_kwargs={'function_call': {'name': 'calculator', 'arguments': '{"expression": "12 * 7"}'}} response_metadata={'prompt_feedback': {'block_reason': 0, 'safety_ratings': []}, 'finish_reason': 'STOP', 'safety_ratings': []} id='run--b8c8ab73-5082-47c0-9bd9-8e46b2ae3ad1-0' tool_calls=[{'name': 'calculator', 'args': {'expression': '12 * 7'}, 'id': '662e4855-57d9-4304-be72-923b453c8577', 'type': 'tool_call'}] usage_metadata={'input_tokens': 19, 'output_tokens': 7, 'total_tokens': 26, 'input_token_details': {'cache_read': 0}}
content='84' name='calculator' id='6354a43a-c65e-44b2-84b5-2a575a7b3cf2' tool_call_id='662e4855-57d9-4304-be72-923b453c8577'
content='The answer is 84.' additional_kwargs={} response_metadata={'prompt_feedback': {'block_reason': 0, 'safety_ratings': []}, 'finish_reason': 'STOP', 'safety_ratings': []} id='run--dd0f2241-aab6-4412-bd78-c36a79038328-0' usage_metadata={'input_tokens': 29, 'output_tokens': 8, 'total_tokens': 37, 'input_token_details': {'cache_read': 0}}

Planificación

A diferencia del enrutamiento, que selecciona una sola rama, la planificación establece una hoja de ruta de varios pasos. Piénsalo como la capacidad del agente de decir: " Primero investigaré GraphQL, luego lo compararé con REST y, por último, sintetizaré mis hallazgos".

Personalmente, la planificación me ha resultado invaluable en tareas como la de asistente de investigación o bots de comparación multifuente. Convierte al agente en un estratega más que en un agente reactivo.

# Set up the Language Model
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
)

# Define the State
class PlannerState(TypedDict):
    task: str
    plan: list[str]
    graphql_results: str  # Results from the GraphQL researcher
    rest_results: str     # Results from the REST researcher
    final_output: str     # The final written summary

# Dummy tools for our researchers
def graphql_search_tool(query: str):
    """A dummy tool that returns fixed info about GraphQL."""
    print(f"GRAPHQL RESEARCHER: Searching for '{query}'")
    return "GraphQL Pros: Efficient data fetching (no over-fetching), single endpoint, strongly typed schema."

def rest_search_tool(query: str):
    """A dummy tool that returns fixed info about REST."""
    print(f"REST RESEARCHER: Searching for '{query}'")
    return "REST API Cons: Can lead to over or under-fetching data, requires multiple round-trips for complex queries, URL-based structure can be rigid."

# Worker Nodes
def graphql_research_worker(state: PlannerState):
    """Worker node that researches GraphQL pros."""
    results = graphql_search_tool("pros of GraphQL")
    return {"graphql_results": results}

def rest_research_worker(state: PlannerState):
    """Worker node that researches REST cons."""
    results = rest_search_tool("cons of REST APIs")
    return {"rest_results": results}

def writer_worker(state: PlannerState):
    """
    Synthesizer node that waits for all research and writes the final output.
    This node acts as the "join" point.
    """
    print("WRITER: Synthesizing results")
    graphql_results = state['graphql_results']
    rest_results = state['rest_results']
    
    writing_prompt = f"""
    Write a short, balanced comparison post based on the following research.

    GraphQL Information:
    {graphql_results}

    REST API Information:
    {rest_results}
    """
    response = llm.invoke(writing_prompt)
    return {"final_output": response.content}

# Define the Planner
def planner(state: PlannerState):
    """Planner node that creates the initial plan."""
    print("PLANNER: Creating a plan for parallel execution")
    # For this example, the plan is hardcoded.
    # In a real app, an LLM would generate this based on the task.
    plan = [
        "Research GraphQL pros",
        "Research REST cons",
        "Write comparison post"
    ]
    return {"plan": plan}

# Build the Graph
workflow = StateGraph(PlannerState)

# Add the nodes
workflow.add_node("planner", planner)
workflow.add_node("graphql_researcher", graphql_research_worker)
workflow.add_node("rest_researcher", rest_research_worker)
workflow.add_node("writer", writer_worker)

# Set the entry point
workflow.set_entry_point("planner")

# Define the parallel edges
# After the planner, both research workers are called.
workflow.add_edge("planner", "graphql_researcher")
workflow.add_edge("planner", "rest_researcher")

# Define the join point
# The writer will only run after BOTH research workers are complete.
workflow.add_edge("graphql_researcher", "writer")
workflow.add_edge("rest_researcher", "writer")

# The graph ends after the writer is done
workflow.add_edge("writer", END)

# Compile the graph
graph = workflow.compile()

Diagrama de flujo de trabajo de planificación:

Planificación del flujo de trabajo con LangGraph

Al transmitir los eventos del gráfico, podemos observar el flujo de ejecución y ver cómo el agente progresa paso a paso.
El resultado final demuestra cómo la Planificación permite un razonamiento estructurado de varios pasos para obtener resultados más completos y coherentes.

# Run the Graph
user_task = "Write a short post comparing the pros of GraphQL with the cons of REST APIs."
initial_state = {"task": user_task}

# Stream the events to see the execution flow
for event in graph.stream(initial_state):
    for key, value in event.items():
        print(f"Node '{key}' output:")
        print("---")
        print(value)
        print("\n" + "="*30 + "\n")

# Get the final output
final_state = graph.invoke(initial_state)
print("Final Comparison Post:")
print(final_state['final_output'])

Producción:

PLANNER: Creating a plan for parallel execution
Node 'planner' output:
---
{'plan': ['Research GraphQL pros', 'Research REST cons', 'Write comparison post']}

==============================

GRAPHQL RESEARCHER: Searching for 'pros of GraphQL'
Node 'graphql_researcher' output:
---
{'graphql_results': 'GraphQL Pros: Efficient data fetching (no over-fetching), single endpoint, strongly typed schema.'}

==============================

REST RESEARCHER: Searching for 'cons of REST APIs'
Node 'rest_researcher' output:
---
{'rest_results': 'REST API Cons: Can lead to over or under-fetching data, requires multiple round-trips for complex queries, URL-based structure can be rigid.'}

==============================

WRITER: Synthesizing results
Node 'writer' output:
---
{'final_output': '## GraphQL vs. REST: A Quick Comparison\n\nChoosing the right API architecture is crucial for efficient data handling. Both GraphQL and REST offer solutions, but cater to different needs and priorities.\n\n**GraphQL shines with its:**\n\n*   **Efficient Data Fetching:** GraphQL allows clients to request only the data they need, avoiding the common REST problem of over-fetching.\n*   **Single Endpoint:** A single endpoint simplifies API interaction and management.\n*   **Strongly Typed Schema:** The schema ensures data consistency and facilitates client-side development.\n\n**REST APIs, while well-established, can be less efficient due to:**\n\n*   **Over/Under-Fetching:** REST endpoints often return more or less data than required, leading to wasted bandwidth and unnecessary processing.\n*   **Multiple Round-Trips:** Complex data requirements may necessitate multiple API calls, increasing latency.\n*   **Rigid URL Structure:** The URL-based structure can be inflexible and challenging to adapt to evolving data needs.\n\n**In Conclusion:**\n\nGraphQL excels in scenarios demanding precise data retrieval and complex queries, while REST remains a viable option for simpler applications where performance is less critical. The best choice depends on the specific requirements and constraints of your project.'}

==============================

PLANNER: Creating a plan for parallel execution
GRAPHQL RESEARCHER: Searching for 'pros of GraphQL'
REST RESEARCHER: Searching for 'cons of REST APIs'
WRITER: Synthesizing results
Final Comparison Post:
## GraphQL vs. REST APIs: A Quick Comparison
Choosing the right API architecture is crucial for any web application. Two popular options are GraphQL and REST APIs, each with its own strengths and weaknesses. Here's a brief comparison:
**GraphQL:**
*   **Pros:** GraphQL excels at efficient data fetching. By allowing clients to request only the specific data they need, it eliminates the problem of over-fetching common in REST APIs. It also offers a single endpoint and a strongly typed schema, contributing to better developer experience and improved data validation.
**REST APIs:**
*   **Cons:** REST APIs can suffer from over or under-fetching, leading to inefficient data transfer. Complex queries often require multiple round-trips to different endpoints, increasing latency. Furthermore, its URL-based structure can sometimes feel rigid and inflexible.
**In conclusion:**
GraphQL shines when data efficiency and precise control over data fetching are paramount. REST APIs, while potentially less efficient in certain scenarios, are a well-established standard with a vast ecosystem of tools and resources. The best choice depends on the specific needs and complexity of your project.

Colaboración entre múltiples agentes

Por último, el patrón de peso pesado: la colaboración entre múltiples agentes.

Aquí, no se trata solo de ejecutar nodos en un grafo; se orquesta un conjunto de agentes. Un supervisor delega tareas, los especialistas gestionan sus dominios y los resultados se incorporan al flujo de trabajo. Se asemeja más al funcionamiento de los equipos humanos: diferentes roles coordinados en torno a un objetivo central.

Seré sincero: aquí es donde las cosas pueden complicarse. La coordinación, las transferencias, las condiciones de terminación y los ciclos de retroalimentación son fundamentales. Pero también es donde residen las mayores oportunidades.
He creado sistemas donde un equipo de agentes supera a cualquier LLM, simplemente porque cada agente fue asignado para hacer una cosa realmente bien.

En esta configuración de ejemplo:

  1. Un nodo supervisor decide qué agente debe manejar la solicitud del usuario.
  2. Los agentes especialistas (clima y vuelos) procesan la solicitud y devuelven los resultados al supervisor.
  3. El flujo de trabajo utiliza Commandobjetos para el enrutamiento dinámico , lo que permite a los agentes actualizar el estado y determinar el siguiente paso sin problemas.

Esta arquitectura ilustra cómo LangGraph puede orquestar múltiples agentes trabajando juntos, lo que permite sistemas de IA escalables y modulares.

# Set up the Language Model
llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    api_key=GOOGLE_API_KEY,
)

# Define the Supervisor's Routing Logic

# This Pydantic model helps the supervisor choose the next agent.
class Router(BaseModel):
    """Decide the next agent to route to."""
    next_agent: Literal["Weather", "Flights", "__end__"]

# Bind the router to the model to get structured output.
supervisor_model = llm.with_structured_output(Router)

def supervisor(state: MessagesState) -> Command:
    """
    The central supervisor that routes to the correct agent or ends the conversation.
    """
    print("--- 🧑‍💼 SUPERVISOR ---")
    
    # The prompt tells the supervisor how to route the user's message.
    prompt = f"""You are a supervisor routing tasks to a specialist agent. 
    Based on the user's request, choose the appropriate agent.
    
    Available agents:
    - Weather: For questions about weather forecasts.
    - Flights: For questions about flight information.
    
    If the user is saying thank you or the conversation is over, choose '__end__'.
    
    User message: "{state['messages'][-1].content}"
    """
    if isinstance(state['messages'][-1], HumanMessage):
      # The supervisor model makes the routing decision.
      response = supervisor_model.invoke(prompt)
      print(f"Supervisor routing to: {response.next_agent}")
      return Command(goto=response.next_agent)
    else:
      return Command(goto='__end__')

# Define the Specialist Agents

def weather_agent(state: MessagesState) -> Command:
    """A specialist agent for handling weather-related queries."""
    print("--- ☀️ WEATHER AGENT ---")
    
    prompt = f"""You are a weather forecaster. Provide a concise mock weather forecast 
    for the location mentioned in the user's message.
    
    User message: "{state['messages'][-1].content}"
    """
    response = llm.invoke(prompt)
    print(f"Response: {response.content}")
    # Return to the supervisor after the agent has run.
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

def flights_agent(state: MessagesState) -> Command:
    """A specialist agent for handling flight-related queries."""
    print("--- ✈️ FLIGHTS AGENT ---")
    
    prompt = f"""You are a flight information assistant. Provide some mock flight
    details for the destination in the user's message. Respond with short concise information.
    
    User message: "{state['messages'][-1].content}"
    """
    response = llm.invoke(prompt)
    print(f"Response: {response.content}")
    # Return to the supervisor after the agent has run.
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

# Build the Graph

builder = StateGraph(MessagesState)

# Add the supervisor and agents as nodes in the graph.
builder.add_node("supervisor", supervisor)
builder.add_node("Weather", weather_agent)
builder.add_node("Flights", flights_agent)

# The START node directs the flow to the supervisor first.
builder.add_edge(START, "supervisor")

# The graph is now complete. The `Command` objects will handle the dynamic routing.
graph = builder.compile()

Diagrama de flujo de trabajo de múltiples agentes:

Flujo de trabajo multiagente con LangGraph

Para ver el patrón de Colaboración Multiagente en acción, podemos ejecutar una sesión interactiva. El usuario introduce un mensaje y el supervisor lo enruta dinámicamente al agente especialista correspondiente.

El graph.stream()método nos permite transmitir respuestas en tiempo real, para que puedas observar cómo el supervisor y los agentes interactúan, manejan las solicitudes y devuelven resultados sin problemas.

while  True : 
    user_input = input ( "Usuario: " ) 
    if user_input.lower() in [ "quit" , "exit" , "q" ]: 
        print ( "¡Adiós!" ) 
        break 
    
    # El método graph.stream() invoca el gráfico y transmite los resultados.
     events = graph.stream({ "messages" :[HumanMessage(content=user_input)]}) 
    for event in events: 
        # Solo imprimimos las respuestas de la IA al usuario. 
        if  "messages"  in event: 
            event[ "messages" ][- 1 ].pretty_print()

Producción:

User: What's the weather in Tunisia?
--- 🧑‍💼 SUPERVISOR ---
Supervisor routing to: Weather
--- ☀️ WEATHER AGENT ---
Response: Good morning, Tunisia! Expect a sunny day across most of the country. Temperatures will be warm, ranging from the mid-20s Celsius along the coast to the low 30s inland. A light breeze will keep things comfortable along the Mediterranean. Enjoy the sunshine!
--- 🧑‍💼 SUPERVISOR ---
User: Is there any flights from Tunisia to USA available today?
--- 🧑‍💼 SUPERVISOR ---
Supervisor routing to: Flights
--- ✈️ FLIGHTS AGENT ---
Response: Yes, there are flights available.

*   **Airline:** Lufthansa
*   **Departure:** Tunis (TUN) 10:00 AM
*   **Arrival:** New York (JFK) 6:00 PM (Next Day)
*   **Stops:** Frankfurt (FRA)
*   **Price:** $850
--- 🧑‍💼 SUPERVISOR ---
User: thank you
--- 🧑‍💼 SUPERVISOR ---
Supervisor routing to: __end__
User: q
Goodbye!

Reflexiones finales

En este blog, te explico los patrones de diseño de agencia clave, mostrándote cómo implementarlos paso a paso. Hay patrones más avanzados disponibles, que puedes explorar en la documentación de LangGraph.

Si quieres profundizar en cada patrón con explicaciones detalladas, consulta mi entrada anterior (abajo), donde los explico a fondo. También puedes explorar todos los ejemplos de código de esta entrada en un notebook de Colab ejecutable aquí .

Gracias por leer Codigo en Casa