AI Agents · レッスン

LangSmith と Langfuse によるトレース分析

トレースを読み取り、遅いツール、誤った判断、エラーパターンを特定します。

レッスン 1/413 ステップ

「LangSmith と Langfuse によるトレース分析」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

エージェントをトレースする理由

エージェントは1回の実行につき、複数のLLM呼び出しやツール呼び出しを行います。トレーシングがなければ、デバッグは推測に頼ることになります。トレーシングでは、入力、出力、トークン使用量、レイテンシ、エラーなど、すべてのステップを記録し、各実行の全体像を把握できます。

LangSmithのセットアップ

LangSmithは、LangChain向けのAnthropicのトレーシングプラットフォームです。環境変数を2つ設定して有効にします。すべてのLangChain呼び出しが自動的にトレースされ、LangSmith UIで確認できるようになります。

import os
from dotenv import load_dotenv

load_dotenv()

# LangSmith tracing configuration
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = os.environ.get('LANGSMITH_API_KEY', 'ls__...')
os.environ['LANGCHAIN_PROJECT'] = 'my-agent-project'

# Now any LangChain code is automatically traced
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-4o-mini', api_key=os.environ.get('OPENAI_API_KEY', 'sk-...'))

# This call is traced automatically
response = llm.invoke([HumanMessage(content='What is 2+2?')])
print(response.content)
# Check trace at: https://smith.langchain.com

実行メタデータの追加

トレースにタグとメタデータを追加すると、LangSmith UIでフィルタリングや検索ができるようになります。異なるエージェントのバージョン、ユーザーID、実験ラベルなどの追跡に役立ちます。

import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langsmith import traceable

os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = 'ls__your-key'
os.environ['LANGCHAIN_PROJECT'] = 'my-agent-project'

llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')

@traceable(name='my-agent-run', tags=['production', 'v2'], metadata={'user_id': '42'})
def run_agent(question: str) -> str:
    response = llm.invoke(
        [HumanMessage(content=question)],
        config={
            'run_name': f'agent-{question[:20]}',
            'tags': ['production'],
            'metadata': {'user_id': '42', 'version': 'v2.1'}
        }
    )
    return response.content

result = run_agent('Explain LangChain tracing')
print(result)

LangSmith UIでトレースを確認する

LangSmithのダッシュボードでは、完全なトレースツリーとともにすべての実行を確認できます。各ノードには、入力、出力、トークン数、レイテンシ、エラーの有無が表示されます。実行を比較したり、タグやプロジェクトでフィルタリングしたりすることもできます。

  • エラーステータスでフィルタリングして、失敗した実行を見つけます
  • レイテンシで並べ替えて、遅いステップを特定します
  • 2つの実行を横に並べて比較し、リグレッションをデバッグします
# Programmatically query LangSmith for run data
from langsmith import Client

client = Client(api_key='ls__your-key')

# List recent runs for a project
runs = list(client.list_runs(
    project_name='my-agent-project',
    execution_order=1,      # Top-level runs only
    error=True,             # Only failed runs
    limit=10
))

for run in runs:
    print(f'Run: {run.name}')
    print(f'  Status: {run.status}')
    print(f'  Latency: {run.end_time - run.start_time if run.end_time else "running"}')
    print(f'  Error: {run.error}')
    print()

カスタムトレーシングにLangfuseを使う

Langfuseは、LangSmithに代わるオープンソースのツールです。どのLLMフレームワークやカスタムコードとも連携できます。Langfuse SDKを使って、トレースとスパンを手動で作成します。

from langfuse import Langfuse

lf = Langfuse(
    public_key='pk-lf-...',
    secret_key='sk-lf-...',
    host='https://cloud.langfuse.com'  # Or your self-hosted URL
)

# Create a trace
trace = lf.trace(
    name='email-agent-run',
    user_id='user-42',
    metadata={'environment': 'production'}
)

# Create a span for entity extraction
span = trace.span(
    name='entity-extraction',
    input={'text': 'Meeting with Alice from Google tomorrow'}
)

# Simulate work
extracted = ['Alice', 'Google']

# End the span with output
span.end(output={'entities': extracted})

print('Trace created in Langfuse')
print(f'View at: https://cloud.langfuse.com/trace/{trace.id}')

LangfuseでLLM呼び出しをトレースする

LLMを呼び出すたびにgenerationスパンを作成します。使用したモデル、プロンプト、生成結果、トークン数を記録でき、コスト分析に最も重要なデータを取得できます。

from langfuse import Langfuse
import openai

lf = Langfuse(public_key='pk-lf-...', secret_key='sk-lf-...')
client = openai.OpenAI(api_key='sk-...')

def traced_llm_call(trace, prompt: str, model: str = 'gpt-4o-mini') -> str:
    generation = trace.generation(
        name='llm-call',
        model=model,
        input=[{'role': 'user', 'content': prompt}]
    )
    
    response = client.chat.completions.create(
        model=model,
        messages=[{'role': 'user', 'content': prompt}]
    )
    content = response.choices[0].message.content
    
    generation.end(
        output=content,
        usage={
            'prompt_tokens': response.usage.prompt_tokens,
            'completion_tokens': response.usage.completion_tokens,
            'total_tokens': response.usage.total_tokens
        }
    )
    return content

trace = lf.trace(name='test-trace')
result = traced_llm_call(trace, 'What is the capital of France?')
print('Result:', result)

エラーとレイテンシで実行をフィルタリングする

LangSmithクライアントを使うと、問題のある実行をプログラムから見つけられます。エラーステータス、レイテンシのしきい値、特定のタグでフィルタリングし、デバッグ対象を絞り込みます。

from langsmith import Client
from datetime import datetime, timedelta

client = Client(api_key='ls__your-key')

def find_slow_runs(project: str, latency_threshold_ms: int = 10000):
    runs = list(client.list_runs(
        project_name=project,
        execution_order=1,
        start_time=datetime.utcnow() - timedelta(hours=24)
    ))
    
    slow_runs = []
    for run in runs:
        if run.end_time and run.start_time:
            duration_ms = (run.end_time - run.start_time).total_seconds() * 1000
            if duration_ms > latency_threshold_ms:
                slow_runs.append({
                    'id': str(run.id),
                    'name': run.name,
                    'duration_ms': round(duration_ms),
                    'tags': run.tags
                })
    
    slow_runs.sort(key=lambda x: x['duration_ms'], reverse=True)
    return slow_runs

print('Find slow runs function defined')
print('Usage: find_slow_runs("my-agent-project", latency_threshold_ms=5000)')

実行を比較する

LangSmithでは、UI上で2つの実行を比較して変更点を確認できます。プログラムからは、実行結果、トークン使用量、レイテンシを比較し、モデルやプロンプトの変更後に発生したリグレッションを検出できます。

from langsmith import Client

client = Client(api_key='ls__your-key')

def compare_runs(run_id_1: str, run_id_2: str) -> dict:
    run1 = client.read_run(run_id_1)
    run2 = client.read_run(run_id_2)
    
    def get_tokens(run):
        if run.total_tokens:
            return run.total_tokens
        return 0
    
    def get_latency_ms(run):
        if run.end_time and run.start_time:
            return (run.end_time - run.start_time).total_seconds() * 1000
        return 0
    
    return {
        'run1': {'id': run_id_1, 'tokens': get_tokens(run1), 'latency_ms': get_latency_ms(run1), 'status': run1.status},
        'run2': {'id': run_id_2, 'tokens': get_tokens(run2), 'latency_ms': get_latency_ms(run2), 'status': run2.status},
        'token_delta': get_tokens(run2) - get_tokens(run1),
        'latency_delta_ms': get_latency_ms(run2) - get_latency_ms(run1)
    }

print('Run comparison function defined')

スコアとフィードバックを追加する

エージェントの実行を手動または自動で評価した後、トレースにスコアやフィードバックを追加します。これにより、ファインチューニングやプロンプト変更の評価に使えるデータセットを作成できます。

from langsmith import Client

client = Client(api_key='ls__your-key')

def score_run(run_id: str, score: float, reasoning: str = ''):
    # score: 0.0 (bad) to 1.0 (perfect)
    client.create_feedback(
        run_id=run_id,
        key='quality',
        score=score,
        comment=reasoning
    )

def auto_evaluate_run(run_id: str, expected_output: str, actual_output: str) -> float:
    # Simple heuristic: check if key terms from expected output are present
    expected_terms = set(expected_output.lower().split())
    actual_terms = set(actual_output.lower().split())
    overlap = len(expected_terms & actual_terms) / max(len(expected_terms), 1)
    score = min(1.0, overlap * 1.5)  # Normalize
    score_run(run_id, score, f'Term overlap: {overlap:.2f}')
    return score

print('Scoring functions defined')
print('Example: score_run("run-id-abc", 0.85, "Good answer but missing one detail")')

構造化されたトレースコンテキスト

セッションID、ユーザーID、エージェントのバージョン、フィーチャーフラグなど、意味のあるコンテキストをトレースに付加します。これにより、トレースを簡単に分類し、異なる設定間でパフォーマンスを比較できます。

import os
from langsmith import traceable
from langchain_core.runnables import RunnableConfig

def build_trace_config(user_id: str, session_id: str, version: str) -> dict:
    return {
        'metadata': {
            'user_id': user_id,
            'session_id': session_id,
            'agent_version': version,
            'environment': os.environ.get('ENV', 'development')
        },
        'tags': [version, os.environ.get('ENV', 'development')],
        'run_name': f'agent-{user_id[:8]}'
    }

@traceable
def run_agent_with_context(question: str, user_id: str, session_id: str):
    config = build_trace_config(user_id, session_id, 'v2.3')
    # Pass config to any LangChain component
    # llm.invoke([HumanMessage(content=question)], config=config)
    print(f'Running agent for user {user_id}, session {session_id}')
    return 'Answer here'

result = run_agent_with_context('Question', 'user-001', 'sess-xyz')
print(result)

アラートを設定する

LangSmithまたはLangfuseでアラートを設定し、エージェントの健全性を監視します。エラー率がしきい値を超えたとき、P99レイテンシが急上昇したとき、または特定のステップが繰り返し失敗したときにアラートを発生させます。

from langsmith import Client
from datetime import datetime, timedelta

client = Client(api_key='ls__your-key')

def check_error_rate(project: str, window_minutes: int = 60, threshold: float = 0.05) -> dict:
    runs = list(client.list_runs(
        project_name=project,
        execution_order=1,
        start_time=datetime.utcnow() - timedelta(minutes=window_minutes)
    ))
    
    if not runs:
        return {'error_rate': 0.0, 'alert': False}
    
    error_count = sum(1 for r in runs if r.status == 'error')
    error_rate = error_count / len(runs)
    
    if error_rate > threshold:
        print(f'ALERT: Error rate {error_rate:.1%} exceeds threshold {threshold:.1%}')
        # Send to Slack/PagerDuty here
    
    return {
        'total_runs': len(runs),
        'error_count': error_count,
        'error_rate': round(error_rate, 4),
        'alert': error_rate > threshold
    }

print('Error rate monitor defined')

理解度チェック:トレーシング

LangSmithとLangfuseを使ったエージェントのトレーシングについて理解度を確認します。

トレーシングのまとめ

LangSmithとLangfuseは、互いに補完し合うツールです。LangSmithはLangChainと緊密に統合され、最小限のセットアップで利用できる一方、Langfuseはどのフレームワークでも動作し、より細かく制御できます。どちらも、エージェントの各ステップについて入力、出力、トークン使用量、レイテンシ、エラーを記録します。フィルタリング、スコアリング、アラートを活用して、本番環境でエージェントの品質を維持します。

無料で開始

AI チューターと学ぶ AI Agents — 無料

ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。

コース
60
レッスン
239

よくある質問

「LangSmith と Langfuse によるトレース分析」レッスンは無料ですか?

はい。「LangSmith と Langfuse によるトレース分析」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「LangSmith と Langfuse によるトレース分析」で何を学びますか?

トレースを読み取り、遅いツール、誤った判断、エラーパターンを特定します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか?

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。

「LangSmith と Langfuse によるトレース分析」レッスンにはどのくらい時間がかかりますか?

ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか?

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. LangSmith と Langfuse によるトレース分析
  2. ステップごとのトークン数とコストのプロファイリング
  3. 遅くてコストの高いステップを特定する
  4. エージェントの失敗に対する根本原因分析
← AI Agentsに戻る