langsmith-telemetry

name	langsmith-telemetry
description	LangSmith 통합 및 LLM Observability 가이드. 토큰 추적, 비용 계산, Run Tree, Tracing 데코레이터, OTEL 연동 구현 시 참조. "langsmith", "tracing", "observability", "token usage", "cost tracking" 키워드로 트리거.

LangSmith Telemetry Guide

개요

LangSmith는 LangChain/LangGraph 애플리케이션의 Observability 플랫폼. LLM 호출, 토큰 사용량, 비용, 레이턴시를 추적.

┌─────────────────────────────────────────────────────────────────────────┐
│                    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                                                  │   │
│  └──────────────────────────────────────────────────────────────────┘   │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

환경 변수

# 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

설치

pip install langsmith langchain-core

설정 확인 유틸리티

# 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)

# 토큰당 가격 (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 기본 사용

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()

커스텀 메타데이터 추가

@traceable(
    name="answer_generation",
    metadata={"feature": "chat", "version": "2.0"},
    tags=["production", "chat"],
)
async def generate_answer(context: dict, question: str) -> str:
    """답변 생성."""
    ...

LLM 호출 추적

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 호출 추적

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):
    """카카오 장소 검색."""
    ...

이미지 생성 추적

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

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 매핑

# 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 연동 (선택)

# 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² 실제 구현

# 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

1. 환경별 프로젝트 분리: LANGCHAIN_PROJECT=eco2-{env}
2. 태그 일관성: feature:xxx, intent:xxx, model:xxx 형식
3. 비용 추적: 모든 LLM 호출에 토큰/비용 메트릭 추가
4. 에러 컨텍스트: 실패 시 상세 정보 포함
5. 샘플링: 프로덕션에서 100% 추적은 비용 부담 → 샘플링 고려

Reference

• LangSmith Docs
• LangSmith Python SDK
• OpenTelemetry Integration