LangGraph + Pi Agent -1-

LangGraph 와 Pi Agent 의 인터페이스를 엮어볼까 싶어서 과정을 정리해 본다.

mkdir lang-harness

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

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

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

LangGraph 개념

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

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

위와 같이 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}

글이 길어져서 다음 포스트에서 이어 나가기로 한다.