Agentes Hierárquicos

Implemente arquiteturas multi-agente hierárquicas com agentes gerentes, sub-agentes, subgrafos e passagem de estado entre níveis hierárquicos.

Agentes hierárquicos organizam o trabalho em níveis: um agente gestor delega tarefas a subagentes, que podem ser eles próprios grafos com nós internos. Isso permite sistemas de agentes escalonáveis, mantíveis e encapsulados.

Por que Hierárquico?

Sistemas multi-agente achatados têm limitações:

Complexidade de estado: Todos os agentes compartilham um único dicionário de estado, que cresce de forma ingovernável
Custo de coordenação: Cada agente vê todas as mensagens, mesmo as irrelevantes
Encapsulamento: Os detalhes internos dos agentes vazam para o estado compartilhado
Escalabilidade: Adicionar agentes aumenta a complexidade linearmente

Arquiteturas hierárquicas resolvem isso com encapsulamento e delegação:

Manager Agent
├── Research Sub-Agent (internal graph)
│   ├── plan_research
│   ├── execute_search
│   └── summarize
├── Write Sub-Agent (internal graph)
│   ├── draft
│   ├── review
│   └── finalize
└── Quality Sub-Agent (internal graph)
    ├── check_facts
    └── validate_output

Subgrafos

Subgrafos são grafos usados como nós dentro de um grafo pai. Eles têm seu próprio estado, nós e arestas: totalmente encapsulados.

python

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# === Subgraph: Research ===
class ResearchState(TypedDict):
    topic: str
    findings: str
    status: str

def plan_research(state: ResearchState) -> dict:
    plan = f"Research plan for: {state['topic']}"
    return {"status": "planned"}

def execute_research(state: ResearchState) -> dict:
    findings = f"Findings about {state['topic']}: ..."
    return {"findings": findings, "status": "completed"}

research_builder = StateGraph(ResearchState)
research_builder.add_node("plan", plan_research)
research_builder.add_node("execute", execute_research)
research_builder.add_edge(START, "plan")
research_builder.add_edge("plan", "execute")
research_builder.add_edge("execute", END)
research_subgraph = research_builder.compile()

ℹ️Nota

O subgrafo de pesquisa tem seu próprio estado (ResearchState). Ele não conhece o estado do grafo pai, e o pai não conhece os detalhes internos da pesquisa.

Grafo Pai Usando Subgrafo

python

class ManagerState(TypedDict):
    query: str
    research_findings: str
    final_report: str
    status: str

def manager_router(state: ManagerState) -> str:
    if not state.get("research_findings"):
        return "research_subgraph"
    return "finalize"

def finalize_report(state: ManagerState) -> dict:
    report = f"Report based on: {state['research_findings']}"
    return {"final_report": report, "status": "completed"}

# Parent graph
builder = StateGraph(ManagerState)
builder.add_node("research_subgraph", research_subgraph)  # Subgraph as a node
builder.add_node("finalize", finalize_report)

builder.add_edge(START, "research_subgraph")
builder.add_edge("research_subgraph", "finalize")
builder.add_edge("finalize", END)

app = builder.compile()

✅Sucesso

Um subgrafo é adicionado como um nó, exatamente como uma função comum. O LangGraph trata o mapeamento de estado entre pai e subgrafo automaticamente.

Mapeamento de Estado entre Pai e Subgrafo

O pai passa um subconjunto do seu estado para o subgrafo. O subgrafo retorna o seu estado, que é mesclado de volta.

python

class ParentState(TypedDict):
    query: str
    research: str
    output: str

class ResearchState(TypedDict):
    query: str          # Mapped from parent
    findings: str       # Mapped back to parent
    depth: str

# When invoke is called on the subgraph from the parent node:
def research_node(parent_state: ParentState) -> dict:
    # Map parent state to subgraph state
    sub_input = {
        "query": parent_state["query"],
        "findings": "",
        "depth": "deep"
    }

    # Invoke the subgraph
    sub_result = research_subgraph.invoke(sub_input)

    # Map subgraph result back to parent state
    return {"research": sub_result["findings"]}

Hierarquia de Múltiplos Níveis

python

# Level 3: Sub-subgraph
class FactCheckState(TypedDict):
    claim: str
    verified: bool

def verify_claim(state: FactCheckState) -> dict:
    verified = "verified" in state["claim"].lower()
    return {"verified": verified}

fact_builder = StateGraph(FactCheckState)
fact_builder.add_node("verify", verify_claim)
fact_builder.add_edge(START, "verify")
fact_builder.add_edge("verify", END)
fact_subgraph = fact_builder.compile()

# Level 2: Subgraph uses fact_subgraph
class ResearchState(TypedDict):
    topic: str
    claims: list
    verified_claims: list
    summary: str

def check_claims(state: ResearchState) -> dict:
    verified = []
    for claim in state["claims"]:
        result = fact_subgraph.invoke({"claim": claim, "verified": False})
        verified.append({"claim": claim, "verified": result["verified"]})
    return {"verified_claims": verified}

research_builder = StateGraph(ResearchState)
research_builder.add_node("check", check_claims)
research_builder.add_edge(START, "check")
research_builder.add_edge("check", END)
research_graph = research_builder.compile()

# Level 1: Parent uses research_graph
builder = StateGraph(ParentState)
builder.add_node("research", research_graph)
builder.add_edge(START, "research")
builder.add_edge("research", END)
app = builder.compile()

📌Importante

Subgrafos em qualquer nível podem conter seus próprios subgrafos. Esse aninhamento permite hierarquias de agentes arbitrariamente complexas, mantendo o encapsulamento em cada nível.

Contexto Compartilhado entre Níveis

Use um objeto de contexto que percorra todos os níveis da hierarquia:

python

class GlobalContext(TypedDict):
    user_id: str
    session_id: str
    constraints: List[str]
    preferences: dict

class TaskState(TypedDict):
    context: GlobalContext  # Passed through all levels
    task: str
    result: str

# Each subgraph receives and passes the context
def research_subgraph_node(state: TaskState) -> dict:
    ctx = state["context"]
    result = perform_research(state["task"], ctx["constraints"])
    return {"result": result}  # Context flows through automatically?

⚠️Aviso

O mapeamento de estado entre pai e subgrafo deve ser explícito. O pai decide quais campos passar ao subgrafo e quais saídas do subgrafo ler.

Execução Paralela de Subgrafos

Execute vários subgrafos em paralelo:

python

def run_parallel_research(state: ManagerState) -> dict:
    topics = [state["query"], f"{state['query']} advanced"]

    results = []
    for topic in topics:
        result = research_subgraph.invoke({
            "topic": topic,
            "findings": "",
            "status": ""
        })
        results.append(result["findings"])

    return {"research_findings": "\n\n".join(results)}

Para paralelismo real, use fan-out:

python

builder.add_edge(START, "research_topic_1")
builder.add_edge(START, "research_topic_2")
builder.add_edge("research_topic_1", "merge")
builder.add_edge("research_topic_2", "merge")

Exemplo Completo de Agente Hierárquico

python

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import List, Annotated
from operator import add

# === Level 2: Search Subgraph ===
class SearchState(TypedDict):
    query: str
    results: List[str]

def search_web(state: SearchState) -> dict:
    return {"results": [f"Result for: {state['query']}"]}

search_builder = StateGraph(SearchState)
search_builder.add_node("search", search_web)
search_builder.add_edge(START, "search")
search_builder.add_edge("search", END)
search_graph = search_builder.compile()

# === Level 2: Analyze Subgraph ===
class AnalyzeState(TypedDict):
    data: List[str]
    analysis: str

def analyze_data(state: AnalyzeState) -> dict:
    return {"analysis": f"Analysis of {len(state['data'])} sources"}

analyze_builder = StateGraph(AnalyzeState)
analyze_builder.add_node("analyze", analyze_data)
analyze_builder.add_edge(START, "analyze")
analyze_builder.add_edge("analyze", END)
analyze_graph = analyze_builder.compile()

# === Level 1: Manager ===
class ManagerState(TypedDict):
    query: str
    results: List[str]
    analysis: str
    final: str
    logs: Annotated[List[str], add]

def search_manager(state: ManagerState) -> dict:
    sub_result = search_graph.invoke({"query": state["query"], "results": []})
    return {"results": sub_result["results"], "logs": ["Search completed"]}

def analyze_manager(state: ManagerState) -> dict:
    sub_result = analyze_graph.invoke({"data": state["results"], "analysis": ""})
    return {"analysis": sub_result["analysis"], "logs": ["Analysis completed"]}

def finalize(state: ManagerState) -> dict:
    return {"final": f"Query: {state['query']}\nAnalysis: {state['analysis']}"}

builder = StateGraph(ManagerState)
builder.add_node("search", search_manager)
builder.add_node("analyze", analyze_manager)
builder.add_node("finalize", finalize)
builder.add_edge(START, "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "finalize")
builder.add_edge("finalize", END)

app = builder.compile()

result = app.invoke({
    "query": "LangGraph hierarchy",
    "results": [],
    "analysis": "",
    "final": "",
    "logs": []
})
print(result["final"])

Benefícios do Design Hierárquico

Benefício	Descrição
Encapsulamento	Os detalhes internos do subgrafo não vazam para o nível pai
Reutilização	Subgrafos podem ser usados em múltiplos grafos pai
Testabilidade	Cada subgrafo é testado de forma independente
Gestão de complexidade	Cada nível lida apenas com suas próprias preocupações
Paralelismo	Múltiplos subgrafos podem rodar em paralelo
Interfaces claras	Contratos explícitos de entrada/saída entre níveis