LangGraph

왜 그래프인가?

기존 LangChain의 체인(Chain)은 선형 실행에 최적화되어 있습니다. A → B → C 순서로 실행되며 분기, 루프, 병렬 실행이 어렵습니다.

LangGraph는 에이전트 실행을 방향성 있는 그래프(DAG + 사이클 허용) 로 모델링합니다. 이를 통해 조건부 분기, 루프, 병렬 실행, Human-in-the-Loop 등 복잡한 에이전트 패턴을 선언적으로 표현할 수 있습니다.

Chain:     입력 → A → B → C → 출력

LangGraph: 입력 → [분류기]
                      ↓          ↓
                   [에이전트A] [에이전트B]
                      ↓          ↓
                   [검증기] ←──────
                      ↓  (실패 시 루프)
                   [출력]

핵심 개념

개념	설명
StateGraph	상태를 공유하는 그래프 정의
Node	상태를 받아 업데이트를 반환하는 함수
Edge	노드 간 연결 (조건부/무조건부)
State	그래프 전체에서 공유되는 타입 정의된 상태
Checkpointer	상태 스냅샷 저장 및 복원

StateGraph 기본 구현

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# 1. 공유 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 메시지는 누적
    task: str
    attempts: int
    result: str | None

# 2. 노드 함수 정의
def agent_node(state: AgentState) -> dict:
    """에이전트가 작업 수행"""
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "attempts": state["attempts"] + 1
    }

def validator_node(state: AgentState) -> dict:
    """결과 검증"""
    last_message = state["messages"][-1]
    is_valid = validate(last_message.content)
    return {"result": last_message.content if is_valid else None}

# 3. 조건부 라우팅
def should_retry(state: AgentState) -> str:
    if state["result"] is not None:
        return "done"
    if state["attempts"] >= 3:
        return "escalate"
    return "retry"

# 4. 그래프 조립
graph = StateGraph(AgentState)

graph.add_node("agent", agent_node)
graph.add_node("validator", validator_node)
graph.add_node("escalate", escalate_to_human)

graph.set_entry_point("agent")
graph.add_edge("agent", "validator")
graph.add_conditional_edges(
    "validator",
    should_retry,
    {
        "done":     END,
        "retry":    "agent",
        "escalate": "escalate",
    }
)

app = graph.compile()

체크포인팅 — 상태 영속화

체크포인터는 그래프 실행 중 상태를 저장합니다. 중단 후 재개, Human-in-the-Loop, 시간 여행 디버깅이 가능해집니다.

from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite 기반 체크포인터
with SqliteSaver.from_conn_string("agent_state.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "session-001"}}

    # 실행 (중간에 중단되어도 상태 저장됨)
    result = await app.ainvoke(
        {"task": "코드 리팩터링", "messages": [], "attempts": 0, "result": None},
        config=config,
    )

    # 나중에 같은 thread_id로 재개
    result2 = await app.ainvoke(
        {"messages": [HumanMessage("계속해주세요")]},
        config=config,
    )

Human-in-the-Loop — interrupt()

LangGraph는 interrupt()를 통해 그래프 실행을 일시 정지하고 사람의 입력을 기다릴 수 있습니다.

from langgraph.types import interrupt

def review_node(state: AgentState) -> dict:
    """사람의 검토가 필요한 시점"""
    decision = interrupt({
        "action": state["pending_action"],
        "reason": "프로덕션 배포 전 확인 필요",
    })

    if decision["approved"]:
        return {"approved": True}
    else:
        return {"approved": False, "feedback": decision.get("feedback")}

interrupt()의 동작 방식

interrupt()가 호출되면 그래프 실행이 일시 정지되고 상태가 체크포인터에 저장됩니다. 사람이 응답을 제공하면 그래프는 정지된 지점부터 재개됩니다. 서버를 재시작해도 상태가 유지됩니다.

미들웨어 아키텍처 — 52.8% → 66.5%

LangGraph 팀은 에이전트 그래프에 미들웨어 레이어를 추가하는 것만으로 Terminal Bench 벤치마크에서 52.8% → 66.5% 성능 향상을 달성했습니다.

핵심 미들웨어:

LocalContext

에이전트가 현재 컨텍스트를 명시적으로 추적합니다.

def local_context_middleware(state: AgentState) -> dict:
    """현재 작업 컨텍스트를 상태에 유지"""
    return {
        "context": {
            "current_file": get_current_file(),
            "recent_changes": get_recent_changes(),
            "test_results": get_last_test_results(),
        }
    }

LoopDetection

무한 루프를 감지하고 탈출합니다.

def loop_detection_middleware(state: AgentState) -> dict:
    recent_actions = state["messages"][-10:]
    if is_repetitive(recent_actions):
        return {
            "messages": [SystemMessage("루프 감지. 다른 접근법을 시도하세요.")],
            "loop_detected": True,
        }
    return {}

ReasoningSandwich

실행 전후에 명시적 추론 단계를 삽입합니다.

def reasoning_sandwich(state: AgentState) -> dict:
    """행동 전: 계획 수립 / 행동 후: 결과 평가"""
    return {
        "messages": [
            SystemMessage("""
            행동 전: 다음 단계를 실행하기 전에
            1) 무엇을 하려는가?
            2) 어떤 위험이 있는가?
            3) 성공 기준은 무엇인가?
            를 명시하세요.
            """)
        ]
    }

PreCompletionChecklist

완료를 선언하기 전 검사 목록을 실행합니다.

COMPLETION_CHECKLIST = [
    "테스트가 모두 통과하는가?",
    "타입 에러가 없는가?",
    "console.log가 남아있지 않은가?",
    "문서화가 완료됐는가?",
]

미들웨어	역할	효과
LocalContext	컨텍스트 손실 방지	반복 실수 감소
LoopDetection	무한 루프 탈출	행 낭비 방지
ReasoningSandwich	명시적 추론 강제	충동적 행동 감소
PreCompletionChecklist	완료 기준 검증	품질 향상

모델 교체 없이 성능 향상

52.8% → 66.5% 개선은 모델을 변경하지 않고 하네스(미들웨어)만 수정하여 달성됐습니다. 이것이 하네스 엔지니어링의 핵심 가치입니다.

요약

LangGraph는 그래프 기반 실행으로 조건부 분기, 루프, 병렬 실행을 선언적으로 표현하고, 체크포인팅으로 상태를 영속화합니다. LocalContext, LoopDetection, ReasoningSandwich, PreCompletionChecklist 미들웨어를 추가하는 것만으로 Terminal Bench에서 13.7%p 성능 향상을 달성한 사례는 하네스 엔지니어링의 강력한 근거입니다.