「動かない」を防ぐ！LangSmithを活用したLLMエージェントの挙動デバッグ術

はじめに

LLMエージェントを本番環境で運用していると、「開発環境では動いていたのに本番で失敗する」「なぜこのプロンプトが選ばれたのか分からない」といった課題に直面します。従来のログでは、LLMの内部状態やエージェントの意思決定プロセスを追跡することが困難でした。

本記事では、LangSmithを使ったLLMエージェントのデバッグ手法を、実装コード例とともに解説します。エージェントの挙動を可視化し、問題を素早く特定するための実践的なテクニックを紹介します。

LangSmithとは？

LangSmithは、LangChainが提供するLLMアプリケーションの開発・デバッグ・モニタリングプラットフォームです。LLMの入出力、エージェントの実行トレース、コスト分析などを一元管理できます。

主な機能として、トレーシング（実行履歴の記録）、デバッグ（問題箇所の特定）、評価（出力品質の測定）、モニタリング（本番環境の監視）を提供します。特にマルチエージェントシステムでは、各エージェントの実行フローを可視化できるため、複雑な挙動の理解に役立ちます。

まとめ

• LangSmithのトレーシング機能で、LLMエージェントの実行フローを完全に可視化できる
• プロンプトのバージョン管理とA/Bテストで、最適なプロンプトを科学的に選定可能
• カスタムメタデータとフィルタリングで、本番障害の原因を素早く特定できる
• コスト分析とレイテンシ監視で、運用コストとパフォーマンスを最適化できる
• 評価データセットを活用し、プロンプト改善の効果を定量的に測定できる

LangSmithの基本セットアップ

まず、LangSmithを使い始めるための基本的なセットアップを行います。

インストールと環境設定

# 必要なパッケージのインストール
pip install langsmith langchain-openai langchain

# 環境変数の設定
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

プロジェクト名を設定することで、複数のプロジェクトを並行して管理できます。開発環境と本番環境で異なるプロジェクト名を使い分けることで、トレースを分離して管理できます。

基本的なトレーシングの実装

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import tool
from langsmith import traceable

# カスタムツールの定義
@tool
def get_weather(location: str) -> str:
    """指定された場所の天気情報を取得します"""
    # 実際にはAPIを呼び出す
    return f"{location}の天気は晴れです"

@tool
def calculate(expression: str) -> str:
    """数式を計算します"""
    try:
        result = eval(expression)
        return f"計算結果: {result}"
    except Exception as e:
        return f"エラー: {str(e)}"

# エージェントのセットアップ
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [get_weather, calculate]

prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは親切なアシスタントです。ツールを使って質問に答えてください。"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# エージェントの実行（自動的にトレースされる）
result = agent_executor.invoke({"input": "東京の天気を教えて、その後10+20を計算して"})
print(result["output"])

このコードを実行すると、LangSmithのダッシュボードに実行トレースが自動的に記録されます。エージェントがどのツールを選択し、どのような順序で実行したかが可視化されます。

トレースの詳細分析とデバッグ

LangSmithのトレース機能を使うと、エージェントの実行過程を詳細に分析できます。

メタデータの追加とフィルタリング

from langsmith import Client
from langchain_core.runnables import RunnableConfig

client = Client()

# カスタムメタデータを追加してトレース
config = RunnableConfig(
    tags=["production", "customer-support"],
    metadata={
        "user_id": "user_123",
        "session_id": "session_456",
        "environment": "production",
        "version": "v2.1"
    }
)

result = agent_executor.invoke(
    {"input": "在庫管理システムの現在の在庫数を確認して"},
    config=config
)

メタデータを活用することで、特定のユーザー、セッション、環境でのトレースを素早く検索できます。本番環境で問題が発生した際、該当するユーザーIDやセッションIDでフィルタリングすることで、問題の再現性を確認できます。

エラートレースの分析

from langsmith.run_helpers import traceable

@traceable(name="custom_processing", run_type="chain")
def process_user_query(query: str, user_context: dict) -> dict:
    """ユーザークエリを処理する関数"""
    try:
        # クエリの前処理
        processed_query = preprocess(query, user_context)
        
        # エージェントの実行
        result = agent_executor.invoke({"input": processed_query})
        
        # 後処理
        final_result = postprocess(result, user_context)
        
        return {
            "success": True,
            "result": final_result,
            "tokens_used": result.get("token_usage", {})
        }
    except Exception as e:
        # エラー情報をトレースに記録
        return {
            "success": False,
            "error": str(e),
            "error_type": type(e).__name__,
            "query": query
        }

def preprocess(query: str, context: dict) -> str:
    """クエリの前処理（簡略化）"""
    return f"{context.get('user_name', 'ユーザー')}さんの質問: {query}"

def postprocess(result: dict, context: dict) -> str:
    """結果の後処理（簡略化）"""
    return result.get("output", "")

# 使用例
user_context = {"user_name": "田中", "user_id": "123"}
result = process_user_query("今日の売上を確認して", user_context)

@traceableデコレータを使うことで、カスタム関数の実行も追跡できます。エラーが発生した場合、その時点でのすべての変数の状態がトレースに記録されるため、問題の原因を特定しやすくなります。

プロンプトのバージョン管理とA/Bテスト

LangSmithを使うと、プロンプトのバージョン管理とA/Bテストを効率的に行えます。

プロンプトハブの活用

from langchain import hub

# プロンプトをLangSmith Hubから取得
prompt = hub.pull("shineos/customer-support-agent")

# プロンプトのバージョンを指定
prompt_v2 = hub.pull("shineos/customer-support-agent:v2")

# エージェントの作成
agent_v2 = create_openai_functions_agent(llm, tools, prompt_v2)
agent_executor_v2 = AgentExecutor(agent=agent_v2, tools=tools)

# A/Bテストのための実行
import random

def run_with_ab_test(user_input: str):
    # ランダムにバージョンを選択
    version = random.choice(["v1", "v2"])
    
    if version == "v1":
        prompt = hub.pull("shineos/customer-support-agent:v1")
    else:
        prompt = hub.pull("shineos/customer-support-agent:v2")
    
    agent = create_openai_functions_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    
    config = RunnableConfig(
        tags=[f"prompt-version-{version}"],
        metadata={"ab_test_version": version}
    )
    
    result = executor.invoke({"input": user_input}, config=config)
    return result, version

# テスト実行
result, version = run_with_ab_test("返品手続きについて教えて")
print(f"使用バージョン: {version}")
print(f"結果: {result['output']}")

プロンプトをLangSmith Hubで管理することで、チーム全体でプロンプトを共有し、バージョン管理できます。A/Bテストの結果はLangSmithのダッシュボードで比較分析できます。

評価データセットの活用

from langsmith import Client

client = Client()

# 評価用データセットの作成
dataset_name = "customer-support-qa"

# データセットが存在しない場合は作成
try:
    dataset = client.read_dataset(dataset_name=dataset_name)
except:
    dataset = client.create_dataset(
        dataset_name=dataset_name,
        description="カスタマーサポート用のQ&Aデータセット"
    )

# テストケースの追加
test_cases = [
    {
        "input": {"input": "注文をキャンセルしたい"},
        "expected_output": "注文キャンセルの手順",
        "metadata": {"category": "order_management"}
    },
    {
        "input": {"input": "配送状況を確認したい"},
        "expected_output": "配送追跡情報",
        "metadata": {"category": "shipping"}
    },
    {
        "input": {"input": "返金はいつされますか"},
        "expected_output": "返金処理のタイムライン",
        "metadata": {"category": "refund"}
    }
]

for case in test_cases:
    client.create_example(
        inputs=case["input"],
        outputs={"expected": case["expected_output"]},
        dataset_id=dataset.id,
        metadata=case["metadata"]
    )

print(f"データセット '{dataset_name}' に {len(test_cases)} 件のテストケースを追加しました")

評価データセットを使うことで、プロンプトの変更が回答品質に与える影響を定量的に測定できます。継続的にテストを実行し、品質の劣化を防げます。

本番環境でのモニタリング戦略

LangSmithを使った本番環境のモニタリング手法を紹介します。

コストとレイテンシの追跡

from datetime import datetime, timedelta
from langsmith import Client

client = Client()

def analyze_production_metrics(days: int = 7):
    """過去N日間の本番メトリクスを分析"""
    
    # 期間の設定
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    
    # トレースの取得
    runs = client.list_runs(
        project_name="my-agent-project",
        start_time=start_time,
        end_time=end_time,
        filter='eq(tags, "production")'
    )
    
    # メトリクスの集計
    total_cost = 0
    total_latency = 0
    error_count = 0
    success_count = 0
    latency_list = []
    
    for run in runs:
        # コストの集計
        if run.total_cost:
            total_cost += run.total_cost
        
        # レイテンシの集計
        if run.latency:
            latency_ms = run.latency
            total_latency += latency_ms
            latency_list.append(latency_ms)
        
        # 成功/失敗のカウント
        if run.error:
            error_count += 1
        else:
            success_count += 1
    
    total_requests = success_count + error_count
    
    # 統計の計算
    avg_latency = total_latency / total_requests if total_requests > 0 else 0
    error_rate = (error_count / total_requests * 100) if total_requests > 0 else 0
    
    # P95レイテンシの計算
    if latency_list:
        latency_list.sort()
        p95_index = int(len(latency_list) * 0.95)
        p95_latency = latency_list[p95_index] if p95_index < len(latency_list) else latency_list[-1]
    else:
        p95_latency = 0
    
    print(f"=== 過去{days}日間の本番メトリクス ===")
    print(f"総リクエスト数: {total_requests}")
    print(f"成功率: {(success_count/total_requests*100):.2f}%")
    print(f"エラー率: {error_rate:.2f}%")
    print(f"総コスト: ${total_cost:.4f}")
    print(f"平均レイテンシ: {avg_latency:.0f}ms")
    print(f"P95レイテンシ: {p95_latency:.0f}ms")
    print(f"1リクエストあたり平均コスト: ${(total_cost/total_requests):.6f}")
    
    return {
        "total_requests": total_requests,
        "error_rate": error_rate,
        "total_cost": total_cost,
        "avg_latency_ms": avg_latency,
        "p95_latency_ms": p95_latency
    }

# メトリクスの分析実行
metrics = analyze_production_metrics(days=7)

定期的にメトリクスを分析することで、コストの急増やレイテンシの劣化を早期に検知できます。アラートを設定し、閾値を超えた場合に通知を受け取る仕組みも構築できます。

エラーパターンの自動検出

from collections import Counter

def detect_error_patterns(days: int = 1):
    """エラーパターンを自動検出"""
    
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    
    # エラーが発生したトレースのみ取得
    error_runs = client.list_runs(
        project_name="my-agent-project",
        start_time=start_time,
        end_time=end_time,
        filter='and(eq(tags, "production"), exists(error))'
    )
    
    error_types = []
    error_messages = []
    failed_tools = []
    
    for run in error_runs:
        if run.error:
            error_types.append(type(run.error).__name__ if hasattr(run.error, '__name__') else 'Unknown')
            error_messages.append(str(run.error)[:100])
        
        # 失敗したツールを特定
        if run.run_type == "tool":
            failed_tools.append(run.name)
    
    # パターンの集計
    error_type_counts = Counter(error_types)
    failed_tool_counts = Counter(failed_tools)
    
    print("=== エラーパターン分析 ===")
    print("\n【エラータイプ別】")
    for error_type, count in error_type_counts.most_common(5):
        print(f"  {error_type}: {count}件")
    
    print("\n【失敗したツール】")
    for tool_name, count in failed_tool_counts.most_common(5):
        print(f"  {tool_name}: {count}件")
    
    return {
        "error_types": dict(error_type_counts),
        "failed_tools": dict(failed_tool_counts)
    }

# エラーパターンの検出
patterns = detect_error_patterns(days=1)

エラーパターンを自動的に分析することで、頻発する問題を優先的に対処できます。特定のツールやAPIで問題が集中している場合、そのツールの実装を重点的に改善できます。

よくある質問

LangSmithは無料で使えますか？

LangSmithには無料プランが用意されており、月間5,000トレースまで無料で利用できます。個人開発や小規模なプロジェクトであれば、無料プランで十分に活用できます。

既存のLangChainコードにLangSmithを導入するのは難しいですか？

環境変数を設定するだけで、既存のLangChainコードに対してトレーシングが自動的に有効になります。コードの変更は最小限で済みます。

トレースデータはどこに保存されますか？

トレースデータはLangSmithのクラウドサーバーに保存されます。セキュリティが懸念される場合は、機密情報を含むデータをトレースから除外する設定が可能です。

LangChain以外のLLMフレームワークでも使えますか？

LangSmithはLangChain専用ですが、LangChain経由で他のLLMプロバイダー（OpenAI、Anthropic、Google、Azure等）を利用できます。また、REST APIを使えば、任意のフレームワークからトレースを送信することも可能です。

本番環境でトレーシングを有効にするとパフォーマンスに影響しますか？

LangSmithのトレーシングは非同期で動作するため、アプリケーションのレスポンスタイムへの影響は最小限です。ただし、大量のトレースを送信する場合は、ネットワーク帯域幅を考慮する必要があります。

エージェントが期待通りに動作しない場合、どこから調査を始めればよいですか？

まず、LangSmithのトレース画面でエージェントの実行フローを確認し、どのステップで問題が発生しているかを特定します。プロンプトの内容、ツールの選択、各ステップの入出力を順番にチェックすることで、問題箇所を絞り込めます。

おわりに

LangSmithを活用することで、LLMエージェントの複雑な挙動を可視化し、デバッグやモニタリングを効率化できます。トレーシング機能により実行フローを詳細に追跡でき、プロンプトのバージョン管理とA/Bテストで最適なプロンプトを科学的に選定できます。

本番環境でのコスト分析やエラーパターンの自動検出により、運用品質を継続的に改善できます。評価データセットを活用した品質測定で、プロンプトの変更が与える影響を定量的に把握できます。

私たちShineosでは、LLMエージェントを活用した業務自動化システムの開発支援を行っています。LangSmithを使った本番運用のベストプラクティスや、トラブルシューティングのノウハウをご提供できます。LLMエージェントの開発・運用でお困りの際は、ぜひお気軽にご相談ください。

参考リンク

• [1] LangSmith Documentation - LangChain
• [2] LangSmith Tracing - LangChain
• [3] LangSmith Evaluation - LangChain