LangGraph 상태 관리: 복잡한 AI 워크플로우를 위한 핵심 전략

AI 에이전트를 구축하다 보면 반드시 마주치는 순간이 있습니다. RuntimeError: maximum recursion depth exceeded 또는 상태 유실로 인한 예기치 않은 동작—이것들은 LangGraph의 상태 관리 개념을 제대로 이해하지 못했을 때 발생하는 전형적인 증상입니다.

저는 HolySheep AI 게이트웨이를 통해 여러 AI 모델을 통합하면서 LangGraph 기반 워크플로우를 수십 차례 구축했습니다. 이 글에서는 복잡한 AI 파이프라인에서 상태를 효과적으로 관리하는 방법과 실제 프로덕션 환경에서 겪는 문제들을 해결하는 구체적인 전략을 다룹니다.

LangGraph 상태 관리 핵심 개념

LangGraph는 상태 다이어그램(State Diagram) 기반으로 AI 워크플로우를建模합니다. 각 노드(Node)는 특정 작업을 수행하고, 엣지(Edge)는 상태 전이를 정의합니다. 핵심은 StateGraph가 상태를 어떻게 관리하고 전달하느냐입니다.

상태 스키마 설계

효과적인 상태 관리의 첫걸음은 적절한 상태 스키마를 설계하는 것입니다. TypedDict를 사용한 타입 안전성을 확보하세요.

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

1단계: 상태 스키마 정의
class AgentState(TypedDict):
    """AI 에이전트의 중앙 상태 정의"""
    messages: Annotated[Sequence[BaseMessage], operator.add]  # 메시지 누적
    current_step: str                                          # 현재 실행 단계
    context: dict                                               # 공유 컨텍스트 데이터
    retry_count: int                                            # 재시도 카운터
    error_log: list[str]                                        # 에러 히스토리

2단계: 그래프 구성
graph = StateGraph(AgentState)

3단계: 노드 함수 정의
def process_node(state: AgentState) -> AgentState:
    """처리 노드 - 상태 업데이트 로직"""
    new_state = state.copy()
    new_state["current_step"] = "processing"
    new_state["retry_count"] = state.get("retry_count", 0) + 1
    
    # HolySheep AI를 통한 AI 호출
    from openai import OpenAI
    
    client = OpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    try:
        response = client.chat.completions.create(
            model="gpt-4.1",
            messages=[
                {"role": "user", "content": f"현재 단계: {state['current_step']}"}
            ]
        )
        new_state["context"]["last_response"] = response.choices[0].message.content
    except Exception as e:
        new_state["error_log"].append(str(e))
        new_state["current_step"] = "error_recovery"
    
    return new_state

4단계: 그래프 빌드 및 컴파일
graph.add_node("process", process_node)
graph.set_entry_point("process")
graph.add_edge("process", END)

compiled_graph = graph.compile()

체크포인팅과 상태 영속성

긴 실행 시간을 가진 AI 워크플로우에서는 중간 상태를 저장하는 것이 필수적입니다. Checkpoint 기능을 사용하면 워크플로우를 일시 중단하고 재개할 수 있습니다.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from langgraph.types import Interrupt, Command

체크포인팅이 적용된 그래프 생성
checkpointer = MemorySaver()

class CheckpointableState(TypedDict):
    messages: Annotated[list[str], operator.add]
    task_id: str
    status: str
    checkpoints: list[dict]

def node_with_checkpoint(state: CheckpointableState) -> CheckpointableState:
    """체크포인트를 활용한 상태 관리 노드"""
    updated = state.copy()
    updated["status"] = "in_progress"
    
    # HolySheep AI - Claude Sonnet 4.5 사용
    from openai import OpenAI
    client = OpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    # 작업 실행
    result = client.chat.completions.create(
        model="claude-sonnet-4-20250514",
        messages=[{"role": "user", "content": "복잡한 분석 작업 수행"}]
    )
    
    # 체크포인트 저장
    updated["checkpoints"].append({
        "step": len(state["messages"]),
        "result": result.choices[0].message.content,
        "timestamp": "2025-01-15"
    })
    
    # 특정 조건에서 인간 승인 대기
    if len(updated["messages"]) >= 5:
        raise Interrupt(value={"reason": "human_approval_required"})
    
    return updated

체크포인팅 적용
graph = StateGraph(CheckpointableState)
graph.add_node("analyze", node_with_checkpoint)
graph.set_entry_point("analyze")
graph.add_edge("analyze", END)

compiled = graph.compile(checkpointer=checkpointer)

실행 및 재개 예시
initial_state = {
    "messages": [],
    "task_id": "task_001",
    "status": "pending",
    "checkpoints": []
}

첫 실행
first_run = compiled.invoke(initial_state, config={"configurable": {"thread_id": "user_123"}})

이후 재개 (사용자 승인 후)
resume_state = compiled.invoke(
    Command(resume={"approved": True}),
    config={"configurable": {"thread_id": "user_123"}}
)

에러 처리와 회복탄력성 설계

AI API 호출은 네트워크 문제, Rate Limit, 모델 일시적 장애 등 다양한 원인으로 실패할 수 있습니다. LangGraph의 예외 처리 기능을 활용하면 자동 재시도 회로를 구성할 수 있습니다.

from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from tenacity import retry, stop_after_attempt, wait_exponential
import time

class ResilientState(TypedDict):
    attempt: int
    max_attempts: int
    last_error: str | None
    data: dict
    success: bool

def resilient_api_call(state: ResilientState) -> ResilientState:
    """재시도 로직이 포함된 API 호출 노드"""
    from openai import OpenAI
    client = OpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    attempt = state["attempt"]
    max_attempts = state.get("max_attempts", 3)
    
    try:
        # HolySheep AI - Gemini 2.5 Flash 활용
        response = client.chat.completions.create(
            model="gemini-2.5-flash",
            messages=[{"role": "user", "content": "데이터 분석 요청"}],
            timeout=30.0
        )
        
        return {
            **state,
            "success": True,
            "data": {"result": response.choices[0].message.content},
            "last_error": None
        }
        
    except Exception as e:
        error_msg = str(e)
        new_state = {
            **state,
            "attempt": attempt + 1,
            "last_error": error_msg,
            "success": False
        }
        
        # 최대 재시도 횟수 도달 시 실패 상태 반환
        if attempt >= max_attempts:
            new_state["data"] = {"error": f"최대 재시도 횟수 초과: {error_msg}"}
            return new_state
        
        # 지수 백오프 대기 후 재시도 (LangGraph가 자동으로 다음 노드로 전이)
        time.sleep(2 ** attempt)  # 1초, 2초, 4초...
        return new_state

조건부 에지 라우팅
def should_retry(state: ResilientState) -> str:
    """재시사 필요 여부 판단"""
    if state.get("success"):
        return "success"
    elif state["attempt"] < state.get("max_attempts", 3):
        return "retry"
    else:
        return "failed"

graph = StateGraph(ResilientState)
graph.add_node("api_call", resilient_api_call)
graph.set_entry_point("api_call")

조건부 엣지 추가
graph.add_conditional_edges(
    "api_call",
    should_retry,
    {
        "retry": "api_call",      # 재시도 루프
        "success": END,           # 성공 시 종료
        "failed": END             # 실패 시 종료
    }
)

compiled = graph.compile()

병렬 처리와 상태 병합

여러 AI 모델을 동시에 호출하고 결과를 병합해야 하는 경우가 많습니다. Send API를 사용하면 상태를 분산 처리할 수 있습니다.

from langgraph.constants import Send
from langgraph.graph import StateGraph, END

class ParallelState(TypedDict):
    query: str
    results: Annotated[dict, merge_results]
    errors: list[str]

def merge_results(left: dict, right: dict) -> dict:
    """병렬 결과 병합"""
    merged = left.copy()
    for key, value in right.items():
        if key in merged:
            if isinstance(merged[key], list):
                merged[key].append(value)
            else:
                merged[key] = [merged[key], value]
        else:
            merged[key] = value
    return merged

def spawn_parallel_tasks(state: ParallelState) -> list[Send]:
    """여러 모델에 병렬 요청"""
    query = state["query"]
    
    # HolySheep AI를 통해 3개 모델 동시 호출
    return [
        Send("gpt_task", {"query": query, "model": "gpt-4.1", "cost_per_token": 8.0}),
        Send("claude_task", {"query": query, "model": "claude-sonnet-4", "cost_per_token": 15.0}),
        Send("gemini_task", {"query": query, "model": "gemini-2.5-flash", "cost_per_token": 2.50})
    ]

def gpt_task(state: dict) -> dict:
    """GPT-4.1 태스크 노드"""
    from openai import OpenAI
    client = OpenAI(api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1")
    
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": state["query"]}]
    )
    
    return {"results": {"gpt": response.choices[0].message.content}}

def claude_task(state: dict) -> dict:
    """Claude 태스크 노드"""
    from openai import OpenAI
    client = OpenAI(api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1")
    
    response = client.chat.completions.create(
        model="claude-sonnet-4-20250514",
        messages=[{"role": "user", "content": state["query"]}]
    )
    
    return {"results": {"claude": response.choices[0].message.content}}

def gemini_task(state: dict) -> dict:
    """Gemini 태스크 노드"""
    from openai import OpenAI
    client = OpenAI(api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1")
    
    response = client.chat.completions.create(
        model="gemini-2.5-flash",
        messages=[{"role": "user", "content": state["query"]}]
    )
    
    return {"results": {"gemini": response.choices[0].message.content}}

그래프 구성
graph = StateGraph(ParallelState)
graph.add_node("gpt_task", gpt_task)
graph.add_node("claude_task", claude_task)
graph.add_node("gemini_task", gemini_task)
graph.add_node("orchestrator", spawn_parallel_tasks)
graph.set_entry_point("orchestrator")
graph.add_conditional_edges("orchestrator", lambda x: x, ["gpt_task", "claude_task", "gemini_task"])

병합 노드 추가
def collect_results(state: ParallelState) -> ParallelState:
    return state

graph.add_node("collect", collect_results)
for node in ["gpt_task", "claude_task", "gemini_task"]:
    graph.add_edge(node, "collect")
graph.add_edge("collect", END)

compiled = graph.compile()

HolySheep AI 게이트웨이 통합

HolySheep AI를 LangGraph와 통합하면 단일 API 키로 다양한 모델을 쉽게 전환하고 비용을 최적화할 수 있습니다. 저는 실무에서 DeepSeek V3.2의 저렴한 가격($0.42/MTok)을 preliminary 분석에 사용하고, 최종 응답에만 GPT-4.1을 활용하는 전략을 자주 사용합니다.

자주 발생하는 오류와 해결

1. ConnectionError: timeout — API 응답 지연

# 문제: LangGraph 워크플로우 실행 중 타임아웃 발생
File "node.py", line 42, in api_call
ConnectionError: timeout

해결: 타임아웃 설정 및 재시도 회로 구현
from openai import OpenAI
from httpx import Timeout

client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    timeout=Timeout(60.0, connect=10.0)  # 총 60초, 연결 10초
)

또는 httpx Client 직접 사용
import httpx

with httpx.Client(
    base_url="https://api.holysheep.ai/v1",
    timeout=60.0,
    headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
) as client:
    response = client.post(
        "/chat/completions",
        json={
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": "Hello"}],
            "max_tokens": 100
        }
    )

2. 401 Unauthorized — 잘못된 API 키 또는 인증 실패

# 문제: API 호출 시 401 에러
openai.AuthenticationError: 401 Incorrect API key provided

해결: API 키 환경 변수 사용 및 검증
import os
from dotenv import load_dotenv

load_dotenv()  # .env 파일에서 API 키 로드

API_KEY = os.getenv("HOLYSHEEP_API_KEY")
if not API_KEY or API_KEY == "YOUR_HOLYSHEEP_API_KEY":
    raise ValueError("""
    HolySheep AI API 키가 설정되지 않았습니다.
    1. https://www.holysheep.ai/register 에서 가입
    2. 대시보드에서 API 키 발급
    3. .env 파일에 HOLYSHEEP_API_KEY=your_key 설정
    """)

client = OpenAI(api_key=API_KEY, base_url="https://api.holysheep.ai/v1")

키 유효성 검증
try:
    client.models.list()
    print("API 키 인증 성공")
except Exception as e:
    print(f"인증 실패: {e}")

3. RuntimeError: maximum recursion depth — 상태 순환 참조

# 문제: 무한 루프导致的 상태 순환
RuntimeError: maximum recursion depth exceeded in comparison

해결: 상태 전이 제한 및 루프 감지 구현
MAX_ITERATIONS = 10

def check_iteration_limit(state: dict, config: dict) -> bool:
    """반복 횟수 제한으로 무한 루프 방지"""
    history = config.get("configurable", {}).get("iteration_history", [])
    
    if len(history) >= MAX_ITERATIONS:
        raise RuntimeError(f"""
        최대 반복 횟수({MAX_ITERATIONS})를 초과했습니다.
        상태 순환이 감지되었습니다. 워크플로우 설정을 검토하세요.
        
        마지막 상태: {state}
        """)
    
    return True

def safe_route(state: dict, config: dict) -> str:
    """안전한 라우팅 with 루프 감지"""
    check_iteration_limit(state, config)
    
    # 현재 상태를 히스토리에 기록
    history = config.get("configurable", {}).get("iteration_history", [])
    history.append(state.get("current_step", "unknown"))
    
    if state.get("done"):
        return END
    return "next_node"

사용 시 config 전달
result = compiled.invoke(
    state,
    config={
        "configurable": {
            "thread_id": "unique_session_id",
            "iteration_history": []
        }
    }
)

4. RateLimitError — API 호출 빈도 제한

# 문제: 과도한 API 호출로 인한 Rate Limit
RateLimitError: Rate limit exceeded for model gpt-4.1

해결: Rate Limit 핸들링 및 백오프
import asyncio
from openai import RateLimitError

async def rate_limit_aware_call(client, model: str, messages: list):
    """Rate Limit을 처리하는 비동기 API 호출"""
    max_retries = 5
    base_delay = 2
    
    for attempt in range(max_retries):
        try:
            response = await client.chat.completions.create(
                model=model,
                messages=messages
            )
            return response
        
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            
            # HolySheep AI Rate Limit 대응
            delay = base_delay * (2 ** attempt)
            print(f"Rate Limit 도달. {delay}초 후 재시도 ({attempt + 1}/{max_retries})")
            await asyncio.sleep(delay)
        
        except Exception as e:
            raise

HolySheep AI의 뛰어난 Rate Limit 정책 활용
(저의 경험상 HolySheep AI는 타사 대비 Rate Limit이 관대한 편입니다)

실전 최적화 팁

• 상태 크기 최소화: 너무 큰 데이터를 상태에 저장하지 마세요. 필요한 경우 S3나 Redis 등 외부 스토리지 활용
• 지연 읽기 활용: Annotated[List, operator.add] 패턴으로 메시지를 누적하면 메모리 이슈 발생 가능
• 모델 선택 전략: preliminary 분석에는 Gemini 2.5 Flash($2.50/MTok), 복잡한 추론에는 Claude Sonnet 4.5($15/MTok), 최종 응답에는 GPT-4.1($8/MTok)
• 체크포인트 빈도: 장기 실행 워크플로우는 각 중요 단계마다 체크포인트를 저장하세요

결론

LangGraph의 상태 관리는 단순히 상태를 저장하는 것을 넘어, AI 워크플로우의 신뢰성, 확장성, 디버깅 가능성을 결정하는 핵심 요소입니다. HolySheep AI 게이트웨이와 결합하면 다양한 모델을 손쉽게 통합하면서 비용도 최적화할 수 있습니다.

저는 HolySheep AI를 통해 월 50만 토큰 이상의 AI API 호출을 관리하는데, 단일 API 키로 여러 모델을 제어할 수 있다는 점이 가장 큰 장점입니다. Rate Limit 관리와 비용 추적도 대시보드에서 바로 확인할 수 있어 프로덕션 환경에서 매우 유용합니다.

👉 HolySheep AI 가입하고 무료 크레딧 받기