LangGraph 살펴보기

Karl

10 7월 2026 • 22 min read

LangGraph 와 Pi Agent 의 인터페이스를 엮어볼까 싶어서 LangGraph 의 프로세스를 알아볼 요량으로 정리해 본다.

mkdir lang-harness

# 현재 디렉터리를 LangGraph 하네스 프로젝트로 만들기
uv init --bare

# 필요한 패키지 추가
uv add langgraph langgraph-checkpoint-sqlite

# 설치 확인
uv run python -c "import langgraph; print('ok')"

LangGraph 개념

LangGraph는 Graph 그대로 Node와 Edge로 구성된 Graph 위에서 상태(State)를 관리하며 실행 흐름을 제어하는 오케스트레이션 (작업들을 어떤 순서와 조건으로 실행할지를 관리하는 역할) 런타임이다.

현재 [상태] 확인
    ↓        ← 엣지 (다음에 어디로 이동할지 결정하는 실행 규칙)
[노드] 실행
    ↓        ← 엣지
[상태] 변경
    ↓        ← 엣지
다음 [노드] 결정
    ↓        ← 엣지
[상태] 저장

Graph는 전체 작업 흐름을 이야기한다

위와 같이 LangGraph는 State, Node, Edge를 조합하여 반복과 조건 분기가 있는 실행 흐름을 만든다.

LangGraph는 특히 다음과 같은 작업에 적합하다.

- 중간에 사용자 승인이 필요한 작업
- 실패하면 이전 단계로 돌아가는 작업
- 프로세스가 종료되어도 이어서 실행할 작업
- 단계마다 다른 입력을 사용해야 하는 작업

위 내용을 보면 코딩 하네스를 만들어도 좋을 것 같은 생각이 든다. 실제로 LangGraph 를 기반으로 만든 하네스가 있다. (deepagents)

StateGraph

compile() 하지 않은 상태의 그래프 정의를 의미한다. 아래 코드 처럼 builder 를 선언하고 node 및 edge 를 정의할 수 있게 한다.

builder = StateGraph(CodingState)

builder.add_node("plan", plan_node)
builder.add_node("approve", approval_node)

builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")

graph = builder.compile()

Checkpointer와 Thread

Checkpointer는 Graph의 State를 실행 단계별로 저장하며 각 작업은 thread_id로 구분한다

task-001
→ 인증 오류 수정 작업

task-002
→ 결제 테스트 수정 작업

Checkpointer를 연결하면 Graph 실행 단계마다 State 스냅샷이 저장되는 기능을 하고 thread_id는 체크포인트를 저장하고 다시 불러오는 기본 식별자로 사용할 수 있다.

Interrupt

interrupt()는 노드 실행을 중단하고 외부 입력을 기다리게 한다.

def approval_node(state):
    # 1. 현재 State에서 계획을 읽는다.
    plan = state["plan"]

    # 2. 사용자 입력을 기다리기 위해 실행을 중단한다.
    # 이 시점에 LangGraph 런타임이 checkpointer를 통해
    # 현재 State와 실행 위치를 저장한다.
    response = interrupt({
        "type": "plan_approval",
        "plan": plan,
    })

    # 3. 재개되면 사용자 입력이 response에 들어온다.
    # 노드가 반환한 값은 기존 State에 병합된다.
    return {
        "approval_result": response,
    }

위 2번에서 현재 상태를 저장 하는 부분이 동작 하려면 이미 Graph가 체크포인터와 같이 컴파일 되어 있는 상태여야 한다.

graph = builder.compile(
    checkpointer=checkpointer
)

하네스와 개념 연결

재미삼아 보는 김에 코딩 하네스를 만든다고 가정하면 아래와 같이 구성 가능하겠다. 여직까지 다 나온 개념들이다.

CodingState = State

plan_node
approval_node
implement_node
verify_node
            = Nodes

계획 → 승인
구현 → 검증
            = Edges

승인 결과에 따른 분기
검증 결과에 따른 분기
            = Conditional Edges

SQLiteSaver
            = Checkpointer

작업 ID
            = thread_id

계획 승인 대기
            = interrupt()

연습용 프로젝트 구조 생성

lang-harness/
├── .venv/                       # uv가 생성하는 가상환경, Git 제외
├── src/
│   └── coding_harness/
│       └── __init__.py
├── runtime/
│   ├── checkpoints/
│   │   └── .gitignore
│   └── logs/
│       └── .gitignore
├── tests/
│   └── __init__.py
├── .gitignore
├── .python-version              # 선택 사항
├── pyproject.toml
└── uv.lock

CodingState 정의

State 를 정의해야 하는데 주제에 맞게 라고 정의를 할 것이고 이는 코딩 작업이 현재 어디까지 진행됐는지를 저장하는 구조이다.

각 노드는 State 를 읽고 자신이 처리한 결과만 반환하고 LangGraph 는 이를 기존 State 에 반영한다.

상태값은 다음과 같이 정의 해보자

pending: 대기
approve: 승인
revise: 계획 변경
cancel: 작업 취소

src/coding_harness/state.py

from typing import Literal, TypedDict


ApprovalResult = Literal[
    "pending",
    "approve",
    "revise",
    "cancel",
]


class CodingState(TypedDict):
    """코딩 작업의 진행 상태."""

    # 사용자가 요청한 작업
    task: str

    # 작업할 프로젝트 경로
    repository_path: str

    # 코드 수정 계획
    plan: str

    # 사용자의 계획 승인 결과
    approval_result: ApprovalResult

    # 실제로 변경된 파일 목록
    changed_files: list[str]

    # 구현 결과 설명
    implementation_summary: str

    # 검증 성공 여부
    verification_passed: bool

    # 검증 실패 또는 보완 내용
    verification_feedback: str

    # 현재 재시도 횟수
    retry_count: int

    # 허용할 최대 재시도 횟수
    max_retries: int


def create_initial_state(
    task: str,
    repository_path: str,
    max_retries: int = 2,
) -> CodingState:
    """새로운 코딩 작업의 초기 State를 만든다."""

    task = task.strip()
    repository_path = repository_path.strip()

    if not task:
        raise ValueError("task는 비어 있을 수 없습니다.")

    if not repository_path:
        raise ValueError("repository_path는 비어 있을 수 없습니다.")

    if max_retries < 0:
        raise ValueError("max_retries는 0 이상이어야 합니다.")

    return {
        "task": task,
        "repository_path": repository_path,
        "plan": "",
        "approval_result": "pending",
        "changed_files": [],
        "implementation_summary": "",
        "verification_passed": False,
        "verification_feedback": "",
        "retry_count": 0,
        "max_retries": max_retries,
    }

State 초기화 확인

create_initial_state 함수를 통해 State 를 만들어보자

from pprint import pprint
from coding_harness.state import create_initial_state

state = create_initial_state(
    task="인증 오류 처리를 수정한다.",
    repository_path="/tmp/example-repository",
    max_retries=2,
)

pprint(state)

실행결과는 아래와 같이 나오고 이 State 를 그래프 실행시에 전달하게 된다.

graph.invoke(coding_state)

아마도 이렇게.

{
    'task': '인증 오류 처리를 수정한다.',
    'repository_path': '/tmp/example-repository',
    'plan': '',
    'approval_result': 'pending',
    'changed_files': [],
    'implementation_summary': '',
    'verification_passed': False,
    'verification_feedback': '',
    'retry_count': 0,
    'max_retries': 2
}

위 처럼 State 가 이런 저런 Node들을 통과하면서 업데이트 되고 Edge 에 따라 또 다시 어떤 Node 를 수행하게 될지가 결정된다.

연습용 Graph 작성

coding_harness/basic_graph.py 에 다음과 같은 함수를 작성해준다.

from pprint import pprint
from langgraph.graph import END, START, StateGraph
from coding_harness.state import CodingState, create_initial_state

def plan_node(state: CodingState) -> dict:
    """사용자 요청을 바탕으로 임시 계획을 작성한다."""

    print("\n[plan_node 실행]")
    print("받은 작업:", state["task"])

    plan = (
        "1. 관련 코드를 조사한다.\n"
        "2. 필요한 파일을 수정한다.\n"
        "3. 테스트를 실행한다."
    )

    return {
        "plan": plan,
    }


def finish_node(state: CodingState) -> dict:
    """계획 노드의 실행 결과를 확인한다."""

    print("\n[finish_node 실행]")
    print("전달받은 계획:")
    print(state["plan"])

    return {
        "implementation_summary": "연습용 계획 작성이 완료되었습니다.",
    }


def create_graph():
    """연습용 LangGraph를 구성한다."""

    builder = StateGraph(CodingState)

    builder.add_node("plan", plan_node)
    builder.add_node("finish", finish_node)

    builder.add_edge(START, "plan")
    builder.add_edge("plan", "finish")
    builder.add_edge("finish", END)

    return builder.compile()


def main() -> None:
    initial_state = create_initial_state(
        task="인증 오류 처리를 수정한다.",
        repository_path="/tmp/example-repository",
        max_retries=2,
    )

    graph = create_graph()

    final_state = graph.invoke(initial_state)

    print("\n[최종 State]")
    pprint(final_state)


if __name__ == "__main__":
    main()

Node

Node는 크게 plan, finish 두개로 구성했다. 별 내용없이 plan node는 State 로 task 를 받고 plan 을 리턴한다, finish node 도 State 로 plan 을 받고 결과를 리턴하는 식이다.

노드에서 완전한 State 를 리턴하지 않는다는 점을 확인하자. 아래처럼 구구절절 리턴할 필요가 없다.

return {
    "task": state["task"],
    "repository_path": state["repository_path"],
    "plan": plan,
    "approval_result": state["approval_result"],
    "changed_files": state["changed_files"],
    ...
}

Graph 빌더

Graph 를 함수가 추가 됐다. builder = StateGraph(CodingState) 처럼 CodingState 를 담은 빌더를 구성하고 add_node 로 node들을 추가, add_edge 로 시작부터 끝까지 node 들을 이어주는 코드를 작성했다. 그리고 나서 builder.compile() 을 해 주고 나서야 CompiledStateGraph 로 변환하게 된다. CompiledStateGraph 는 invoke(), stream() 등의 실행 메서드를 가진다.

즉, StateGraph는 설계용 빌더고 compile() 하고 나서야 실행 가능한 CompiledStateGraph 가 된다.

연습용 빌더 테스트

python -m coding_harness.basic_graph

[plan_node 실행]
받은 작업: 인증 오류 처리를 수정한다.

[finish_node 실행]
전달받은 계획:
1. 관련 코드를 조사한다.
2. 필요한 파일을 수정한다.
3. 테스트를 실행한다.

[최종 State]
{'approval_result': 'pending',
 'changed_files': [],
 'implementation_summary': '연습용 계획 작성이 완료되었습니다.',
 'max_retries': 2,
 'plan': '1. 관련 코드를 조사한다.\n2. 필요한 파일을 수정한다.\n3. 테스트를 실행한다.',
 'repository_path': '/tmp/example-repository',
 'retry_count': 0,
 'task': '인증 오류 처리를 수정한다.',
 'verification_feedback': '',
 'verification_passed': False}

네 개 노드와 조건 분기

위 밋밋한 Graph 에 조건 분기를 넣어보자

START
  ↓
plan
  ↓
approve
  ├─ approve → implement
  ├─ revise  → plan
  └─ cancel  → END
                   ↑
implement          │
  ↓                │
verify ────────────┘
  ├─ 성공 → END
  ├─ 실패·재시도 가능 → implement
  └─ 재시도 초과 → END

시나리오는 다음과 같다.

계획 작성
  → 자동 승인
  → 임시 구현
  
  → 첫 검증 실패
  → 구현 재실행
  
  → 두 번째 검증 성공
  → 종료

여기서는 라우팅 함수라는게 등장하는데 State 상태에 따라 분기를 도와주는 함수라고 보면 되는 데 아래와 같이 정의를 하고 다음 실행위치만 결정한다.

def route_verification(state: CodingState) -> str:
    if state["verification_passed"]:
        return "end"

    return "implement"

또한 라우팅 함수는 다음과 같이 Graph 에 정의한다.

builder.add_conditional_edges(
    "verify",            # 1. 어느 노드가 끝난 뒤 분기할지
    route_verification,  # 2. 다음 경로를 판단하는 함수
    {                    # 3. 판단 결과와 실제 이동할 노드의 연결표
                         # 예) 라우터_반환값: 실제_이동할_노드
        "implement": "implement",   
        "end": END,
    }

)

분기 형태의 Graph 작성

coding_harness/harness_graph.py 에 다음과 같은 함수를 작성해준다.

from pprint import pprint
from typing import Literal

from langgraph.graph import END, START, StateGraph

from coding_harness.state import CodingState, create_initial_state


def plan_node(state: CodingState) -> dict:
    """저장소 조사 결과를 바탕으로 작업 계획을 작성한다."""

    print("\n[계획 노드]")
    print("작업:", state["task"])
    print("저장소:", state["repository_path"])

    plan = (
        "1. 관련 코드와 테스트를 조사한다.\n"
        "2. 오류 처리 로직을 수정한다.\n"
        "3. 정적 분석과 테스트를 실행한다."
    )

    return {
        "plan": plan,
        "approval_result": "pending",
    }


def approval_node(state: CodingState) -> dict:
    """계획 승인을 임시로 자동 처리한다."""

    print("\n[승인 노드]")
    print("승인할 계획:")
    print(state["plan"])

    # 이후 interrupt()를 사용한 실제 사용자 입력으로 교체한다.
    temporary_result = "approve"

    print("임시 승인 결과:", temporary_result)

    return {
        "approval_result": temporary_result,
    }


def implement_node(state: CodingState) -> dict:
    """승인된 계획에 따라 임시 구현 결과를 생성한다."""

    attempt_number = state["retry_count"] + 1

    print("\n[구현 노드]")
    print("구현 시도:", attempt_number)
    print("승인된 계획:")
    print(state["plan"])

    if state["verification_feedback"]:
        print("직전 검증 피드백:", state["verification_feedback"])

    # 이후 Pi Agent 호출로 교체한다.
    return {
        "changed_files": [
            "lib/auth/auth_service.dart",
        ],
        "implementation_summary": (
            f"{attempt_number}번째 구현 시도에서 "
            "인증 오류 처리 로직을 수정함"
        ),
    }


def verify_node(state: CodingState) -> dict:
    """구현 결과를 임시 검증한다."""

    print("\n[검증 노드]")
    print("변경 파일:", state["changed_files"])

    # 첫 번째 검증은 의도적으로 실패시킨다.
    if state["retry_count"] == 0:
        print("검증 결과: 실패")

        return {
            "verification_passed": False,
            "verification_feedback": (
                "인증 취소 테스트가 실패했습니다. "
                "취소 예외 처리 조건을 다시 확인하세요."
            ),
            "retry_count": state["retry_count"] + 1,
        }

    # 두 번째 검증은 성공시킨다.
    print("검증 결과: 성공")

    return {
        "verification_passed": True,
        "verification_feedback": "모든 검증을 통과했습니다.",
    }


def route_approval(
    state: CodingState,
) -> Literal["implement", "plan", "end"]:
    """승인 결과에 따라 다음 노드를 결정한다."""

    result = state["approval_result"]

    if result == "approve":
        return "implement"

    if result == "revise":
        return "plan"

    return "end"


def route_verification(
    state: CodingState,
) -> Literal["implement", "end"]:
    """검증 결과와 재시도 횟수에 따라 다음 노드를 결정한다."""

    if state["verification_passed"]:
        return "end"

    if state["retry_count"] <= state["max_retries"]:
        return "implement"

    return "end"


def create_graph():
    """코딩 하네스 그래프를 구성한다."""

    builder = StateGraph(CodingState)

    # 노드 등록
    builder.add_node("plan", plan_node)
    builder.add_node("approve", approval_node)
    builder.add_node("implement", implement_node)
    builder.add_node("verify", verify_node)

    # 고정 엣지
    builder.add_edge(START, "plan")
    builder.add_edge("plan", "approve")
    builder.add_edge("implement", "verify")

    # 승인 결과 조건 분기
    builder.add_conditional_edges(
        "approve",
        route_approval,
        {
            "implement": "implement",
            "plan": "plan",
            "end": END,
        },
    )

    # 검증 결과 조건 분기
    builder.add_conditional_edges(
        "verify",
        route_verification,
        {
            "implement": "implement",
            "end": END,
        },
    )

    return builder.compile()


def main() -> None:
    initial_state = create_initial_state(
        task="인증 오류 처리를 수정한다.",
        repository_path="/tmp/example-repository",
        max_retries=2,
    )

    graph = create_graph()

    final_state = graph.invoke(initial_state)

    print("\n[최종 State]")
    pprint(final_state)


if __name__ == "__main__":
    main()

실행 순서는 다음과 같다

approval_node 실행
    ↓
approval_result 업데이트
    ↓
route_approval 실행
    ↓
반환값에 해당하는 노드로 이동

PYTHONPATH=src uv run python -m coding_harness.harness_graph

실행결과는 아래와 같다. 의도적으로 1차 구현을 검증노드에서 실패 시키고 다시 구현 검증을 거치는 프로세스이다.

[계획 노드]
작업: 인증 오류 처리를 수정한다.
저장소: /tmp/example-repository

[승인 노드]
승인할 계획:
1. 관련 코드와 테스트를 조사한다.
2. 오류 처리 로직을 수정한다.
3. 정적 분석과 테스트를 실행한다.
임시 승인 결과: approve

  [구현 노드]
    구현 시도: 1
    승인된 계획:
    1. 관련 코드와 테스트를 조사한다.
    2. 오류 처리 로직을 수정한다.
    3. 정적 분석과 테스트를 실행한다.

[검증 노드]
변경 파일: ['lib/auth/auth_service.dart']
검증 결과: 실패

  [구현 노드]
    구현 시도: 2
    승인된 계획:
    1. 관련 코드와 테스트를 조사한다.
    2. 오류 처리 로직을 수정한다.
    3. 정적 분석과 테스트를 실행한다.
    직전 검증 피드백: 인증 취소 테스트가 실패했습니다. 취소 예외 처리 조건을 다시 확인하세요.

[검증 노드]
변경 파일: ['lib/auth/auth_service.dart']
검증 결과: 성공

[최종 State]
{'approval_result': 'approve',
 'changed_files': ['lib/auth/auth_service.dart'],
 'implementation_summary': '2번째 구현 시도에서 인증 오류 처리 로직을 수정함',
 'max_retries': 2,
 'plan': '1. 관련 코드와 테스트를 조사한다.\n2. 오류 처리 로직을 수정한다.\n3. 정적 분석과 테스트를 실행한다.',
 'repository_path': '/tmp/example-repository',
 'retry_count': 1,
 'task': '인증 오류 처리를 수정한다.',
 'verification_feedback': '모든 검증을 통과했습니다.',
 'verification_passed': True}

interrupt 추가

아래는 지난 코드에 interrupt() 를 추가하기 위해 checkpoint 를 넣고 부분적으로 소스를 수정했다.

from pprint import pprint
from typing import Literal

# [추가] interrupt 시점의 그래프 상태를 메모리에 저장하기 위한 체크포인터
from langgraph.checkpoint.memory import InMemorySaver

from langgraph.graph import END, START, StateGraph

# [추가] Command는 중단된 그래프 재개에 사용하고,
# interrupt는 사용자 승인 지점에서 그래프를 중단하는 데 사용한다.
from langgraph.types import Command, interrupt

from coding_harness.state import CodingState, create_initial_state


def plan_node(state: CodingState) -> dict:
    """저장소 조사 결과를 바탕으로 작업 계획을 작성한다."""

    print("\n[계획 노드]")
    print("작업:", state["task"])
    print("저장소:", state["repository_path"])

    # 현재는 LangGraph 흐름을 확인하기 위한 고정 계획이다.
    # 실제 하네스에서는 이 부분을 Pi Agent 호출로 교체한다.
    plan = (
        "1. 관련 코드와 테스트를 조사한다.\n"
        "2. 오류 처리 로직을 수정한다.\n"
        "3. 정적 분석과 테스트를 실행한다."
    )

    # [추가] 사용자가 이전 승인 단계에서 revise를 선택했다면
    # 전달된 계획 수정 요청을 확인한다.
    revision_feedback = state.get("revision_feedback", "")

    # [추가] 계획 수정 요청이 있으면 계획에 반영한다.
    if revision_feedback:
        print("계획 수정 요청:", revision_feedback)

        # 현재 예제에서는 수정 요청을 기존 계획 끝에 추가한다.
        # 실제 Pi Agent 연동 시에는 기존 계획 전체를 다시 작성해야 한다.
        plan += (
            "\n4. 다음 사용자 수정 요청을 반영한다: "
            f"{revision_feedback}"
        )

    return {
        "plan": plan,

        # [수정] 새 계획이 작성될 때마다 승인 상태를 pending으로 초기화한다.
        "approval_result": "pending",
    }


def approval_node(state: CodingState) -> dict:
    """현재 계획을 사용자에게 보여주고 승인을 기다린다."""

    # [수정] 단순 문자열 대신 action과 feedback을 포함하는
    # 딕셔너리 응답을 받을 수 있도록 변경했다.
    #
    # interrupt()는 터미널에서 직접 입력받지 않는다.
    # 그래프 실행을 중단하고 현재 승인 요청 정보를 호출자에게 반환한다.
    response = interrupt(
        {
            "type": "plan_approval",
            "plan": state["plan"],
            "choices": [
                "approve",
                "revise",
                "cancel",
            ],
        }
    )

    # Command(resume=...)에서 다음 구조를 전달받는다.
    #
    # {
    #     "action": "approve" | "revise" | "cancel",
    #     "feedback": "계획 수정 요청"
    # }

    # [추가] 예상하지 못한 resume 값이 전달되는 것을 방지한다.
    if not isinstance(response, dict):
        raise ValueError(
            "승인 응답은 딕셔너리 형식이어야 합니다."
        )

    action = response.get("action")

    # [추가] 승인 결과가 허용된 값인지 확인한다.
    if action not in {"approve", "revise", "cancel"}:
        raise ValueError(
            f"올바르지 않은 승인 결과입니다: {action}"
        )

    return {
        # [수정] 기존에는 interrupt 반환 문자열을 그대로 저장했지만,
        # 이제 response 딕셔너리의 action 값을 저장한다.
        "approval_result": action,

        # [추가] revise 선택 시 사용자가 입력한 계획 수정 요청을 저장한다.
        "revision_feedback": response.get("feedback", ""),
    }


def implement_node(state: CodingState) -> dict:
    """승인된 계획에 따라 임시 구현 결과를 생성한다."""

    # retry_count는 이미 발생한 검증 실패 횟수다.
    # 따라서 현재 구현 시도 번호는 retry_count + 1이다.
    attempt_number = state["retry_count"] + 1

    print("\n[구현 노드]")
    print("구현 시도:", attempt_number)
    print("승인된 계획:")
    print(state["plan"])

    # 이전 검증에 실패해 구현 노드로 돌아온 경우
    # 실패 원인을 다음 구현에 참고한다.
    if state["verification_feedback"]:
        print(
            "직전 검증 피드백:",
            state["verification_feedback"],
        )

    # 현재는 고정된 임시 결과를 반환한다.
    # 실제 하네스에서는 이 부분을 Pi Agent 호출로 교체한다.
    return {
        "changed_files": [
            "lib/auth/auth_service.dart",
        ],
        "implementation_summary": (
            f"{attempt_number}번째 구현 시도에서 "
            "인증 오류 처리 로직을 수정함"
        ),
    }


def verify_node(state: CodingState) -> dict:
    """구현 결과를 임시로 검증한다."""

    print("\n[검증 노드]")
    print("변경 파일:", state["changed_files"])

    # 반복 흐름을 확인하기 위해 첫 번째 검증은 실패시킨다.
    if state["retry_count"] == 0:
        print("검증 결과: 실패")

        return {
            "verification_passed": False,
            "verification_feedback": (
                "인증 취소 테스트가 실패했습니다. "
                "취소 예외 처리 조건을 다시 확인하세요."
            ),
            "retry_count": state["retry_count"] + 1,
        }

    # 두 번째 검증부터는 성공시키는 임시 로직이다.
    print("검증 결과: 성공")

    return {
        "verification_passed": True,
        "verification_feedback": (
            "모든 검증을 통과했습니다."
        ),
    }


# [추가] 검증 성공과 취소, 실패 종료를 구분하기 위해
# 별도의 종료 처리 노드를 추가했다.
def complete_node(state: CodingState) -> dict:
    """검증이 성공했을 때 최종 상태를 완료로 변경한다."""

    print("\n[작업 완료]")
    print(state["verification_feedback"])

    return {
        "final_status": "completed",
    }


# [추가] 사용자가 cancel을 선택했을 때
# 명시적인 취소 상태를 저장하는 노드다.
def cancel_node(state: CodingState) -> dict:
    """사용자가 작업을 취소했을 때 상태를 변경한다."""

    print("\n[작업 취소]")
    print("사용자가 계획 승인을 취소했습니다.")

    return {
        "final_status": "cancelled",
    }


# [추가] 검증 재시도 한도를 초과했을 때
# 명시적인 실패 상태를 저장하는 노드다.
def fail_node(state: CodingState) -> dict:
    """최대 재시도 횟수를 초과했을 때 실패 상태로 변경한다."""

    print("\n[작업 실패]")
    print("최대 재시도 횟수를 초과했습니다.")
    print("마지막 검증 결과:", state["verification_feedback"])

    return {
        "final_status": "failed",
    }


def route_approval(
    state: CodingState,
) -> Literal["approve", "revise", "cancel"]:
    """승인 결과에 따라 다음 실행 경로를 선택한다."""

    # [수정] 기존에는 implement, plan, end 같은 실제 노드명을 반환했다.
    #
    # 이제 사용자 승인 결과인 approve, revise, cancel을 그대로 반환하고,
    # add_conditional_edges()의 매핑에서 실제 이동 노드를 결정한다.
    return state["approval_result"]


def route_verification(
    state: CodingState,
) -> Literal["success", "retry", "failed"]:
    """검증 결과와 재시도 횟수에 따라 경로를 선택한다."""

    # [수정] 기존의 단순 end 결과를 success와 failed로 구분했다.
    if state["verification_passed"]:
        return "success"

    # retry_count는 이미 발생한 검증 실패 횟수다.
    #
    # max_retries가 2라면:
    #
    # 첫 번째 실패: retry_count = 1 → 재시도
    # 두 번째 실패: retry_count = 2 → 재시도
    # 세 번째 실패: retry_count = 3 → 실패 종료
    if state["retry_count"] <= state["max_retries"]:
        return "retry"

    return "failed"


def create_graph():
    """코딩 하네스의 LangGraph 실행 흐름을 구성한다."""

    builder = StateGraph(CodingState)

    # ------------------------------------------------------------------
    # 노드 등록
    # ------------------------------------------------------------------

    builder.add_node("plan", plan_node)
    builder.add_node("approve", approval_node)
    builder.add_node("implement", implement_node)
    builder.add_node("verify", verify_node)

    # [추가] 종료 이유를 상태에 명시하기 위한 종료 처리 노드
    builder.add_node("complete", complete_node)
    builder.add_node("cancel", cancel_node)
    builder.add_node("fail", fail_node)

    # ------------------------------------------------------------------
    # 고정 엣지
    # ------------------------------------------------------------------

    builder.add_edge(START, "plan")
    builder.add_edge("plan", "approve")
    builder.add_edge("implement", "verify")

    # [추가] 각 종료 처리 노드가 실행된 후 그래프를 종료한다.
    builder.add_edge("complete", END)
    builder.add_edge("cancel", END)
    builder.add_edge("fail", END)

    # ------------------------------------------------------------------
    # 계획 승인 결과에 따른 조건부 분기
    # ------------------------------------------------------------------

    builder.add_conditional_edges(
        "approve",
        route_approval,
        {
            # [수정] 왼쪽은 사용자 승인 결과이고,
            # 오른쪽은 실제 이동할 그래프 노드다.
            "approve": "implement",
            "revise": "plan",
            "cancel": "cancel",
        },
    )

    # ------------------------------------------------------------------
    # 검증 결과에 따른 조건부 분기
    # ------------------------------------------------------------------

    builder.add_conditional_edges(
        "verify",
        route_verification,
        {
            # [수정] 기존의 implement/end 두 경로를
            # 성공, 재시도, 실패 세 경로로 분리했다.
            "success": "complete",
            "retry": "implement",
            "failed": "fail",
        },
    )

    # [추가] interrupt() 발생 시 현재 그래프 상태와 실행 위치를
    # 메모리에 저장하기 위한 체크포인터다.
    #
    # InMemorySaver는 Python 프로세스가 종료되면 상태가 사라진다.
    # 실제 하네스에서는 SQLite 체크포인터로 교체할 수 있다.
    checkpointer = InMemorySaver()

    # [수정] 체크포인터 없이 compile()하던 구조에서
    # checkpointer를 전달하도록 변경했다.
    return builder.compile(
        checkpointer=checkpointer,
    )


# [추가] 사용자 승인 입력과 resume 데이터를 만드는 로직을
# main()에서 별도 함수로 분리했다.
def read_approval_from_user(
    interrupt_value: dict,
) -> dict:
    """인터럽트 정보를 출력하고 사용자 승인 응답을 생성한다."""

    print("\n[사용자 승인 요청]")
    print(interrupt_value["plan"])
    print("선택:", interrupt_value["choices"])

    while True:
        approval = input(
            "approve / revise / cancel: "
        ).strip().lower()

        if approval in {
            "approve",
            "revise",
            "cancel",
        }:
            break

        print(
            "approve, revise, cancel 중 "
            "하나를 입력하세요."
        )

    feedback = ""

    # [추가] revise 선택 시 구체적인 계획 수정 내용을 입력받는다.
    if approval == "revise":
        feedback = input(
            "계획 수정 내용을 입력하세요: "
        ).strip()

    # [수정] 기존의 단순 문자열 대신 action과 feedback을
    # 함께 포함하는 딕셔너리를 resume 값으로 반환한다.
    return {
        "action": approval,
        "feedback": feedback,
    }


def main() -> None:
    """그래프를 실행하고 인터럽트 발생 시 사용자 입력을 처리한다."""

    initial_state = create_initial_state(
        task="인증 오류 처리를 수정한다.",
        repository_path="/tmp/example-repository",
        max_retries=2,
    )

    # [추가] 체크포인터가 현재 작업 상태를 식별할 수 있도록
    # 고정된 thread_id를 설정한다.
    #
    # 그래프를 재개할 때도 반드시 동일한 thread_id를 사용해야 한다.
    config = {
        "configurable": {
            "thread_id": "coding-task-001",
        }
    }

    # 노드와 엣지가 연결된 실행 가능 그래프를 생성한다.
    graph = create_graph()

    # [수정] 기존에는 graph.invoke(initial_state)만 호출했지만,
    # 체크포인터를 사용하기 위해 config를 함께 전달한다.

    # 그래프는 중단되고 결과에 __interrupt__가 포함된다.
    result = graph.invoke(
        initial_state,
        config=config,
    )

    # [추가] revise를 여러 번 선택할 수 있으므로
    # 인터럽트가 없어질 때까지 반복해서 사용자 응답을 처리한다.
    while "__interrupt__" in result:
        # 현재 예제는 한 번에 하나의 인터럽트만 처리
        interrupt_value = result[
            "__interrupt__"
        ][0].value

        # 사용자에게 계획을 보여주고 승인 응답을 받는다.
        resume_value = read_approval_from_user(
            interrupt_value
        )

        # [추가] Command(resume=...)를 사용해
        # 중단된 approval_node를 동일한 thread_id에서 재개한다.
        #
        # 동일한 thread_id를 사용해 중단된 그래프를 재개한다.
        #
        # resume_value는 approval_node 내부 interrupt()의
        # 반환값으로 전달된다.
        result = graph.invoke(
            Command(resume=resume_value),
            config=config,
        )

    print("\n[최종 State]")
    pprint(result)


if __name__ == "__main__":
    main()

여기 까지 LangGraph 기본 구성에 대해 알아봤다.

다음 포스트에서는 Pi Agent 에서 LangGraph 를 활용하는 포스트를 써보고 싶다.
어떻게 구성해야 할지 생각을 안해봤는데 아마도 /langgraph-harness 뒤에 자연어로 명령을 하면 하네스로 코드를 구현하는 command 형식으로 추가되지 않을까 싶다. 🤔

아니면 그 반대로 python 에서 pi rpc 를 호출하거나...