Projeto Intermediário: Sistema de Pesquisa Multi-Agente

Construa um sistema de pesquisa multi-agente pronto para produção com um agente supervisor coordenando agentes de pesquisa especializados para geração de relatórios abrangentes.

Construa um sistema de pesquisa multi-agente que coordena agentes especializados para pesquisar temas, analisar descobertas e gerar relatórios abrangentes. Este projeto reúne estado avançado, persistência, padrões multi-agente e streaming.

Arquitetura do 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

Passo 1: Definir Estado e Mensagens

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

Passo 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 diferentes otimizam o custo. Tarefas simples (pesquisa, revisão) usam gpt-4o-mini. Raciocínio complexo (planejamento, análise) usa gpt-4o.

Passo 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 Planejador

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 Pesquisador

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()}]}

Passo 4: Roteador e Construção do 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())

✅Sucesso

O padrão de laço do supervisor: START → supervisor → agente → supervisor → agente → ... → finalizador → END. Cada agente devolve o controle ao supervisor depois de concluir seu trabalho.

Passo 5: Executar o 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)

Passo 6: Adicionando Revisão Humana

Adicione uma etapa opcional de revisão 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
})

💡Dica

Adicionar revisão humana em pontos estratégicos (antes da finalização) pega erros enquanto mantém a maior parte do fluxo de trabalho autônoma.

Diagrama do Sistema

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