LangGraph 심화

기본기는 16장에서 끝났다. 이번엔 가 다른 그래프 라이브러리와 진짜 차별되는 네 기능 — , , , — 을 차례로 본다.

큰 그래프 문제

가 커지면 한 파일에 30개 노드가 쌓인다. 가독성도 무너지고 테스트도 어렵다.

해법은 모듈화다. 작은 를 만들어 처럼 끼워 넣는다. 이것이 다.

다이어그램 로딩…

서브그래프 합성

는 그 자체로 컴파일된 이고, 부모 에서 그냥 함수처럼 add_node로 등록한다. 키가 겹치면 자동 매핑된다.

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

class State(TypedDict):
  messages: Annotated[list, add_messages]

# 검색 서브그래프 — 독립 컴파일
search = StateGraph(State)
search.add_node("query", lambda s: {"messages": [{"role":"tool","content":"hits..."}]})
search.add_edge(START, "query")
search.add_edge("query", END)
search_app = search.compile()

# 부모는 서브그래프를 노드처럼 사용
parent = StateGraph(State)
parent.add_node("search", search_app)  # compiled subgraph
parent.add_node("respond", lambda s: {"messages": [{"role":"assistant","content":"done"}]})
parent.add_edge(START, "search")
parent.add_edge("search", "respond")
parent.add_edge("respond", END)
app = parent.compile()

import { StateGraph, START, END, MessagesAnnotation } from '@langchain/langgraph'

// 검색 서브그래프 — 독립 컴파일
const search = new StateGraph(MessagesAnnotation)
.addNode('query', () => ({
  messages: [{ role: 'tool', content: 'hits...' }],
}))
.addEdge(START, 'query')
.addEdge('query', END)
.compile()

// 부모는 서브그래프를 노드처럼 사용
const app = new StateGraph(MessagesAnnotation)
.addNode('search', search)
.addNode('respond', () => ({
  messages: [{ role: 'assistant', content: 'done' }],
}))
.addEdge(START, 'search')
.addEdge('search', 'respond')
.addEdge('respond', END)
.compile()

서로 다른 상태 스키마

부모와 의 가 완전히 다를 때는, 에서 받은 부모 상태를 서브 상태로 변환해 invoke하고, 결과를 다시 부모 상태로 풀어 반환한다. 간 계약을 한 함수에서 끝낸다.

# 부모 상태 != 서브 상태일 때
def call_sub(parent_state):
  sub_input = {"q": parent_state["messages"][-1]["content"]}
  sub_output = search_app.invoke(sub_input)
  return {"messages": [{"role": "tool", "content": str(sub_output)}]}

parent.add_node("search", call_sub)

// 부모 상태 != 서브 상태일 때
const callSub = async (s: typeof MessagesAnnotation.State) => {
const subInput = { q: (s.messages.at(-1)?.content ?? '') as string }
const subOutput = await search.invoke(subInput)
return { messages: [{ role: 'tool', content: JSON.stringify(subOutput) }] }
}

HITL — 사람을 끼워 넣기

위험한 행동 직전, 를 잠깐 멈추고 사람의 확인을 받는다. 이게 다.

는 두 가지 방법을 제공한다. 정적 interrupt_before=["node"]와 동적 함수다. 후자가 v0.2 이후의 표준이며 어느 에서도 호출 가능하다.

동적 인터럽트

는 안에서 interrupt(value)를 호출하는 순간 가 정지하고, value가 클라이언트로 빠져나간다. 사람이 결정한 입력을 Command(resume=...)로 다시 넣으면 그 자리에서 재개한다. 가 있어야 동작한다.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def review(state):
  decision = interrupt({"draft": state["draft"]})  # 여기서 정지
  return {"approved": decision == "ok"}

g.add_node("review", review)
app = g.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "t1"}}
out = app.invoke({"draft": "..."}, config)  # interrupt 발생
# 사람이 확인 후 재개
final = app.invoke(Command(resume="ok"), config)

import { interrupt, Command, MemorySaver } from '@langchain/langgraph'

const review = (s: { draft: string }) => {
const decision = interrupt({ draft: s.draft })  // 여기서 정지
return { approved: decision === 'ok' }
}

const app = g.compile({ checkpointer: new MemorySaver() })
const config = { configurable: { thread_id: 't1' } }
await app.invoke({ draft: '...' }, config)
const final = await app.invoke(new Command({ resume: 'ok' }), config)

시간여행 — 과거로 되감기

가 있으면 이 공짜다. get_state_history로 과거 스냅샷을 받고, 그중 하나의 config를 그대로 invoke에 넘기면 그 시점부터 다시 실행한다. 디버깅과 분기 실험에 강력하다.

update_state로 과거 상태를 살짝 고친 뒤 재실행하면, “만약 그때 답이 달랐다면?”을 그래프 위에서 직접 확인한다. 의 결정을 사후 검증하는 표준 워크플로다.

# 1) 히스토리 훑기
history = list(app.get_state_history(config))
target = history[2]  # 세 번째로 오래된 스냅샷

# 2) 과거 상태를 살짝 수정
new_cfg = app.update_state(
  target.config,
  {"messages": [{"role": "user", "content": "다른 질문"}]},
)

# 3) 그 자리에서 재실행
app.invoke(None, new_cfg)

// 1) 히스토리 훑기
const history: any[] = []
for await (const s of app.getStateHistory(config)) history.push(s)
const target = history[2]

// 2) 과거 상태를 살짝 수정
const newCfg = await app.updateState(target.config, {
messages: [{ role: 'user', content: '다른 질문' }],
})

// 3) 그 자리에서 재실행
await app.invoke(null, newCfg)

장기 메모리 — Store

스레드 간을 가로지르는 기억은 가 못 한다. 그래서 v0.2부터 가 1급 API로 들어왔다.

세션이 바뀌어도 살아남아야 할 사용자 프로필·선호·요약 — 이걸 다 로 본다. 자세한 백엔드는 22장에서 다룬다.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

def remember(state, *, store):
  # 네임스페이스 (user_id,)에 키-값 저장
  store.put(("user-42",), key="preference", value={"lang": "ko"})
  return {}

def recall(state, *, store):
  item = store.get(("user-42",), "preference")
  return {"pref": item.value if item else None}

app = g.compile(checkpointer=InMemorySaver(), store=store)

import { InMemoryStore, MemorySaver } from '@langchain/langgraph'

const store = new InMemoryStore()

const remember = async (_s: unknown, cfg: { store: InMemoryStore }) => {
await cfg.store.put(['user-42'], 'preference', { lang: 'ko' })
return {}
}

const recall = async (_s: unknown, cfg: { store: InMemoryStore }) => {
const item = await cfg.store.get(['user-42'], 'preference')
return { pref: item?.value ?? null }
}

const app = g.compile({ checkpointer: new MemorySaver(), store })

스트리밍 — 세 가지 모드

은 stream(...)에 stream_mode 인자를 바꿔 끼우면 모드가 바뀐다. 진행 중에 무엇을 흘려보낼지 결정한다.

모드	단위	용도
`values`	매 단계 전체	디버깅·트레이스
`updates`	별 변경분	일반 UI 업데이트
`messages`	LLM	챗 UI 실시간 출력

여러 모드를 동시에 받을 수도 있다. stream_mode=["updates","messages"]처럼 리스트로 넘긴다. 로 그대로 흘려보내면 챗봇 UI가 완성된다.

스트리밍 코드

astream 한 호출에 여러 모드를 끼우면 한 이터레이터로 업데이트와 LLM 토큰이 함께 흘러나온다. 엔드포인트에 붙이기 가장 쉬운 형태다.

# 노드별 업데이트 + 토큰을 함께 받는다
async for kind, chunk in app.astream(
  {"messages": [{"role": "user", "content": "안녕"}]},
  config,
  stream_mode=["updates", "messages"],
):
  if kind == "messages":
      token, meta = chunk  # (AIMessageChunk, metadata)
      print(token.content, end="", flush=True)
  else:
      print("\n[update]", chunk)

// 노드별 업데이트 + 토큰을 함께 받는다
const stream = await app.stream(
{ messages: [{ role: 'user', content: '안녕' }] },
{ ...config, streamMode: ['updates', 'messages'] },
)
for await (const [kind, chunk] of stream) {
if (kind === 'messages') {
  const [token] = chunk as [{ content: string }, unknown]
  process.stdout.write(token.content)
} else {
  console.log('\n[update]', chunk)
}
}

ToolNode와 prebuilt

는 결과를 자동으로 tool 로 빚어 상태에 꽂아 준다. 의 create_react_agent는 LLM·도구·메모리·인터럽트를 한 함수로 묶는다.

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_react_agent(
  model="anthropic:claude-sonnet-4-5",
  tools=[add],
  checkpointer=InMemorySaver(),
  interrupt_before=["tools"],  # 도구 직전마다 사람 승인
)

import { createReactAgent } from '@langchain/langgraph/prebuilt'
import { MemorySaver } from '@langchain/langgraph'

const agent = createReactAgent({
llm: new ChatAnthropic({ model: 'claude-sonnet-4-5' }),
tools: [add],
checkpointSaver: new MemorySaver(),
interruptBefore: ['tools'],
})

운영 체크리스트

가 커지면 다음을 점검한다.

백엔드를 Postgres로 옮겼는가
ID 정책이 명확한가 (사용자·세션·요청 중 무엇)
에 들어갈 데이터와 에 머무를 데이터의 경계가 분명한가
모드가 UI 요구와 맞는가 (updates vs messages)
지점이 감사 로그로 남는가

이 다섯이 통과되면 LangGraph 프로덕션은 거의 끝났다.

다음 장으로

는 그래프 중심. 다음은 역할 중심 — 로 가자. 같은 멀티 문제를 “역할과 태스크”로 바라본다.