LLMアプリのデバッグが難しい理由

LLMデバッグ特有の難しさ

非決定性によって再現が難しくなる

import json
import time

def logged_llm_call(client, messages, model, temperature, **kwargs):
    request_id = f'{int(time.time() * 1000)}-{id(messages)}'
    
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        **kwargs
    )
    
    # Log EVERYTHING needed to reproduce this exact call
    log_entry = {
        'request_id': request_id,
        'model': model,
        'temperature': temperature,
        'messages': messages,
        'response': response.choices[0].message.content,
        'finish_reason': response.choices[0].finish_reason,
        'usage': response.usage.model_dump(),
        'timestamp': time.time()
    }
    write_to_trace_store(log_entry)
    return response

サイレントな失敗：壊れているのではなく間違っている

# This succeeds with HTTP 200 but returns wrong information
response = client.chat.completions.create(
    model='gpt-4o',
    messages=[{'role': 'user', 'content': 'What is the boiling point of water at sea level?'}]
)

output = response.choices[0].message.content
# response.status_code: None (not relevant - always 200 if we got here)
# No exception thrown
# But if output is '90 degrees Celsius', it is WRONG and your app will serve bad data

# You need semantic validation:
def validate_boiling_point_answer(text: str) -> bool:
    return '100' in text  # Rough check - real validation is more sophisticated

マルチステップチェーン：どこで間違ったのか

# Without tracing: you see only the final wrong answer
def rag_pipeline_naive(query):
    chunks = retrieve(query)         # step 1 - might return bad chunks
    context = format_context(chunks) # step 2 - might truncate key info
    answer = generate(query, context) # step 3 - LLM gets bad context
    return answer  # WRONG - but why?

# With tracing: you can see each step's input and output
def rag_pipeline_traced(query):
    with trace_span('retrieve') as span:
        chunks = retrieve(query)
        span.set_attribute('num_chunks', len(chunks))
        span.set_attribute('top_chunk_score', chunks[0]['score'] if chunks else 0)
    
    with trace_span('format_context') as span:
        context = format_context(chunks)
        span.set_attribute('context_length', len(context))
    
    with trace_span('generate') as span:
        answer = generate(query, context)
        span.set_attribute('answer_length', len(answer))
    
    return answer  # Now you can diagnose: was retrieve the problem?

トークン数とコストの予想外の増加

COST_PER_1K = {'gpt-4o': {'input': 0.005, 'output': 0.015},
               'gpt-4o-mini': {'input': 0.000150, 'output': 0.000600}}

def compute_cost(model: str, usage) -> float:
    pricing = COST_PER_1K.get(model, {'input': 0.005, 'output': 0.015})
    input_cost = (usage.prompt_tokens / 1000) * pricing['input']
    output_cost = (usage.completion_tokens / 1000) * pricing['output']
    return input_cost + output_cost

def instrumented_call(client, model, messages):
    response = client.chat.completions.create(model=model, messages=messages)
    cost = compute_cost(model, response.usage)
    
    # Alert if single call is unexpectedly expensive
    if cost > 0.10:  # more than 10 cents for one call
        print(f'WARNING: Expensive LLM call: ${cost:.4f} ({response.usage.prompt_tokens} prompt tokens)')
    
    metrics.record('llm_cost_usd', cost, tags={'model': model})
    metrics.record('llm_prompt_tokens', response.usage.prompt_tokens)
    return response

レイテンシ：どのステップが遅いのか

import time
from contextlib import contextmanager

@contextmanager
def timed(name: str, metrics_client):
    start = time.monotonic()
    try:
        yield
    finally:
        elapsed_ms = (time.monotonic() - start) * 1000
        metrics_client.histogram(f'step_latency_ms', elapsed_ms, tags={'step': name})
        if elapsed_ms > 2000:  # flag steps taking more than 2 seconds
            print(f'SLOW STEP [{name}]: {elapsed_ms:.0f}ms')

# Usage
def rag_with_timing(query, metrics):
    with timed('embed_query', metrics):
        query_embedding = embed(query)
    
    with timed('vector_search', metrics):
        chunks = vector_db.search(query_embedding, top_k=5)
    
    with timed('llm_generate', metrics):
        answer = generate(query, chunks)
    
    return answer

実際に必要な情報

from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class LLMTrace:
    request_id: str
    session_id: str
    step_name: str
    model: str
    temperature: float
    system_prompt: str
    user_messages: list[dict]
    response: str
    finish_reason: str
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    latency_ms: float
    tool_calls: list[dict] = field(default_factory=list)
    error: Optional[str] = None
    timestamp: float = field(default_factory=time.time)

    def is_anomalous(self) -> bool:
        return (
            self.cost_usd > 0.10 or
            self.latency_ms > 10000 or
            self.finish_reason == 'length' or  # was cut off
            self.error is not None
        )

リクエストIDによるトレースの関連付け

import uuid
from contextvars import ContextVar

# Thread-safe request ID propagation using context variables
request_id_var: ContextVar[str] = ContextVar('request_id', default='unknown')

def handle_user_request(query: str):
    # Set request ID at the entry point
    req_id = str(uuid.uuid4())[:8]
    request_id_var.set(req_id)
    return rag_pipeline(query)

def get_current_request_id() -> str:
    return request_id_var.get()

# Every LLM call logs with the same request_id
def log_llm_call(model, prompt, response):
    logger.info('LLM call', extra={
        'request_id': get_current_request_id(),  # automatically correlates all calls
        'model': model,
        'prompt_length': len(prompt),
        'response_length': len(response)
    })

LLM可観測性スタック

品質低下のアラート

可観測性の取り組みを始める

クイックチェック

レッスンのまとめ