🎯 LangGraph

핵심 구조

직선을 넘는 3요소

LCEL vs LangGraph

직선 파이프라인의 한계를 순환 그래프로 해결

StateGraph 3요소

State + Node + Edge — 이 3개로 모든 워크플로우를 표현

문의 하나의 그래프 여정

"배송이 3일째 안 와요" — 분류 노드의 유형 판정과 조건 엣지가 고르는 갈림길

State 스냅샷 — 노드를 지날 때마다

"배송이 3일째 안 와요" 문의의 여정을 데이터로 — 노드가 반환한 갱신분이 State에 병합되는 과정

State 설계 원칙

그래프의 데이터 계약 — 필드 최소 · 불변 갱신 · 자동 누적

핵심 포인트

TypedDict

상태 스키마를 타입으로 선언 — 필드 구조가 곧 그래프의 계약

최소 필드

필요한 값만 State에 — 필드가 늘수록 메모리와 디버깅 부담 증가

불변 갱신

노드는 State를 직접 수정하지 않고 갱신분을 반환 — 흐름 추적 용이

Annotated[list, add]

리스트 필드 자동 누적 — messages처럼 쌓이는 값에 사용

코드

from typing import TypedDict

# 그래프 전체가 공유하는 상태 스키마
class RAGState(TypedDict):
    question: str
    documents: list
    relevance_score: float
    answer: str
    retry_count: int

# 노드는 State를 받아 갱신분만 반환하는 순수 함수
def retrieve(state: RAGState):
    docs = retriever.invoke(state["question"])
    return {"documents": docs}

StateGraph 코드 패턴

add_node → add_edge → compile() — 3단계로 그래프 조립

from langgraph.graph import StateGraph, MessagesState, START, END

# 노드 = State를 받아 업데이트를 반환하는 순수 함수
def retrieve(state): ...
def generate(state): ...

# State 스키마로 그래프 선언 — MessagesState는 messages를 자동 누적
graph = StateGraph(MessagesState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")  # START/END 특수 노드로 진입·종료 지정
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
app = graph.compile()  # 컴파일하면 invoke/stream 가능한 Runnable이 됨

RAG 그래프 패턴

검증과 재시도를 노드로 만드는 세 패턴

Self-RAG

생성 후 "답변이 문서에 근거하는가" 검증 → 불합격 시 재생성

재시도 루프 안전장치

retry_count와 best effort — 무한 루프 없는 자기 교정

핵심 포인트

relevance_score >= 0.7

합격선 — LLM이 매긴 관련성 점수로 생성 진행 여부 판정

retry_count

재작성 횟수를 State에 기록 — 최대 2회로 제한해 무한 루프 차단

best effort

한계 도달 시 현재 문서로 답변 생성 — 멈추지 않고 마무리

판단용 경량 모델

관련성 판정은 Haiku급 경량 모델로 — 추가 호출 비용 최소화

코드

# 조건 분기 함수 — 다음 노드 이름을 문자열로 반환
def decide_next(state: RAGState) -> str:
    if state["relevance_score"] >= 0.7:
        return "generate"   # 충분 → 답변 생성
    if state["retry_count"] >= 2:
        return "generate"   # 재시도 한계 → best effort
    return "rewrite"        # 부족 → 쿼리 재작성

# 재작성 노드가 카운터를 올린다 — 안전장치의 실체
def rewrite_query(state: RAGState):
    new_query = rewrite_chain.invoke({"question": state["question"]})
    return {"rewritten_query": new_query,
            "retry_count": state["retry_count"] + 1}

CRAG + Query Router

검색 전 라우팅 + 검색 후 관련성 검증

1. 1
   Query Router

   질문 분류 → 벡터/키워드/웹 검색 분기

2. 2
   Retrieve

   선택된 경로로 문서 검색

3. 3
   Grade Docs

   관련성 LLM 판정 — relevant / not relevant

4. 4
   분기

   관련 문서 있음 → Generate, 없음 → Web Search 폴백

5. 5
   Generate

   검증된 문서 기반 최종 답변 생성

ReAct 에이전트

추론→도구→관찰 순환의 구현

ReAct = StateGraph + ToolNode

추론→도구→관찰을 그래프 순환으로 구현

ReAct 에이전트 코드

StateGraph로 도구 순환 루프 직접 구현

from langgraph.prebuilt import ToolNode

# tool_calls를 실행하고 결과를 messages에 추가하는 빌트인 노드
tool_node = ToolNode([get_weather, search])

def should_continue(state):
    last = state["messages"][-1]
    return "tools" if last.tool_calls else "end"  # 도구 호출이 없으면 루프 종료

graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue,
    {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")  # 순환!

create_agent vs 직접 구성

표준 루프는 한 줄로, 루프 안을 바꿔야 하면 StateGraph로

멀티에이전트

협업 패턴과 사람의 개입 지점

Supervisor vs Swarm

작업 분해 방식에 따라 패턴을 선택

langgraph-supervisor로 구축

중앙 위임 패턴을 패키지 하나로 — 직접 구현 전에 먼저 검토

pip install langgraph-supervisor

별도 패키지(0.0.31) — LangGraph 1.x 호환 공식 확인

create_supervisor()

하위 에이전트 목록으로 Supervisor 그래프를 바로 생성

중앙 상태 관리

전체 상태는 Supervisor가 보유 — 하위 에이전트는 배정받은 태스크만

직접 구현 전환점

배정 로직이 특수해지면 StateGraph + 조건 엣지로 커스텀

HITL 승인 게이트

interrupt()로 그래프 일시 중지 → 사용자 확인 → Command(resume)로 재개

1. 1
   invoke()

   사용자 요청으로 그래프 실행 시작

2. 2
   interrupt_before

   위험 노드 진입 전 자동 중지 — Checkpointer에 저장

3. 3
   사용자 확인

   "결제를 승인하시겠습니까?" 표시

4. 4
   Command(resume)

   승인 → True로 그래프 재개, 거부 → 다른 경로

HITL 메시지 왕복

중단 · 저장 · 재개가 오가는 실제 순서 — 클라이언트 · 그래프 · Checkpointer · 승인자

Deep Agents와 프로덕션

추상화 선택과 배포 전 점검

LangGraph vs Deep Agents

직접 구현 vs 고수준 추상화 — 시작은 Deep Agents, 제어는 LangGraph

Checkpointer — 상태 영속화

스냅샷이 만드는 세 가지 능력 — HITL 재개 · 장애 복구 · 시간여행

AgentCore Runtime으로 배포

컴파일된 그래프를 관리형 런타임에 올리는 4단계

1. 1
   스캐폴드 생성

   agentcore create --framework LangChain_LangGraph — 프레임워크 옵션으로 LangGraph 프로젝트 골격을 만듭니다.

2. 2
   그래프 컴파일

   app = graph.compile(checkpointer=...) — 이때 Checkpointer는 프로덕션 백엔드로 교체합니다.

3. 3
   진입점 래핑

   BedrockAgentCoreApp의 @app.entrypoint로 호출 진입점을 선언합니다. bedrock-agentcore 패키지는 langgraph extras를 제공합니다.

4. 4
   배포

   agentcore deploy 한 번으로 패키징과 배포가 끝납니다. 오토스케일 · 내장 관측성 · 버전 관리는 Runtime이 제공합니다.

프로덕션 체크리스트

LangGraph 에이전트를 프로덕션에 배포하기 전 확인

1. 1
   Checkpointer 교체

   MemorySaver(개발) → PostgresSaver(프로덕션, DynamoDB는 커뮤니티 구현) — 상태 영속화 필수

2. 2
   LangSmith 연동

   노드별 실행 시간·토큰·에러 시각화 — 병목 발견

3. 3
   interrupt 설계

   고위험 노드에 interrupt_before 설정 — HITL 경로 확보

4. 4
   배포 선택

   AgentCore Runtime(관리형) 또는 ECS/Lambda(자체 운영)

에이전트의 두뇌 회로

State·Node·Edge 3요소 위의 Self-RAG·ReAct·Supervisor·HITL — 그래프가 곧 에이전트 설계