Este artículo es una guía práctica para desarrolladores sobre cómo crear agentes de IA autónomos en Python. No repetiremos la teoría sobre qué son LangChain y LangGraph. En su lugar, nos centraremos en el código, la arquitectura y la resolución de problemas del mundo real.
Objetivo: Construir dos proyectos desde cero:
- Un Agente Clasificador: Un agente de múltiples pasos con estado gestionado, pero sin herramientas externas.
- Un Agente Asistente: Un agente completo con acceso al sistema de archivos y búsqueda web a través del protocolo MCP, construido sobre lógica cíclica.
Cubriremos las mejores prácticas: gestión de la configuración, selección de modelos y manejo de errores para crear sistemas robustos.
In Questo Articolo
Breve apunte sobre los conceptos: El Agente y el puente MCP
Antes de sumergirnos en el código, establezcamos dos conceptos:
- Agente de IA: Un programa construido en torno a un bucle de «razonamiento-acción». Recibe una tarea, utiliza un LLM para decidir qué hacer a continuación (p. ej., llamar a una herramienta), ejecuta la acción y repite el ciclo hasta que la tarea se completa.
- MCP (Model Context Protocol): Un estándar que actúa como un puente entre la lógica del agente y las herramientas externas. Permite que el agente trabaje con archivos, APIs o búsquedas de manera unificada, sin preocuparse por los detalles de su implementación.
Parte 1: Configurando un entorno robusto
Paso 1: Entorno virtual y dependencias
Crea y activa un entorno virtual. Luego, crea un archivo requirements.txt:
# Frameworks principales
langchain
langgraph
# Adaptadores de modelos
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # Para Ollama
# Herramientas y protocolos
langchain-mcp-adapters
mcp
ollama
# Utilidades
python-dotenv
tenacity # Para un manejo de errores robustoInstala las dependencias:
pip install -r requirements.txt```
#### Paso 2: Configuración de claves de API
Crea un archivo `.env` para almacenar tus claves:```
OPENAI_API_KEY="sk-..."
GOOGLE_API_KEY="AIzaSy..."
MISTRAL_API_KEY="..."
BRAVE_API_KEY="..." # Para la herramienta de búsqueda web vía MCPPaso 3: El patrón «Fábrica de Modelos»
Para cambiar de forma flexible entre modelos en la nube y locales sin modificar el código del agente, usaremos el patrón de fábrica (factory pattern).
# llm_factory.py
import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama
load_dotenv()
class ModelProvider(Enum):
OPENAI = "openai"
GEMINI = "gemini"
MISTRAL_API = "mistral_api"
OLLAMA = "ollama"
def get_llm(provider: ModelProvider, model_name: str = None):
"""Una fábrica para crear instancias de LLM."""
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or "gpt-4o-mini", temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or "gemini-1.5-flash", temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or "mistral-large-latest", temperature=0)
elif provider == ModelProvider.OLLAMA:
# Asegúrate de tener Ollama corriendo con el modelo requerido
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or "mistral", temperature=0)
raise ValueError(f"Proveedor de modelo desconocido: {provider}")
# Ejemplo de uso
if __name__ == "__main__":
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke("Explica el concepto de RAG en tres frases.")
print(response.content)Parte 2: Proyecto 1 — Agente de Clasificación de Vacantes
Este agente demuestra cómo usar LangGraph para crear un grafo lineal con estado gestionado. Recibirá una descripción de trabajo y la clasificará secuencialmente según tres parámetros.
Paso 1: Definiendo el Estado
El estado es la «memoria» de nuestro grafo, que se pasa de un nodo al siguiente.
# vacancy_classifier.py
from typing import TypedDict, Dict
class ClassificationState(TypedDict):
"""Estado para el agente clasificador."""
description: str # Texto fuente
job_type: str # Tipo de trabajo (proyecto/permanente)
category: str # Profesión
search_type: str # Objetivo (busca trabajo/ejecutor)
classification_log: list # Registro de depuraciónPaso 2: Implementando los Nodos del Grafo
Cada nodo es una función que toma el estado, realiza su parte del trabajo y devuelve el estado actualizado.
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider
class VacancyClassifierAgent:
def __init__(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name="gpt-4o-mini")
async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
"""Nodo 1: Determina el tipo de trabajo."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determina el tipo de trabajo. La respuesta debe ser 'por proyecto' o 'permanente'."),
("human", "Descripción del trabajo:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["job_type"] = result.content.strip()
state["classification_log"].append("Tipo de trabajo determinado.")
return state
async def _classify_category(self, state: ClassificationState) -> ClassificationState:
"""Nodo 2: Determina la categoría de la profesión."""
# Las categorías se pueden cargar desde un archivo o base de datos
categories = ["Desarrollador Python", "Diseñador", "Marketero", "Animador 3D"]
prompt = ChatPromptTemplate.from_messages([
("system", f"Elige la categoría más adecuada de la lista: {', '.join(categories)}."),
("human", "Descripción del trabajo:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["category"] = result.content.strip()
state["classification_log"].append("Categoría determinada.")
return state
async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
"""Nodo 3: Determina el objetivo de la búsqueda."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determina el objetivo del autor. La respuesta debe ser 'busca trabajo' o 'busca ejecutor'."),
("human", "Descripción del trabajo:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["search_type"] = result.content.strip()
state["classification_log"].append("Objetivo de búsqueda determinado.")
return statePaso 3: Ensamblando y Ejecutando el Grafo
Ensamblamos los nodos en un único flujo de trabajo.
# ... continuación de la clase VacancyClassifierAgent ...
from langgraph.graph import StateGraph, END
def build_graph(self):
"""Ensambla el grafo de estados."""
workflow = StateGraph(ClassificationState)
workflow.add_node("job_type_classifier", self._classify_job_type)
workflow.add_node("category_classifier", self._classify_category)
workflow.add_node("search_type_classifier", self._classify_search_type)
workflow.set_entry_point("job_type_classifier")
workflow.add_edge("job_type_classifier", "category_classifier")
workflow.add_edge("category_classifier", "search_type_classifier")
workflow.add_edge("search_type_classifier", END)
return workflow.compile()
async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()
description = "Buscamos un desarrollador Python experimentado para unirse a nuestro equipo a tiempo completo para trabajar en un proyecto fintech."
initial_state = ClassificationState(
description=description,
job_type="", category="", search_type="",
classification_log=[]
)
final_state = await graph.ainvoke(initial_state)
print("--- Resultado de la Clasificación ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(main())Parte 3: Proyecto 2 — Agente Asistente con Herramientas (MCP)
Este agente demuestra una lógica cíclica, donde puede llamar repetidamente a herramientas para resolver una tarea.
Paso 1: Gestión de la Configuración
Para agentes que interactúan con el mundo exterior, una configuración robusta es esencial.
# mcp_agent_config.py
from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider
@dataclass
class AgentConfig:
workdir: str = "./agent_workdir"
model_provider: ModelProvider = ModelProvider.OLLAMA
def __post_init__(self):
"""Validación post-inicialización."""
os.makedirs(self.workdir, exist_ok=True)Paso 2: Definiendo el Estado para el Diálogo
El estado ahora almacenará el historial de mensajes.
# mcp_agent.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]Paso 3: Implementando el Grafo Cíclico
El grafo constará de dos nodos principales y una arista condicional que crea el bucle de «razonamiento-acción».
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig
class MCPAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None
async def setup_tools(self):
"""Inicializa las herramientas vía MCP."""
mcp_config = {
"filesystem": {
"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
"transport": "stdio"
},
# Añade brave-search si tienes una BRAVE_API_KEY
}
mcp_client = MultiServerMCPClient(mcp_config)
self.tools = await mcp_client.get_tools()
self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])
# Vincula las herramientas al modelo
self.llm = self.llm.bind_tools(self.tools)
def _should_continue(self, state: AgentState):
"""Arista condicional: decide si llamar a una herramienta."""
last_message = state['messages'][-1]
if not last_message.tool_calls:
return "end"
return "continue"
def _call_model(self, state: AgentState):
"""Nodo 1: Llama al LLM para tomar una decisión."""
response = self.llm.invoke(state['messages'])
return {"messages": [response]}
def _call_tool(self, state: AgentState):
"""Nodo 2: Ejecuta la llamada a la herramienta."""
last_message = state['messages'][-1]
tool_call = last_message.tool_calls[0]
action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
response = self.tool_executor.invoke(action)
return {"messages": [response]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("agent", self._call_model)
workflow.add_node("action", self._call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
self._should_continue,
{"continue": "action", "end": END}
)
workflow.add_edge("action", "agent")
return workflow.compile()Paso 4: Ejecutando e Interactuando
# ... continuación de mcp_agent.py ...
import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
"""Ejecuta una tarea con manejo de errores."""
return await graph.ainvoke({"messages": [HumanMessage(content=task)]})
async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # o OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()
task = "Crea un archivo llamado 'hola.txt' en el directorio de trabajo y escribe '¡Hola, mundo!' en él."
result = await run_agent_task(graph, task)
print("\n--- Respuesta Final del Agente ---")
print(result['messages'][-1].content)
if __name__ == "__main__":
asyncio.run(main())Aquí, hemos añadido el decorador tenacity para mayor robustez: si la llamada al agente falla debido a un error de red temporal, se reintentará automáticamente.
Conclusión
Hemos construido dos tipos de agentes utilizando prácticas modernas:
- Un grafo lineal es excelente para tareas con una secuencia clara de pasos, como procesos ETL o análisis de múltiples etapas.
- Un grafo cíclico es la base para crear asistentes interactivos y agentes autónomos capaces de resolver problemas complejos con herramientas.
Los patrones arquitectónicos presentados —la fábrica de modelos, la gestión de la configuración, la separación de la lógica en nodos y el uso de grafos de estados— son los fundamentos para construir sistemas de IA escalables y robustos.