LangGraph 시각화 도구와 디버깅 워크플로우 완전 가이드

서론: AI 에이전트 개발의 핵심, LangGraph

2026년 최신 AI 모델 비용 비교 분석

월 1,000만 토큰 기준 비용 비교표

LangGraph 프로젝트 구조와 핵심 개념

langgraph-agent/
├── app.py                 # 메인 애플리케이션 진입점
├── graph/
│   ├── __init__.py
│   ├── state.py          # 상태 스키마 정의
│   ├── nodes.py          # 노드 함수들
│   ├── edges.py          # 조건부 엣지 로직
│   └── compiler.py       # 그래프 컴파일
├── tools/
│   ├── __init__.py
│   ├── search.py         # 검색 도구
│   ├── calculator.py     # 계산 도구
│   └── holysheep_client.py  # HolySheep AI 클라이언트
├── visualization/
│   ├── renderer.py       # 그래프 시각화
│   └── debugger.py       # 디버깅 유틸리티
├── config.py             # 설정 파일
└── requirements.txt

HolySheep AI 통합 LangGraph 에이전트 구현

import os
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

HolySheep AI API 설정
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

HolySheep AI를 지원하는 LangChain ChatOpenAI 클라이언트
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL,
    temperature=0.7,
    max_tokens=2000
)

상태 스키마 정의
class AgentState(TypedDict):
    messages: Annotated[Sequence[HumanMessage | AIMessage | SystemMessage], "메시지 히스토리"]
    intent: str | None
    confidence: float | None
    tool_results: dict | None

def create_agent_graph():
    """LangGraph 에이전트 그래프 생성"""
    
    # 그래프 빌더 초기화
    workflow = StateGraph(AgentState)
    
    # 노드 등록
    workflow.add_node("understand_intent", understand_intent_node)
    workflow.add_node("route_request", route_request_node)
    workflow.add_node("search_info", search_info_node)
    workflow.add_node("calculate", calculate_node)
    workflow.add_node("generate_response", generate_response_node)
    
    # 진입점 설정
    workflow.set_entry_point("understand_intent")
    
    # 엣지 정의
    workflow.add_edge("understand_intent", "route_request")
    workflow.add_conditional_edges(
        "route_request",
        route_decision,
        {
            "search": "search_info",
            "calculate": "calculate",
            "respond": "generate_response"
        }
    )
    
    # 종료 노드 연결
    workflow.add_edge("search_info", "generate_response")
    workflow.add_edge("calculate", "generate_response")
    workflow.add_edge("generate_response", END)
    
    return workflow.compile()

의도 이해 노드
def understand_intent_node(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1].content
    
    response = llm.invoke([
        SystemMessage(content="""사용자의 메시지를 분석하여 의도를 파악하세요.
        가능한 의도: search(정보 검색 필요), calculate(계산 필요), respond(일반 응답 가능)""")
    ] + [HumanMessage(content=last_message)])
    
    intent = response.content.strip().lower()
    confidence = 0.9 if any(keyword in intent for keyword in ["search", "계산", "정보"]) else 0.7
    
    return {
        "intent": intent,
        "confidence": confidence,
        "messages": state["messages"] + [response]
    }

print("✅ HolySheep AI 통합 LangGraph 에이전트 초기화 완료")

LangGraph 시각화 도구 구성

import json
from pathlib import Path
from langgraph.visualization import MermaidRenderer
from IPython.display import display, Image, Markdown

class LangGraphVisualizer:
    """LangGraph 그래프 시각화 및 디버깅 도구"""
    
    def __init__(self, graph, output_dir: str = "./visualizations"):
        self.graph = graph
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    def generate_mermaid_diagram(self, filename: str = "graph_structure.md"):
        """Mermaid 형식으로 그래프 구조 시각화"""
        mermaid_code = self.graph.get_graph().draw_mermaid()
        
        output_path = self.output_dir / filename
        with open(output_path, "w", encoding="utf-8") as f:
            f.write("```mermaid\n")
            f.write(mermaid_code)
            f.write("\n```")
        
        print(f"✅ Mermaid 다이어그램 저장됨: {output_path}")
        return output_path
    
    def generate_png(self, filename: str = "graph.png"):
        """PNG 이미지로 그래프 내보내기"""
        try:
            from langgraph.visualization import plot_graph
            graph_image = plot_graph(self.graph.get_graph())
            output_path = self.output_dir / filename
            graph_image.savefig(output_path, dpi=150, bbox_inches="tight")
            print(f"✅ PNG 이미지 저장됨: {output_path}")
            return output_path
        except Exception as e:
            print(f"⚠️ PNG 내보내기 실패: {e}")
            return None
    
    def visualize_execution_trace(self, trace_data: dict, filename: str = "trace.json"):
        """실행 추적 데이터 시각화"""
        trace_path = self.output_dir / filename
        with open(trace_path, "w", encoding="utf-8") as f:
            json.dump(trace_data, f, indent=2, ensure_ascii=False)
        print(f"✅ 추적 데이터 저장됨: {trace_path}")
        return trace_path
    
    def create_debug_dashboard(self, state_history: list):
        """디버깅 대시보드 생성"""
        print("\n" + "=" * 60)
        print("🔍 LangGraph 실행 디버깅 대시보드")
        print("=" * 60)
        
        for i, state in enumerate(state_history):
            print(f"\n[Step {i + 1}]")
            print(f"  Intent: {state.get('intent', 'N/A')}")
            print(f"  Confidence: {state.get('confidence', 0):.2%}")
            print(f"  Messages: {len(state.get('messages', []))}개")
            
            if state.get("tool_results"):
                print(f"  Tool Results: {json.dumps(state['tool_results'], ensure_ascii=False)[:100]}...")
        
        print("\n" + "=" * 60)

시각화 도구 사용 예제
if __name__ == "__main__":
    from app import create_agent_graph
    
    graph = create_agent_graph()
    visualizer = LangGraphVisualizer(graph)
    
    # Mermaid 다이어그램 생성
    visualizer.generate_mermaid_diagram()
    print("\n📊 Mermaid 코드를 Mermaid Live Editor에 붙여넣어 시각화하세요")

디버깅 워크플로우最佳实践

import logging
from functools import wraps
import time
from typing import Callable, Any
from langgraph.errors import GraphRecursionError
import openai

로깅 설정
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("langgraph_debug")

class HolySheepDebugger:
    """HolySheep AI 기반 LangGraph 디버깅 유틸리티"""
    
    def __init__(self, api_key: str, cost_tracker: dict = None):
        self.api_key = api_key
        self.cost_tracker = cost_tracker or {"total_tokens": 0, "cost": 0.0}
        self.trace_history = []
        
    def track_api_call(self, model: str, tokens: int, latency_ms: float):
        """API 호출 추적 및 비용 계산"""
        prices = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.5,
            "deepseek-v3.2": 0.42
        }
        
        price = prices.get(model, 8.0)
        cost = (tokens / 1_000_000) * price
        
        self.cost_tracker["total_tokens"] += tokens
        self.cost_tracker["cost"] += cost
        
        logger.info(f"🔄 API 호출 추적 - Model: {model}, Tokens: {tokens}, "
                   f"Latency: {latency_ms:.0f}ms, Cost: ${cost:.4f}")
    
    def debug_node_execution(self, node_name: str):
        """노드 실행 디코레이터"""
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            def wrapper(*args, **kwargs):
                logger.debug(f"🔵 노드 시작: {node_name}")
                start_time = time.time()
                
                try:
                    result = func(*args, **kwargs)
                    elapsed = (time.time() - start_time) * 1000
                    logger.debug(f"🟢 노드 완료: {node_name} ({elapsed:.0f}ms)")
                    return result
                    
                except Exception as e:
                    elapsed = (time.time() - start_time) * 1000
                    logger.error(f"🔴 노드 오류: {node_name} ({elapsed:.0f}ms) - {str(e)}")
                    raise
                    
            return wrapper
        return decorator
    
    def handle_recursion_error(self, error: GraphRecursionError):
        """재귀 깊이 초과 에러 처리"""
        logger.error(f"⚠️ 재귀 깊이 초과: {error}")
        print("\n💡 해결 방법:")
        print("1. graph.compile() 시 max_iterations 매개변수 확인")
        print("2. 상태 전이 로직에서 무한 루프 확인")
        print("3. HolySheep AI 연결 상태 점검")
        return {"error": "max_recursion_exceeded", "message": str(error)}
    
    def generate_debug_report(self) -> str:
        """디버깅 리포트 생성"""
        report = f"""
        ╔════════════════════════════════════════════════════╗
        ║          LangGraph 디버깅 리포트                    ║
        ╠════════════════════════════════════════════════════╣
        ║ 총 토큰 사용량: {self.cost_tracker['total_tokens']:,}                   ║
        ║ 총 비용: ${self.cost_tracker['cost']:.4f}                         ║
        ║ 추적된 실행: {len(self.trace_history)}건                          ║
        ╚════════════════════════════════════════════════════╝
        """
        return report

def create_debuggable_graph():
    """디버깅 가능한 그래프 생성"""
    debugger = HolySheepDebugger(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # 디버깅 노드 데코레이터 적용
    @debugger.debug_node_execution("understand_intent")
    def debug_understand_intent(state):
        # 디버깅 로직
        logger.info(f"상태 메시지 수: {len(state.get('messages', []))}")
        return state
    
    # 추가 노드们也 동일하게 데코레이터 적용
    return debugger

print("✅ 디버깅 워크플로우 설정 완료")
print("\n📋 주요 디버깅 포인트:")
print("   1. 노드별 실행 시간 추적")
print("   2. API 호출 비용 모니터링")
print("   3. 상태 전이 추적")
print("   4. 에러 스택트레이스 캡처")

HolySheep AI 다중 모델 전환 전략

from enum import Enum
from typing import Literal

class ModelType(Enum):
    """HolySheep AI 지원 모델"""
    GPT_4_1 = "gpt-4.1"
    CLAUDE_SONNET = "claude-sonnet-4.5"
    GEMINI_FLASH = "gemini-2.5-flash"
    DEEPSEEK = "deepseek-v3.2"

class HolySheepRouter:
    """작업 유형별 최적 모델 라우팅"""
    
    PRICES = {
        ModelType.GPT_4_1: 8.0,
        ModelType.CLAUDE_SONNET: 15.0,
        ModelType.GEMINI_FLASH: 2.5,
        ModelType.DEEPSEEK: 0.42
    }
    
    @classmethod
    def select_model(cls, task_complexity: str, budget_mode: bool = False) -> ModelType:
        """작업 복잡도에 따른 모델 선택"""
        
        if budget_mode:
            # 비용 최적화 모드
            return ModelType.DEEPSEEK
        
        complexity_mapping = {
            "simple": ModelType.GEMINI_FLASH,
            "medium": ModelType.GPT_4_1,
            "complex": ModelType.CLAUDE_SONNET,
            "reasoning": ModelType.GPT_4_1
        }
        
        return complexity_mapping.get(task_complexity, ModelType.GPT_4_1)
    
    @classmethod
    def estimate_cost(cls, model: ModelType, input_tokens: int, output_tokens: int) -> float:
        """비용 추정"""
        price = cls.PRICES[model]
        total_tokens = input_tokens + output_tokens
        cost = (total_tokens / 1_000_000) * price
        return cost
    
    @classmethod
    def create_model_client(cls, model_type: ModelType, api_key: str):
        """HolySheep AI 모델 클라이언트 생성"""
        from langchain_openai import ChatOpenAI
        
        return ChatOpenAI(
            model=model_type.value,
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1",
            temperature=0.7,
            max_tokens=2000
        )

def smart_routing_node(state: AgentState):
    """스마트 라우팅 노드 - HolySheep AI"""
    
    messages = state.get("messages", [])
    if not messages:
        return state
    
    last_message = messages[-1].content
    message_length = len(last_message)
    has_technical_terms = any(term in last_message for term in ["코드", "함수", "API", "프로그래밍"])
    
    # 복잡도 판단
    if message_length < 50 and not has_technical_terms:
        complexity = "simple"
    elif message_length > 200 or has_technical_terms:
        complexity = "complex"
    else:
        complexity = "medium"
    
    # HolySheep AI를 통한 모델 선택
    selected_model = HolySheepRouter.select_model(complexity)
    estimated_cost = HolySheepRouter.estimate_cost(selected_model, 500, 200)
    
    print(f"📊 모델 선택: {selected_model.value}")
    print(f"💰 예상 비용: ${estimated_cost:.4f}")
    
    # 선택된 모델로 LLM 클라이언트 생성
    llm = HolySheepRouter.create_model_client(
        selected_model, 
        "YOUR_HOLYSHEEP_API_KEY"
    )
    
    response = llm.invoke(messages + [HumanMessage(content="분석을 수행하세요.")])
    
    return {
        **state,
        "messages": state["messages"] + [response],
        "selected_model": selected_model.value,
        "estimated_cost": estimated_cost
    }

print("✅ HolySheep AI 다중 모델 라우팅 시스템 구성 완료")

자주 발생하는 오류와 해결책

오류 1: API 연결 실패 및 인증 오류

AuthenticationError: Incorrect API key provided. 
Expected format: sk-holysheep-...

import os
from dotenv import load_dotenv

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

환경 변수 확인
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")

if not HOLYSHEEP_API_KEY:
    raise ValueError("""
    ❌ HolySheheep API 키가 설정되지 않았습니다.
    
    해결 방법:
    1. .env 파일 생성: touch .env
    2. .env 파일에 추가: HOLYSHEEP_API_KEY=sk-holysheep-your-key-here
    3. Python에서 로드: from dotenv import load_dotenv; load_dotenv()
    4. HolySheep AI 대시보드에서 API 키 발급: https://www.holysheep.ai/register
    """)

API 키 형식 검증
if not HOLYSHEEP_API_KEY.startswith("sk-holysheep-"):
    raise ValueError(f"""
    ❌ 잘못된 API 키 형식입니다.
    현재: {HOLYSHEEP_API_KEY[:15]}...
    예상: sk-holysheep-xxxxx...
    
    HolySheep AI에서 올바른 API 키를 발급받으세요.
    """)

연결 테스트
from langchain_openai import ChatOpenAI

test_client = ChatOpenAI(
    model="deepseek-v3.2",
    api_key=HOLYSHEEP_API_KEY,
    base_url="https://api.holysheep.ai/v1"
)

try:
    response = test_client.invoke([HumanMessage(content="test")])
    print("✅ HolySheep AI 연결 성공!")
except Exception as e:
    print(f"❌ 연결 실패: {e}")
    print("💡 네트워크 연결을 확인하거나 HolySheep AI 상태를 점검하세요.")

오류 2: LangGraph 재귀 깊이 초과

GraphRecursionError: Budget exceeded, try something else in 

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

def create_safe_graph():
    """재귀 오류를 방지하는 안전한 그래프 구성"""
    
    workflow = StateGraph(AgentState)
    
    # ... 노드 정의 ...
    
    # 무한 루프 방지를 위한 checkpointer 설정
    checkpointer = MemorySaver()
    
    # 컴파일 시 최대 반복 횟수 명시적 설정
    compiled_graph = workflow.compile(
        checkpointer=checkpointer,
        interrupt_before=["generate_response"],  # 특정 노드에서 인터럽트 가능
    )
    
    return compiled_graph

def execute_with_budget(graph, input_state, max_steps: int = 10):
    """단계별 실행으로 무한 루프 방지"""
    
    config = {"configurable": {"thread_id": "debug-session"}}
    step_count = 0
    
    try:
        # 초기 상태 설정
        graph.update_state(config, input_state)
        
        while step_count < max_steps:
            # 단일 단계만 실행
            snapshot = graph.get_state(config)
            
            if not snapshot.next:
                print("✅ 그래프 실행 완료")
                break
            
            # 다음 노드 실행
            graph.invoke(None, config)
            step_count += 1
            print(f"📍 Step {step_count}: {snapshot.next}")
            
        else:
            print(f"⚠️ 최대 단계 수({max_steps}) 도달, 실행 중단")
            print("💡 상태 전이 로직을 점검하여 무한 루프 가능성을 확인하세요.")
            
    except Exception as e:
        print(f"❌ 실행 중 오류 발생: {e}")
        raise
        
    return graph.get_state(config)

모니터링 디코레이터
from functools import wraps

def monitored_node(max_calls: int = 5):
    """노드 호출 횟수 제한 디코레이터"""
    call_counts = {}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            node_name = func.__name__
            call_counts[node_name] = call_counts.get(node_name, 0) + 1
            
            if call_counts[node_name] > max_calls:
                raise RecursionError(
                    f"노드 '{node_name}' 호출 횟수({call_counts[node_name]})가 "
                    f"최대값({max_calls})을 초과했습니다."
                )
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

@monitored_node(max_calls=3)
def search_info_node(state):
    """정보 검색 노드 (최대 3회 호출 제한)"""
    # 검색 로직
    pass

print("✅ 재귀 깊이 보호 메커니즘 적용 완료")

RateLimitError: Rate limit reached for gpt-4.1 in organization...
Context length exceeded. Maximum context length is 128000 tokens.

from langchain_core.messages import trim_messages, SystemMessage

class ContextManager:
    """컨텍스트 윈도우 및 토큰 관리 유틸리티"""
    
    def __init__(self, max_tokens: int = 6000):
        self.max_tokens = max_tokens
    
    def trim_message_history(self, messages, strategy: str = "last"):
        """메시지 히스토리 트리밍"""
        
        if strategy == "last":
            # 최근 메시지만 유지
            trimmed = trim_messages(
                messages,
                max_tokens=self.max_tokens,
                strategy="last",
                include_system=True,
                allow_partial=True
            )
        elif strategy == "summary":
            # 이전 대화를 요약으로 변환
            # (실제 구현 시 LLM을 사용한 요약 로직 필요)
            if len(messages) > 4:
                summary_prompt = "이전 대화를 2문장으로 요약하세요."
                # 요약 생성 로직...
                pass
        
        return trimmed
    
    def estimate_tokens(self, text: str) -> int:
        """대략적인 토큰 수 추정 (한글 기준)"""
        # 한글은 영어보다 토큰 효율이 낮음
        return len(text) // 2
    
    def safe_invoke(self, llm, messages, fallback_strategy: str = "trim"):
        """안전한 LLM 호출"""
        
        try:
            return llm.invoke(messages)
            
        except Exception as e:
            if "context length" in str(e).lower():
                print("⚠️ 컨텍스트 길이 초과, 메시지 트리밍 시도...")
                
                if fallback_strategy == "trim":
                    trimmed_messages = self.trim_message_history(messages)
                    return llm.invoke(trimmed_messages)
                    
                elif fallback_strategy == "switch_model":
                    # 더 큰 컨텍스트를 지원하는 모델로 전환
                    print("💡 DeepSeek V3.2로 모델 전환 (128K 컨텍스트)")
                    # DeepSeek 클라이언트로 전환
                    pass
                    
            elif "rate limit" in str(e).lower():
                print("⚠️ 속도 제한 도달, 재시도 대기...")
                import time
                time.sleep(5)
                return self.safe_invoke(llm, messages)
            
            raise

class HolySheepRateLimiter:
    """HolySheep AI 속도 제한 관리"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.rpm = requests_per_minute
        self.request_times = []
    
    def wait_if_needed(self):
        """속도 제한에 도달하면 대기"""
        import time
        from datetime import datetime, timedelta
        
        now = datetime.now()
        cutoff = now - timedelta(minutes=1)
        
        # 1분 이내 요청 필터링
        self.request_times = [t for t in self.request_times if t > cutoff]
        
        if len(self.request_times) >= self.rpm:
            wait_time = (self.request_times[0] - cutoff).total_seconds()
            print(f"⏳ 속도 제한 대기: {wait_time:.1f}초")
            time.sleep(wait_time + 0.5)
        
        self.request_times.append(datetime.now())

사용 예제
rate_limiter = HolySheepRateLimiter(requests_per_minute=30)
context_manager = ContextManager(max_tokens=4000)

def safe_llm_call(llm, messages):
    """안전한 LLM 호출 래퍼"""
    rate_limiter.wait_if_needed()
    return context_manager.safe_invoke(llm, messages)

print("✅ 토큰 및 속도 제한 관리 시스템 적용 완료")

TypeError: Expected 'messages' to be a list of messages, got 'str' instead
ValueError: Invalid state schema: missing required field 'intent'

from typing import TypedDict, Annotated, List, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import add_states_IN

class AgentState(TypedDict, total=False):
    """명시적 타입 정의로 상태 불일치 방지"""
    
    messages: Annotated[List[BaseMessage], "메시지 목록 (최소 1개)"]
    intent: Optional[str]
    confidence: Optional[float]
    tool_results: Optional[dict]
    metadata: Optional[dict]

def validate_state(state: dict) -> bool:
    """상태 유효성 검증"""
    
    required_fields = ["messages"]
    
    for field in required_fields:
        if field not in state:
            raise ValueError(f"필수 필드 누락: {field}")
        
        if field == "messages":
            if not isinstance(state[field], list):
                raise TypeError(
                    f"messages는 list여야 합니다. 현재: {type(state[field])}"
                )
            if not all(isinstance(m, BaseMessage) for m in state[field]):
                raise TypeError(
                    "messages의 모든 요소는 BaseMessage 타입이어야 합니다."
                )
    
    return True

def safe_node_wrapper(func):
    """노드 함수에 상태 검증 추가"""
    @wraps(func)
    def wrapper(state):
        # 입력 상태 검증
        validate_state(state)
        
        # 함수 실행
        result = func(state)
        
        # 출력 상태 검증
        if result:
            validate_state(result)
        
        return result
    
    return wrapper

@safe_node_wrapper
def generate_response_node(state: AgentState) -> AgentState:
    """응답 생성 노드"""
    
    messages = state.get("messages", [])
    
    # 메시지가 비어있으면 기본값 반환
    if not messages:
        return {
            **state,
            "messages": messages + [AIMessage(content="")]
        }
    
    # 문자열이 메시지 목록에 있으면 변환
    cleaned_messages = []
    for msg in messages:
        if isinstance(msg, str):
            cleaned_messages.append(HumanMessage(content=msg))
        elif isinstance(msg, dict):
            cleaned_messages.append(HumanMessage(content=msg.get("content", "")))
        else:
            cleaned_messages.append(msg)
    
    return {
        **state,
        "messages": cleaned_messages
    }

print("✅ 상태 스키마 검증 시스템 적용 완료")