LangGraph: Tool Calling과 Bounded Execution을 활용한 코드 검증 에이전트 구축하기

자율적인 코드 수정 작업에는 일반적인 선형 프롬프트 체인 이상의 기능이 필요합니다. 에이전트가 리포지토리 코드를 수정할 때는 파일을 읽고, 변경 사항을 적용하고, 테스트 스위트를 실행하고, 출력을 점검하며, 회귀(Regression) 문제가 발생했을 때 스스로 수정할 수 있어야 합니다. 그러나 ReAct(Reasoning and Acting) 패러다임에 기반한 순환형 에이전트 패턴은 적절한 제어가 없으면 운영상 위험을 초래할 수 있습니다. 의존성 오류나 지속적인 테스트 실패를 만난 에이전트는 비용이 많이 드는 무한 실행 루프에 빠질 수 있습니다.

Google AI의 LangGraph ReAct 아키텍처 문서를 바탕으로, 본 가이드에서는 핵심 상태 그래프(State Graph) 메커니즘을 수정하여 실행 범위가 제한된 특화 코딩 에이전트를 구축합니다. 이 구현체는 코드 편집을 실행하고, 자동화된 테스트 스위트로 변경 사항을 검증하며, 엄격한 반복 횟수 예산(Iteration budget)을 적용합니다.

---

핵심 아키텍처: State, Nodes, Conditional Edges

LangGraph는 에이전트 워크플로우를 방향성 그래프(Directed Graph)로 모델링합니다. 정적인 방향성 비순환 그래프(DAG)와 달리, LangGraph는 반복적인 에이전트 추론에 필수적인 순환(Cycle) 구조를 기본적으로 지원합니다.

          [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"

(참고: 선택한 대상 모델이 OpenAI 호환 엔드포인트를 통해 구조화된 도구 호출 매개변수를 지원하는지 항상 확인하세요).

---

1단계: 실행 제한이 적용된 Agent State 정의

일반적인 대화형 워크플로우에서는 MessagesState가 대화 기록 추가를 처리합니다. 코딩 작업의 경우, Python의 TypedDict를 사용하여 AgentState를 명시적으로 정의하고, 메시지 시퀀스와 함께 명시적 카운터(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단계: 파일 시스템 및 검증 도구 구현

에이전트가 소스 파일을 검사하고, 업데이트를 적용하며, 리포지토리 테스트를 실행할 수 있는 도구가 필요합니다. LangChain의 @tool 데코레이터를 사용하여 이러한 작업을 캡슐화합니다.

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 문서 및 빠른 시작 가이드를 확인하세요.

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단계: 그래프 노드 및 루프 가드(Loop Guard) 구성

노드는 로직을 실행하고 실행 단계를 증가시킵니다. 에이전트가 실패한 테스트를 수정하지 못하고 반복해서 루프를 돌 경우, 엣지 로직이 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 Execution): 로컬 환경에서 pytest와 같은 임의의 셸 명령을 직접 실행하면 환경이 손상되거나 안전하지 않은 스크립트가 실행될 위험이 있습니다. 임시 도커 컨테이너나 보안 샌드박스 내에서 테스트 실행을 격리하세요.
• 상태 토큰 제한(State Token Limits): 여러 페이지에 달하는 스택 트레이스처럼 도구 출력 결과가 지나치게 길어지면 컨텍스트 윈도우가 빠르게 소진될 수 있습니다. 테스트 출력 메시지를 상태 기록에 저장하기 전에 적절히 잘라내야(truncate) 합니다.
• 모델 비용 관리: 반복적인 ReAct 루프는 토큰 소비를 급증시킵니다. OneMux 요금제 대시보드에서 사용량을 모니터링하고, OneMux 모델 목록에서 비용이 낮은 추론 모델을 평가하여 작업당 비용을 최적화하세요.

---

참고 자료

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (2026년 9월 2일 업데이트).