オンラインモニタリング - Weights & Biases Documentation

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

Weave で統合する: 本番ダッシュボード

このノートブックでは、Weave の API と関数を使用して、Weave の Traces ビューの拡張として、本番モニタリング向けのカスタムダッシュボードを作成する方法を説明します。このガイドは、本番環境で LLM アプリケーションを運用しており、デフォルトの Traces ビューだけでは得られないパフォーマンス、コスト、ユーザーフィードバックを、用途に合わせて可視化したい開発者や ML エンジニアを対象としています。このノートブックでは、次の内容に重点を置きます。

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

このノートブックを終えるころには、Weave プロジェクトのトレースデータ、コスト、フィードバックを単一のビューに集約した、実際に使えるカスタムダッシュボードが完成します。streamlit をインストールして production dashboard script を実行すれば、ご自身の Weave プロジェクトでこのダッシュボードを試せます。 Weave を使用した本番ダッシュボードの例

セットアップ

まず、以下のパッケージをインストールします。

!pip install streamlit pandas plotly weave

実装

以下のセクションでは、Weave クライアントの初期化、call データの取得、ダッシュボード用の可視化の生成について順を追って説明します。

Weave クライアントを初期化してコストを定義する

まず、Weave クライアントを初期化し、各モデルのコストを追加する関数を設定します。この手順は、後続のコストクエリで各 Call のトークン単価を適切に対応付けるために必須です。 W&B には多くの一般的なモデル向けの標準コストが含まれており、独自のカスタムコストやカスタムモデルを追加することもできます。次の例では、いくつかのモデルにカスタムコストを追加し、それ以外には標準コストを使用する方法を示します。コストは、Weave で各 Call についてトラッキングされたトークン数に基づいて計算されます。多くの LLM ベンダーのライブラリでは、Weave がトークン使用量を自動的にトラッキングしますが、任意の Call に対してカスタムのトークン数を返すこともできます。カスタムモデルのトークン数とコスト計算の定義方法について詳しくは、custom cost cookbookを参照してください。

PROJECT_NAME = "wandb-smle/weave-cookboook-demo"
python
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)

Weave から Call データを取得する

クライアントを初期化し、コストを設定したら、次のステップは Weave から Call データを取得することです。Call データを取得する方法は 2 つあります。

Call ごとにデータを取得する
高レベル API を使用する

以下のセクションでは、それぞれの方法について説明します。

データを Call ごとに取得する

Weave からデータにアクセスする最初の方法は、フィルターされた Call のリストを取得し、必要なデータを Call ごとに抽出することです。これを行うには、calls_query_stream API を使用して Weave から Call データを取得します。

calls_query_stream API: Weave から Call データを取得する API です。
filter dictionary: Call データを取得するためのフィルターパラメーターを含む辞書です。詳しくは、CallSchema リファレンスを参照してください。
expand_columns list: Call データ内で展開する列を含むリストです。
sort_by list: Call データの並べ替えパラメーターを含むリストです。
include_costs boolean: Call データにコストを含めるかどうかを示す真偽値です。
include_feedback boolean: Call データにフィードバックを含めるかどうかを示す真偽値です。

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)
        )  # 取得するCallの数が多すぎる場合に制限する
        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)
python
# 生データはCallオブジェクトのリスト
pd.DataFrame([call.dict() for call in calls]).head(3)

Weaveからの戻り値を使ってCallを処理します。関連する情報を抽出し、辞書のリストに格納します。次に、その辞書のリストを 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)
python
df_calls = process_calls(calls)
df_calls.head(3)

高レベル API の使用

個々のcallをすべて確認する代わりに、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 ごとに最初の行のみを表示する
df_costs

入力データを集約して可視化を生成する

Call データとコストを DataFrame として利用できるため、plotly を使って可視化を生成できます。これは用途に応じて自由にカスタマイズできる、出発点となるダッシュボードです。より高度な例については、knowledge-worker-weave リポジトリ内の 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

# すべてのプロットのソースコードを参照
python
plot_feedback_pie_chart(df_calls["Thumbs Up"].sum(), df_calls["Thumbs Down"].sum())
python
plot_model_cost_distribution(df_costs)

結論

このクックブックでは、Weave の API と関数を使用して、カスタムの本番モニタリングダッシュボードを作成する方法を紹介しました。Weave は、高速なインテグレーションによってデータの取り込みを効率化するとともに、カスタムプロセス向けにデータを抽出できることに重点を置いています。

データ入力:
- @weave-op() デコレータによるフレームワーク非依存のトレースと、CSV から Call をインポートするオプション (関連する import cookbook を参照してください) 。
- さまざまなプログラミングフレームワークや言語から Weave にログするための Service API endpoint。詳細は Service API リファレンスを参照してください。
データ出力:
- データを CSV、TSV、JSONL、または JSON 形式でダウンロードできます。詳細は Service API リファレンスを参照してください。
- データへのプログラムによるアクセスを使用してエクスポートします。このクックブックで説明したように、エクスポートパネルの「Use Python」セクションを参照してください。詳細は Querying and exporting calls を参照してください。

このカスタムダッシュボードは、Weave に標準搭載されている Traces ビューを拡張し、本番環境の LLM アプリケーションを用途に応じてモニタリングできるようにします。より複雑なダッシュボードについては、agent-dev-collection リポジトリの Streamlit example を確認してください。ここでは独自の Weave プロジェクト URL を追加できます。

​Weave で統合する: 本番ダッシュボード

​セットアップ

​実装

​Weave クライアントを初期化してコストを定義する

​Weave から Call データを取得する

​データを Call ごとに取得する

​高レベル API の使用

​入力データを集約して可視化を生成する

​結論

Weave で統合する: 本番ダッシュボード

セットアップ

実装

Weave クライアントを初期化してコストを定義する

Weave から Call データを取得する

データを Call ごとに取得する

高レベル API の使用

入力データを集約して可視化を生成する

結論