LangGraph 1.0/1.1의 StateGraph, 듀러블 상태, 조건부 엣지, 휴먼인더루프, type-safe 스트리밍을 실전 예제와 함께 분석합니다.
2장에서 다룬 LCEL은 순차적, 병렬적 체인을 구성하는 데 탁월합니다. 그러나 에이전트 워크플로우에서는 LCEL만으로 표현하기 어려운 패턴들이 존재합니다.
LangGraph는 이러한 패턴을 방향성 그래프(Directed Graph)로 표현합니다. 노드는 처리 단계, 엣지는 전이 조건을 나타내며, 상태가 그래프를 따라 흐르는 구조입니다.
StateGraph는 LangGraph의 핵심 구성 요소입니다. 상태(State)를 정의하고, 노드(Node)가 상태를 변환하며, 엣지(Edge)가 전이 조건을 결정합니다.
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""에이전트 상태 스키마"""
messages: Annotated[list, add_messages]
current_step: str
retry_count: intAnnotated 타입과 리듀서 함수(add_messages)를 사용하여 상태 업데이트 방식을 선언적으로 정의합니다. add_messages는 기존 메시지 리스트에 새 메시지를 추가하는 리듀서입니다.
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
model = ChatOpenAI(model="gpt-4o")
# 노드 함수 정의
def call_agent(state: AgentState) -> dict:
"""에이전트 노드: LLM을 호출하여 응답 생성"""
response = model.invoke(state["messages"])
return {"messages": [response], "current_step": "agent"}
def call_tools(state: AgentState) -> dict:
"""도구 노드: 마지막 메시지의 도구 호출 실행"""
last_message = state["messages"][-1]
# 도구 실행 로직
results = execute_tools(last_message.tool_calls)
return {"messages": results, "current_step": "tools"}
# 라우팅 함수
def should_call_tools(state: AgentState) -> str:
"""도구 호출이 필요한지 판단"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("agent", call_agent)
graph.add_node("tools", call_tools)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_call_tools)
graph.add_edge("tools", "agent") # 도구 결과 후 에이전트로 복귀
app = graph.compile()StateGraph의 노드 함수는 전체 상태를 받아 변경할 부분만 딕셔너리로 반환합니다. 반환된 값은 리듀서를 통해 기존 상태와 병합됩니다. 이 패턴 덕분에 각 노드가 독립적으로 상태의 일부만 업데이트할 수 있습니다.
대화형 에이전트처럼 메시지 리스트가 유일한 상태인 경우, MessageGraph를 사용하면 보일러플레이트를 줄일 수 있습니다.
from langgraph.graph import MessageGraph
graph = MessageGraph()
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)
graph.add_conditional_edges("agent", should_call_tools)
graph.add_edge("tools", "agent")
app = graph.compile()
# 메시지 리스트를 직접 전달
result = app.invoke([HumanMessage("서울의 오늘 날씨는?")])LangGraph의 차별화된 기능 중 하나가 듀러블 상태(Durable State)입니다. 워크플로우의 각 단계에서 자동으로 상태를 저장(체크포인팅)하여, 장시간 실행이나 실패 복구를 지원합니다.
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
# SQLite 체크포인터 (개발/테스트용)
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
app = graph.compile(checkpointer=checkpointer)
# thread_id로 대화 세션 관리
config = {"configurable": {"thread_id": "user-123"}}
# 첫 번째 메시지
result = app.invoke(
{"messages": [HumanMessage("프로젝트 일정을 잡아줘")]},
config=config,
)
# 같은 thread_id로 이어서 대화 (이전 상태 자동 복원)
result = app.invoke(
{"messages": [HumanMessage("다음 주 월요일로 변경해줘")]},
config=config,
)# 특정 스레드의 상태 이력 조회
for state in app.get_state_history(config):
print(f"Step: {state.metadata['step']}")
print(f"Messages: {len(state.values['messages'])}")
print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
print("---")
# 특정 체크포인트로 롤백
past_config = {
"configurable": {
"thread_id": "user-123",
"checkpoint_id": "checkpoint-abc",
}
}
result = app.invoke(
{"messages": [HumanMessage("다시 시작하겠습니다")]},
config=past_config,
)프로덕션 환경에서는 PostgresSaver를 사용하세요. SQLite는 동시성 제한이 있어 멀티 유저 환경에서는 적합하지 않습니다. PostgresSaver는 연결 풀링과 비동기 I/O를 지원합니다.
def route_by_intent(state: AgentState) -> str:
"""의도에 따라 다른 노드로 라우팅"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
elif state.get("needs_human_approval"):
return "human_review"
elif state["retry_count"] > 3:
return "fallback"
else:
return END
graph.add_conditional_edges(
"agent",
route_by_intent,
{
"tools": "tools",
"human_review": "human_review",
"fallback": "fallback_handler",
END: END,
},
)에이전트가 도구를 호출하고 결과를 평가한 뒤 다시 도구를 호출하는 사이클은 LangGraph에서 자연스럽게 표현됩니다. 단, 무한 루프를 방지하기 위한 제한이 필요합니다.
def check_iteration(state: AgentState) -> str:
"""반복 횟수를 확인하여 종료 여부 결정"""
if state["retry_count"] >= 5:
return "max_retries_exceeded"
return "continue"
# 재시도 카운터 증가 노드
def increment_retry(state: AgentState) -> dict:
return {"retry_count": state["retry_count"] + 1}또한 LangGraph는 recursion_limit 설정을 통해 그래프 레벨에서 최대 재귀 깊이를 제한할 수 있습니다.
result = app.invoke(
{"messages": [HumanMessage("복잡한 분석을 해줘")]},
config={"recursion_limit": 25},
)특정 작업(결제 승인, 민감한 데이터 접근 등)에서 사람의 확인이 필요한 경우, LangGraph의 인터럽트(Interrupt) 기능을 사용합니다.
from langgraph.types import interrupt, Command
def sensitive_action(state: AgentState) -> dict:
"""민감한 작업 실행 전 사람의 승인 요청"""
action = state["pending_action"]
# 실행 중단 및 승인 요청
approval = interrupt(
{
"action": action,
"message": f"다음 작업을 승인하시겠습니까? {action['description']}",
}
)
if approval == "approved":
result = execute_action(action)
return {"messages": [f"작업 완료: {result}"]}
else:
return {"messages": ["작업이 거부되었습니다."]}
# 그래프에 인터럽트 노드 추가
graph.add_node("sensitive_action", sensitive_action)클라이언트 측에서는 중단된 그래프를 재개합니다.
config = {"configurable": {"thread_id": "user-123"}}
# 첫 실행 - 인터럽트에서 중단됨
result = app.invoke(
{"messages": [HumanMessage("계좌에서 1000만원을 이체해줘")]},
config=config,
)
# result에 interrupt 정보 포함
# 사용자 승인 후 재개
result = app.invoke(
Command(resume="approved"),
config=config,
)복잡한 워크플로우를 관리 가능한 단위로 분할하기 위해 서브그래프를 활용합니다.
# 연구 에이전트 서브그래프
research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_web)
research_graph.add_node("summarize", summarize_results)
research_graph.add_edge(START, "search")
research_graph.add_edge("search", "summarize")
research_graph.add_edge("summarize", END)
research_agent = research_graph.compile()
# 작성 에이전트 서브그래프
writing_graph = StateGraph(WritingState)
writing_graph.add_node("draft", write_draft)
writing_graph.add_node("review", review_draft)
writing_graph.add_edge(START, "draft")
writing_graph.add_edge("draft", "review")
writing_graph.add_edge("review", END)
writing_agent = writing_graph.compile()
# 메인 그래프에서 서브그래프를 노드로 사용
main_graph = StateGraph(MainState)
main_graph.add_node("research", research_agent)
main_graph.add_node("writing", writing_agent)
main_graph.add_edge(START, "research")
main_graph.add_edge("research", "writing")
main_graph.add_edge("writing", END)서브그래프의 상태 스키마가 메인 그래프와 다를 경우, 노드 함수에서 명시적으로 상태를 변환해야 합니다. 서브그래프 간 상태 매핑이 불명확하면 런타임 에러가 발생할 수 있습니다.
LangGraph 1.1은 Pydantic과 dataclass를 활용한 타입 안전 스트리밍을 도입했습니다.
from pydantic import BaseModel
class StreamEvent(BaseModel):
node: str
data: dict
metadata: dict
# 이벤트 스트리밍
async for event in app.astream_events(
{"messages": [HumanMessage("분석해줘")]},
config=config,
version="v2",
):
if event["event"] == "on_chat_model_stream":
# 토큰 단위 스트리밍
print(event["data"]["chunk"].content, end="", flush=True)
elif event["event"] == "on_chain_end":
# 노드 완료 이벤트
print(f"\nNode {event['name']} completed")LangGraph 애플리케이션을 LangSmith Fleet으로 배포하면 에이전트 아이덴티티, 공유, 권한 관리가 가능합니다.
# langgraph.json
{
"graphs": {
"my_agent": {
"graph": "src.agent:graph",
"config": {
"model": "gpt-4o",
"temperature": 0.7
}
}
}
}# LangGraph CLI로 배포
langgraph deploy --config langgraph.json4장에서는 데이터 중심의 접근 방식을 취하는 LlamaIndex를 분석합니다. 데이터 커넥터와 노드 파서, 다양한 인덱스 유형, 쿼리 엔진의 작동 원리, 그리고 이벤트 드리븐 Workflows 1.0까지 살펴보겠습니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
LlamaIndex의 데이터 커넥터, 인덱스 유형, 쿼리 엔진, 그리고 이벤트 드리븐 Workflows 1.0을 실전 예제와 함께 분석합니다.
LangChain 1.0의 아키텍처, LCEL 파이프 문법, 미들웨어, 콘텐츠 블록, OpenTelemetry 통합을 실전 예제와 함께 분석합니다.
Microsoft Semantic Kernel의 멀티 언어 아키텍처, 플러그인 시스템, 플래너, Azure 통합, 엔터프라이즈 보안과 거버넌스를 분석합니다.