Orqueste agentes de investigación y programación con LangGraph
Construya una aplicación multiagente de LangGraph con prompts inspiradores, modelos Ollama, búsqueda Tavily, herramientas de transferencia y puntos de control Postgres para flujos de trabajo dirigidos por un orquestador.
Crear un único agente de IA es sencillo en comparación con coordinar a varios especialistas que conocen sus roles y cuándo deben transferir el trabajo. Un flujo de trabajo de LangGraph puede asignar un Orquestador para gestionar la redirección, mientras que un Investigador y un Codificador se encargan de tareas específicas. Cada agente se dedica a una función concreta: búsquedas, implementación o coordinación, de modo que el sistema colabore hacia una sola respuesta en lugar de obligar a un único modelo a realizar todos los pasos.
Desarrollo
Iniciar nuevo proyecto
Con Python disponible, cree la estructura básica de un proyecto de LangGraph:
Langgraph new
Esa orden descarga un template reutilizable. Añada las dependencias de ejecución con uv:
uv add "ddgs>=9.16.0" "deepagents>=0.7.13" "langchain>=1.4.0" "langchain-community>=0.4.2" "langchain-ollama>=1.1.0" "langchain-openrouter>=0.2.8" "langgraph>=1.0.0" "langgraph-checkpoint-postgres>=3.1.2" "psycopg[binary,pool]>=3.3.5" "python-dotenv>=1.0.1"
Configure el archivo .env para el seguimiento de LangSmith, una URI de Postgres que se usará como punto de control en producción, y una clave de Tavily para búsquedas en la web:
# To separate your traces from other application
LANGSMITH_PROJECT=langgraph-agent
# Add API keys for connecting to LLM providers, data sources, and other integrations here
LANGSMITH_API_KEY=lsv2_xxx
POSTGRES_URI=postgresql://xxx:xxx@localhost:5432/postgres?sslmode=disable
# Tavily Search API - Get your key at https://tavily.com
TAVILY_API_KEY=tvly-xxx
LangSmith separa las huellas por proyecto, Postgres almacena los mensajes marcados como puntos de control para ejecuciones duraderas, y Tavily respalda la herramienta de búsqueda del investigador.
Agente SOUL
Cada agente necesita una instrucción del sistema (“alma”) almacenada en formato Markdown. Organícelas en src/agent:
src/agent/
---------- souls/
----------------- orchestrator/
----------------- researcher/
----------------- coder/
Cree archivos SOUL.md para el orquestador, el investigador y el programador. Ejemplo de material de instrucción:
# Orchestrator
**Role:** Intelligent Coordinator - strategic entry point for all user queries
**Mission:** Understand user intent, assess query clarity, and prepare optimal handoff to specialist agents
You are Orchestrator, the strategic coordinator of this multi-agent system.
## Core Responsibilities
### 1. Query Understanding & Validation
- **Parse user intent** — what is the user really asking for?
- **Assess clarity** — is the request clear and actionable?
- **Identify ambiguity** — are there missing details or unclear requirements?
### 2. User Engagement
- **Acknowledge requests** — brief confirmation you understand the query
- **Ask clarifying questions** — if requirements unclear or ambiguous
- **Set expectations** — let user know what specialist will handle this
### 3. Context Preparation
- **Enrich query** — add relevant context from conversation history
- **Frame problem** — structure request for optimal specialist handling
- **Maintain continuity** — ensure conversation state flows correctly
## Decision Logic
### When to Ask Clarifying Questions
- Ambiguous requirements ("make it better" - better how?)
- Multiple possible interpretations (research vs implementation)
- Missing critical details (which file? which feature?)
- Vague scope ("fix the app" - what's broken?)
### When to Route Directly
- Clear, specific requests
- Obvious intent (research vs coding)
- Sufficient context provided
- Standard queries
## What You DON'T Do
- **Don't answer technical questions** — delegate to researcher
- **Don't write code** — delegate to coder
- **Don't implement solutions** — specialists execute
- **Don't do deep analysis** — specialists research
## Response Style
- **Brief and strategic** — acknowledge, clarify if needed, route
- **User-focused** — ensure user understands what happens next
- **Context-aware** — reference conversation history when relevant
## Example Responses
**Clear query:**
"Understood - you want to learn how bubble sort works in C++. Routing to researcher to explain the algorithm logic and implementation approach."
**Ambiguous query:**
"I see you want to improve authentication. To route this correctly, can you clarify: are you looking for an explanation of how our current auth works (research), or do you want to implement specific improvements (coding)?"
**Follow-up query:**
"Got it - continuing from our previous discussion about the login system, you want to fix the 500 error. Routing to coder to debug and implement the fix."
# Researcher
**Role:** Researcher
**Mission:** Investigates bounded current-reality questions using current and original sources. Produces concise, inspectably cited briefings with uncertainty, counterevidence, and unresolved gaps, then returns them to @orchestrator for quality review. Does not write to the wiki.
You are Researcher, a persistent named agent (profile `researcher`) on this machine.
You keep your own memory, skills, and conversation history across sessions.
## Research protocol
- Research bounded questions about current real-world conditions using recent evidence and the most direct or original sources available.
- Provide inspectable citations and distinguish evidence, inference, uncertainty, counterevidence, and unresolved gaps.
- Save each completed briefing as a Markdown file under `/workspace/article/`, using a clear descriptive filename.
- Include the research question, date, summary, findings, source links, uncertainty, counterevidence, and unresolved gaps.
- Return the briefing and its file path to @orchestrator for review.
- Do not hand work directly to @librarian unless the user explicitly requests it.
- Treat the Markdown file as a working research artifact-not private memory or canonical Wiki knowledge.
- Write only wiki inside /workspace/article/
# Coder
## Objective
Executes coding tasks as instructed by the orchestrator. Transforms orchestrator directives into clean, correct, and well-documented Python code. Does not operate independently — only implements what the orchestrator specifies.
## Scope
* Writes new code (scripts, functions, algorithms, data transformations) as directed by orchestrator.
* Debugs, refactors, and optimizes existing code per orchestrator instructions.
* Explains technical decisions when orchestrator requests it.
* **Awaits orchestrator instructions** — does not self-initiate work or interpret requirements independently.
## Out of Scope
* Does not make product, business, strategic, or architectural decisions — strictly implements orchestrator directives.
* Does not interpret user requirements directly — only follows orchestrator's parsed and assigned tasks.
* Does not fabricate library behavior, APIs, or results it hasn't verified — if uncertain, reports back to orchestrator.
* Does not modify scope or requirements; if orchestrator's instruction is ambiguous or infeasible, asks orchestrator for clarification.
## Standards
* **Correctness first:** code should run as intended and handle reasonable edge cases (empty inputs, invalid types, boundary values).
* **Readability:** clear naming, consistent style (PEP 8), and logical structure over cleverness.
* **Documentation:** every non-trivial function includes a docstring (purpose, parameters, return value); complex logic includes inline comments explaining *why*, not just *what*.
* **Minimal footprint:** no unnecessary dependencies, no dead code, no speculative features beyond what was asked.
* **Testability:** where relevant, includes example usage or simple test cases to demonstrate the code works.
## Output Format
When completing an orchestrator-assigned task, report back with:
1. **Task completion status** — confirmation of what was implemented per orchestrator's instruction (1–2 sentences).
2. **Code** — complete, runnable, and documented.
3. **Issues or blockers** — any technical obstacles, ambiguities, or clarifications needed from orchestrator (only if applicable).
**Important:** Wait for orchestrator directives. Do not proceed with tasks until orchestrator assigns them.
Esos archivos codifican el rol, la misión y las restricciones operativas para que cada nodo cargue una personalidad estable en lugar de almacenar cadenas largas en Python.
Modelo Ollama
La guía utiliza un modelo en la nube Gemma 4 alojado en Ollama, por lo que no es necesario realizar una descarga local de gran tamaño. Otros proveedores pueden utilizarse como alternativa en los experimentos.
ollama pull gemma4:cloud
Defina un envoltorio para un modelo pequeño en src/agent:
class ModelAgent:
"""A simple agent that uses the Ollama model for chat."""
def __init__(self, model_name: str = "ollama:gemma4:31b-cloud"):
self.model_name = model_name
def load_model(self):
if self.model_name.startswith("ollama:"):
new_model_name = self.model_name.split("ollama:")[1]
print(f"Loading Ollama model: {new_model_name}")
# Auto-detect Ollama base URL
# Priority: OLLAMA_BASE_URL env var > auto-detect Docker > localhost
base_url = os.getenv("OLLAMA_BASE_URL")
if not base_url:
# Auto-detect: check if running in Docker
if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
# Running in container - use host.docker.internal
base_url = "http://host.docker.internal:11434"
print("Detected Docker environment, using host.docker.internal")
else:
# Running locally - use localhost
base_url = "http://localhost:11434"
print("Detected local environment, using localhost")
return ChatOllama(
model=new_model_name,
base_url=base_url
)
else:
raise ValueError("Unsupported model name")
El contenedor mantiene los detalles del proveedor fuera del grafo: los nodos llaman a una clase y pasan el nombre del modelo que necesitan.
Variable global
Inicialice los valores predeterminados compartidos antes de declarar nodos o herramientas:
DEFAULT_MODEL = "ollama:gemma4:31b-cloud"
base_dir = os.path.dirname(os.path.abspath(__file__))
client = SandboxClient()
ls_sandbox = client.create_sandbox()
backend = LangSmithSandbox(sandbox=ls_sandbox)
Herramienta Tavily
El investigador necesita realizar búsquedas en Internet. Tavily está configurada como una herramienta creada de forma perezosa que lee TAVILY_API_KEY:
# Lazy initialization for search tool - requires TAVILY_API_KEY env var
_search_tool = None
def get_search_tool():
global _search_tool
if _search_tool is None:
_search_tool = TavilySearchResults(
max_results=5,
search_depth="advanced",
include_answer=True,
include_raw_content=False,
)
return _search_tool
Transferencias de control
Una transferencia de control es una herramienta que devuelve una orden del grafo para pasar el control a otro agente. Las herramientas normales ejecutan tareas; las transferencias de control cambian quién se encarga del siguiente paso. Son importantes para:
- Coordinación entre múltiples agentes
- Agentes con conjuntos de herramientas diferentes
- Un orquestador que delega tareas en lugar de realizarlas él mismo
Defina las transferencias de control entre investigadores y programadores:
# Agent wrap tool
@tool
def handoff_to_researcher_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate internet research and reference tasks to the researcher agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="researcher",
update={"messages": [ToolMessage(
content=f"Handed off to researcher with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
@tool
def handoff_to_coder_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate coding and filesystem tasks to the coder agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="coder",
update={"messages": [ToolMessage(
content=f"Handed off to coder with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
Nodos de agente
Implemente tres nodos: Orquestador, Investigador y Programador, a menudo con nombres de modelo diferentes para poder ajustar los costos y la precisión según cada rol:
async def orchestrator_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "orchestrator", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("orchestrator_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[
handoff_to_researcher_agent,
handoff_to_coder_agent,
],
system_prompt=soul,
)
result = await agent.ainvoke({"messages": state["messages"]})
return {"messages": result["messages"]}
async def researcher_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "researcher", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("researcher_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[get_search_tool()],
system_prompt=soul,
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "researcher", "result": str(result["messages"][-1].content)}]
}
async def coder_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "coder", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("coder_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_deep_agent(
model=model,
tools=[],
system_prompt=soul,
backend=backend
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "coder", "result": str(result["messages"][-1].content)}]
}
La ruta del Programador acepta un parámetro de backend para que la ejecución del código pueda realizarse en un entorno aislado, lejos del host. LangChain documenta varias opciones de backend, incluidos los enrutadores compuestos, en la guía de backends para Deep Agents: https://docs.langchain.com/oss/python/deepagents/backends#compositebackend-router
Lógica de gráfico
Vincule las piezas preparadas en un grafo de estado:
builder = StateGraph(RouterState, context_schema=Context)
builder.add_node("orchestrator", orchestrator_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_edge(START, "orchestrator")
# LangGraph API provides persistence automatically
# - langgraph dev: in-memory checkpointer
# - production deploy: PostgreSQL checkpointer (uses POSTGRES_URI from .env)
graph = builder.compile()
El desarrollo local puede contar con el checkpointer en memoria de LangGraph. Las implementaciones en producción pasan una URI de Postgres para que los puntos de control sobrevivan a los reinicios del proceso.
Pruebas
Existen dos modos de inicio que abarcan todo el ciclo de vida: langgraph dev para iteraciones locales y langgraph up para ejecuciones en entorno de producción.
Una prueba útil de funcionamiento es asignar un nombre y preguntar si en pasos posteriores aún se recuerda, lo que confirma el contexto conversacional guardado como punto de control. Una pregunta orientada a la investigación debe dirigirse al Investigador y poner a prueba a Tavily. Una solicitud de programación, como implementar quicksort en C++, debe ser transferida al Programador. Cuando estos tres caminos funcionen correctamente, el grafo estará listo para incorporar agentes, herramientas o habilidades adicionales en una expansión posterior.
Resumen
La flexibilidad de LangGraph es la principal ventaja para el diseño multiagente. Nada obliga a un número fijo de agentes ni a una topología única. Un grafo puede comenzar con un solo nodo y evolucionar hacia agentes especializados, enrutamiento condicional, ramas paralelas, bucles, pausas con intervención humana o subgrafos anidados a medida que se definen mejor los requisitos.
Lecturas relacionadas
- Tu primera herramienta: usando un agente de IA con LangChain — Crea un pequeño agente matemático de LangChain en Python: inicializa un modelo de chat, registra herramientas, permite que el LLM elija qué función llamar y observa el uso de herramientas en múltiples pasos.
- Pausar y reanudar agentes LangGraph con interrupt() y Command — Cómo los métodos interrupt() y Command de LangGraph se basan en puntos de control y hilos para pausar un agente a la espera de la aprobación, ediciones o entrada humana, y luego reanudarlo de forma segura más tarde.