オンライン・モニタリング

W&B Weave を使用したオンラインモニタリング方法について学ぶ

これはインタラクティブな ノートブック です。ローカルで実行するか、以下のリンクを使用できます：

• Google Colab で開く
• GitHub でソースを表示

​

Weave との統合：プロダクションダッシュボード

GenAI ツールキットの環境は急速に進化しており、新しい フレームワーク、ツール、アプリケーション が常に登場しています。Weave は、GenAI のモニタリングと評価に関するあらゆるニーズに応えるワンストップショップを目指しています。これは同時に、既存の プラットフォーム と統合したり、プロジェクト や組織の特定のニーズに合わせて Weave を拡張したりする必要がある場合があることも意味しています。

このクックブックでは、Weave の強力な API と関数を活用して、Weave の トレース ビューの拡張として、プロダクションモニタリング用のカスタム ダッシュボード を作成する方法をデモします。以下の点に焦点を当てます：

• Weave から トレース、コスト、フィードバック、その他の メトリクス を取得する
• ユーザーフィードバックとコスト分布の集計ビューを作成する
• トークン使用量とレイテンシの経時変化を 可視化 する

streamlit をインストールし、このプロダクションダッシュボードスクリプト を実行することで、自身の Weave プロジェクト で ダッシュボード を試すことができます！

​

1. セットアップ

このチュートリアルに沿って進めるには、以下のパッケージをインストールするだけです：

!pip install streamlit pandas plotly weave

​

2. 実装

​

2.1 Weave クライアントの初期化とコストの定義

まず、Weave クライアントを初期化し、各 モデル のコストを追加する関数を設定します。

• 多くの標準的な モデル の標準コストが含まれていますが、独自のカスタムコストやカスタム モデル を簡単に追加することもできます。以下では、いくつかの モデル にカスタムコストを追加し、残りは標準コストを使用する方法を示します。
• コストは、Weave でトラッキングされた各コール（call）のトークンに基づいて計算されます。多くの LLM ベンダーライブラリでは、トークン使用量を自動的にトラッキングしますが、任意のコールに対してカスタムのトークン数を返すことも可能です。カスタム モデル のトークンカウントとコスト計算の定義方法については、こちらのクックブックを参照してください - custom cost cookbook

PROJECT_NAME = "wandb-smle/weave-cookboook-demo"

import weave

MODEL_NAMES = [
    # モデル名, プロンプトコスト, コンプリーションコスト
    ("gpt-4o-2024-05-13", 0.03, 0.06),
    ("gpt-4o-mini-2024-07-18", 0.03, 0.06),
    ("gemini/gemini-1.5-flash", 0.00025, 0.0005),
    ("gpt-4o-mini", 0.03, 0.06),
    ("gpt-4-turbo", 0.03, 0.06),
    ("claude-3-haiku-20240307", 0.01, 0.03),
    ("gpt-4o", 0.03, 0.06),
]

def init_weave_client(project_name):
    try:
        client = weave.init(project_name)
        for model, prompt_cost, completion_cost in MODEL_NAMES:
            client.add_cost(
                llm_id=model,
                prompt_token_cost=prompt_cost,
                completion_token_cost=completion_cost,
            )
    except Exception as e:
        print(f"Failed to initialize Weave client for project '{project_name}': {e}")
        return None
    else:
        return client

client = init_weave_client(PROJECT_NAME)

​

2.2 Weave からのコールデータの取得

Weave からコール（call）データを取得するには、2つのオプションがあります：

1. コールごとにデータを取得する
2. ハイレベルな API を使用する

​

2.2.1 コールごとにデータを取得する

Weave から データ にアクセスする最初のオプションは、フィルタリングされたコールのリストを取得し、必要なデータをコールごとに抽出することです。そのためには、calls_query_stream API を使用して Weave からコールの ログ データを取得できます：

• calls_query_stream API: Weave からコールの ログ データを取得できる API です。
• filter 辞書: コール データを取得するためのフィルタ パラメータ が含まれます。詳細は こちら を参照してください。
• expand_columns リスト: コール データ内で展開するカラムを指定します。
• sort_by リスト: コール データのソート パラメータ を含みます。
• include_costs ブール値: コール データにコストを含めるかどうかを指定します。
• include_feedback ブール値: コール データにフィードバックを含めるかどうかを指定します。

import itertools
from datetime import datetime, timedelta

import pandas as pd

def fetch_calls(client, project_id, start_time, trace_roots_only, limit):
    filter_params = {
        "project_id": project_id,
        "filter": {"started_at": start_time, "trace_roots_only": trace_roots_only},
        "expand_columns": ["inputs.example", "inputs.model"],
        "sort_by": [{"field": "started_at", "direction": "desc"}],
        "include_costs": True,
        "include_feedback": True,
    }
    try:
        calls_stream = client.server.calls_query_stream(filter_params)
        calls = list(
            itertools.islice(calls_stream, limit)
        )  # 取得するコール数が多すぎる場合に備えて制限をかける
        print(f"Fetched {len(calls)} calls.")
    except Exception as e:
        print(f"Error fetching calls: {e}")
        return []
    else:
        return calls

calls = fetch_calls(client, PROJECT_NAME, datetime.now() - timedelta(days=1), True, 100)

# 生データは Call オブジェクトのリストです
pd.DataFrame([call.dict() for call in calls]).head(3)

Weave からの戻り値を使用すると、コールの プロセッシング は非常に簡単です。関連する 情報を抽出し、辞書 のリストに保存します。その後、辞書 のリストを pandas の DataFrame に変換して返します。

import json
from datetime import datetime

import pandas as pd

def process_calls(calls):
    records = []
    for call in calls:
        feedback = call.summary.get("weave", {}).get("feedback", [])
        thumbs_up = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👍"
        )
        thumbs_down = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👎"
        )
        latency = call.summary.get("weave", {}).get("latency_ms", 0)

        records.append(
            {
                "Call ID": call.id,
                "Trace ID": call.trace_id,  # トレースを取得するために使用できるユニークな ID
                "Display Name": call.display_name,  # UIやプログラムで設定可能なオプション名
                "Latency (ms)": latency,
                "Thumbs Up": thumbs_up,
                "Thumbs Down": thumbs_down,
                "Started At": pd.to_datetime(getattr(call, "started_at", datetime.min)),
                "Inputs": json.dumps(call.inputs, default=str),
                "Outputs": json.dumps(call.output, default=str),
            }
        )
    return pd.DataFrame(records)

df_calls = process_calls(calls)
df_calls.head(3)

​

2.2.2 ハイレベルな API の使用

すべてのコールを個別に調べる代わりに、Weave は モデル のコスト、フィードバック、およびその他の メトリクス に直接 アクセス するためのハイレベルな API も提供しています。 例えば、コストについては、query_costs API を使用して、プロジェクト で使用されているすべての LLM のコストを取得できます。

# コスト API を使用してコストを取得
costs = client.query_costs()
df_costs = pd.DataFrame([cost.dict() for cost in costs])
df_costs["total_cost"] = (
    df_costs["prompt_token_cost"] + df_costs["completion_token_cost"]
)

# 各 llm_id に対して最初の1行のみを表示
df_costs

​

2.4 インプットの収集と可視化の生成

次に、plotly を使用して 可視化 を生成できます。これは最も基本的な ダッシュボード ですが、好みに合わせてカスタマイズできます！より複雑な例については、こちら の Streamlit の例を確認してください。

import plotly.express as px
import plotly.graph_objects as go

def plot_feedback_pie_chart(thumbs_up, thumbs_down):
    fig = go.Figure(
        data=[
            go.Pie(
                labels=["Thumbs Up", "Thumbs Down"],
                values=[thumbs_up, thumbs_down],
                marker={"colors": ["#66b3ff", "#ff9999"]},
                hole=0.3,
            )
        ]
    )
    fig.update_traces(textinfo="percent+label", hoverinfo="label+percent")
    fig.update_layout(showlegend=False, title="Feedback Summary")
    return fig

def plot_model_cost_distribution(df):
    fig = px.bar(
        df,
        x="llm_id",
        y="total_cost",
        color="llm_id",
        title="Cost Distribution by Model",
    )
    fig.update_layout(xaxis_title="Model", yaxis_title="Cost (USD)")
    return fig

# すべてのプロットについてはソースコードを参照してください

plot_feedback_pie_chart(df_calls["Thumbs Up"].sum(), df_calls["Thumbs Down"].sum())

plot_model_cost_distribution(df_costs)

​

まとめ

このクックブックでは、Weave の API と関数を使用してカスタムのプロダクションモニタリング ダッシュボード を作成する方法をデモしました。Weave は現在、データの簡単な入力のための迅速な インテグレーション と、カスタム プロセス のためのデータ抽出に重点を置いています。

• データ入力：
  • @weave-op() デコレータによる フレームワーク に依存しない トレース と、CSV からのコールのインポート機能（関連する import cookbook を参照）。
  • 様々な プログラミングフレームワーク や言語から Weave に ログ を記録するためのサービス API エンドポイント。詳細は こちら を参照。
• データ出力：
  • CSV、TSV、JSONL、JSON 形式での簡単なデータダウンロード。詳細は こちら を参照。
  • プログラムによる データアクセス を使用した簡単なエクスポート。このクックブックで説明されているエクスポート パネル の “Use Python” セクションを参照してください。詳細は こちら を参照してください。

このカスタム ダッシュボード は Weave ネイティブの トレース ビューを拡張し、プロダクション環境における LLM アプリケーション のカスタマイズされたモニタリングを可能にします。より複雑な ダッシュボード を見たい場合は、自身の Weave プロジェクト の URL を追加できる このリポジトリ の Streamlit の例をチェックしてください。