Para crear un agente IA con Python y LangGraph necesitas instalar langgraph y langchain-openai, definir un estado tipado, crear nodos como funciones Python, compilar el grafo con StateGraph y ejecutarlo con .invoke(). En menos de 100 líneas de código puedes tener un agente funcional que consulta FAQs y escala a un humano cuando no sabe la respuesta.
En este tutorial construimos un agente IA completo paso a paso: desde cero hasta un agente de soporte que consulta una base de conocimiento, decide si puede responder o necesita ayuda humana, y mantiene el contexto de la conversación. Todo el código es funcional y listo para usar.
Requisitos previos
Antes de empezar, asegúrate de tener:
- Python 3.11 o superior instalado
- Una API key de OpenAI (o cualquier proveedor compatible: Anthropic, Google, etc.)
- Conocimientos básicos de Python: clases, funciones, type hints
- Un entorno virtual activado (recomendado: venv o conda)
Instala las dependencias:
pip install langgraph langchain-openai langchain-core python-dotenvConceptos clave: grafos, nodos y estados
LangGraph es un framework para construir agentes como grafos dirigidos. Antes de escribir código, entiende estos tres conceptos:
Estado (State)
El estado es un diccionario tipado (usando TypedDict) que viaja por todos los nodos del grafo. Cada nodo puede leerlo y modificarlo. Es la memoria del agente durante la conversación.
Nodos (Nodes)
Los nodos son funciones Python que reciben el estado y devuelven un diccionario con las actualizaciones. Cada nodo representa una acción: llamar al LLM, consultar una base de datos, decidir el siguiente paso.
Aristas (Edges)
Las aristas conectan nodos. Pueden ser fijas (siempre van de A a B) o condicionales (van a B o C según el estado). Las aristas condicionales son lo que da inteligencia al flujo.
El flujo típico de un agente LangGraph es:
START → nodo_entrada → [decisión condicional] → nodo_acción → nodo_respuesta → END
Paso 1: Definir el estado del agente
Empezamos definiendo qué información necesita mantener nuestro agente:
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
class AgentState(TypedDict):
# Lista de mensajes de la conversación
messages: Annotated[List[BaseMessage], operator.add]
# Respuesta final del agente
response: str
# Si necesita escalar a un humano
needs_human: bool
# Número de intentos de respuesta
attempts: intUsamos Annotated[List[BaseMessage], operator.add] para que LangGraph acumule los mensajes en lugar de sobrescribirlos en cada nodo. Esto es esencial para mantener el contexto conversacional.
Paso 2: Crear los nodos del agente
Ahora definimos las funciones que actuarán como nodos. Cada función recibe el estado y devuelve un dict con las actualizaciones:
Nodo 1: Consultar la base de conocimiento
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
# Base de conocimiento simulada (en producción, esto sería RAG sobre tu documentación)
FAQ_DATABASE = {
"precio": "Nuestros planes empiezan desde 299€/mes. Puedes ver todos los precios en /precios.",
"prueba": "Ofrecemos un piloto gratuito de 2 semanas sin compromiso. Solicítalo en /piloto.",
"integraciones": "Nos integramos con Salesforce, HubSpot, Odoo, WhatsApp Business y más de 50 herramientas.",
"tiempo": "El tiempo de implementación típico es de 2 a 4 semanas desde la firma del contrato.",
}
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def consultar_faq(state: AgentState) -> dict:
# Busca en la base de conocimiento y genera una respuesta.
messages = state["messages"]
ultima_pregunta = messages[-1].content if messages else ""
# Construir contexto con la FAQ relevante
contexto_faq = "
".join([f"- {k}: {v}" for k, v in FAQ_DATABASE.items()])
system_prompt = f"Eres un agente de soporte de Sphyrna Solutions.
Responde SOLO con información de la siguiente base de conocimiento.
Si la pregunta no está cubierta, responde exactamente: "NO_SE"
Base de conocimiento:
{contexto_faq}"
response = llm.invoke([
SystemMessage(content=system_prompt),
*messages
])
respuesta = response.content
necesita_humano = respuesta.strip() == "NO_SE"
return {
"response": respuesta if not necesita_humano else "",
"needs_human": necesita_humano,
"attempts": state.get("attempts", 0) + 1,
"messages": [AIMessage(content=respuesta)] if not necesita_humano else [],
}Nodo 2: Escalar a un humano
def escalar_a_humano(state: AgentState) -> dict:
# Genera un mensaje de escalado cuando el agente no sabe la respuesta.
respuesta_escalado = (
"No tengo información suficiente para responder a tu pregunta con precisión. "
"He notificado a nuestro equipo y te contactarán en menos de 2 horas. "
"También puedes escribirnos directamente a [email protected]."
)
return {
"response": respuesta_escalado,
"needs_human": True,
"messages": [AIMessage(content=respuesta_escalado)],
}Paso 3: Definir la lógica de enrutamiento
La función de enrutamiento decide qué nodo ejecutar a continuación según el estado actual:
def decidir_siguiente_paso(state: AgentState) -> str:
# Decide si responder o escalar según el estado.
if state.get("needs_human", False):
return "escalar"
if state.get("attempts", 0) >= 2:
# Después de 2 intentos sin éxito, escalar
return "escalar"
return END # Fin del flujo si tenemos respuestaPaso 4: Compilar y ejecutar el grafo
Ahora ensamblamos todas las piezas en el grafo:
from langgraph.graph import StateGraph, START, END
def compilar_agente() -> object:
# Compila el grafo del agente y devuelve un ejecutable.
# Crear el grafo con nuestro estado
workflow = StateGraph(AgentState)
# Añadir nodos
workflow.add_node("consultar_faq", consultar_faq)
workflow.add_node("escalar", escalar_a_humano)
# Definir flujo: empezar siempre consultando FAQ
workflow.add_edge(START, "consultar_faq")
# Arista condicional: después de consultar, decidir si escalar o terminar
workflow.add_conditional_edges(
"consultar_faq",
decidir_siguiente_paso,
{
"escalar": "escalar",
END: END,
}
)
# Después de escalar, siempre terminar
workflow.add_edge("escalar", END)
# Compilar y devolver el agente ejecutable
return workflow.compile()
# Crear el agente
agente = compilar_agente()El método .compile() valida el grafo (detecta ciclos infinitos, nodos sin salida, etc.) y devuelve un objeto ejecutable. Si hay errores en la definición del grafo, los verás aquí antes de ejecutar.
Ejemplo completo: agente de soporte funcionando
Juntamos todo en un script ejecutable:
#!/usr/bin/env python3
# Agente de soporte con LangGraph - ejemplo completo funcional.
import os
from dotenv import load_dotenv
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
import operator
load_dotenv()
# --- Estado ---
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
response: str
needs_human: bool
attempts: int
# --- Base de conocimiento ---
FAQ_DATABASE = {
"precio": "Nuestros planes empiezan desde 299€/mes. Más info en /precios.",
"prueba": "Piloto gratuito de 2 semanas sin compromiso. Solicítalo en /piloto.",
"integraciones": "Integramos con Salesforce, HubSpot, Odoo, WhatsApp Business y más.",
"tiempo": "Implementación en 2-4 semanas desde firma del contrato.",
}
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# --- Nodos ---
def consultar_faq(state: AgentState) -> dict:
contexto = "
".join([f"- {k}: {v}" for k, v in FAQ_DATABASE.items()])
system_prompt = f"Eres agente de soporte de Sphyrna Solutions.
Responde SOLO con info de la base de conocimiento. Si no sabes, responde: "NO_SE"
Base de conocimiento:
{contexto}"
response = llm.invoke([SystemMessage(content=system_prompt), *state["messages"]])
respuesta = response.content
necesita_humano = respuesta.strip() == "NO_SE"
return {
"response": respuesta if not necesita_humano else "",
"needs_human": necesita_humano,
"attempts": state.get("attempts", 0) + 1,
"messages": [AIMessage(content=respuesta)] if not necesita_humano else [],
}
def escalar_a_humano(state: AgentState) -> dict:
msg = ("No tengo información suficiente. Nuestro equipo te contactará "
"en menos de 2 horas. También puedes escribirnos a [email protected].")
return {"response": msg, "needs_human": True, "messages": [AIMessage(content=msg)]}
# --- Enrutamiento ---
def decidir_siguiente_paso(state: AgentState) -> str:
if state.get("needs_human") or state.get("attempts", 0) >= 2:
return "escalar"
return END
# --- Compilar grafo ---
workflow = StateGraph(AgentState)
workflow.add_node("consultar_faq", consultar_faq)
workflow.add_node("escalar", escalar_a_humano)
workflow.add_edge(START, "consultar_faq")
workflow.add_conditional_edges("consultar_faq", decidir_siguiente_paso,
{"escalar": "escalar", END: END})
workflow.add_edge("escalar", END)
agente = workflow.compile()
# --- Ejecutar ---
def chat(pregunta: str) -> str:
estado_inicial = {
"messages": [HumanMessage(content=pregunta)],
"response": "",
"needs_human": False,
"attempts": 0,
}
resultado = agente.invoke(estado_inicial)
return resultado["response"]
if __name__ == "__main__":
print("Agente de soporte Sphyrna Solutions")
print("Escribe 'salir' para terminar
")
while True:
pregunta = input("Tú: ").strip()
if pregunta.lower() in ("salir", "exit", "quit"):
break
respuesta = chat(pregunta)
print(f"Agente: {respuesta}
")
Diagrama de flujo del agente
Este es el flujo de decisión del agente que acabamos de construir:
┌─────────────────────────────────────────────────────┐
│ INICIO │
└───────────────────────┬─────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ NODO: consultar_faq │
│ • Construye contexto con FAQ_DATABASE │
│ • Llama al LLM con el contexto + mensajes │
│ • Si respuesta = "NO_SE" → needs_human = True │
└───────────────────────┬─────────────────────────────┘
▼
┌─────────────────────────┐
│ ¿needs_human = True? │
│ ¿attempts >= 2? │
└──────┬──────────┬───────┘
SÍ │ │ NO
▼ ▼
┌──────────────────┐ ┌──────────┐
│ NODO: escalar │ │ END │
│ Mensaje escalado │ └──────────┘
└────────┬─────────┘
▼
┌──────┐
│ END │
└──────┘
Tips para llevar este agente a producción
El ejemplo funciona, pero para un entorno de producción real necesitas añadir:
1. Persistencia de estado (memoria entre sesiones)
LangGraph soporta checkpointers para guardar el estado entre invocaciones. Usa SqliteSaver para desarrollo o PostgresSaver para producción:
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")
agente = workflow.compile(checkpointer=memory)
# Ahora puedes continuar conversaciones con thread_id
config = {"configurable": {"thread_id": "usuario-123"}}
resultado = agente.invoke(estado, config=config)2. RAG en lugar de FAQ estática
En producción, sustituye el diccionario FAQ_DATABASE por un sistema RAG (Retrieval-Augmented Generation) que busca en tus documentos reales. Consulta nuestro artículo sobre RAG para entender cómo implementarlo.
3. Manejo de errores y timeouts
Añade try/except en los nodos, configura timeouts en el LLM y define un nodo de error para casos inesperados. En producción, los fallos de la API del LLM son más comunes de lo que parece.
4. Observabilidad con LangSmith
LangSmith es la plataforma de observabilidad oficial de LangChain/LangGraph. Con una simple variable de entorno puedes ver cada ejecución, nodo por nodo, con los inputs, outputs y tiempos:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "tu-api-key-langsmith"5. Tests unitarios para cada nodo
Los nodos son funciones Python puras: reciben un estado y devuelven un dict. Esto los hace trivialmente testables:
def test_escalar_a_humano():
estado = {"messages": [], "response": "", "needs_human": True, "attempts": 2}
resultado = escalar_a_humano(estado)
assert resultado["needs_human"] == True
assert len(resultado["response"]) > 0
assert len(resultado["messages"]) == 1Preguntas frecuentes sobre LangGraph y agentes IA con Python
¿LangGraph es mejor que LangChain para agentes?
LangGraph y LangChain son complementarios. LangChain provee los bloques básicos (LLMs, herramientas, embeddings). LangGraph añade la capa de orquestación con grafos, que es superior a las cadenas lineales de LangChain para agentes complejos con flujos condicionales, loops y estado persistente. Para agentes con toma de decisiones, LangGraph es la elección correcta en 2026.
¿Qué diferencia hay entre un agente LangGraph y un chatbot?
Un chatbot es reactivo: responde a lo que el usuario dice según reglas o un LLM directo. Un agente LangGraph es activo y razonador: puede ejecutar herramientas, tomar decisiones en múltiples pasos, mantener estado entre turnos y escalar inteligentemente. La diferencia práctica es que el agente puede resolver tareas complejas que requieren varios pasos, mientras que el chatbot solo responde al mensaje actual.
¿Puedo usar LangGraph con modelos distintos a OpenAI?
Sí. LangGraph es agnóstico al proveedor de LLM. Puedes usar ChatAnthropic (Claude), ChatGoogleGenerativeAI (Gemini), ChatOllama (modelos locales) o cualquier modelo compatible con la interfaz de LangChain. Solo necesitas cambiar la importación y la inicialización del LLM.
¿Cuánto cuesta ejecutar este agente en producción?
Con GPT-4o-mini, el coste es aproximadamente 0.15$/millón de tokens de entrada y 0.60$/millón de salida. Una conversación de 10 turnos consume ~2.000-5.000 tokens. Calcula ~0.003$ por conversación, es decir, unos 3€ por cada 1.000 conversaciones. Para volúmenes altos, considera modelos locales o optimizar el contexto enviado al LLM.
¿Quieres un agente IA sin escribir el código tú?
Si necesitas un agente IA conversacional para tu empresa pero no quieres gestionar la infraestructura técnica, en Sphyrna Solutions lo construimos, integramos y mantenemos por ti. Piloto gratuito en menos de una semana.