Habilidade: Arquiteto e Atualizador Multi-Agente

Visão Geral

Esta habilidade transforma Claude em um Arquiteto Sênior de Sistemas Multi-Agente de IA, especializado em LangGraph, LangChain e DeepAgents. Ela fornece fluxos de trabalho estruturados para criar e atualizar sistemas multi-agente de nível de produção — incluindo agentes supervisores, planejadores, pesquisadores, codificadores e pipelines autônomos com memória. Use-a sempre que precisar projetar, construir, depurar ou escalar qualquer sistema de IA multi-agente.

Se esta habilidade adaptar material de um repositório GitHub externo, declare ambos:

• source_repo: owner/repo
• source_type: official ou source_type: community

Quando Usar Esta Habilidade

• Use quando precisar criar um novo agente ou fluxo de trabalho multi-agente do zero
• Use ao trabalhar com grafos de estado, nós, arestas ou roteamento condicional do LangGraph
• Use quando o usuário perguntar sobre comunicação entre agentes, sistemas de memória ou pipelines de chamada de ferramentas
• Use ao depurar ou otimizar um sistema de agente LangChain/LangGraph existente
• Use ao projetar papéis de agente supervisor, planejador, de pesquisa, de codificação ou de validação
• Use ao integrar DeepAgents com planejamento hierárquico e delegação

Como Funciona

Passo 1: Entender o Objetivo

Antes de escrever qualquer código, esclareça:

• Qual é o objetivo de negócio que este sistema de agentes deve alcançar?
• Quais papéis de agente são necessários (supervisor, planejador, pesquisador, codificador, validador)?
• Quais ferramentas cada agente requer?
• Qual estratégia de memória é necessária (Redis, Vector DB, LangChain Memory)?
• Qual protocolo de comunicação conecta os agentes (estado compartilhado, passagem de mensagens)?

Passo 2: Definir o Esquema de Estado

Todos os agentes compartilham um objeto de estado tipado passado através do grafo:

from typing import TypedDict

class AgentState(TypedDict):
    user_goal: str
    tasks: list[str]
    completed_tasks: list[str]
    next_agent: str
    context: dict
    step_count: int          # guards against infinite loops
    error: str | None

Passo 3: Definir Nós de Agente

Cada agente é uma função assíncrona que lê do estado e retorna um estado atualizado:

import logging
from langchain_openai import ChatOpenAI

logger = logging.getLogger(__name__)

async def research_node(state: AgentState) -> AgentState:
    logger.info("research_node: starting")
    llm = ChatOpenAI(model="gpt-4o")
    result = await llm.bind_tools(research_tools).ainvoke(state["user_goal"])
    state["context"]["research"] = result.content
    state["next_agent"] = "coder"
    return state

Passo 4: Construir o LangGraph

Conecte os nós com arestas e roteamento condicional:

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

def build_graph() -> StateGraph:
    graph = StateGraph(AgentState)

    graph.add_node("supervisor", supervisor_node)
    graph.add_node("research",   research_node)
    graph.add_node("coder",      coding_node)
    graph.add_node("validator",  validation_node)
    graph.add_node("tools",      ToolNode(all_tools))

    graph.set_entry_point("supervisor")

    graph.add_conditional_edges(
        "supervisor",
        route_next,
        {"research": "research", "coder": "coder", "end": END}
    )

    graph.add_edge("research",  "supervisor")
    graph.add_edge("coder",     "validator")
    graph.add_edge("validator", "supervisor")

    return graph.compile()

def route_next(state: AgentState) -> str:
    if state["step_count"] > 20:
        return "end"
    return state["next_agent"]

Passo 5: Adicionar Memória

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_memory(session_id: str):
    return RedisChatMessageHistory(
        session_id=session_id,
        url=os.getenv("REDIS_URL"),
        ttl=3600
    )

Passo 6: Executar o Grafo

async def run(user_goal: str, session_id: str):
    graph = build_graph()
    initial_state = AgentState(
        user_goal=user_goal,
        tasks=[],
        completed_tasks=[],
        next_agent="supervisor",
        context={},
        step_count=0,
        error=None,
    )
    return await graph.ainvoke(initial_state)

Passo 7: Expor via FastAPI (opcional)

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class RunRequest(BaseModel):
    goal: str
    session_id: str

@app.post("/run")
async def run_agent(req: RunRequest):
    result = await run(req.goal, req.session_id)
    return {"result": result}

…