LangGraph란
LangChain 위에 구축된 상태 기반 그래프 프레임워크입니다. 단순한 체인(chain)으로는 표현하기 어려운 복잡한 AI 워크플로우를 노드(node)와 엣지(edge)로 설계할 수 있습니다. 조건 분기, 반복, 병렬 실행 같은 패턴이 필요할 때 LangGraph가 힘을 발휘합니다.
LangChain의 체인이 “직선 파이프라인”이라면, LangGraph는 “순환이 가능한 그래프”입니다.
| 특성 | LangChain 체인 | LangGraph |
|---|---|---|
| 실행 흐름 | 직선 (A → B → C) | 그래프 (분기, 반복, 순환) |
| 상태 관리 | 딕셔너리 전달 | TypedDict 기반 상태 객체 |
| 조건 분기 | RunnableBranch (제한적) | 조건부 엣지 (자유로움) |
| 인간 개입 | 어려움 | interrupt/resume 내장 |
| 적합한 용도 | 단순 QA, 요약 | 에이전트, 멀티스텝 워크플로우 |
핵심 개념
LangGraph의 세 가지 핵심 구성 요소를 이해하면 어떤 워크플로우든 설계할 수 있습니다.
State(상태): 그래프 전체에서 공유되는 데이터. TypedDict로 정의하며, 각 노드가 이 상태를 읽고 업데이트합니다.
Node(노드): 실제 작업을 수행하는 함수. LLM 호출, API 요청, 데이터 처리 등 무엇이든 될 수 있습니다.
Edge(엣지): 노드 간 연결. 무조건 이동(일반 엣지)과 조건에 따른 분기(조건부 엣지)가 있습니다.
설치와 기본 구조
pip install langgraph langchain-openai가장 단순한 LangGraph 예제로 구조를 파악해봅시다.
# LangGraph 기본 구조 — 두 노드를 연결하는 그래프
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 1단계: 상태 정의
class State(TypedDict):
question: str # 사용자 질문
category: str # 분류 결과
answer: str # 최종 답변
# 2단계: 노드 함수 정의
def classify(state: State) -> dict:
"""질문을 카테고리로 분류"""
question = state["question"]
# 간단한 키워드 분류 (실전에서는 LLM 사용)
if "코드" in question or "에러" in question:
category = "technical"
else:
category = "general"
return {"category": category}
def generate_answer(state: State) -> dict:
"""카테고리에 따라 답변 생성"""
if state["category"] == "technical":
answer = f"기술 질문입니다: {state['question']} → 코드 예제와 함께 답변합니다."
else:
answer = f"일반 질문입니다: {state['question']} → 개념 위주로 답변합니다."
return {"answer": answer}
# 3단계: 그래프 구성
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("answer", generate_answer)
# 4단계: 엣지 연결
graph.add_edge(START, "classify") # 시작 → 분류
graph.add_edge("classify", "answer") # 분류 → 답변
graph.add_edge("answer", END) # 답변 → 종료
# 5단계: 컴파일 및 실행
app = graph.compile()
result = app.invoke({"question": "Python 에러 해결 방법", "category": "", "answer": ""})
print(result["answer"])
# 기술 질문입니다: Python 에러 해결 방법 → 코드 예제와 함께 답변합니다.StateGraph에 노드를 등록하고, 엣지로 연결한 뒤, compile()로 실행 가능한 앱을 만듭니다. 이 패턴은 모든 LangGraph 프로젝트의 기본 골격입니다.
조건부 엣지 — 분기 로직
실전에서는 상태에 따라 다른 경로로 이동해야 합니다. add_conditional_edges로 라우팅 함수를 지정합니다.
# 조건부 엣지로 분기하는 리서치 에이전트
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
class ResearchState(TypedDict):
query: str
search_results: str
quality: str # "good" 또는 "poor"
final_report: str
retry_count: int
def search(state: ResearchState) -> dict:
"""웹 검색 수행 (시뮬레이션)"""
query = state["query"]
# 실전에서는 Tavily, Exa 등 검색 API 호출
results = f"'{query}'에 대한 검색 결과 3건"
return {"search_results": results}
def evaluate(state: ResearchState) -> dict:
"""검색 결과 품질 평가"""
# LLM으로 품질 판단 (여기서는 재시도 횟수로 시뮬레이션)
retry = state.get("retry_count", 0)
quality = "good" if retry >= 1 else "poor"
return {"quality": quality, "retry_count": retry + 1}
def refine_query(state: ResearchState) -> dict:
"""검색어 개선"""
return {"query": f"{state['query']} (상세)"}
def write_report(state: ResearchState) -> dict:
"""최종 보고서 작성"""
return {"final_report": f"리서치 보고서: {state['search_results']} — 품질: {state['quality']}"}
# 라우팅 함수: 품질에 따라 분기
def route_by_quality(state: ResearchState) -> Literal["refine", "report"]:
if state["quality"] == "poor":
return "refine" # 품질 나쁨 → 검색어 개선 후 재검색
return "report" # 품질 좋음 → 보고서 작성
# 그래프 구성
graph = StateGraph(ResearchState)
graph.add_node("search", search)
graph.add_node("evaluate", evaluate)
graph.add_node("refine", refine_query)
graph.add_node("report", write_report)
graph.add_edge(START, "search")
graph.add_edge("search", "evaluate")
# 조건부 엣지: evaluate 후 품질에 따라 분기
graph.add_conditional_edges(
"evaluate",
route_by_quality,
{"refine": "refine", "report": "report"},
)
graph.add_edge("refine", "search") # 개선 후 재검색 (순환!)
graph.add_edge("report", END)
app = graph.compile()
result = app.invoke({
"query": "LangGraph 사용법",
"search_results": "",
"quality": "",
"final_report": "",
"retry_count": 0,
})
print(result["final_report"])
# 리서치 보고서: 'LangGraph 사용법 (상세)'에 대한 검색 결과 3건 — 품질: good핵심은 evaluate → refine → search → evaluate로 **순환(cycle)**이 가능하다는 점입니다. 품질이 충족될 때까지 자동으로 재시도합니다. 이것이 단순 체인과 LangGraph의 가장 큰 차이입니다.
LLM 도구 호출 에이전트
LangGraph의 가장 대표적인 패턴은 **도구 호출 에이전트(tool-calling agent)**입니다. LLM이 도구 사용 여부를 스스로 결정하고, 결과를 받아 다음 행동을 판단합니다.
# LangGraph 도구 호출 에이전트
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
# 도구 정의
@tool
def get_weather(city: str) -> str:
"""도시의 현재 날씨를 조회합니다."""
# 실전에서는 날씨 API 호출
weather_data = {"서울": "맑음 22°C", "부산": "흐림 19°C", "제주": "비 17°C"}
return weather_data.get(city, f"{city}: 정보 없음")
@tool
def get_population(city: str) -> str:
"""도시의 인구를 조회합니다."""
pop_data = {"서울": "약 950만명", "부산": "약 330만명", "제주": "약 68만명"}
return pop_data.get(city, f"{city}: 정보 없음")
# 상태: 메시지 리스트 (add_messages로 누적)
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# LLM에 도구 바인딩
tools = [get_weather, get_population]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
def call_model(state: AgentState) -> dict:
"""LLM 호출 — 도구를 쓸지 직접 답할지 LLM이 결정"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 그래프 구성
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools)) # 도구 실행 노드
graph.add_edge(START, "agent")
# agent 노드 후: 도구 호출이 있으면 tools로, 없으면 종료
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent") # 도구 결과를 다시 agent에게
app = graph.compile()
# 실행
result = app.invoke({
"messages": [HumanMessage(content="서울 날씨와 인구를 알려줘")]
})
# 최종 응답 출력
for msg in result["messages"]:
print(f"[{msg.type}] {msg.content[:100]}")
# [human] 서울 날씨와 인구를 알려줘
# [ai] ← 도구 호출 결정 (content 비어있음)
# [tool] 맑음 22°C
# [tool] 약 950만명
# [ai] 서울의 현재 날씨는 맑음 22°C이며, 인구는 약 950만명입니다.tools_condition은 LangGraph가 제공하는 내장 라우터로, LLM 응답에 도구 호출(tool_calls)이 포함되어 있으면 tools 노드로, 아니면 END로 보냅니다.
인간 개입 (Human-in-the-Loop)
민감한 작업 전에 사용자 확인을 받아야 할 때 interrupt 기능을 사용합니다.
# 인간 개입 패턴 (개념 코드)
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# 체크포인터 — 상태를 저장하여 중단/재개 가능
checkpointer = MemorySaver()
# 그래프 컴파일 시 interrupt_before 지정
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["tools"], # tools 노드 실행 전 중단
)
# 첫 실행 — tools 노드 직전에 멈춤
config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke(
{"messages": [HumanMessage(content="서울 날씨 알려줘")]},
config=config,
)
# → agent가 도구 호출을 결정하지만, 실행 전에 중단됨
# 사용자가 확인 후 재개
result = app.invoke(None, config=config) # None으로 이어서 실행
# → tools 노드 실행 → agent → 최종 답변interrupt_before로 특정 노드 실행 전에 그래프를 멈출 수 있습니다. 체크포인터가 상태를 저장하므로, 나중에 invoke(None)으로 중단 지점부터 이어서 실행할 수 있습니다.
설계 팁과 주의사항
무한 루프 방지: 순환 그래프에서는 반드시 탈출 조건을 넣어야 합니다. retry_count 같은 카운터를 두거나, recursion_limit 파라미터를 설정하세요.
# recursion_limit으로 최대 반복 횟수 제한
result = app.invoke(initial_state, {"recursion_limit": 10})상태 설계가 핵심: 그래프의 모든 노드가 공유하는 State를 잘 설계해야 합니다. 필요한 데이터만 포함하고, Annotated로 리듀서(add_messages 등)를 지정하여 상태 업데이트 방식을 명확히 하세요.
디버깅: app.get_graph().draw_mermaid()로 그래프를 시각화할 수 있습니다. 복잡한 워크플로우에서 엣지 연결이 올바른지 확인하는 데 유용합니다.
LangGraph vs 직접 구현: 노드가 3개 이하이고 분기가 없다면 LangChain 체인이나 순수 Python으로 충분합니다. LangGraph는 조건 분기, 순환, 인간 개입이 필요한 복잡한 워크플로우에서 진가를 발휘합니다.