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

미들웨어 아키텍처와 통제된 평가

미들웨어는 에이전트 그래프의 도구 선택, 컨텍스트 구성, 재시도, 안전 검사를 분리하는 계층입니다. Terminal-Bench 같은 평가에서 미들웨어 변경 효과를 보려면 모델·과제·환경·예산을 고정한 실험이 필요합니다. 공개 리더보드의 수치만으로 특정 미들웨어의 효과를 단정할 수는 없습니다.

핵심 미들웨어:

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 | 완료 기준 검증 | 품질 향상 |

모델 교체 없이 성능 향상

하네스(미들웨어) 변경의 가치는 같은 평가 계약 아래에서 성공률, 성공당 비용, 지연 시간, 실패 trace가 어떻게 달라지는지로 판단합니다. 단일 공개 점수는 그 자체로 인과 근거가 아닙니다.

요약

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