langsmith-observability

LangSmith - LLM 可観測性プラットフォーム

言語モデルと AI アプリケーションのデバッグ、評価、モニタリングのための開発プラットフォームです。

LangSmith を使用する場面

LangSmith を使用する場面：

• LLM アプリケーションの問題（プロンプト、チェーン、エージェント）のデバッグ
• データセットに対してモデル出力を体系的に評価する
• 本番環境の LLM システムをモニタリングする
• AI 機能の回帰テストを構築する
• レイテンシ、トークン使用量、コストを分析する
• プロンプトエンジニアリングで共同作業する

主な機能：

• トレーシング：すべての LLM 呼び出しの入力、出力、レイテンシを記録します
• 評価：組み込みの評価器とカスタム評価器による体系的なテスト
• データセット：本番環境のトレースから、または手動でテストセットを作成します
• モニタリング：本番環境のメトリクス、エラー、コストを追跡します
• 統合：OpenAI、Anthropic、LangChain、LlamaIndex と連携します

代わりに以下を使用してください：

• Weights & Biases：ディープラーニングの実験追跡、モデルのトレーニング
• MLflow：汎用的な ML ライフサイクル、モデルレジストリを重視
• Arize/WhyLabs：ML モニタリング、データドリフトの検出

クイックスタート

インストール

pip install langsmith

# Set environment variables
export LANGSMITH_API_KEY="your-api-key"
export LANGSMITH_TRACING=true

@traceable による基本的なトレーシング

from langsmith import traceable
from openai import OpenAI

client = OpenAI()

@traceable
def generate_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Automatically traced to LangSmith
result = generate_response("What is machine learning?")

OpenAI ラッパー（自動トレーシング）

from langsmith.wrappers import wrap_openai
from openai import OpenAI

# Wrap client for automatic tracing
client = wrap_openai(OpenAI())

# All calls automatically traced
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

主な概念

実行とトレース

実行は、単一の実行単位（LLM 呼び出し、チェーン、ツール）です。実行は階層的なトレースを形成し、実行フロー全体を示します。

from langsmith import traceable

@traceable(run_type="chain")
def process_query(query: str) -> str:
    # Parent run
    context = retrieve_context(query)  # Child run
    response = generate_answer(query, context)  # Child run
    return response

@traceable(run_type="retriever")
def retrieve_context(query: str) -> list:
    return vector_store.search(query)

@traceable(run_type="llm")
def generate_answer(query: str, context: list) -> str:
    return llm.invoke(f"Context: {context}\n\nQuestion: {query}")

プロジェクト

プロジェクトでは、関連する実行を整理します。環境変数またはコードで設定します。

import os
os.environ["LANGSMITH_PROJECT"] = "my-project"

# Or per-function
@traceable(project_name="my-project")
def my_function():
    pass

クライアント API

from langsmith import Client

client = Client()

# List runs
runs = list(client.list_runs(
    project_name="my-project",
    filter='eq(status, "success")',
    limit=100
))

# Get run details
run = client.read_run(run_id="...")

# Create feedback
client.create_feedback(
    run_id="...",
    key="correctness",
    score=0.9,
    comment="Good answer"
)

データセットと評価

データセットを作成する

from langsmith import Client

client = Client()

# Create dataset
dataset = client.create_dataset("qa-test-set", description="QA evaluation")

# Add examples
client.create_examples(
    inputs=[
        {"question": "What is Python?"},
        {"question": "What is ML?"}
    ],
    outputs=[
        {"answer": "A programming language"},
        {"answer": "Machine learning"}
    ],
    dataset_id=dataset.id
)

評価を実行する

from langsmith import evaluate

def my_model(inputs: dict) -> dict:
    # Your model logic
    return {"answer": generate_answer(inputs["question"])}

def correctness_evaluator(run, example):
    prediction = run.outputs["answer"]
    reference = example.outputs["answer"]
    score = 1.0 if reference.lower() in prediction.lower() else 0.0
    return {"key": "correctness", "score": score}

results = evaluate(
    my_model,
    data="qa-test-set",
    evaluators=[correctness_evaluator],
    experiment_prefix="v1"
)

print(f"Average score: {results.aggregate_metrics['correctness']}")

組み込みの評価器

from langsmith.evaluation import LangChainStringEvaluator

# Use LangChain evaluators
results = evaluate(
    my_model,
    data="qa-test-set",
    evaluators=[
        LangChainStringEvaluator("qa"),
        LangChainStringEvaluator("cot_qa")
    ]
)

高度なトレーシング

トレーシングコンテキスト

from langsmith import tracing_context

with tracing_context(
    project_name="experiment-1",
    tags=["production", "v2"],
    metadata={"version": "2.0"}
):
    # All traceable calls inherit context
    result = my_function()

手動での実行

from langsmith import trace

with trace(
    name="custom_operation",
    run_type="tool",
    inputs={"query": "test"}
) as run:
    result = do_something()
    run.end(outputs={"result": result})

入力／出力を処理する

def sanitize_inputs(inputs: dict) -> dict:
    if "password" in inputs:
        inputs["password"] = "***"
    return inputs

@traceable(process_inputs=sanitize_inputs)
def login(username: str, password: str):
    return authenticate(username, password)

サンプリング

import os
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"  # 10% sampling

LangChain との統合

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Tracing enabled automatically with LANGSMITH_TRACING=true
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

chain = prompt | llm

# All chain runs traced automatically
response = chain.invoke({"input": "Hello!"})

本番環境のモニタリング

Hub プロンプト

from langsmith import Client

client = Client()

# Pull prompt from hub
prompt = client.pull_prompt("my-org/qa-prompt")

# Use in application
result = prompt.invoke({"question": "What is AI?"})

非同期クライアント

from langsmith import AsyncClient

async def main():
    client = AsyncClient()

    runs = []
    async for run in client.list_runs(project_name="my-project"):
        runs.append(run)

    return runs

フィードバックの収集

from langsmith import Client

client = Client()

# Collect user feedback
def record_feedback(run_id: str, user_rating: int, comment: str = None):
    client.create_feedback(
        run_id=run_id,
        key="user_rating",
        score=user_rating / 5.0,  # Normalize to 0-1
        comment=comment
    )

# In your application
record_feedback(run_id="...", user_rating=4, comment="Helpful response")

テストとの統合

Pytest との統合

from langsmith import test

@test
def test_qa_accuracy():
    result = my_qa_function("What is Python?")
    assert "programming" in result.lower()

CI/CD での評価

from langsmith import evaluate

def run_evaluation():
    results = evaluate(
        my_model,
        data="regression-test-set",
        evaluators=[accuracy_evaluator]
    )

    # Fail CI if accuracy drops
    assert results.aggregate_metrics["accuracy"] >= 0.9, \
        f"Accuracy {results.aggregate_metrics['accuracy']} below threshold"

ベストプラクティス

1. 一貫した命名 - プロジェクト名／実行名に一貫した命名規則を使用します
2. メタデータを追加する - バージョン、環境、ユーザー情報を含めます
3. 本番環境でサンプリングする - サンプリングレートを使ってデータ量を調整します
4. データセットを作成する - 本番環境の興味深い事例からテストセットを構築します
5. 評価を自動化する - CI/CD パイプラインで評価を実行します
6. コストをモニタリングする - トークン使用量とレイテンシの傾向を追跡します

よくある問題

トレースが表示されない場合：

import os
# Ensure tracing is enabled
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-key"

# Verify connection
from langsmith import Client
client = Client()
print(client.list_projects())  # Should work

トレーシングによるレイテンシが高い場合：

# Enable background batching (default)
from langsmith import Client
client = Client(auto_batch_tracing=True)

# Or use sampling
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"

ペイロードが大きい場合：

# Hide sensitive/large fields
@traceable(
    process_inputs=lambda x: {k: v for k, v in x.items() if k != "large_field"}
)
def my_function(data):
    pass

参考資料

• 高度な使用方法 - カスタム評価器、分散トレーシング、Hub プロンプト
• トラブルシューティング - よくある問題、デバッグ、パフォーマンス

リソース

• ドキュメント：https://docs.smith.langchain.com
• Python SDK：https://github.com/langchain-ai/langsmith-sdk
• ウェブアプリ：https://smith.langchain.com
• バージョン：0.2.0+
• ライセンス：MIT

エージェントの作業手順

1. 環境を確認します。 ハードウェア、フレームワークとサーバーのバージョン、モデル形式、想定負荷を確認します。
2. 入力を特定します。 データまたはユーザーから、形式、識別子、パラメーターを確認します。結果に影響する値は推測せず、質問してください。
3. まず小規模に実行します。 規模を拡大する前に、単一のリクエストを処理または記録するか、エンドツーエンドで実行します。
4. 上記の手順と参照情報を使用して、タスク全体を実行します。
5. 結果を検証します。 レイテンシ、スループット、出力の正確性を参照データと照合して測定し、リソース使用量とコストを確認します。
6. 報告します。 実行した内容（バージョン、コマンド、パラメーター）、確認した内容、未確定の点を明記します。

発生したこと	対応
サーバーが起動に失敗する、または OOM が発生する	モデルサイズとメモリのバランス、量子化設定、並列処理設定を確認します。
これらの手順に記載されている関数、フラグ、エンドポイントがインストール済みバージョンにない	インストール済みバージョンの公式ドキュメント（help()、--help、公式ドキュメント）を確認して適宜対応し、ユーザーに伝えます。API をでっち上げてはいけません。
必須の入力、識別子、パラメーターが曖昧	ユーザーに質問するか、実行前に仮定を明示します。

誠実性に関するルール

• 結果、パラメーター、識別子、引用、統計情報を捏造してはいけません。実行または検証できない場合は、その旨を明確に伝えます。
• サービスや認証情報を公開してはいけません。プロビジョニングの前に、クラウドのコストを確認します。
• ここに記載されているバージョン固有の情報は古くなっている可能性があるため、インストール済みバージョンの公式ドキュメントで確認します。
• 費用が発生する操作、共有 GPU やクラウドクォータを消費する操作、個人データや患者データに触れる操作、取り消しできない操作の前には、ユーザーに確認します。

関連スキル

• phoenix-observability：LLM のトレース、評価、監視のためのオープンソース AI オブザーバビリティプラットフォームです。
• llamaindex：RAG を使用した LLM アプリケーションを構築するためのデータフレームワークです。
• langchain：エージェント、チェーン、RAG を使用した LLM 搭載アプリケーションを構築するためのフレームワークです。