LangGraph: Como Construir um Agente com Chamada de Ferramentas e Execução Limitada para Verificação de Código

A modificação autônoma de código exige muito mais do que uma cadeia linear de prompts padrão. Quando um agente altera um repositório, ele precisa ler arquivos, aplicar edições, executar suítes de testes, inspecionar saídas e se autocorrigir diante de regressões. No entanto, padrões cíclicos de agentes — com frequência baseados no paradigma ReAct (Reasoning and Acting) — trazem riscos operacionais se não forem devidamente limitados. Um agente que se depare com dependências quebradas ou falhas persistentes em testes pode entrar em loops de execução caros e intermináveis.

Com base na arquitetura ReAct documentada pelo Google AI para LangGraph, este guia adapta a mecânica essencial de grafos de estado em um agente de programação especializado e delimitado. A implementação executa alterações de código, valida as modificações contra suítes automatizadas de testes e impõe limites rígidos de iterações.

---

Arquitetura Central: Estado, Nós e Arestas Condicionais

O LangGraph modela fluxos de trabalho de agentes como grafos direcionados. Diferente dos grafos acíclicos direcionados (DAGs) estáticos, o LangGraph suporta ciclos de forma nativa, o que é fundamental para o raciocínio iterativo do agente.

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

O grafo apoia-se em três elementos arquiteturais fundamentais:

1. Estado (State): Uma estrutura de dados tipada e centralizada que armazena o histórico de mensagens, metadados de execução e contadores de passos explícitos.
2. Nós (Nodes): Funções Python simples que recebem o estado atual, executam computações isoladas (chamando o modelo ou acionando uma ferramenta) e retornam atributos de estado atualizados.
3. Arestas Condicionais (Conditional Edges): Funções de decisão que inspecionam o estado (como a presença de chamadas de ferramentas ou o alcance do teto de iterações) para decidir se devem encaminhar a execução para a ferramenta ou finalizar em END.

---

Pré-requisitos e Dependências

Certifique-se de que o seu ambiente de desenvolvimento contenha os pacotes necessários de orquestração de grafos e drivers de modelos:

pip install langgraph langchain-core langchain-openai

Para conectar o agente aos provedores de inferência por meio de um gateway compatível com OpenAI, configure suas variáveis de ambiente. Usar o OneMux simplifica o gerenciamento unificado de chaves e o controle de gastos entre múltiplos modelos:

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(Observação: certifique-se sempre de que o modelo de destino escolhido suporte parâmetros de tool calling estruturados por meio de endpoints compatíveis com a OpenAI).

---

Passo 1: Definindo o Estado Bounded do Agente

Em fluxos de conversa padrão, o MessagesState gerencia a inclusão do histórico de mensagens. Para tarefas de programação, definimos explicitamente o AgentState com o TypedDict do Python para rastrear sequências de mensagens junto com um contador explícito (step_count) e um teto máximo (max_steps).

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

O redutor add_messages assegura que as novas listas de mensagens sejam anexadas de forma limpa ao histórico da conversa, em vez de sobrescrever observações anteriores.

---

Passo 2: Implementando Ferramentas de Sistema de Arquivos e Verificação

O agente necessita de ferramentas para inspecionar arquivos-fonte, aplicar atualizações e executar os testes do repositório. Encapsulamos essas operações utilizando o decorador @tool do LangChain.

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

Passo 3: Configuração do Modelo via API Unificada

Inicializamos o modelo utilizando ChatOpenAI, apontando para o endpoint proxy do provedor unificado. Consulte a Documentação do OneMux e o Guia de Início Rápido para verificar a compatibilidade de parâmetros para function calling.

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

Passo 4: Construindo os Nós do Grafo e Guardas de Loop

Os nós executam a lógica e incrementam os limites. Caso um agente entre em loop contínuo sem conseguir corrigir um teste com falha, a lógica da aresta detecta que step_count >= max_steps e faz o roteamento direto para o encerramento do grafo.

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

Passo 5: Compilação e Montagem do Grafo

Junte os componentes usando StateGraph e compile o fluxo de trabalho:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

Passo 6: Executando Modificações de Código com Verificação por Testes

Para executar o fluxo de trabalho, defina uma tarefa de programação em conjunto com limites estritos de execução:

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

Limitações e Preparação para Produção

Ao implantar agentes que alteram código em ambientes de produção, considere as seguintes medidas de proteção:

• Execução em Sandbox: Executar comandos shell arbitrários como pytest localmente traz riscos de corrupção do ambiente ou execução de scripts inseguros. Isole a execução de testes em contêineres Docker efêmeros ou sandboxes protegidas.
• Limites de Tokens no Estado: Saídas extensas de ferramentas — como stack traces de várias páginas — podem esgotar rapidamente a janela de contexto. Trunque as mensagens de saída de testes antes de gravá-las no histórico de estado.
• Gerenciamento de Custos do Modelo: Loops iterativos ReAct multiplicam o consumo de tokens. Acompanhe o uso pelo painel de Preços do OneMux e avalie modelos de raciocínio mais econômicos na página de Modelos do OneMux para otimizar o custo por tarefa.

---

Fontes

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Atualizado em 2 de setembro de 2026).