📌Este es el Nivel Intermedio. Si aún no completaste el nivel básico, te recomendamos empezar por ahí: Curso Básico de LangGraph — Primeros Pasos .
¿Qué aprenderás en este nivel?
- Darle a tu agente acceso a búsqueda real en internet (Tavily)
- Conectar tu agente a API externa
- Crear flujos condicionales avanzados
- Usar subgrafos para organizar agentes complejos
- Implementar RAG (el agente lee tus documentos propios)
- Desplegar tu agente como una API con FastAPI
Requisitos previos
- ✅ Haber completado el Nivel Básico de LangGraph
-
✅ Tu entorno virtual
agente-envcreado y funcionando -
✅ Archivo
agente.pycon el agente básico funcionando -
✅ Clave API de OpenAI configurada en tu archivo
.env
Módulo 1: Darle Acceso a Internet con Tavily

En el nivel básico tu agente solo sabía lo que el modelo tenía entrenado. Ahora le vamos a dar acceso a búsqueda real en internet usando Tavily, una API de búsqueda diseñada para agentes de IA.
Paso 1 — Obtener tu clave API de Tavily
- Ve a app.tavily.com y crea una cuenta gratuita
- En el panel copia tu clave API
-
Agrégala a tu archivo
.env:
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
TAVILY_API_KEY=tvly-xxxxxxxxxxxxxxxxxxxxxxxx
Paso 2 — Instalar Tavily
En tu terminal, con el entorno virtual activado, ejecuta:
pip install tavily-python langchain-community
Paso 3 — Crear el archivo del agente con búsqueda
En VS Code, crea un archivo nuevo llamado agente_intermedio.pyy escribe lo siguiente:
from dotenv import load_dotenv
import os
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_community.tools.tavily_search import TavilySearchResults
load_dotenv()
# Definimos el estado
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
# Inicializamos el modelo
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Creamos la herramienta de búsqueda (max 3 resultados)
search_tool = TavilySearchResults(max_results=3)
tools = [search_tool]
# Vinculamos las herramientas al modelo
llm_with_tools = llm.bind_tools(tools)
Paso 4 — Definir los nudos y el gráfico
# Nodo del agente
def agent_node(state: AgentState):
messages = state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# Función para decidir si buscar o terminar
def should_search(state: AgentState):
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
# Nodo ejecutor de herramientas
tool_node = ToolNode(tools)
# Construimos el grafo
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_search, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
app = workflow.compile()
Paso 5 — Probar el agente con búsqueda
# Probamos con una pregunta que requiere información actual
result = app.invoke({
"messages": [HumanMessage(content="¿Cuáles son las últimas noticias sobre inteligencia artificial en 2026?")]
})
print(result["messages"][-1].content)
Ejecuta en tu terminal:
python agente_intermedio.py
✅ Tu agente ahora busca en internet en tiempo real antes de responder.
Módulo 2: Conectar tu Agente a una API Externa
Vamos a crear una herramienta personalizada que consulta una API externa. Usaremos la API pública de clima Open-Meteo (gratuita, sin clave API).
Paso 1 — Instalar la librería de peticiones HTTP
pip install requests
Paso 2 — Crear la herramienta de clima
Agrega esto a tu archivo agente_intermedio.py, antes de la lista de tools:
import requests
from langchain_core.tools import tool
@tool
def obtener_clima(ciudad: str) -> str:
"""
Obtiene el clima actual de una ciudad.
Usa esta herramienta cuando el usuario pregunte por el clima o temperatura.
"""
# Primero obtenemos las coordenadas de la ciudad
geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={ciudad}&count=1&language=es"
geo_response = requests.get(geo_url).json()
if not geo_response.get("results"):
return f"No encontré la ciudad: {ciudad}"
lat = geo_response["results"][0]["latitude"]
lon = geo_response["results"][0]["longitude"]
nombre = geo_response["results"][0]["name"]
# Luego consultamos el clima
clima_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t_weather=true"
clima_response = requests.get(clima_url).json()
temp = clima_response["current_weather"]["temperature"]
viento = clima_response["current_weather"]["windspeed"]
return f"En {nombre}: {temp}°C, viento a {viento} km/h"
Paso 3 — Agregar la herramienta al agente
Actualiza la lista de herramientas para incluir ambas:
# Ahora el agente tiene búsqueda web Y consulta de clima
tools = [search_tool, obtener_clima]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
Paso 4 — Probar con una pregunta de clima
result = app.invoke({
"messages": [HumanMessage(content="¿Qué temperatura hace ahora en Bogotá?")]
})
print(result["messages"][-1].content)
✅ El agente decide automáticamente qué herramienta usar según la pregunta.
Módulo 3: Flujos Condicionales Avanzados
Hasta ahora el agente decide entre "usar herramienta" o "terminar". Ahora vamos a crear un flujo donde el agente puede tomar múltiples caminos según la intención del usuario.
Paso 1 — Crear un nuevo archivoagente_rutas.py
from dotenv import load_dotenv
from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
load_dotenv()
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
categoria: str # Nueva variable de estado
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
Paso 2 — Crear el nodo clasificador
def clasificar_pregunta(state: AgentState):
"""
Este nodo analiza la pregunta y decide a qué especialista enviarla.
"""
messages = state["messages"]
ultima_pregunta = messages[-1].content
prompt = f"""
Analiza esta pregunta y responde SOLO con una de estas palabras:
- TECNICA (si es sobre programación, tecnología o IA)
- NEGOCIO (si es sobre marketing, ventas o estrategia)
- GENERAL (cualquier otra cosa)
Pregunta: {ultima_pregunta}
Respuesta:"""
response = llm.invoke([HumanMessage(content=prompt)])
categoria = response.content.strip().upper()
return {"categoria": categoria}
Paso 3 — Crear los nodos especialistas
def experto_tecnico(state: AgentState):
system = SystemMessage(content="Eres un experto en programación e inteligencia artificial. Responde de forma técnica y precisa.")
response = llm.invoke([system] + state["messages"])
return {"messages": [response]}
def experto_negocio(state: AgentState):
system = SystemMessage(content="Eres un experto en marketing digital y estrategia de negocios. Responde con enfoque comercial.")
response = llm.invoke([system] + state["messages"])
return {"messages": [response]}
def experto_general(state: AgentState):
system = SystemMessage(content="Eres un asistente general amigable y útil.")
response = llm.invoke([system] + state["messages"])
return {"messages": [response]}
Paso 4 — Definir el enrutador condicional
def enrutar(state: AgentState) -> Literal["tecnico", "negocio", "general"]:
categoria = state.get("categoria", "GENERAL")
if "TECNICA" in categoria:
return "tecnico"
elif "NEGOCIO" in categoria:
return "negocio"
else:
return "general"
Paso 5 — Construir el gráfico con múltiples rutas
workflow = StateGraph(AgentState)
# Agregamos todos los nodos
workflow.add_node("clasificador", clasificar_pregunta)
workflow.add_node("tecnico", experto_tecnico)
workflow.add_node("negocio", experto_negocio)
workflow.add_node("general", experto_general)
# El punto de entrada es el clasificador
workflow.set_entry_point("clasificador")
# Después del clasificador, enruta según la categoría
workflow.add_conditional_edges(
"clasificador",
enrutar,
{
"tecnico": "tecnico",
"negocio": "negocio",
"general": "general"
}
)
# Todos los expertos terminan el flujo
workflow.add_edge("tecnico", END)
workflow.add_edge("negocio", END)
workflow.add_edge("general", END)
app = workflow.compile()
# Prueba con diferentes tipos de preguntas
preguntas = [
"¿Cómo funciona un transformer en deep learning?",
"¿Cómo puedo aumentar mis ventas en Instagram?",
"¿Cuál es la capital de Francia?"
]
for pregunta in preguntas:
print(f"\nPregunta: {pregunta}")
result = app.invoke({"messages": [HumanMessage(content=pregunta)]})
print(f"Respuesta: {result['messages'][-1].content[:200]}...")
✅ Ahora tienes un agente que enruta preguntas a diferentes expertos automáticamente.
Módulo 4: RAG — El Agente Lee tus Propios Documentos
RAG (Retrieval-Augmented Generation) permite que tu agente lea y responda preguntas sobre tus propios archivos (PDF, documentos de texto, etc.).
Paso 1 — Instalar las dependencias de RAG
pip install langchain-community chromadb pypdf sentence-transformers
Paso 2 — Prepara tu documento
Crea una carpeta de llamada documentosdentro de tu proyecto y coloca ahí cualquier archivo PDF que quieras que tu agente pueda leer. Para este ejemplo, cree un archivo documentos/empresa.txtcon este contenido de prueba:
Empresa: Ciberclicks
Fundada en: 2020
Servicios: Marketing Digital, Cursos de IA, Desarrollo Web
Mercados: Colombia, México, Estados Unidos
Misión: Democratizar el marketing digital en LATAM
Paso 3 — Crear el sistema RAG
Crea un archivo nuevo llamado agente_rag.py:
from dotenv import load_dotenv
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
load_dotenv()
# --- PASO A: Cargar y procesar el documento ---
print("Cargando documentos...")
loader = TextLoader("documentos/empresa.txt", encoding="utf-8")
documentos = loader.load()
# Dividimos el documento en fragmentos pequeños
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
fragmentos = splitter.split_documents(documentos)
# Creamos la base de datos vectorial en memoria
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(fragmentos, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
print("Documentos cargados correctamente ✅")
# --- PASO B: Crear la herramienta de búsqueda en documentos ---
@tool
def buscar_en_documentos(pregunta: str) -> str:
"""
Busca información en los documentos internos de la empresa.
Usa esta herramienta cuando pregunten sobre la empresa, sus servicios o datos internos.
"""
resultados = retriever.invoke(pregunta)
if not resultados:
return "No encontré información relevante en los documentos."
contenido = "\n\n".join([doc.page_content for doc in resultados])
return f"Información encontrada:\n{contenido}"
# --- PASO C: Construir el agente RAG ---
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [buscar_en_documentos]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
def agent_node(state: AgentState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_use_tools(state: AgentState):
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return END
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_use_tools, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
app = workflow.compile()
# Probamos el agente RAG
result = app.invoke({
"messages": [HumanMessage(content="¿En qué mercados opera Ciberclicks?")]
})
print(result["messages"][-1].content)
✅ Tu agente ahora puede leer y responder preguntas sobre tus propios documentos.
Módulo 5: Desplegar tu Agente como una API con FastAPI
Hasta ahora ejecutas tu agente desde la terminal. Ahora vamos a convertirlo en una API web para que cualquier aplicación pueda usar.
Paso 1 — Instalar FastAPI y Uvicorn
pip install fastapi uvicorn
Paso 2 — Crear el archivo de la API
Crea un archivo nuevo llamado api.py:
from dotenv import load_dotenv
from typing import TypedDict, Annotated
import operator
from fastapi import FastAPI
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_community.tools.tavily_search import TavilySearchResults
load_dotenv()
# --- Construimos el agente (igual que antes) ---
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
search_tool = TavilySearchResults(max_results=3)
tools = [search_tool]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
def agent_node(state: AgentState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_search(state: AgentState):
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return END
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_search, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
agente = workflow.compile()
# --- Creamos la API con FastAPI ---
app = FastAPI(title="Mi Agente de IA", version="1.0")
# Modelo de entrada
class Pregunta(BaseModel):
mensaje: str
session_id: str = "default"
# Modelo de salida
class Respuesta(BaseModel):
respuesta: str
session_id: str
@app.get("/")
def inicio():
return {"estado": "API del Agente de IA funcionando ✅"}
@app.post("/chat", response_model=Respuesta)
def chat(pregunta: Pregunta):
result = agente.invoke({
"messages": [HumanMessage(content=pregunta.mensaje)]
})
respuesta = result["messages"][-1].content
return Respuesta(respuesta=respuesta, session_id=pregunta.session_id)
Paso 3 — Ejecutar la API
En tu terminal, con el entorno virtual activado, ejecuta:
uvicorn api:app --reload --port 8000
Verás algo como:
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process
INFO: Application startup complete.
Paso 4 — Probar la API
Abre tu navegador y ve a http://127.0.0.1:8000/docs— verás la documentación interactiva de tu API generada automáticamente por FastAPI.
También puedes probarla desde la terminal con:
curl -X POST "http://127.0.0.1:8000/chat" \
-H "Content-Type: application/json" \
-d '{"mensaje": "¿Cuáles son las noticias de IA hoy?", "session_id": "usuario-001"}'
✅ Tu agente ahora es una lista de API para integrarse con cualquier aplicación web, móvil o sistema externo.
Resumen del Nivel Intermedio
- ✅ Agente con búsqueda real en internet (Tavily)
- ✅ Conexión a APIs externas con herramientas personalizadas
- ✅ Flujos condicionales con múltiples rutas y expertos
- ✅ RAG: el agente lee tus propios documentos
- ✅ Despliegue como API REST con FastAPI
Próximos Pasos — Nivel Avanzado
- Agentes multi-agente: varios agentes colaborando entre sí
- Memoria persistente con bases de datos (PostgreSQL, Redis)
- Despliegue en la nube (AWS, Google Cloud, Railway)
- Agentes con voz: entrada y salida de audio
- Monitoreo y observabilidad con LangSmith
- Seguridad y autenticación en APIs de agentes
¡Excelente trabajo llegando hasta aquí! Cada módulo que completas te acerca más a construir agentes de IA de nivel profesional. 🚀
© 2026 Ciberclicks. Todos los derechos reservados. Prohibida su reproducción sin autorización.
0 comentarios