Proyecto Intermedio: Sistema de Investigación Multi-Agente

Construye un sistema de investigación multi-agente listo para producción con un agente supervisor coordinando agentes de investigación especializados para generación de informes completos.

Construye un sistema de investigación multiagente que coordina agentes especializados para investigar temas, analizar hallazgos y generar informes integrales. Este proyecto reúne estado avanzado, persistencia, patrones multiagente y transmisión.

Arquitectura del Sistema

User Query
    ↓
Supervisor Agent ──→ Planner Agent
    ↓                      ↓
Researcher Agent ←── Research Plan
    ↓
Analyst Agent ──→ Synthesizes findings
    ↓
Writer Agent ──→ Generates final report
    ↓
Reviewer Agent ──→ Quality check
    ↓
Final Report

Paso 1: Definir Estado y Mensajes

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any, Optional
from operator import add
from datetime import datetime
import json

# Structured agent message
class AgentMessage(TypedDict):
    sender: str
    recipient: str
    content: str
    message_type: str  # plan, research, analysis, draft, review, final
    timestamp: str

class ResearchState(TypedDict):
    query: str                                    # Original user query
    messages: Annotated[List[AgentMessage], add]  # Agent conversation log
    research_plan: str                            # Structured research plan
    research_findings: List[str]                  # Raw research data
    analysis: str                                 # Synthesized analysis
    draft_report: str                             # Written report
    review_feedback: str                          # Quality review
    final_report: str                             # Approved final report
    next_agent: str                               # Supervisor's decision
    errors: Annotated[List[str], add]             # Error log
    status: str                                   # Overall status

Paso 2: Inicializar LLMs

python

# Different models for different agents
planner_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
researcher_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
analyst_llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
reviewer_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
supervisor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

ℹ️Nota

Modelos distintos optimizan el costo. Las tareas simples (investigación, revisión) usan gpt-4o-mini. El razonamiento complejo (planificación, análisis) usa gpt-4o.

Paso 3: Definir Agentes

Agente Supervisor

python

def supervisor_agent(state: ResearchState) -> dict:
    context = "\n".join(
        f"{m['sender']} → {m['recipient']}: {m['content'][:200]}"
        for m in state["messages"][-5:]
    ) if state["messages"] else "No activity yet."

    prompt = f"""Research Query: {state['query']}

Current Status: {state.get('status', 'starting')}
Recent Activity:
{context}

Available agents and their order:
1. planner — Creates a research plan
2. researcher — Executes the research
3. analyst — Analyzes research findings
4. writer — Writes the report draft
5. reviewer — Reviews the draft for quality
6. complete — Task is finished

Which agent should work next? Respond with one word."""

    response = supervisor_llm.invoke(prompt)
    next_agent = response.content.strip().lower()

    if next_agent not in ["planner", "researcher", "analyst", "writer", "reviewer"]:
        next_agent = "complete"

    return {"next_agent": next_agent,
            "messages": [{"sender": "supervisor",
                          "recipient": next_agent,
                          "content": f"Proceed to {next_agent}",
                          "message_type": "instruction",
                          "timestamp": datetime.now().isoformat()}]}

Agente Planificador

python

def planner_agent(state: ResearchState) -> dict:
    prompt = f"""Create a detailed research plan for: {state['query']}

The plan should include:
1. Key topics to investigate
2. Search queries to execute
3. Data sources to consult
4. Analysis methodology

Format as a structured plan with numbered sections."""

    response = planner_llm.invoke(prompt)
    plan = response.content

    return {"research_plan": plan,
            "status": "planned",
            "messages": [{"sender": "planner",
                          "recipient": "supervisor",
                          "content": plan,
                          "message_type": "plan",
                          "timestamp": datetime.now().isoformat()}]}

Agente Investigador

python

def researcher_agent(state: ResearchState) -> dict:
    plan = state.get("research_plan", state["query"])

    prompt = f"""Research Plan: {plan}

Conduct thorough research on each topic. For each area:
1. Find key facts and statistics
2. Note different perspectives
3. Identify authoritative sources
4. Highlight recent developments

Provide detailed findings for each topic."""

    response = researcher_llm.invoke(prompt)

    return {"research_findings": state.get("research_findings", []) + [response.content],
            "status": "researched",
            "messages": [{"sender": "researcher",
                          "recipient": "supervisor",
                          "content": response.content[:500],
                          "message_type": "research",
                          "timestamp": datetime.now().isoformat()}]}

Agente Analista

python

def analyst_agent(state: ResearchState) -> dict:
    findings = "\n\n".join(state.get("research_findings", ["No findings"]))

    prompt = f"""Research Findings:
{findings}

Analyze these findings and provide:
1. Key insights and patterns
2. Contradictions or debates
3. Gaps in the research
4. Implications and conclusions
5. Recommendations for the report

Be critical and thorough."""

    response = analyst_llm.invoke(prompt)

    return {"analysis": response.content,
            "status": "analyzed",
            "messages": [{"sender": "analyst",
                          "recipient": "supervisor",
                          "content": response.content[:500],
                          "message_type": "analysis",
                          "timestamp": datetime.now().isoformat()}]}

Agente Escritor

python

def writer_agent(state: ResearchState) -> dict:
    report_prompt = f"""Write a comprehensive research report.

Query: {state['query']}
Analysis: {state['analysis']}
Research Findings: {json.dumps(state.get('research_findings', []))}

The report should include:
1. Executive Summary
2. Introduction
3. Methodology
4. Key Findings (with data and evidence)
5. Analysis and Discussion
6. Conclusions
7. Recommendations
8. References

Write in a professional, academic style. Use markdown formatting."""

    response = writer_llm.invoke(report_prompt)

    return {"draft_report": response.content,
            "status": "drafted",
            "messages": [{"sender": "writer",
                          "recipient": "supervisor",
                          "content": "Draft report completed",
                          "message_type": "draft",
                          "timestamp": datetime.now().isoformat()}]}

Agente Revisor

python

def reviewer_agent(state: ResearchState) -> dict:
    review_prompt = f"""Review this research report for quality:

Report Draft:
{state['draft_report']}

Original Query:
{state['query']}

Check for:
1. Accuracy of claims and data
2. Completeness — does it answer the query?
3. Clarity and organization
4. Grammar and style
5. Missing sections or information
6. Bias or unsupported statements

Provide specific, actionable feedback."""

    response = reviewer_llm.invoke(review_prompt)

    return {"review_feedback": response.content,
            "status": "reviewed",
            "messages": [{"sender": "reviewer",
                          "recipient": "supervisor",
                          "content": response.content[:500],
                          "message_type": "review",
                          "timestamp": datetime.now().isoformat()}]}

Agente Finalizador

python

def finalizer_agent(state: ResearchState) -> dict:
    final_prompt = f"""Based on the draft report and review feedback, produce the final report.

Draft: {state['draft_report']}
Review Feedback: {state['review_feedback']}

Incorporate the feedback and polish the report. Ensure it is comprehensive,
well-structured, and directly addresses the query: {state['query']}"""

    response = writer_llm.invoke(final_prompt)

    return {"final_report": response.content,
            "status": "completed",
            "messages": [{"sender": "finalizer",
                          "recipient": "supervisor",
                          "content": "Final report ready",
                          "message_type": "final",
                          "timestamp": datetime.now().isoformat()}]}

Paso 4: Enrutador y Construcción del Grafo

python

def supervisor_router(state: ResearchState) -> str:
    if state.get("status") == "completed":
        return "finalize"
    return state["next_agent"]

builder = StateGraph(ResearchState)

# Add nodes
builder.add_node("supervisor", supervisor_agent)
builder.add_node("planner", planner_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("analyst", analyst_agent)
builder.add_node("writer", writer_agent)
builder.add_node("reviewer", reviewer_agent)
builder.add_node("finalizer", finalizer_agent)

# Build edges
builder.add_edge(START, "supervisor")

# Supervisor routes to any agent
builder.add_conditional_edges(
    "supervisor",
    supervisor_router,
    {
        "planner": "planner",
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "reviewer": "reviewer",
        "finalize": "finalizer"
    }
)

# All agents return to supervisor
builder.add_edge("planner", "supervisor")
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("writer", "supervisor")
builder.add_edge("reviewer", "supervisor")
builder.add_edge("finalizer", END)

# Compile with persistence
app = builder.compile(checkpointer=MemorySaver())

✅Éxito

El patrón de bucle del supervisor: START → supervisor → agente → supervisor → agente → ... → finalizador → END. Cada agente devuelve el control al supervisor después de terminar su trabajo.

Paso 5: Ejecutar el Sistema

python

def research_topic(query: str, thread_id: str = "research-1") -> str:
    config = {"configurable": {"thread_id": thread_id}}

    # Stream the execution
    for event in app.stream(
        {
            "query": query,
            "messages": [],
            "research_plan": "",
            "research_findings": [],
            "analysis": "",
            "draft_report": "",
            "review_feedback": "",
            "final_report": "",
            "next_agent": "",
            "errors": [],
            "status": "starting"
        },
        config,
        stream_mode="updates"
    ):
        for node, update in event.items():
            if node == "__end__":
                continue
            if "status" in update:
                print(f"[{update['status'].upper()}] {node} completed")

    # Get the final result
    final = app.get_state(config)
    return final.values.get("final_report", "No report generated.")


# Run the research system
report = research_topic(
    "What are the environmental impacts of quantum computing?",
    "research-quantum-1"
)
print(report)

Paso 6: Agregando Revisión Humana

Añade un paso opcional de revisión humana:

python

def human_review_node(state: ResearchState) -> dict:
    response = interrupt({
        "draft": state["draft_report"],
        "prompt": "Review the draft. Approve or request changes."
    })

    if response.get("approved"):
        return {"status": "approved"}
    return {"status": "rejected",
            "review_feedback": response.get("feedback", "Revise")}

def review_router(state: ResearchState) -> str:
    if state.get("status") == "approved":
        return "finalize"
    return "revise"

# Add human review to graph
builder.add_node("human_review", human_review_node)
builder.add_edge("reviewer", "human_review")
builder.add_conditional_edges("human_review", review_router, {
    "finalize": "finalizer",
    "revise": "writer"  # Send back to writer with feedback
})

💡Consejo

Añadir revisión humana en puntos estratégicos (antes de la finalización) detecta errores mientras se mantiene la mayor parte del flujo de trabajo autónoma.

Diagrama del Sistema

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon