LangSmith를 활용한 추적

LangSmith란 무엇인가요?

# Install: pip install langsmith
import os

# Set environment variables to enable automatic tracing
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = 'lsv2_...your_key_here...'
os.environ['LANGCHAIN_PROJECT'] = 'my-rag-app'  # project name in LangSmith UI

# That's all - LangChain now sends traces to LangSmith automatically
# No code changes needed to your chain or agent

코드 변경 없이 자동 추적하기

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# This chain is automatically traced - no extra code needed
llm = ChatOpenAI(model='gpt-4o')
prompt = ChatPromptTemplate.from_template('Answer this question: {question}')
chain = prompt | llm | StrOutputParser()

# This call creates a trace in LangSmith showing:
# - The formatted prompt (with question substituted)
# - The LLM call with model, temperature, token counts
# - The parsed output
# - End-to-end latency and cost
result = chain.invoke({'question': 'What is RAG?'})
print(result)

RAG 처리 과정 추적하기

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnablePassthrough

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={'k': 5})

rag_chain = (
    {'context': retriever, 'question': RunnablePassthrough()}
    | ChatPromptTemplate.from_template('Context: {context}\n\nQuestion: {question}\n\nAnswer:')
    | ChatOpenAI(model='gpt-4o')
    | StrOutputParser()
)

# LangSmith traces EVERY step:
# 1. Retriever: query embedding + vector search + returned documents (with scores)
# 2. Prompt: formatted template with context injected
# 3. LLM: full prompt, response, tokens, latency, cost
# 4. Parser: final string output
answer = rag_chain.invoke('What is the capital of France?')

추적에 메타데이터 추가하기

from langchain_core.runnables import RunnableConfig

def handle_user_request(user_id: str, query: str, ab_variant: str):
    config = RunnableConfig(
        tags=['production', ab_variant],
        metadata={
            'user_id': user_id,
            'ab_variant': ab_variant,
            'feature': 'rag_qa'
        }
    )
    
    result = rag_chain.invoke(query, config=config)
    return result

# In LangSmith UI you can now:
# - Filter traces by user_id to debug a specific user's issue
# - Compare latency between ab_variant='A' and ab_variant='B'
# - See all traces tagged 'production' vs 'staging'

수동 구간 만들기

from langsmith import Client, traceable

client = Client()

# Decorate functions to auto-trace them
@traceable(name='preprocess_query')
def preprocess_query(raw_query: str) -> str:
    # This step is now traced even though it doesn't use LangChain
    cleaned = raw_query.strip().lower()
    cleaned = ' '.join(cleaned.split())  # normalize whitespace
    return cleaned

@traceable(name='fetch_user_history')
def fetch_user_history(user_id: str) -> list[str]:
    # Database call - also traced
    return db.query('SELECT message FROM chat_history WHERE user_id = ? ORDER BY timestamp DESC LIMIT 5', user_id)

# All three steps appear in the same trace
def handle_request(user_id: str, raw_query: str):
    query = preprocess_query(raw_query)         # traced
    history = fetch_user_history(user_id)       # traced
    result = rag_chain.invoke({'query': query, 'history': history})  # traced by LangChain
    return result

LangSmith에서 추적 평가하기

from langsmith.evaluation import evaluate, LangChainStringEvaluator

# Create an evaluator that uses an LLM to judge correctness
correctness_evaluator = LangChainStringEvaluator(
    'qa',
    config={'llm': ChatOpenAI(model='gpt-4o')}
)

# Run evaluation against a dataset of traced examples
results = evaluate(
    rag_chain,
    data='my-rag-test-set',      # name of dataset in LangSmith
    evaluators=[correctness_evaluator],
    experiment_prefix='rag-v2-chunking-test'
)

print('Evaluation results:')
print(f'Correctness: {results.results["correctness"].mean():.2f}')
print(f'Average latency: {results.results["latency"].mean():.2f}s')

추적에서 시험 데이터 세트 만들기

from langsmith import Client

client = Client()

# Create a dataset from existing traces
dataset = client.create_dataset('rag-regression-tests')

# Add examples from production traces (by trace ID)
for trace_id in failed_trace_ids:
    run = client.read_run(trace_id)
    client.create_example(
        inputs=run.inputs,
        outputs={'answer': run.outputs.get('output', '')},
        dataset_id=dataset.id,
        metadata={'source': 'production_failure', 'date': run.start_time.isoformat()}
    )

print(f'Added {len(failed_trace_ids)} examples to regression test dataset')

추적 필터링 및 검색하기

from langsmith import Client

client = Client()

# Find slow traces (useful for performance investigation)
slow_runs = client.list_runs(
    project_name='my-rag-app',
    filter='gt(latency, 5)',  # latency > 5 seconds
    limit=20
)

# Find error traces
error_runs = client.list_runs(
    project_name='my-rag-app',
    filter='eq(error, true)',
    limit=50
)

# Find traces from a specific user
user_runs = client.list_runs(
    project_name='my-rag-app',
    filter='has(metadata, user_id="user_abc123")',
    limit=100
)

for run in slow_runs:
    print(f'Slow run: {run.id}, latency: {run.end_time - run.start_time}')

LangSmith에서 실험 비교하기

from langsmith.evaluation import evaluate

test_dataset = 'my-rag-eval-set'

# Run experiment A: chunk size 500
results_a = evaluate(
    rag_pipeline_v1,
    data=test_dataset,
    evaluators=[correctness_evaluator, relevance_evaluator],
    experiment_prefix='chunk-500'
)

# Run experiment B: chunk size 1000
results_b = evaluate(
    rag_pipeline_v2,
    data=test_dataset,
    evaluators=[correctness_evaluator, relevance_evaluator],
    experiment_prefix='chunk-1000'
)

# Compare in LangSmith UI: Experiments tab shows A vs B side by side
# Or compare programmatically:
print(f'Correctness - v1: {results_a.results["correctness"].mean():.2f}, v2: {results_b.results["correctness"].mean():.2f}')

운영 환경에서의 LangSmith

import os

# Production configuration
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_ENDPOINT'] = 'https://api.smith.langchain.com'
os.environ['LANGCHAIN_PROJECT'] = 'production'

# Enable async tracing (non-blocking - does not add latency to requests)
os.environ['LANGCHAIN_CALLBACKS_BACKGROUND'] = 'true'

# Optional: sample 10% of traces to reduce cost in high-traffic scenarios
import random

def should_trace() -> bool:
    return random.random() < 0.10  # 10% sampling rate

def handle_request(query):
    config = RunnableConfig()
    if not should_trace():
        config = RunnableConfig(callbacks=[])  # disable tracing for this request
    return rag_chain.invoke(query, config=config)

LangSmith와 사용자 지정 로그 기록 비교

빠른 확인

강의 요약