LangGraph: создание агента с вызовом инструментов и ограничением шагов для верификации кода

Автономное внесение изменений в кодовую базу требует гораздо большего, чем стандартная линейная цепочка промптов. Когда агент модифицирует репозиторий, ему необходимо читать файлы, применять правки, запускать наборы тестов, анализировать вывод и самостоятельно исправлять регрессии при их появлении. Однако циклические паттерны агентов — часто основанные на парадигме ReAct (Reasoning and Acting) — несут эксплуатационные риски при отсутствии ограничений. Столкнувшись с нарушенными зависимостями или непроходящими тестами, агент может войти в бесконечный и дорогостоящий цикл выполнения.

Опираясь на архитектуру ReAct для LangGraph из документации Google AI, это руководство адаптирует базовую механику графов состояния для специализированного агента написания кода с жесткими ограничениями. Реализация выполняет правки кода, проверяет изменения с помощью автоматических тестов и соблюдает строгий бюджет итераций.

---

Базовая архитектура: State, Nodes и Conditional Edges

LangGraph моделирует рабочие процессы агента в виде ориентированных графов. В отличие от статических направленных ациклических графов (DAG), LangGraph нативно поддерживает циклы, необходимые для итеративного процесса рассуждений агента.

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

Граф опирается на три архитектурных компонента:

1. State (Состояние): централизованная типизированная структура данных, хранящая историю сообщений, метаданные выполнения и явные счетчики шагов.
2. Nodes (Узлы): обычные функции Python, которые получают текущее состояние, выполняют изолированные вычисления (обращение к модели или запуск инструмента) и возвращают обновленные атрибуты состояния.
3. Conditional Edges (Условные ребра): функции принятия решений, проверяющие состояние (например, наличие вызовов инструментов или превышение лимита итераций), чтобы определить, перенаправить ли выполнение к исполнителю инструментов или завершить процесс на узле END.

---

Предварительные требования и зависимости

Убедитесь, что в вашей среде разработки установлены необходимые пакеты для оркестрации графов и взаимодействия с моделями:

pip install langgraph langchain-core langchain-openai

Для подключения агента к провайдерам инференса через OpenAI-совместимый шлюз настройте переменные окружения. Использование OneMux упрощает централизованное управление ключами и контроль расходов по разным моделям:

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

(Примечание: обязательно убедитесь, что выбранная вами целевая модель поддерживает структурированные параметры tool-calling через OpenAI-совместимые эндпоинты).

---

Шаг 1: Определение ограниченного состояния агента

В стандартных диалоговых сценариях MessagesState отвечает за добавление истории чата. Для задач разработки мы явно определяем AgentState с помощью TypedDict в Python, чтобы отслеживать последовательности сообщений вместе с явным счетчиком (step_count) и пороговым лимитом (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

Редьюсер add_messages гарантирует корректное добавление новых списков сообщений в историю диалога без перезаписи предыдущих наблюдений.

---

Шаг 2: Реализация инструментов файловой системы и верификации

Агенту требуются инструменты для чтения исходных файлов, применения обновлений и запуска тестов репозитория. Мы оборачиваем эти операции с помощью декоратора @tool из 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}

---

Шаг 3: Настройка модели через единый API

Инициализируем модель с помощью ChatOpenAI, указав прокси-эндпоинт единого провайдера. Ознакомьтесь с OneMux Docs и Quickstart Guide, чтобы проверить совместимость параметров вызова функций.

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)

---

Шаг 4: Построение узлов графа и защита от циклов

Узлы выполняют логику и увеличивают счетчик шагов. Если агент повторяет цикл без успешного прохождения теста, логика ребра определяет условие step_count >= max_steps и направляет выполнение напрямую к завершению графа.

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"

---

Шаг 5: Сборка и компиляция графа

Объедините компоненты с помощью StateGraph и скомпилируйте рабочий процесс:

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

---

Шаг 6: Применение правок кода с проверкой тестами

Для запуска процесса сформулируйте задачу по коду вместе с жесткими лимитами выполнения:

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.

---

Ограничения и подготовка к продакшену

Развертывая агентов, модифицирующих код, в продакшен-средах, предусмотрите следующие меры безопасности:

• Изоляция исполнения (Sandboxing): запуск произвольных команд оболочки (например, pytest) локально несет риск повреждения окружения или выполнения небезопасного кода. Изолируйте выполнение тестов во временных Docker-контейнерах или защищенных песочницах.
• Лимиты токенов в состоянии: объемный вывод инструментов — например, многостраничные трейсы ошибок — способен быстро исчерпать контекстное окно. Усекайте сообщения с результатами тестов перед сохранением в историю состояния.
• Управление расходами на модель: итеративные циклы ReAct многократно увеличивают потребление токенов. Отслеживайте использование через панель OneMux Pricing и подбирайте более экономичные модели для рассуждений в каталоге OneMux Models, чтобы оптимизировать стоимость решения задач.

---

Источники

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Обновлено 2 сентября 2026 г.).