LangGraph Docs 한글 번역

이강현·2025년 7월 21일

LangGraph 완전 가이드 (한국어)

개발자를 위한 LangGraph 공식 문서 번역본
원본: https://langchain-ai.github.io/langgraph/guides/
번역: Manus AI
최종 업데이트: 2025년 7월 21일

LangGraph 소개

LangGraph는 대형 언어 모델(LLM)을 사용하여 상태 저장형 멀티 액터 애플리케이션을 구축하기 위한 라이브러리입니다. 이는 에이전트와 멀티 에이전트 워크플로우를 만들기 위해 설계되었으며, 사이클과 제어 가능성에 중점을 둡니다.

주요 특징

LangGraph는 다음과 같은 핵심 가치를 제공합니다:

사이클 지원: 대부분의 에이전트 아키텍처에서 필수적인 사이클을 지원합니다. 이는 DAG(Directed Acyclic Graph) 기반 솔루션과 차별화되는 핵심 기능입니다.

제어 가능성: 에이전트가 어떻게 작동하는지에 대한 매우 낮은 수준의 제어를 제공합니다. 이는 에이전트의 안정성과 신뢰성을 보장하는 데 중요합니다.

지속성: 그래프의 상태를 자동으로 저장하여 오류 복구, 인간 개입, 시간 여행 디버깅 등을 지원합니다.

인간 개입: 에이전트가 작업을 수행하기 전에 인간의 승인을 받거나 수정할 수 있는 기능을 제공합니다.

스트리밍: 실시간으로 에이전트의 진행 상황을 모니터링할 수 있습니다.

언제 LangGraph를 사용해야 하는가?

LangGraph는 다음과 같은 경우에 특히 유용합니다:

복잡한 에이전트 워크플로우: 여러 단계와 조건부 로직이 필요한 경우
멀티 에이전트 시스템: 여러 에이전트가 협력해야 하는 경우
인간 개입이 필요한 시스템: 중요한 결정에 인간의 승인이 필요한 경우
상태 관리가 중요한 애플리케이션: 대화 기록이나 컨텍스트를 유지해야 하는 경우
오류 복구가 중요한 시스템: 실패 시 이전 상태로 복구해야 하는 경우

에이전트 개발

에이전트란 무엇인가?

에이전트는 세 가지 핵심 구성 요소로 이루어집니다:

대형 언어 모델(LLM): 추론과 의사결정을 담당하는 핵심 엔진
도구(Tools): 에이전트가 외부 시스템과 상호작용할 수 있게 하는 함수들
프롬프트(Prompt): 에이전트의 행동을 안내하는 지시사항

에이전트는 다음과 같은 루프로 작동합니다:

1. 현재 상황 분석
2. 사용할 도구 선택
3. 도구 실행 및 결과 관찰
4. 결과를 바탕으로 다음 행동 결정
5. 목표 달성까지 반복

사전 구축된 컴포넌트 사용하기

LangGraph는 빠른 개발을 위한 사전 구축된 컴포넌트들을 제공합니다. 이를 통해 복잡한 오케스트레이션, 메모리 관리, 인간 피드백 처리를 직접 구현할 필요 없이 에이전트를 구축할 수 있습니다.

기본 에이전트 생성

가장 간단한 에이전트를 만드는 방법:

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# LLM 모델 설정
model = ChatOpenAI("gpt-4")

# 도구 정의
def get_weather(location: str) -> str:
    """특정 위치의 날씨 정보를 가져옵니다."""
    # 실제 날씨 API 호출 로직
    return f"{location}의 현재 날씨는 맑음입니다."

# 에이전트 생성
agent = create_react_agent(
    model,
    tools=[get_weather],
)

# 에이전트 실행
response = agent.invoke({
    "messages": [{"role": "user", "content": "서울의 날씨는 어때?"}]
})

고급 에이전트 구성

더 복잡한 에이전트를 위한 추가 옵션들:

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

class WeatherResponse(BaseModel):
    location: str
    temperature: float
    condition: str
    humidity: int

def pre_model_hook(messages):
    """모델 호출 전 메시지 전처리"""
    # 메시지 압축이나 컨텍스트 정리
    return messages

def post_model_hook(response):
    """모델 응답 후 후처리"""
    # 가드레일이나 추가 검증 로직
    return response

agent = create_react_agent(
    model=ChatOpenAI("gpt-4"),
    tools=[get_weather],
    pre_model_hook=pre_model_hook,
    post_model_hook=post_model_hook,
    response_format=WeatherResponse,  # 구조화된 출력
)

에이전트 실행하기

기본 실행 방법

에이전트는 동기와 비동기 두 가지 방식으로 실행할 수 있습니다:

동기 실행:

# 전체 응답 받기
response = agent.invoke({
    "messages": [{"role": "user", "content": "안녕하세요"}]
})

# 스트리밍으로 실행
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "안녕하세요"}]
}):
    print(chunk)

비동기 실행:

import asyncio

async def run_agent():
    # 비동기 전체 응답
    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": "안녕하세요"}]
    })
    
    # 비동기 스트리밍
    async for chunk in agent.astream({
        "messages": [{"role": "user", "content": "안녕하세요"}]
    }):
        print(chunk)

asyncio.run(run_agent())

입력 형식

에이전트 입력은 다양한 형식을 지원합니다:

형식	예시	설명
문자열	`{"messages": "안녕하세요"}`	HumanMessage로 자동 변환
메시지 딕셔너리	`{"messages": {"role": "user", "content": "안녕하세요"}}`	단일 메시지
메시지 배열	`{"messages": [{"role": "user", "content": "안녕하세요"}]}`	여러 메시지
사용자 정의 상태	`{"messages": [...], "user_id": "123"}`	추가 컨텍스트 포함

출력 형식

에이전트 출력은 다음을 포함합니다:

messages: 전체 대화 기록 (사용자 입력, 에이전트 응답, 도구 호출 등)
structured_response: 구조화된 출력이 설정된 경우
사용자 정의 상태 필드들 (설정된 경우)

실행 제어

무한 루프를 방지하기 위한 실행 제한 설정:

from langgraph.errors import GraphRecursionError

max_iterations = 5
recursion_limit = 2 * max_iterations + 1

try:
    response = agent.invoke(
        {"messages": [{"role": "user", "content": "복잡한 작업을 해주세요"}]},
        {"recursion_limit": recursion_limit}
    )
except GraphRecursionError:
    print("최대 반복 횟수에 도달했습니다.")

패키지 생태계

LangGraph는 다양한 용도에 맞는 전문 패키지들을 제공합니다:

패키지	용도	설치 명령
`langgraph-prebuilt`	기본 에이전트 생성	`pip install -U langgraph langchain`
`langgraph-supervisor`	수퍼바이저 에이전트	`pip install -U langgraph-supervisor`
`langgraph-swarm`	스웜 멀티 에이전트	`pip install -U langgraph-swarm`
`langchain-mcp-adapters`	MCP 서버 통합	`pip install -U langchain-mcp-adapters`
`langmem`	메모리 관리	`pip install -U langmem`
`agentevals`	성능 평가	`pip install -U agentevals`

실제 사용 예제

1. 간단한 질의응답 에이전트

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

def search_web(query: str) -> str:
    """웹에서 정보를 검색합니다."""
    # 실제 검색 API 호출
    return f"'{query}'에 대한 검색 결과입니다."

def calculate(expression: str) -> str:
    """수학 계산을 수행합니다."""
    try:
        result = eval(expression)
        return f"계산 결과: {result}"
    except:
        return "계산할 수 없는 식입니다."

agent = create_react_agent(
    ChatOpenAI("gpt-4"),
    tools=[search_web, calculate]
)

# 사용
response = agent.invoke({
    "messages": [{"role": "user", "content": "2024년 올림픽은 어디서 열렸고, 참가국 수를 2로 나눈 값은?"}]
})

2. 상태를 가진 대화 에이전트

from typing_extensions import TypedDict
from typing import Annotated
from operator import add

class ConversationState(TypedDict):
    messages: list
    user_preferences: dict
    conversation_history: Annotated[list, add]

def personalized_response(state: ConversationState):
    """사용자 선호도를 고려한 개인화된 응답"""
    preferences = state.get("user_preferences", {})
    # 개인화 로직
    return {"messages": [...]}

# 상태 기반 에이전트 구성
# (더 복잡한 예제는 Graph API 섹션에서 다룹니다)

LangGraph APIs

LangGraph는 워크플로우를 구축하기 위한 세 가지 주요 API를 제공합니다:

Graph API: 그래프 패러다임을 사용한 저수준 제어
Functional API: 함수형 패러다임을 사용한 고수준 추상화
Runtime: 실행 환경 관리

Graph API

Graph API는 LangGraph의 핵심으로, 에이전트 워크플로우를 그래프로 모델링합니다.

핵심 개념

1. 상태(State)
그래프의 현재 스냅샷을 나타내는 공유 데이터 구조입니다.

from typing_extensions import TypedDict
from typing import Annotated
from operator import add

class AgentState(TypedDict):
    messages: list  # 대화 메시지들
    user_input: str  # 사용자 입력
    current_step: str  # 현재 단계
    results: Annotated[list, add]  # 누적 결과 (리듀서 사용)

2. 노드(Nodes)
실제 작업을 수행하는 Python 함수들입니다.

def analyze_input(state: AgentState) -> AgentState:
    """사용자 입력을 분석하는 노드"""
    user_input = state["user_input"]
    
    # 입력 분석 로직
    analysis_result = f"분석 완료: {user_input}"
    
    return {
        "current_step": "analysis_complete",
        "results": [analysis_result]
    }

def generate_response(state: AgentState) -> AgentState:
    """응답을 생성하는 노드"""
    results = state["results"]
    
    # 응답 생성 로직
    response = f"결과를 바탕으로 한 응답: {results}"
    
    return {
        "messages": [{"role": "assistant", "content": response}],
        "current_step": "complete"
    }

3. 엣지(Edges)
다음에 실행할 노드를 결정하는 함수들입니다.

def should_continue(state: AgentState) -> str:
    """다음 단계를 결정하는 조건부 엣지"""
    current_step = state["current_step"]
    
    if current_step == "analysis_complete":
        return "generate_response"
    elif current_step == "complete":
        return "END"
    else:
        return "analyze_input"

그래프 구축하기

from langgraph.graph import StateGraph, START, END

# 1. 그래프 생성
workflow = StateGraph(AgentState)

# 2. 노드 추가
workflow.add_node("analyze_input", analyze_input)
workflow.add_node("generate_response", generate_response)

# 3. 엣지 추가
workflow.add_edge(START, "analyze_input")
workflow.add_conditional_edges(
    "analyze_input",
    should_continue,
    {
        "generate_response": "generate_response",
        "END": END
    }
)
workflow.add_edge("generate_response", END)

# 4. 그래프 컴파일
app = workflow.compile()

고급 상태 관리

리듀서(Reducers) 사용하기

리듀서는 상태 업데이트가 어떻게 적용될지 결정합니다:

from operator import add
from typing import Annotated

class AdvancedState(TypedDict):
    # 기본 리듀서 (덮어쓰기)
    current_user: str
    
    # 추가 리듀서 (리스트에 추가)
    conversation_history: Annotated[list, add]
    
    # 사용자 정의 리듀서
    scores: Annotated[dict, lambda x, y: {**x, **y}]

다중 스키마 사용하기

내부 통신과 외부 인터페이스를 분리할 수 있습니다:

class InputState(TypedDict):
    user_input: str

class OutputState(TypedDict):
    final_response: str

class InternalState(TypedDict):
    user_input: str
    final_response: str
    internal_data: str  # 외부에 노출되지 않음

# 그래프 생성 시 스키마 지정
workflow = StateGraph(
    InternalState,
    input_schema=InputState,
    output_schema=OutputState
)

실제 그래프 예제

복잡한 에이전트 워크플로우 예제:

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Annotated, Literal
from operator import add

class ResearchState(TypedDict):
    query: str
    search_results: Annotated[list, add]
    analysis: str
    final_report: str
    current_step: str

def web_search(state: ResearchState) -> ResearchState:
    """웹 검색 수행"""
    query = state["query"]
    # 실제 검색 로직
    results = [f"검색 결과 1: {query}", f"검색 결과 2: {query}"]
    
    return {
        "search_results": results,
        "current_step": "search_complete"
    }

def analyze_results(state: ResearchState) -> ResearchState:
    """검색 결과 분석"""
    results = state["search_results"]
    analysis = f"분석: {len(results)}개의 결과를 찾았습니다."
    
    return {
        "analysis": analysis,
        "current_step": "analysis_complete"
    }

def generate_report(state: ResearchState) -> ResearchState:
    """최종 보고서 생성"""
    query = state["query"]
    analysis = state["analysis"]
    
    report = f"'{query}'에 대한 연구 보고서:\n{analysis}"
    
    return {
        "final_report": report,
        "current_step": "complete"
    }

def route_next(state: ResearchState) -> Literal["analyze", "report", "end"]:
    """다음 단계 라우팅"""
    step = state["current_step"]
    
    if step == "search_complete":
        return "analyze"
    elif step == "analysis_complete":
        return "report"
    else:
        return "end"

# 그래프 구축
research_workflow = StateGraph(ResearchState)

research_workflow.add_node("search", web_search)
research_workflow.add_node("analyze", analyze_results)
research_workflow.add_node("report", generate_report)

research_workflow.add_edge(START, "search")
research_workflow.add_conditional_edges(
    "search",
    route_next,
    {
        "analyze": "analyze",
        "report": "report",
        "end": END
    }
)
research_workflow.add_conditional_edges(
    "analyze",
    route_next,
    {
        "analyze": "analyze",
        "report": "report", 
        "end": END
    }
)
research_workflow.add_edge("report", END)

research_app = research_workflow.compile()

# 실행
result = research_app.invoke({
    "query": "인공지능의 최신 동향",
    "search_results": [],
    "analysis": "",
    "final_report": "",
    "current_step": "start"
})

Functional API

Functional API는 그래프 구조를 직접 관리하지 않고도 워크플로우를 구축할 수 있게 해줍니다.

from langgraph.functional import task, workflow

@task
def process_input(input_data: str) -> str:
    """입력 처리 태스크"""
    return f"처리된 입력: {input_data}"

@task  
def generate_output(processed_data: str) -> str:
    """출력 생성 태스크"""
    return f"최종 출력: {processed_data}"

@workflow
def simple_workflow(user_input: str) -> str:
    """간단한 워크플로우"""
    processed = process_input(user_input)
    result = generate_output(processed)
    return result

# 실행
result = simple_workflow("안녕하세요")

Runtime (Pregel)

LangGraph의 런타임은 Google의 Pregel 시스템에서 영감을 받았습니다.

실행 모델

슈퍼 스텝: 그래프 노드에 대한 단일 반복
메시지 전달: 노드 간 상태 전달 메커니즘
병렬 실행: 독립적인 노드들의 동시 실행
수렴 조건: 모든 노드가 비활성화될 때 실행 종료

실행 제어

# 체크포인터와 함께 컴파일
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 설정과 함께 실행
config = {
    "configurable": {
        "thread_id": "conversation_1",
        "recursion_limit": 50
    }
}

result = app.invoke(input_data, config)

핵심 기능

LangGraph의 핵심 기능들은 OSS와 Platform 버전 모두에서 사용할 수 있습니다.

스트리밍 (Streaming)

실시간 업데이트를 통해 반응적이고 투명한 사용자 경험을 제공합니다.

스트리밍 유형

워크플로우 진행 상황: 각 노드 실행 후 상태 업데이트
LLM 토큰: 언어 모델 토큰이 생성되는 대로 스트리밍
사용자 정의 업데이트: 사용자 정의 신호 방출

스트리밍 모드

모드	설명	사용 사례
`values`	전체 상태	완전한 상태 추적
`updates`	상태 변화분만	효율적인 업데이트
`messages`	LLM 토큰 + 메타데이터	실시간 대화
`custom`	사용자 정의 데이터	특별한 알림
`debug`	상세한 추적 정보	디버깅

기본 스트리밍 사용법

# 동기 스트리밍
for chunk in app.stream(
    {"messages": [{"role": "user", "content": "안녕하세요"}]},
    stream_mode="updates"
):
    print(f"업데이트: {chunk}")

# 비동기 스트리밍
async for chunk in app.astream(
    {"messages": [{"role": "user", "content": "안녕하세요"}]},
    stream_mode="values"
):
    print(f"현재 상태: {chunk}")

고급 스트리밍 기능

도구에서 진행 상황 알림:

from langgraph.prebuilt import create_react_agent

def long_running_task(query: str) -> str:
    """시간이 오래 걸리는 작업"""
    import time
    
    # 진행 상황 알림
    for i in range(5):
        time.sleep(1)
        # 사용자 정의 스트림 이벤트 방출
        yield f"진행 상황: {i+1}/5 완료"
    
    return f"작업 완료: {query}"

agent = create_react_agent(
    model,
    tools=[long_running_task]
)

# 사용자 정의 스트림 모드로 실행
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "긴 작업을 해주세요"}]},
    stream_mode="custom"
):
    print(chunk)

서브그래프 스트리밍:

# 부모 그래프와 서브그래프 모두에서 스트리밍
for chunk in parent_graph.stream(
    input_data,
    stream_mode="updates",
    subgraphs=True  # 서브그래프 출력도 포함
):
    if chunk.get("subgraph"):
        print(f"서브그래프 업데이트: {chunk}")
    else:
        print(f"메인 그래프 업데이트: {chunk}")

지속성 (Persistence)

체크포인터를 통한 내장 지속성으로 강력한 기능들을 지원합니다.

체크포인터 설정

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# 메모리 체크포인터 (개발용)
memory_checkpointer = InMemorySaver()

# SQLite 체크포인터 (프로덕션용)
sqlite_checkpointer = SqliteSaver("checkpoints.db")

# 그래프 컴파일 시 체크포인터 지정
app = workflow.compile(checkpointer=sqlite_checkpointer)

스레드와 체크포인트

스레드 관리:

# 스레드 ID로 실행
config = {"configurable": {"thread_id": "user_123_conversation"}}

# 첫 번째 실행
result1 = app.invoke({"messages": [{"role": "user", "content": "안녕하세요"}]}, config)

# 같은 스레드에서 계속 실행 (상태 유지됨)
result2 = app.invoke({"messages": [{"role": "user", "content": "이전 대화를 기억하나요?"}]}, config)

상태 조회:

# 현재 상태 가져오기
current_state = app.get_state(config)
print(f"현재 상태: {current_state.values}")
print(f"다음 실행될 노드: {current_state.next}")

# 상태 기록 가져오기
state_history = list(app.get_state_history(config))
for i, state in enumerate(state_history):
    print(f"단계 {i}: {state.values}")

상태 수정:

# 상태 직접 업데이트
app.update_state(
    config,
    {"user_preference": "한국어"},
    as_node="user_input"  # 특정 노드로서 업데이트
)

# 특정 체크포인트에서 재시작
checkpoint_config = {
    "configurable": {
        "thread_id": "user_123",
        "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"
    }
}
result = app.invoke(new_input, checkpoint_config)

메모리 (Memory)

단기 및 장기 메모리를 통한 상태 저장 동작을 지원합니다.

단기 메모리 (세션 기반)

from typing_extensions import TypedDict
from typing import Annotated
from operator import add

class ConversationState(TypedDict):
    messages: Annotated[list, add]  # 대화 기록 누적
    user_context: dict  # 세션 컨텍스트
    current_topic: str  # 현재 주제

def remember_context(state: ConversationState) -> ConversationState:
    """컨텍스트 기억 및 업데이트"""
    messages = state["messages"]
    
    # 최근 메시지에서 주제 추출
    if messages:
        last_message = messages[-1]["content"]
        # 주제 추출 로직
        topic = extract_topic(last_message)
        
        return {
            "current_topic": topic,
            "user_context": {"last_interaction": "now"}
        }
    
    return {}

장기 메모리 (영구 저장)

from langmem import MemoryStore

# 메모리 저장소 설정
memory_store = MemoryStore()

def store_long_term_memory(state: ConversationState) -> ConversationState:
    """장기 메모리에 중요한 정보 저장"""
    user_id = state.get("user_id")
    important_info = extract_important_info(state["messages"])
    
    # 장기 메모리에 저장
    memory_store.put(
        namespace=f"user_{user_id}",
        key="preferences",
        value=important_info
    )
    
    return state

def recall_long_term_memory(state: ConversationState) -> ConversationState:
    """장기 메모리에서 정보 회수"""
    user_id = state.get("user_id")
    
    # 장기 메모리에서 검색
    memories = memory_store.search(
        namespace=f"user_{user_id}",
        query="사용자 선호도",
        limit=5
    )
    
    return {
        "user_context": {"memories": memories}
    }

의미적 메모리 검색

# 벡터 기반 의미적 검색
semantic_memories = memory_store.search(
    namespace="user_123",
    query="사용자가 좋아하는 음식",
    limit=3,
    similarity_threshold=0.8
)

for memory in semantic_memories:
    print(f"관련 기억: {memory.value} (유사도: {memory.score})")

인간 개입 (Human-in-the-loop)

워크플로우의 어느 지점에서든 인간의 승인이나 수정을 받을 수 있습니다.

기본 인간 개입

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

def sensitive_action(state: AgentState) -> AgentState:
    """민감한 작업 수행"""
    action = state["proposed_action"]
    
    # 인간 승인 대기
    return {
        "status": "awaiting_approval",
        "proposed_action": action
    }

def execute_action(state: AgentState) -> AgentState:
    """승인된 작업 실행"""
    if state.get("approved"):
        # 실제 작업 수행
        result = perform_action(state["proposed_action"])
        return {"result": result, "status": "completed"}
    else:
        return {"status": "cancelled"}

# 인터럽트 설정
workflow = StateGraph(AgentState)
workflow.add_node("sensitive_action", sensitive_action)
workflow.add_node("execute_action", execute_action)

# 특정 노드에서 인터럽트
app = workflow.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["execute_action"]  # 실행 전 중단
)

# 실행
config = {"configurable": {"thread_id": "approval_flow"}}
result = app.invoke({"proposed_action": "중요한 작업"}, config)

# 인간이 검토 후 승인
app.update_state(config, {"approved": True})

# 실행 재개
final_result = app.invoke(None, config)

동적 인터럽트

from langgraph.types import interrupt

def conditional_interrupt(state: AgentState) -> AgentState:
    """조건에 따른 동적 인터럽트"""
    risk_level = assess_risk(state["action"])
    
    if risk_level > 0.8:
        # 높은 위험도일 때만 인터럽트
        interrupt("높은 위험도 작업입니다. 승인이 필요합니다.")
    
    return {"risk_assessed": True}

시간 여행 (Time Travel)

그래프 실행의 특정 지점으로 돌아가서 디버깅하거나 다른 경로를 시도할 수 있습니다.

기본 시간 여행

# 상태 기록 확인
history = list(app.get_state_history(config))

# 특정 체크포인트로 돌아가기
target_checkpoint = history[2]  # 3번째 단계로 돌아가기

rollback_config = {
    "configurable": {
        "thread_id": "user_123",
        "checkpoint_id": target_checkpoint.config["configurable"]["checkpoint_id"]
    }
}

# 해당 지점에서 다시 실행
new_result = app.invoke({"new_input": "다른 시도"}, rollback_config)

분기 생성

# 현재 상태에서 새로운 분기 생성
branch_config = {
    "configurable": {
        "thread_id": "user_123_branch_1",  # 새로운 스레드 ID
        "checkpoint_id": current_checkpoint_id
    }
}

# 분기에서 다른 실험 수행
experiment_result = app.invoke({"experiment": "새로운 접근"}, branch_config)

서브그래프 (Subgraphs)

모듈식 그래프 구축을 통해 복잡한 워크플로우를 관리 가능한 단위로 분해할 수 있습니다.

서브그래프 정의

# 서브그래프 정의
def create_analysis_subgraph():
    subgraph = StateGraph(AnalysisState)
    
    subgraph.add_node("collect_data", collect_data)
    subgraph.add_node("process_data", process_data)
    subgraph.add_node("generate_insights", generate_insights)
    
    subgraph.add_edge(START, "collect_data")
    subgraph.add_edge("collect_data", "process_data")
    subgraph.add_edge("process_data", "generate_insights")
    subgraph.add_edge("generate_insights", END)
    
    return subgraph.compile()

# 메인 그래프에서 서브그래프 사용
main_workflow = StateGraph(MainState)
analysis_subgraph = create_analysis_subgraph()

def run_analysis(state: MainState) -> MainState:
    """서브그래프 실행"""
    analysis_input = prepare_analysis_input(state)
    analysis_result = analysis_subgraph.invoke(analysis_input)
    
    return {
        "analysis_complete": True,
        "analysis_results": analysis_result
    }

main_workflow.add_node("analysis", run_analysis)

멀티 에이전트 (Multi-agent)

복잡한 워크플로우를 여러 전문 에이전트로 분해하여 처리할 수 있습니다.

수퍼바이저 패턴

from langgraph_supervisor import create_supervisor

# 전문 에이전트들 정의
research_agent = create_react_agent(model, tools=[web_search, academic_search])
writing_agent = create_react_agent(model, tools=[grammar_check, style_guide])
review_agent = create_react_agent(model, tools=[fact_check, quality_assess])

# 수퍼바이저 생성
supervisor = create_supervisor(
    agents={
        "researcher": research_agent,
        "writer": writing_agent,
        "reviewer": review_agent
    },
    system_prompt="당신은 팀을 조율하는 수퍼바이저입니다."
)

# 멀티 에이전트 워크플로우 실행
result = supervisor.invoke({
    "task": "AI 윤리에 대한 보고서를 작성해주세요",
    "requirements": ["학술적 근거", "명확한 문체", "사실 확인"]
})

스웜 패턴

from langgraph_swarm import create_swarm

# 동등한 에이전트들의 스웜
agent_swarm = create_swarm([
    ("agent_1", specialist_agent_1),
    ("agent_2", specialist_agent_2),
    ("agent_3", specialist_agent_3)
])

# 스웜 실행 (에이전트들이 협력하여 문제 해결)
swarm_result = agent_swarm.invoke({
    "problem": "복잡한 다면적 문제",
    "collaboration_mode": "consensus"
})

플랫폼 전용 기능

LangGraph Platform에서만 사용할 수 있는 고급 기능들입니다.

인증 및 접근 제어

사용자 인증과 권한 관리를 통해 안전한 에이전트 시스템을 구축할 수 있습니다.

# 사용자 인증 설정
from langgraph.platform.auth import authenticate_user

@authenticate_user(required_roles=["admin", "power_user"])
def sensitive_operation(state: AgentState) -> AgentState:
    """관리자만 접근 가능한 작업"""
    return perform_admin_task(state)

어시스턴트 (Assistants)

LangGraph 그래프와 상호작용하는 어시스턴트를 구축할 수 있습니다.

# 어시스턴트 설정
assistant_config = {
    "name": "연구 어시스턴트",
    "description": "학술 연구를 도와주는 AI 어시스턴트",
    "graph": research_workflow,
    "tools": [web_search, paper_search, citation_generator]
}

더블 텍스팅 처리

연속된 메시지를 효율적으로 처리합니다.

# 더블 텍스팅 설정
double_text_config = {
    "merge_strategy": "append",  # 메시지 병합 전략
    "timeout": 2.0,  # 대기 시간 (초)
    "max_messages": 5  # 최대 병합 메시지 수
}

웹훅 (Webhooks)

외부 시스템과의 통합을 위한 웹훅을 설정할 수 있습니다.

# 웹훅 설정
webhook_config = {
    "url": "https://your-system.com/webhook",
    "events": ["graph_complete", "error_occurred"],
    "headers": {"Authorization": "Bearer your-token"}
}

크론 작업 (Cron Jobs)

정기적인 작업을 스케줄링할 수 있습니다.

# 크론 작업 설정
cron_job = {
    "schedule": "0 9 * * 1",  # 매주 월요일 오전 9시
    "graph": daily_report_workflow,
    "input": {"report_type": "weekly_summary"}
}

서버 사용자 정의

LangGraph 그래프를 실행하는 서버를 사용자 정의할 수 있습니다.

데이터 관리

그래프에서 사용하는 데이터를 관리할 수 있습니다.

배포 (Deployment)

LangGraph 그래프를 서버에 배포할 수 있습니다.

실습 예제

예제 1: 고객 서비스 챗봇

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing_extensions import TypedDict
from typing import Annotated, Literal
from operator import add

class CustomerServiceState(TypedDict):
    messages: Annotated[list, add]
    customer_id: str
    issue_type: str
    priority: Literal["low", "medium", "high"]
    resolution_status: str
    agent_notes: Annotated[list, add]

def classify_issue(state: CustomerServiceState) -> CustomerServiceState:
    """고객 문의 분류"""
    last_message = state["messages"][-1]["content"]
    
    # 문의 유형 분류 로직
    if "환불" in last_message or "취소" in last_message:
        issue_type = "refund"
        priority = "high"
    elif "배송" in last_message:
        issue_type = "shipping"
        priority = "medium"
    else:
        issue_type = "general"
        priority = "low"
    
    return {
        "issue_type": issue_type,
        "priority": priority,
        "agent_notes": [f"문의 분류 완료: {issue_type}"]
    }

def handle_refund(state: CustomerServiceState) -> CustomerServiceState:
    """환불 처리"""
    customer_id = state["customer_id"]
    
    # 환불 처리 로직
    refund_result = process_refund(customer_id)
    
    response = f"환불 처리가 완료되었습니다. 처리 번호: {refund_result['transaction_id']}"
    
    return {
        "messages": [{"role": "assistant", "content": response}],
        "resolution_status": "resolved",
        "agent_notes": [f"환불 처리 완료: {refund_result['transaction_id']}"]
    }

def handle_shipping(state: CustomerServiceState) -> CustomerServiceState:
    """배송 문의 처리"""
    customer_id = state["customer_id"]
    
    # 배송 정보 조회
    shipping_info = get_shipping_info(customer_id)
    
    response = f"배송 상태: {shipping_info['status']}, 예상 도착일: {shipping_info['eta']}"
    
    return {
        "messages": [{"role": "assistant", "content": response}],
        "resolution_status": "resolved",
        "agent_notes": [f"배송 정보 제공: {shipping_info['tracking_number']}"]
    }

def handle_general(state: CustomerServiceState) -> CustomerServiceState:
    """일반 문의 처리"""
    last_message = state["messages"][-1]["content"]
    
    # LLM을 사용한 일반 응답 생성
    response = generate_general_response(last_message)
    
    return {
        "messages": [{"role": "assistant", "content": response}],
        "resolution_status": "resolved",
        "agent_notes": ["일반 문의 응답 제공"]
    }

def route_issue(state: CustomerServiceState) -> str:
    """문의 유형에 따른 라우팅"""
    issue_type = state["issue_type"]
    
    if issue_type == "refund":
        return "handle_refund"
    elif issue_type == "shipping":
        return "handle_shipping"
    else:
        return "handle_general"

# 고객 서비스 워크플로우 구축
cs_workflow = StateGraph(CustomerServiceState)

cs_workflow.add_node("classify_issue", classify_issue)
cs_workflow.add_node("handle_refund", handle_refund)
cs_workflow.add_node("handle_shipping", handle_shipping)
cs_workflow.add_node("handle_general", handle_general)

cs_workflow.add_edge(START, "classify_issue")
cs_workflow.add_conditional_edges(
    "classify_issue",
    route_issue,
    {
        "handle_refund": "handle_refund",
        "handle_shipping": "handle_shipping",
        "handle_general": "handle_general"
    }
)
cs_workflow.add_edge("handle_refund", END)
cs_workflow.add_edge("handle_shipping", END)
cs_workflow.add_edge("handle_general", END)

# 체크포인터와 함께 컴파일
checkpointer = SqliteSaver("customer_service.db")
cs_app = cs_workflow.compile(checkpointer=checkpointer)

# 사용 예제
config = {"configurable": {"thread_id": "customer_12345"}}

result = cs_app.invoke({
    "messages": [{"role": "user", "content": "주문한 상품을 환불하고 싶습니다."}],
    "customer_id": "12345",
    "issue_type": "",
    "priority": "low",
    "resolution_status": "pending",
    "agent_notes": []
}, config)

print(f"응답: {result['messages'][-1]['content']}")
print(f"해결 상태: {result['resolution_status']}")

예제 2: 연구 보고서 생성 에이전트

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from typing_extensions import TypedDict
from typing import Annotated
from operator import add

class ResearchState(TypedDict):
    topic: str
    research_query: str
    sources: Annotated[list, add]
    key_findings: Annotated[list, add]
    outline: str
    draft_report: str
    final_report: str
    current_step: str

def generate_research_queries(state: ResearchState) -> ResearchState:
    """연구 주제를 바탕으로 검색 쿼리 생성"""
    topic = state["topic"]
    
    # LLM을 사용하여 다양한 검색 쿼리 생성
    queries = [
        f"{topic} 최신 연구",
        f"{topic} 동향 분석",
        f"{topic} 사례 연구",
        f"{topic} 전문가 의견"
    ]
    
    return {
        "research_query": queries[0],  # 첫 번째 쿼리부터 시작
        "current_step": "query_generated"
    }

def search_sources(state: ResearchState) -> ResearchState:
    """다양한 소스에서 정보 검색"""
    query = state["research_query"]
    
    # 웹 검색
    web_results = web_search(query)
    
    # 학술 논문 검색
    academic_results = academic_search(query)
    
    # 뉴스 검색
    news_results = news_search(query)
    
    all_sources = web_results + academic_results + news_results
    
    return {
        "sources": all_sources,
        "current_step": "sources_collected"
    }

def analyze_sources(state: ResearchState) -> ResearchState:
    """수집된 소스 분석 및 핵심 발견사항 추출"""
    sources = state["sources"]
    
    key_findings = []
    for source in sources:
        # 각 소스에서 핵심 정보 추출
        finding = extract_key_information(source)
        if finding:
            key_findings.append(finding)
    
    return {
        "key_findings": key_findings,
        "current_step": "analysis_complete"
    }

def create_outline(state: ResearchState) -> ResearchState:
    """보고서 개요 작성"""
    topic = state["topic"]
    findings = state["key_findings"]
    
    # 발견사항을 바탕으로 보고서 구조 생성
    outline = generate_report_outline(topic, findings)
    
    return {
        "outline": outline,
        "current_step": "outline_created"
    }

def write_draft(state: ResearchState) -> ResearchState:
    """초안 작성"""
    outline = state["outline"]
    findings = state["key_findings"]
    sources = state["sources"]
    
    # 개요와 발견사항을 바탕으로 초안 작성
    draft = generate_draft_report(outline, findings, sources)
    
    return {
        "draft_report": draft,
        "current_step": "draft_complete"
    }

def review_and_finalize(state: ResearchState) -> ResearchState:
    """검토 및 최종 보고서 완성"""
    draft = state["draft_report"]
    
    # 문법 검사 및 스타일 개선
    reviewed_report = review_and_improve(draft)
    
    # 인용 추가
    final_report = add_citations(reviewed_report, state["sources"])
    
    return {
        "final_report": final_report,
        "current_step": "complete"
    }

def should_continue(state: ResearchState) -> str:
    """다음 단계 결정"""
    step = state["current_step"]
    
    if step == "query_generated":
        return "search"
    elif step == "sources_collected":
        return "analyze"
    elif step == "analysis_complete":
        return "outline"
    elif step == "outline_created":
        return "draft"
    elif step == "draft_complete":
        return "finalize"
    else:
        return "end"

# 연구 워크플로우 구축
research_workflow = StateGraph(ResearchState)

research_workflow.add_node("generate_queries", generate_research_queries)
research_workflow.add_node("search", search_sources)
research_workflow.add_node("analyze", analyze_sources)
research_workflow.add_node("outline", create_outline)
research_workflow.add_node("draft", write_draft)
research_workflow.add_node("finalize", review_and_finalize)

research_workflow.add_edge(START, "generate_queries")
research_workflow.add_conditional_edges(
    "generate_queries",
    should_continue,
    {
        "search": "search",
        "analyze": "analyze",
        "outline": "outline",
        "draft": "draft",
        "finalize": "finalize",
        "end": END
    }
)

# 모든 노드에서 조건부 엣지 추가
for node in ["search", "analyze", "outline", "draft", "finalize"]:
    research_workflow.add_conditional_edges(
        node,
        should_continue,
        {
            "search": "search",
            "analyze": "analyze", 
            "outline": "outline",
            "draft": "draft",
            "finalize": "finalize",
            "end": END
        }
    )

research_app = research_workflow.compile()

# 사용 예제
research_result = research_app.invoke({
    "topic": "인공지능의 윤리적 고려사항",
    "research_query": "",
    "sources": [],
    "key_findings": [],
    "outline": "",
    "draft_report": "",
    "final_report": "",
    "current_step": "start"
})

print("연구 보고서 생성 완료!")
print(f"최종 보고서 길이: {len(research_result['final_report'])} 문자")

참고 자료

공식 문서 및 리소스

패키지 설치

# 기본 LangGraph
pip install -U langgraph langchain

# 추가 패키지들
pip install -U langgraph-supervisor
pip install -U langgraph-swarm  
pip install -U langchain-mcp-adapters
pip install -U langmem
pip install -U agentevals

# 데이터베이스 체크포인터
pip install -U langgraph[sqlite]
pip install -U langgraph[postgres]

주요 개념 요약

개념	설명	사용 시기
State	그래프의 공유 데이터 구조	모든 그래프에서 필수
Nodes	실제 작업을 수행하는 함수	비즈니스 로직 구현
Edges	실행 흐름을 제어하는 함수	조건부 분기가 필요할 때
Checkpointer	상태 지속성 관리	대화 기록, 오류 복구 필요 시
Streaming	실시간 업데이트 제공	사용자 경험 개선 필요 시
Human-in-the-loop	인간 개입 지점 설정	중요한 결정에 승인 필요 시
Memory	단기/장기 기억 관리	개인화, 컨텍스트 유지 필요 시
Subgraphs	모듈식 그래프 구성	복잡한 워크플로우 분해 시
Multi-agent	여러 에이전트 협력	전문화된 역할 분담 필요 시

모범 사례

상태 설계: 필요한 최소한의 정보만 상태에 포함하고, 적절한 리듀서를 사용하세요.
오류 처리: 각 노드에서 예외 상황을 적절히 처리하고, 체크포인터를 활용해 복구 가능하게 만드세요.
성능 최적화: 불필요한 상태 업데이트를 피하고, 병렬 실행 가능한 노드들을 식별하세요.
테스트: 각 노드를 독립적으로 테스트하고, 전체 워크플로우의 다양한 경로를 검증하세요.
모니터링: 스트리밍과 로깅을 활용해 에이전트의 동작을 모니터링하세요.

문제 해결

일반적인 문제들:

무한 루프: recursion_limit 설정으로 방지
상태 충돌: 적절한 리듀서 함수 사용
메모리 누수: 체크포인터 정리 및 상태 크기 관리
성능 저하: 불필요한 상태 업데이트 최소화

디버깅 팁:

# 디버그 모드로 실행
for chunk in app.stream(input_data, stream_mode="debug"):
    print(f"디버그 정보: {chunk}")

# 상태 기록 확인
history = list(app.get_state_history(config))
for i, state in enumerate(history):
    print(f"단계 {i}: {state.values}")

이 가이드를 통해 LangGraph의 강력한 기능들을 활용하여 복잡하고 지능적인 에이전트 시스템을 구축할 수 있습니다. 각 기능을 단계적으로 학습하고 실제 프로젝트에 적용해보시기 바랍니다.

추가 질문이나 도움이 필요하시면 LangGraph 커뮤니티나 공식 문서를 참조하세요.

빠른 시작 가이드

5분 만에 첫 번째 에이전트 만들기

LangGraph를 처음 사용하는 개발자를 위한 단계별 가이드입니다.

1단계: 설치

pip install -U langgraph langchain langchain-openai

2단계: 환경 설정

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

3단계: 첫 번째 에이전트 생성

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# 간단한 도구 정의
def get_current_time() -> str:
    """현재 시간을 반환합니다."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def calculate(expression: str) -> str:
    """수학 계산을 수행합니다."""
    try:
        result = eval(expression)
        return f"계산 결과: {result}"
    except Exception as e:
        return f"계산 오류: {str(e)}"

# LLM 모델 설정
model = ChatOpenAI(model="gpt-3.5-turbo")

# 에이전트 생성
agent = create_react_agent(
    model,
    tools=[get_current_time, calculate]
)

# 에이전트 실행
response = agent.invoke({
    "messages": [{"role": "user", "content": "현재 시간을 알려주고, 2+3을 계산해주세요."}]
})

print(response["messages"][-1]["content"])

4단계: 실행 결과

에이전트가 도구를 사용하여 현재 시간을 확인하고 계산을 수행한 후 결과를 반환합니다.

다음 단계

상태 관리 학습: Graph API를 사용하여 복잡한 워크플로우 구축
지속성 추가: 체크포인터를 사용하여 대화 기록 저장
스트리밍 구현: 실시간 응답으로 사용자 경험 개선
인간 개입 설정: 중요한 결정에 승인 프로세스 추가

자주 묻는 질문 (FAQ)

Q: LangGraph와 LangChain의 차이점은 무엇인가요?

A: LangChain은 LLM 애플리케이션을 위한 일반적인 프레임워크이고, LangGraph는 특히 에이전트와 멀티 에이전트 워크플로우에 특화된 라이브러리입니다. LangGraph는 사이클, 상태 관리, 인간 개입 등 에이전트에 필요한 고급 기능을 제공합니다.

Q: 프로덕션 환경에서 사용할 수 있나요?

A: 네, LangGraph는 프로덕션 환경에서 사용할 수 있도록 설계되었습니다. 체크포인터를 통한 상태 지속성, 오류 복구, 모니터링 등 프로덕션에 필요한 기능들을 제공합니다.

Q: 어떤 LLM 모델을 사용할 수 있나요?

A: LangGraph는 LangChain과 호환되는 모든 LLM 모델을 사용할 수 있습니다. OpenAI, Anthropic, Google, 로컬 모델 등 다양한 모델을 지원합니다.

Q: 비용은 어떻게 관리하나요?

A: 실행 제한(recursion_limit) 설정, 효율적인 프롬프트 설계, 캐싱 활용 등을 통해 비용을 관리할 수 있습니다. 또한 스트리밍을 통해 불필요한 토큰 생성을 줄일 수 있습니다.

Q: 멀티 에이전트 시스템의 성능은 어떤가요?

A: LangGraph는 병렬 실행을 지원하여 독립적인 에이전트들이 동시에 작업할 수 있습니다. 하지만 에이전트 간 통신이 많을수록 지연시간이 증가할 수 있으므로 적절한 설계가 중요합니다.

커뮤니티 및 지원

공식 채널

GitHub: langchain-ai/langgraph
Discord: LangChain 공식 Discord 서버
Twitter: @LangChainAI

라이선스 및 법적 고지

LangGraph는 MIT 라이선스 하에 배포됩니다. 상업적 사용이 가능하며, 자세한 내용은 공식 라이선스를 참조하세요.

이 번역 문서는 원본 문서의 내용을 한국어로 번역한 것이며, 최신 정보는 항상 공식 문서를 참조하시기 바랍니다.

이강현

백엔드 개발자 지망생입니다.

이전 포스트

인덱스 튜닝

다음 포스트

LangGraph Docs 한글 번역

LangGraph 완전 가이드 (한국어)

목차

LangGraph 소개

주요 특징

언제 LangGraph를 사용해야 하는가?

에이전트 개발

에이전트란 무엇인가?

사전 구축된 컴포넌트 사용하기

기본 에이전트 생성

고급 에이전트 구성

에이전트 실행하기

기본 실행 방법

입력 형식

출력 형식

실행 제어

패키지 생태계

실제 사용 예제

1. 간단한 질의응답 에이전트

2. 상태를 가진 대화 에이전트

LangGraph APIs

Graph API

핵심 개념

그래프 구축하기

고급 상태 관리

실제 그래프 예제

Functional API

Runtime (Pregel)

실행 모델

실행 제어

핵심 기능

스트리밍 (Streaming)

스트리밍 유형

스트리밍 모드

기본 스트리밍 사용법

고급 스트리밍 기능

지속성 (Persistence)

체크포인터 설정

스레드와 체크포인트

메모리 (Memory)

단기 메모리 (세션 기반)

장기 메모리 (영구 저장)

의미적 메모리 검색

인간 개입 (Human-in-the-loop)

기본 인간 개입

동적 인터럽트

시간 여행 (Time Travel)

기본 시간 여행

분기 생성

서브그래프 (Subgraphs)

서브그래프 정의

멀티 에이전트 (Multi-agent)

수퍼바이저 패턴

스웜 패턴

플랫폼 전용 기능

인증 및 접근 제어

어시스턴트 (Assistants)

더블 텍스팅 처리

웹훅 (Webhooks)

크론 작업 (Cron Jobs)

서버 사용자 정의

데이터 관리

배포 (Deployment)

실습 예제

예제 1: 고객 서비스 챗봇

예제 2: 연구 보고서 생성 에이전트

참고 자료

공식 문서 및 리소스

패키지 설치

주요 개념 요약

모범 사례

문제 해결

빠른 시작 가이드

5분 만에 첫 번째 에이전트 만들기

1단계: 설치

2단계: 환경 설정

3단계: 첫 번째 에이전트 생성

4단계: 실행 결과

다음 단계

자주 묻는 질문 (FAQ)