제 경험상 LangGraph로 복잡한 AI 에이전트를 구축한 후 가장 큰 도전은 항상 "이 에이전트가 정확히 무엇을 하고 있는가?"라는 질문에 답하는 것이었습니다. 상태 전이가 제대로 이루어지고 있는지, 각 단계에서 어떤 결정이 내려졌는지, 왜 특정 경로를 선택했는지를 파악하지 못하면 디버깅은 순조와 같다.
저는 6개월 전 이커머스 AI 고객 서비스 프로젝트를 진행하면서 이 문제를 체감했습니다. 사용자의 복잡한 요청(반품 + 환불 + 배송 추적)을 처리하는 다중 단계 에이전트를 구축했는데, 디버깅 없이Production에 배포했다가 처음 문제가 발생했을 때整整 3일을 소모하며 원인을 찾았습니다. 그 후 LangGraph 모니터링과 시각화 분석 도구를 체계적으로 도입했는데, 이후 Similar한 문제 발생 시 평균 30분 내 원인을 파악할 수 있게 되었습니다.
LangGraph 상태 관리와 Checkpointer
LangGraph의 핵심 강점은 상태 머신 기반 아키텍처입니다. 각 노드 실행 후 상태가 어떻게 변하는지 추적하는 것이 모니터링의 기본입니다. MemorySaver를 사용하면 메모리 내 체크포인트를 저장하고, PostgresSaver를 사용하면 영구 저장소에 실행 기록을 보관할 수 있습니다.
import os
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator
HolySheep AI 설정
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4.1",
base_url="https://api.holysheep.ai/v1",
api_key=os.environ["HOLYSHEEP_API_KEY"]
)
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
current_step: str
intent: str | None
extracted_entities: dict
체크포인터 설정 (개발 환경: 메모리, 운영 환경: PostgreSQL)
if os.getenv("ENVIRONMENT") == "production":
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:password@localhost:5432/langgraph_checkpoints"
)
else:
checkpointer = MemorySaver()
def should_continue(state: AgentState) -> str:
"""라우팅 결정 로직"""
if state.get("intent") == "end":
return END
return "continue"
def process_node(state: AgentState) -> AgentState:
"""각 노드에서 상태 변경 추적"""
print(f"[DEBUG] Step: {state['current_step']}, Intent: {state.get('intent')}")
return state
상태 그래프 구축
workflow = StateGraph(AgentState)
workflow.add_node("intent_classifier", process_node)
workflow.add_node("entity_extractor", process_node)
workflow.add_node("action_executor", process_node)
workflow.set_entry_point("intent_classifier")
workflow.add_edge("intent_classifier", "entity_extractor")
workflow.add_edge("entity_extractor", "action_executor")
workflow.add_edge("action_executor", END)
compiled_graph = workflow.compile(checkpointer=checkpointer)
특정 thread_id로 실행 history 조회 가능
config = {"configurable": {"thread_id": "session_12345"}}
for state in compiled_graph.get_state_history(config):
print(f"Checkpoint {state['config']['checkpoint_id']}: {state['values']}")위 코드에서 저는 각 노드 실행 시 현재 상태를 로깅하고, get_state_history() 메서드로 특정 세션의 전체 실행 이력을 조회합니다. 이를 통해 사용자의 요청이 어떤 경로를 통해 처리되었는지 시간순으로 역추적할 수 있습니다.
Phoenix를 활용한 분산 추적 및 시각화
-production 환경에서는 더 강력한 추적이 필요합니다. Arize Phoenix는 LangGraph 에이전트의 실행轨迹을 실시간으로 시각화하는 도구입니다. 저는 Phoenix를 사용하여 각 노드 실행 시간, LLM 호출 횟수, 토큰 사용량을 모니터링합니다.
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from langgraph.callbacks.tracing.opentelemetry import OpenTelemetryRecorderCallback
Phoenix Cloud 또는 self-hosted 설정
os.environ["PHOENIX_API_KEY"] = "your-phoenix-api-key"
from langgraph_sdk import get_client
from langgraph_sdk.schema import HumanMessage
HolySheep AI를 LangGraph Native Tool로 활용
from langchain_holysheep import HolySheepLLM
holysheep_llm = HolySheepLLM(
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1"
)
OpenTelemetry 기반 추적 설정
provider = TracerProvider()
processor = BatchSpanProcessor(
OTLPSpanExporter(endpoint="https://llm.observe.holysheep.ai/v1/traces")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
async def run_monitored_agent(user_input: str, session_id: str):
"""모니터링이 적용된 에이전트 실행"""
client = get_client()
# Phoenix 추적 콜백 활성화
async with OpenTelemetryRecorderCallback(tracer) as callback:
assistant = await client.assistants.create(
graph_id="ecommerce-service-agent",
config={
"configurable": {
"thread_id": session_id,
"llm": holysheep_llm
},
"callbacks": [callback]
}
)
# 에이전트 실행
thread = await client.threads.create()
await client.runs.wait(
thread_id=thread["thread_id"],
assistant_id=assistant["assistant_id"],
input={"messages": [{"role": "user", "content": user_input}]}
)
# Phoenix 대시보드에서 시각화된 실행轨迹 확인
# https://app.phoenix.arize.com/traces/{trace_id}
return callback.trace_id
모니터링 결과 분석
async def analyze_execution(trace_id: str):
"""실행 데이터 분석"""
from arize.pandas.tracing import read_traces
traces_df = read_traces(
export_endpoint="https://llm.observe.holysheep.ai/v1/traces",
trace_ids=[trace_id]
)
print("=== 실행 분석 리포트 ===")
print(f"총 노드 실행 수: {len(traces_df)}")
print(f"평균 응답 시간: {traces_df['latency_ms'].mean():.2f}ms")
print(f"총 토큰 사용: {traces_df['total_tokens'].sum()}")
# HolySheep AI 비용 계산
gpt41_cost = traces_df[traces_df['model'] == 'gpt-4.1']['total_tokens'].sum() * 8 / 1_000_000
print(f"예상 비용: ${gpt41_cost:.4f}")Phoenix의 가장 큰 장점은 실행 그래프를 시각적으로 표현해준다는 점입니다. 각 노드가 어떤 순서로 실행되었는지, 각 단계의 소요 시간, 에러 발생 지점을 한눈에 파악할 수 있습니다. 저는 이 도구를 사용하여 고객 서비스 에이전트의 평균 처리 시간을 45% 절감했습니다.
LangSmith 연동을 통한 상세 로그 분석
LangChain 생태계에서는 LangSmith가 표준 모니터링 도구입니다. HolySheep AI의 모델과 LangSmith를 연동하면 각 LLM 호출의 상세한 로그를 추적할 수 있습니다.
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-langsmith-key"
from langsmith import traceable
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.postgres import PostgresSaver
HolySheep AI LLM 설정
llm = ChatOpenAI(
model="claude-sonnet-4-20250514",
base_url="https://api.holysheep.ai/v1",
api_key=os.environ["HOLYSHEEP_API_KEY"]
)
모니터링 대상 툴 정의
@traceable(name="order-lookup", tags=["ecommerce", "production"])
def lookup_order(order_id: str) -> dict:
"""주문 조회 툴 - 각 호출이 LangSmith에 기록됨"""
return {
"order_id": order_id,
"status": "shipped",
"eta": "2024-01-20"
}
@traceable(name="refund-calculation", tags=["ecommerce", "finance"])
def calculate_refund(order_id: str, reason: str) -> float:
"""환불 금액 계산"""
base = 499.00
if reason == "defective":
return base + 5.00 # 왕복 배송비
return base
tools = [lookup_order, calculate_refund]
LangSmith 추적이 활성화된 에이전트
agent = create_react_agent(llm, tools)
@traceable(name="customer-service-handler")
def handle_customer_request(user_message: str, session_id: str) -> str:
"""고객 서비스 핸들러 - 전체 실행轨迹 추적"""
# 상태 저장소에서 이전 대화 이력 로드
checkpointer = PostgresSaver.from_conn_string(
os.environ["DATABASE_URL"]
)
config = {
"configurable": {"thread_id": session_id},
"recursion_limit": 50 # 무한 루프 방지
}
# 에이전트 실행
response = agent.invoke(
{"messages": [("user", user_message)]},
config=config
)
# 토큰 사용량 및 비용 로깅
usage = response.get("usage_metadata", {})
input_tokens = usage.get("input_tokens", 0)
output_tokens = usage.get("output_tokens", 0)
# HolySheep AI 요금 계산 (Claude Sonnet 4.5: $15/MTok)
cost = (input_tokens + output_tokens) / 1_000_000 * 15
print(f"[COST] Session {session_id}: ${cost:.6f}")
return response["messages"][-1].content
배치 테스트로 성능 벤치마킹
def benchmark_agent():
"""에이전트 성능 벤치마크"""
test_cases = [
("반품 요청したい", "user_001"),
("배송 조혼知りたい", "user_002"),
("환불 진행해주세요", "user_003"),
]
results = []
for message, user_id in test_cases:
import time
start = time.time()
result = handle_customer_request(message, user_id)
latency = time.time() - start
results.append({
"user_id": user_id,
"message": message,
"latency": latency,
"response_length": len(result)
})
# 결과 분석
import pandas as pd
df = pd.DataFrame(results)
print(df.to_string())
print(f"\n평균 지연 시간: {df['latency'].mean()*1000:.2f}ms")
print(f"P95 지연 시간: {df['latency'].quantile(0.95)*1000:.2f}ms")LangSmith 대시보드에서는 각 툴 호출의 입력/출력, 토큰 사용량, 실행 시간, 그리고 에러 메시지를 확인할 수 있습니다. 저는 특히 @traceable 데코레이터를 사용하여 중요 함수의 실행 빈도와 성능 추이를 주기적으로 검토합니다.
커스텀 콜백으로 실시간 모니터링 구현
특정 모니터링 요구사항이 있을 경우 커스텀 콜백을 구현할 수 있습니다. 저는 Prometheus 메트릭을 수집하여 Grafana 대시보드에 연결하는 커스텀 콜백을 사용합니다.
from langgraph.callbacks.base import BaseCallbackHandler
from prometheus_client import Counter, Histogram, Gauge
import json
from datetime import datetime
Prometheus 메트릭 정의
AGENT_INVOCATIONS = Counter(
'agent_invocations_total',
'Total agent invocations',
['agent_name', 'intent']
)
NODE_EXECUTION_TIME = Histogram(
'node_execution_seconds',
'Node execution time',
['node_name']
)
ACTIVE_SESSIONS = Gauge(
'active_sessions_count',
'Number of active agent sessions'
)
STATE_CHANGES = Counter(
'state_changes_total',
'State changes',
['from_state', 'to_state']
)
class PrometheusCallbackHandler(BaseCallbackHandler):
"""Prometheus 메트릭을 수집하는 커스텀 콜백"""
def __init__(self):
super().__init__()
self.current_node = None
self.node_start_time = None
def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id, tags, metadata, **kwargs):
"""체인 시작 시 메트릭 초기화"""
agent_name = metadata.get("agent_name", "unknown")
ACTIVE_SESSIONS.inc()
# 입력 로깅
if "messages" in inputs:
last_message = inputs["messages"][-1]
print(f"[START] Agent: {agent_name}, Input: {last_message[:100]}")
def on_chain_end(self, outputs, *, run_id, inputs, **kwargs):
"""체인 종료 시 세션 카운터 감소"""
ACTIVE_SESSIONS.dec()
def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id, tags, metadata, **kwargs):
"""툴 실행 시작"""
tool_name = serialized.get("name", "unknown")
self.current_node = tool_name
self.node_start_time = datetime.now()
print(f"[TOOL] Executing: {tool_name}")
def on_tool_end(self, output, *, run_id, inputs, outputs, **kwargs):
"""툴 실행 완료 및 소요 시간 기록"""
if self.node_start_time:
duration = (datetime.now() - self.node_start_time).total_seconds()
NODE_EXECUTION_TIME.labels(node_name=self.current_node).observe(duration)
print(f"[TOOL] {self.current_node} completed in {duration*1000:.2f}ms")
def on_llm_start(self, serialized, messages, *, run_id, parent_run_id, tags, metadata, **kwargs):
"""LLM 호출 시작"""
model = serialized.get("name", "unknown")
print(f"[LLM] Model: {model}, Input tokens: {len(str(messages))}")
def on_llm_end(self, response, *, run_id, inputs, outputs, **kwargs):
"""LLM 응답 완료 및 토큰 사용량 기록"""
# HolySheep AI 비용 추적
usage = response.llm_output.get("usage", {}) if hasattr(response, "llm_output") else {}
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# Claude Sonnet 4.5 ($15/MTok) 기준 비용 계산
cost = (prompt_tokens + completion_tokens) / 1_000_000 * 15
print(f"[LLM] Cost: ${cost:.6f}, Tokens: {prompt_tokens + completion_tokens}")
def on_graph_node_start(self, node_id, data, **kwargs):
"""그래프 노드 시작"""
print(f"[NODE] Starting: {node_id}")
def on_graph_node_end(self, node_id, data, **kwargs):
"""그래프 노드 종료 및 상태 변경 기록"""
print(f"[NODE] Completed: {node_id}")
def on_graph_edge(self, source, target, data, **kwargs):
"""상태 전이(edge) 발생 시"""
STATE_CHANGES.labels(from_state=source, to_state=target).inc()
Prometheus + Grafana 설정 예시
def setup_monitoring_dashboard():
"""Grafana 대시보드 자동 설정"""
dashboard_config = {
"title": "LangGraph Agent Monitoring",
"panels": [
{
"title": "요청 처리량",
"type": "graph",
"targets": [
{"expr": "rate(agent_invocations_total[5m])"}
]
},
{
"title": "노드별 실행 시간",
"type": "heatmap",
"targets": [
{"expr": "histogram_quantile(0.95, node_execution_seconds_bucket)"}
]
},
{
"title": "활성 세션 수",
"type": "stat",
"targets": [
{"expr": "active_sessions_count"}
]
}
]
}
# Grafana API로 대시보드 생성
import requests
response = requests.post(
"https://grafana.example.com/api/dashboards/db",
json={"dashboard": dashboard_config},
headers={"Authorization": "Bearer your-grafana-api-key"}
)
print(f"Dashboard created: {response.status_code}")
모니터링 적용
monitoring_handler = PrometheusCallbackHandler()
compiled_graph = workflow.compile(
checkpointer=checkpointer,
callbacks=[monitoring_handler]
)커스텀 콜백의 장점은 모니터링 로직을 완전히 제어할 수 있다는 점입니다. 저는 Prometheus 메트릭을 수집하여 Grafana에서 실시간 대시보드를 구성하고, 특정 임계치를 초과하면 Slack으로 알림을 보내는 체계를 구축했습니다. 이로 인해午夜突发故障 시 평균 대응 시간이 2시간에서 15분으로 단축되었습니다.
자주 발생하는 오류와 해결책
1. MemorySaver 상태 누락 문제
개발 환경에서 MemorySaver를 사용할 때 프로세스 재시작 후 상태가 모두 유실되는 문제가 발생합니다.
# ❌ 잘못된 접근 - 매번 새 인스턴스 생성
def get_graph():
return workflow.compile(checkpointer=MemorySaver()) # 상태 유실!
✅ 해결책 - 싱글톤 패턴 또는 영구 저장소 사용
from functools import lru_cache
@lru_cache(maxsize=1)
def get_checkpointer():
if os.getenv("ENVIRONMENT") == "production":
return PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
return MemorySaver() # 개발용, 프로덕션에서는 PostgresSaver 권장
def get_graph():
return workflow.compile(checkpointer=get_checkpointer())
또는 모든 환경에서 Redis 사용
def get_redis_checkpointer():
from langgraph.checkpoint.redis import RedisSaver
import redis
redis_client = redis.from_url(os.environ["REDIS_URL"])
return RedisSaver(redis_client)
compiled_graph = workflow.compile(checkpointer=get_redis_checkpointer())2. 토큰 초과로 인한 무한 루프
LLM 응답이 매우 길어지거나 에이전트가 같은 노드를 반복 실행하는 경우가 있습니다.
# ❌ recursion_limit 미설정 - 무한 루프 위험
compiled_graph = workflow.compile(checkpointer=checkpointer)
✅ 해결책 - 명확한 recursion_limit 설정
from langgraph.errors import NodeInterrupt
MAX_STEPS = 20 # 최대 20단계까지만 허용
def safety_check(state: AgentState) -> AgentState:
"""안전장치: 허용된 단계 수 초과 시 중단"""
step_count = len(state.get("messages", []))
if step_count > MAX_STEPS:
raise NodeInterrupt(
f"Maximum steps ({MAX_STEPS}) exceeded. "
f"Last state: {state.get('current_step')}"
)
return state
노드 추가
workflow.add_node("safety_check", safety_check)
workflow.set_entry_point("safety_check")
또는 라우팅 조건으로 처리
def check_recursion(state: AgentState) -> bool:
return len(state["messages"]) < MAX_STEPS
workflow.add_conditional_edges(
"agent",
lambda state: check_recursion(state),
{True: "continue", False: END}
)
compiled_graph = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["action_executor"], # 특정 노드 전 중단점 설정
interrupt_after=["safety_check"] # 특정 노드 후 중단점 설정
)3. HolySheep AI API 연결 타임아웃
네트워크 문제나 API 일시적 장애로 연결이 실패하는 경우를 처리해야 합니다.
# ❌ 타임아웃 미설정 - 무한 대기
llm = ChatOpenAI(
model="gpt-4.1",
base_url="https://api.holysheep.ai/v1",
api_key=os.environ["HOLYSHEEP_API_KEY"]
)
✅ 해결책 - 재시도 로직과 타임아웃