LangSmith Telemetry Guide
개요
LangSmith는 LangChain/LangGraph 애플리케이션의 Observability 플랫폼. LLM 호출, 토큰 사용량, 비용, 레이턴시를 추적.
code
┌─────────────────────────────────────────────────────────────────────────┐ │ LangSmith Observability Stack │ ├─────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────────────────────────────────────────────────────────┐ │ │ │ Application │ │ │ │ - LangGraph Pipeline │ │ │ │ - LLM Clients (OpenAI, Gemini) │ │ │ │ - Tool Executions │ │ │ └──────────────────────────────────────────────────────────────────┘ │ │ │ │ │ @traceable decorators │ │ RunTree context │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────────────────┐ │ │ │ LangSmith API │ │ │ │ - Run traces │ │ │ │ - Token counts │ │ │ │ - Latency metrics │ │ │ │ - Cost estimates │ │ │ └──────────────────────────────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────────────────┐ │ │ │ LangSmith Dashboard │ │ │ │ - Trace visualization │ │ │ │ - Token usage graphs │ │ │ │ - Cost analysis │ │ │ │ - Error tracking │ │ │ └──────────────────────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────┘
환경 변수
bash
# LangSmith 활성화 (필수) LANGCHAIN_TRACING_V2=true # API Key (필수) LANGCHAIN_API_KEY=ls-xxxxx # 프로젝트명 LANGCHAIN_PROJECT=eco2-chat-worker # Endpoint (기본값 사용 권장) LANGCHAIN_ENDPOINT=https://api.smith.langchain.com # OTEL 연동 (선택) LANGSMITH_OTEL_ENABLED=true
설치
bash
pip install langsmith langchain-core
설정 확인 유틸리티
python
# apps/chat_worker/infrastructure/telemetry/langsmith.py
import os
LANGSMITH_ENABLED: bool = os.environ.get("LANGCHAIN_TRACING_V2", "").lower() == "true"
LANGSMITH_API_KEY: str | None = os.environ.get("LANGCHAIN_API_KEY")
LANGSMITH_PROJECT: str = os.environ.get("LANGCHAIN_PROJECT", "eco2-chat-worker")
def is_langsmith_enabled() -> bool:
"""LangSmith 활성화 여부 확인."""
return LANGSMITH_ENABLED and bool(LANGSMITH_API_KEY)
def get_langsmith_config() -> dict:
"""LangSmith 설정 반환."""
return {
"enabled": is_langsmith_enabled(),
"project": LANGSMITH_PROJECT,
"endpoint": os.environ.get("LANGCHAIN_ENDPOINT", "https://api.smith.langchain.com"),
}
모델 가격표 (2025)
python
# 토큰당 가격 (USD per 1M tokens)
MODEL_PRICING: dict[str, dict[str, float]] = {
# OpenAI
"gpt-5.2": {"input": 1.75, "output": 14.00},
"gpt-5.2-codex": {"input": 2.00, "output": 16.00},
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"o1": {"input": 15.00, "output": 60.00},
"o1-mini": {"input": 3.00, "output": 12.00},
# Gemini
"gemini-3-pro-preview": {"input": 2.00, "output": 12.00},
"gemini-3-flash-preview": {"input": 0.50, "output": 3.00},
"gemini-2.0-flash": {"input": 0.10, "output": 0.40},
"gemini-1.5-pro": {"input": 1.25, "output": 5.00},
"gemini-1.5-flash": {"input": 0.075, "output": 0.30},
# Anthropic
"claude-opus-4": {"input": 15.00, "output": 75.00},
"claude-sonnet-4": {"input": 3.00, "output": 15.00},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
}
# 이미지 생성 가격
IMAGE_MODEL_PRICING: dict[str, dict[str, float]] = {
"gemini-3-pro-image-preview": {"1k": 0.134, "4k": 0.24, "default": 0.134},
"gpt-image-1.5": {"1024x1024": 0.04, "1024x1792": 0.08, "default": 0.04},
"dall-e-3": {"1024x1024": 0.04, "1024x1792": 0.08, "default": 0.04},
}
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""토큰 사용량 기반 비용 계산."""
pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6)
def calculate_image_cost(model: str, resolution: str = "default") -> float:
"""이미지 생성 비용 계산."""
pricing = IMAGE_MODEL_PRICING.get(model, {})
return pricing.get(resolution, pricing.get("default", 0))
Tracing 데코레이터
@traceable 기본 사용
python
from langsmith import traceable
@traceable(name="classify_intent")
async def classify_intent(message: str) -> dict:
"""의도 분류 (자동 추적)."""
response = await llm.generate_structured(
prompt=message,
response_schema=IntentClassification,
)
return response.model_dump()
커스텀 메타데이터 추가
python
@traceable(
name="answer_generation",
metadata={"feature": "chat", "version": "2.0"},
tags=["production", "chat"],
)
async def generate_answer(context: dict, question: str) -> str:
"""답변 생성."""
...
LLM 호출 추적
python
from functools import wraps
from langsmith import traceable
from langsmith.run_trees import RunTree
def traceable_llm(model: str):
"""LLM 호출 전용 데코레이터.
토큰 사용량과 비용을 자동 추적.
"""
def decorator(func):
@wraps(func)
@traceable(name=f"llm:{model}", run_type="llm")
async def wrapper(*args, **kwargs):
result = await func(*args, **kwargs)
# RunTree에 메트릭 추가
run_tree = RunTree.get_current_run_tree()
if run_tree and hasattr(result, "usage"):
usage = result.usage
track_token_usage(
run_tree,
model=model,
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
)
return result
return wrapper
return decorator
def track_token_usage(
run_tree: RunTree,
model: str,
input_tokens: int,
output_tokens: int,
):
"""토큰 사용량을 RunTree에 기록."""
cost = calculate_cost(model, input_tokens, output_tokens)
run_tree.extra = run_tree.extra or {}
run_tree.extra["metrics"] = {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
"cost_usd": cost,
"model": model,
}
# 사용 예시
@traceable_llm(model="gpt-5.2")
async def generate_with_gpt(prompt: str):
return await openai_client.chat.completions.create(
model="gpt-5.2",
messages=[{"role": "user", "content": prompt}],
)
Tool 호출 추적
python
def traceable_tool(name: str, tool_type: str = "tool"):
"""Tool 호출 추적 데코레이터."""
def decorator(func):
@wraps(func)
@traceable(name=f"tool:{name}", run_type="tool", metadata={"tool_type": tool_type})
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
return decorator
@traceable_tool(name="kakao_search", tool_type="external_api")
async def search_places(query: str, lat: float, lon: float):
"""카카오 장소 검색."""
...
이미지 생성 추적
python
def traceable_image(model: str):
"""이미지 생성 추적 데코레이터."""
def decorator(func):
@wraps(func)
@traceable(name=f"image:{model}", run_type="tool")
async def wrapper(*args, **kwargs):
result = await func(*args, **kwargs)
run_tree = RunTree.get_current_run_tree()
if run_tree:
resolution = kwargs.get("resolution", "default")
cost = calculate_image_cost(model, resolution)
run_tree.extra = run_tree.extra or {}
run_tree.extra["metrics"] = {
"cost_usd": cost,
"model": model,
"resolution": resolution,
}
return result
return wrapper
return decorator
LangGraph Run Config
python
def get_run_config(
job_id: str,
session_id: str | None = None,
user_id: str | None = None,
intent: str | None = None,
tags: list[str] | None = None,
) -> dict:
"""LangGraph 실행 config 생성.
LangSmith에서 추적할 메타데이터 포함.
"""
run_tags = list(tags) if tags else []
# Intent 태그 추가
if intent:
run_tags.append(f"intent:{intent}")
return {
"run_name": f"chat:{job_id}",
"tags": run_tags,
"metadata": {
"job_id": job_id,
"user_id": user_id,
"intent": intent,
},
"configurable": {
"thread_id": session_id if session_id else None,
},
}
# 사용 예시
config = get_run_config(
job_id="job-123",
session_id="sess-456",
user_id="user-789",
intent="waste",
tags=["env:production", "model:gpt-5.2"],
)
result = await pipeline.ainvoke(state, config=config)
Feature-Intent 매핑
python
# LangSmith 대시보드에서 기능별 필터링을 위한 매핑
INTENT_FEATURE_MAP: dict[str, dict[str, Any]] = {
"waste": {
"feature": "rag",
"subagents": ["waste_rag", "weather"],
"has_feedback": True,
"description": "분리배출 RAG 검색",
},
"bulk_waste": {
"feature": "external_api",
"subagents": ["bulk_waste", "weather"],
"has_feedback": False,
"description": "대형폐기물 (행정안전부 API)",
},
"location": {
"feature": "external_api",
"subagents": ["location"],
"has_feedback": False,
"description": "장소 검색 (카카오맵)",
},
"character": {
"feature": "grpc",
"subagents": ["character"],
"has_feedback": False,
"description": "캐릭터 응답 (gRPC)",
},
"weather": {
"feature": "external_api",
"subagents": ["weather"],
"has_feedback": False,
"description": "날씨 정보 (기상청)",
},
"web_search": {
"feature": "web_search",
"subagents": ["web_search"],
"has_feedback": False,
"description": "웹 검색 (OpenAI/Gemini)",
},
"image_generation": {
"feature": "image",
"subagents": ["image_generation"],
"has_feedback": False,
"description": "이미지 생성 (Gemini Imagen)",
},
"general": {
"feature": "llm",
"subagents": ["general"],
"has_feedback": False,
"description": "일반 LLM 응답",
},
}
def get_feature_tags(intent: str) -> list[str]:
"""Intent에 해당하는 feature 태그 반환."""
mapping = INTENT_FEATURE_MAP.get(intent, {})
tags = []
if feature := mapping.get("feature"):
tags.append(f"feature:{feature}")
for subagent in mapping.get("subagents", []):
tags.append(f"subagent:{subagent}")
return tags
OTEL 연동 (선택)
python
# OpenTelemetry와 LangSmith 통합
import os
if os.environ.get("LANGSMITH_OTEL_ENABLED", "").lower() == "true":
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
# OTEL 설정
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
# LangSmith + OTEL 연결
from langsmith.wrappers import wrap_opentelemetry
wrap_opentelemetry()
Eco² 실제 구현
python
# apps/chat_worker/application/commands/process_chat.py
class ProcessChatCommand:
"""채팅 처리 Command."""
async def execute(self, job: ChatJob) -> ChatResult:
"""채팅 처리 실행."""
# LangSmith config 생성
config = get_run_config(
job_id=job.job_id,
session_id=job.thread_id,
user_id=job.user_id,
intent=None, # 아직 모름
tags=[
f"env:{os.environ.get('ENV', 'development')}",
f"model:{job.llm_model or 'default'}",
],
)
# State 초기화
state = ChatState(
job_id=job.job_id,
user_id=job.user_id,
thread_id=job.thread_id,
message=job.message,
image_url=job.image_url,
user_location=job.user_location,
llm_model=job.llm_model,
llm_provider=job.llm_provider,
)
# Pipeline 실행 (LangSmith 자동 추적)
result = await self._pipeline.ainvoke(state, config=config)
# Intent 알게 된 후 태그 업데이트 (run_tree 사용)
if is_langsmith_enabled():
from langsmith import Client
client = Client()
# 현재 run에 intent 태그 추가
# (실제로는 run_tree.patch() 사용)
intent = result.get("intent", "unknown")
feature_tags = get_feature_tags(intent)
# client.update_run(run_id, tags=feature_tags)
return ChatResult(
answer=result.get("answer", ""),
intent=result.get("intent"),
)
Best Practices
- •환경별 프로젝트 분리:
LANGCHAIN_PROJECT=eco2-{env} - •태그 일관성:
feature:xxx,intent:xxx,model:xxx형식 - •비용 추적: 모든 LLM 호출에 토큰/비용 메트릭 추가
- •에러 컨텍스트: 실패 시 상세 정보 포함
- •샘플링: 프로덕션에서 100% 추적은 비용 부담 → 샘플링 고려