メインコンテンツへスキップ
Weave Calls Screenshot
Weave Calls Screenshot
Weave Calls Screenshot
Call は Weave における基本的な構成要素です。これらは関数の単一の実行を表し、以下を含みます。
  • Inputs (引数)
  • Outputs (戻り値)
  • Metadata (実行時間、例外、LLM 使用量など)
Call は OpenTelemetry データモデルにおける span に似ています。ひとつの Call は以下のことが可能です。
  • Trace (同一の実行コンテキスト内での Call の集合) に属する
  • 親および子の Call を持ち、ツリー構造を形成する

Creating Calls

Weave で Call を作成するには、主に3つの方法があります。

1. LLM ライブラリの自動トラッキング

Weave は、openaianthropiccoheremistral などの 一般的な LLM ライブラリへの呼び出しを自動的にトラッキング します。プログラムの開始時に weave.init('project_name') を呼び出すだけです。
weave.initautopatch_settings 引数を使用して、Weave のデフォルトのトラッキング動作を 制御 できます。
import weave

from openai import OpenAI
client = OpenAI()

# Weave Tracing を初期化
weave.init('intro-example')

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "How are you?"
        }
    ],
    temperature=0.8,
    max_tokens=64,
    top_p=1,
)
Call の summary 辞書にメトリクスやその他の実行後の値を保存できます。実行中に call.summary を変更すると、追加した値は Call 終了時に Weave が計算したサマリーデータとマージされます。

2. 関数のデコレートとラッピング

しかし、LLM アプリケーションには、トラッキングしたい追加のロジック(前処理/後処理、プロンプトなど)が含まれていることがよくあります。
Weave では @weave.op デコレータを使用して、これらの呼び出しを手動でトラッキングできます。例:
import weave

# Weave Tracing を初期化
weave.init('intro-example')

# 関数をデコレートする
@weave.op
def my_function(name: str):
    return f"Hello, {name}!"

# 関数を呼び出す -- Weave が自動的に入力と出力をトラッキングします
print(my_function("World"))
クラスのメソッド をトラッキングすることも可能です。

同期および非同期ジェネレータ関数の Trace

Weave は、深くネストされたパターンを含む同期および非同期ジェネレータ関数の両方のトレースをサポートしています。
ジェネレータは値を遅延評価(レイジー)で生成するため、出力はジェネレータが完全に消費されたとき(例:リストに変換されたとき)にのみログに記録されます。 トレースに出力を確実に含めるには、ジェネレータを完全に消費(例:list() を使用)してください。
from typing import Generator
import weave

weave.init("my-project")

# この関数はシンプルな同期ジェネレータを使用しています。
# Weave はこの呼び出しと入力 (`x`) をトレースしますが、
# 出力値はジェネレータが消費された(例:`list()` 経由)後にのみ取得されます。
@weave.op
def basic_gen(x: int) -> Generator[int, None, None]:
    yield from range(x)

# ジェネレータパイプライン内で使用される通常の同期関数。
# この呼び出しも Weave によって独立してトレースされます。
@weave.op
def inner(x: int) -> int:
    return x + 1

# 別のトレース対象関数 (`inner`) を呼び出す同期ジェネレータ。
# 生成される各値は、`inner` への個別のトレースされた呼び出しから取得されます。
@weave.op
def nested_generator(x: int) -> Generator[int, None, None]:
    for i in range(x):
        yield inner(i)

# 上記のジェネレータを構成するより複雑なジェネレータ。
# ここでのトレースは階層的な呼び出しツリーを生成します:
# - `deeply_nested_generator` (親)
#   - `nested_generator` (子)
#     - `inner` (孫)
@weave.op
def deeply_nested_generator(x: int) -> Generator[int, None, None]:
    for i in range(x):
        for j in nested_generator(i):
            yield j

# Weave が出力を取得するにはジェネレータが *消費* される必要があります。
# これは同期ジェネレータと非同期ジェネレータの両方に当てはまります。
res = deeply_nested_generator(4)
list(res)  # すべてのネストされた呼び出しと yield のトレースをトリガーします
Weave でのジェネレータ関数のトレース

実行中の Call オブジェクトのハンドルの取得

Call オブジェクト自体のハンドルを取得すると便利な場合があります。これを行うには op.call メソッドを呼び出します。このメソッドは結果と Call オブジェクトの両方を返します。例:
result, call = my_function.call("World")
その後、call を使用して追加のプロパティを設定、更新、または取得できます(最も一般的なのは、フィードバックに使用する Call の ID の取得です)。
op がクラスのメソッドである場合、最初の引数としてインスタンスを op に渡す必要があります(以下の例を参照)。
# インスタンスを最初の引数として渡していることに注意してください。
print(instance.my_method.call(instance, "World"))
import weave

# Weave Tracing を初期化
weave.init("intro-example")

class MyClass:
    # メソッドをデコレートする
    @weave.op
    def my_method(self, name: str):
        return f"Hello, {name}!"

instance = MyClass()

# メソッドを呼び出す -- Weave が自動的に入力と出力をトラッキングします
instance.my_method.call(instance, "World")

実行時の Call 表示名の設定

Call の表示名を上書きしたい場合があります。これは以下の4つの方法のいずれかで実現できます。
  1. op 呼び出し時に表示名を変更する:
result = my_function("World", __weave={"display_name": "My Custom Display Name"})
__weave 辞書を使用すると、Op の表示名よりも優先される Call 表示名が設定されます。
  1. Call ごとに表示名を変更する。これには Op.call メソッドを使用して Call オブジェクトを取得し、Call.set_display_name を使用して表示名を設定します。
result, call = my_function.call("World")
call.set_display_name("My Custom Display Name")
  1. 特定の Op のすべての Call に対して表示名を変更する:
@weave.op(call_display_name="My Custom Display Name")
def my_function(name: str):
    return f"Hello, {name}!"
  1. call_display_name には、Call オブジェクトを受け取り文字列を返す関数を指定することもできます。関数が呼び出されると Call オブジェクトが自動的に渡されるため、関数名、Call の入力、フィールドなどに基づいて動的に名前を生成できます。
  2. 一般的なユースケースの一つは、関数名にタイムスタンプを追加することです。
    from datetime import datetime
    
    @weave.op(call_display_name=lambda call: f"{call.func_name}__{datetime.now()}")
    def func():
        return ...
    
  3. .attributes を使用してカスタムメタデータをログに記録することもできます。
    def custom_attribute_name(call):
        model = call.attributes["model"]
        revision = call.attributes["revision"]
        now = call.attributes["date"]
    
        return f"{model}__{revision}__{now}"
    
    @weave.op(call_display_name=custom_attribute_name)
    def func():
        return ...
    
    with weave.attributes(
        {
            "model": "finetuned-llama-3.1-8b",
            "revision": "v0.1.2",
            "date": "2024-08-01",
        }
    ):
        func()  # 表示名は "finetuned-llama-3.1-8b__v0.1.2__2024-08-01" になります
    
        with weave.attributes(
            {
                "model": "finetuned-gpt-4o",
                "revision": "v0.1.3",
                "date": "2024-08-02",
            }
        ):
            func()  # 表示名は "finetuned-gpt-4o__v0.1.3__2024-08-02" になります
    
テクニカルノート: “Call” は “Op” によって生成されます。Op とは @weave.op でデコレートされた関数またはメソッドのことです。 デフォルトでは、Op の名前は関数名であり、関連付けられた Call も同じ表示名を持ちます。上の例は、特定の Op のすべての Call に対して表示名を上書きする方法を示しています。Op 自体の名前を上書きしたい場合もあります。これは以下の2つの方法のいずれかで実現できます。
  1. Call がログに記録される前に Op の name プロパティを設定する
my_function.name = "My Custom Op Name"
  1. op デコレータの name オプションを設定する
@weave.op(name="My Custom Op Name)

並列(マルチスレッド)関数呼び出しのトレース

デフォルトでは、並列呼び出しはすべて個別のルート Call として Weave に表示されます。同じ親 op の下で正しくネストさせるには、ThreadPoolExecutor を使用します。以下のコードサンプルは ThreadPoolExecutor の使用例を示しています。 最初の関数 funcx を受け取り x+1 を返すシンプルな op です。2番目の関数 outer は入力リストを受け取る別の op です。 outer の内部で ThreadPoolExecutorexc.map(func, inputs) を使用することで、func への各呼び出しが同じ親トレースコンテキストを保持します。
import weave

@weave.op
def func(x):
    return x+1

@weave.op
def outer(inputs):
    with weave.ThreadPoolExecutor() as exc:
        exc.map(func, inputs)

# Weave プロジェクト名を更新
client = weave.init('my-weave-project')
outer([1,2,3,4,5])
Weave UI では、インクリメント処理が並列に実行されていても、1つの親 Call と5つのネストされた子 Call として表示され、完全な階層的トレースが得られます。outer の1つの親 Call と5つのネストされた子 Call を示す Trace UI

3. 手動での Call トラッキング

API を直接使用して手動で Call を作成することもできます。
import weave

# Weave Tracing を初期化
client = weave.init('intro-example')

def my_function(name: str):
    # Call を開始
    call = client.create_call(op="my_function", inputs={"name": name})

    # ... 関数コード ...

    # Call を終了
    client.finish_call(call, output="Hello, World!")

# 関数を呼び出す
print(my_function("World"))

4. クラスおよびオブジェクトメソッドのトラッキング

クラスおよびオブジェクトメソッドをトラッキングすることもできます。
weave.op を使用して、クラスの任意のメソッドをトラッキングします。
import weave

# Weave Tracing を初期化
weave.init("intro-example")

class MyClass:
    # メソッドをデコレートする
    @weave.op
    def my_method(self, name: str):
        return f"Hello, {name}!"

instance = MyClass()

# メソッドを呼び出す -- Weave が自動的に入力と出力をトラッキングします
print(instance.my_method("World"))

Viewing Calls

ウェブアプリで Call を表示するには:
  1. プロジェクトの Traces タブに移動します。
  2. リストから表示したい Call を探します。
  3. Call をクリックして詳細ページを開きます。
詳細ページには、Call の入力、出力、実行時間、および追加のメタデータが表示されます。ウェブアプリでの Call 表示

weave.Markdown によるレンダリングされたトレースのカスタマイズ

weave.Markdown を使用すると、元のデータを失うことなく、トレース情報の表示方法をカスタマイズできます。これにより、基盤となるデータ構造を維持したまま、入力と出力を読みやすいフォーマットされたコンテンツブロックとしてレンダリングできます。
トレースデータをフォーマットするには、@weave.op デコレータ内で postprocess_inputs および postprocess_output 関数を使用します。以下のコードサンプルは、ポストプロセッサを使用して Weave で絵文字や読みやすいフォーマットで Call をレンダリングします。
import weave

def postprocess_inputs(query) -> weave.Markdown:
    search_box = f"""
**Search Query:**
``+`
{query}
``+`
"""
    return {"search_box": weave.Markdown(search_box),
            "query": query}

def postprocess_output(docs) -> weave.Markdown:
    formatted_docs = f"""
# {docs[0]["title"]}

{docs[0]["content"]}

[Read more]({docs[0]["url"]})

---

# {docs[1]["title"]}

{docs[1]["content"]}

[Read more]({docs[1]["url"]})
"""
    return weave.Markdown(formatted_docs)

@weave.op(
    postprocess_inputs=postprocess_inputs,
    postprocess_output=postprocess_output,
)
def rag_step(query):
    # S&P 500 企業の新聞記事の例
    docs = [
        {
            "title": "OpenAI",
            "content": "OpenAI is a company that makes AI models.",
            "url": "https://www.openai.com",
        },
        {
            "title": "Google",
            "content": "Google is a company that makes search engines.",
            "url": "https://www.google.com",
        },
    ]
    return docs

if __name__ == "__main__":
    weave.init('markdown_renderers')
    rag_step("Tell me about OpenAI")
以下のスクリーンショットでは、フォーマットされていない出力とフォーマットされた出力のそれぞれの違いを確認できます。 コードサンプルを使用して Weave UI でレンダリングされた Call

Updating Calls

Call は作成後にほぼ不変(イミュータブル)ですが、以下の変更はサポートされています。 これらの変更はすべて、UI の Call 詳細ページから実行できます。
ウェブアプリでの Call 更新

表示名の設定

Call の表示名を設定するには、Call.set_display_name() メソッドを使用できます。
import weave

# クライアントを初期化
client = weave.init("your-project-name")

# ID で特定の Call を取得
call = client.get_call("call-uuid-here")

# Call の表示名を設定
call.set_display_name("My Custom Display Name")
実行時に Call の表示名を設定 することも可能です。

フィードバックの追加

詳細は フィードバックドキュメント を参照してください。

Call の削除

Python API を使用して Call を削除するには、Call.delete メソッドを使用できます。
import weave

# クライアントを初期化
client = weave.init("your-project-name")

# ID で特定の Call を取得
call = client.get_call("call-uuid-here")

# Call を削除
call.delete()

複数の Call の削除

Python API を使用して Call を一括削除するには、Call ID のリストを delete_calls() に渡します。
import weave

# クライアントを初期化
client = weave.init("my-project")

# クライアントからすべての Call を取得
all_calls = client.get_calls()

# 最初の 1000 個の Call オブジェクトのリストを取得
first_1000_calls = all_calls[:1000]

# 最初の 1000 個の Call ID のリストを取得
first_1000_calls_ids = [c.id for c in first_1000_calls]

# ID を指定して最初の 1000 個の Call オブジェクトを削除
client.delete_calls(call_ids=first_1000_calls_ids)

Querying and exporting Calls

多くの Call のスクリーンショット
プロジェクトの /calls ページ (“Traces” タブ) には、プロジェクト内のすべての Call のテーブルビューが表示されます。そこでは以下のことが可能です。
  • ソート
  • フィルタリング
  • エクスポート
Calls テーブルビュー
エクスポートモーダル(上記)では、データをさまざまな形式でエクスポートできるほか、選択した Call に対応する Python および CURL のコードスニペットも表示されます。 UI でビューを作成してから、生成されたコードスニペットを通じてエクスポート API について学ぶのが最も簡単な方法です。
Python API を使用して Call を取得するには、client.get_calls メソッドを使用できます。
import weave

# クライアントを初期化
client = weave.init("your-project-name")

# Call を取得
calls = client.get_calls(filter=...)

Call スキーマ

フィールドの完全なリストについては スキーマ を参照してください。
プロパティ説明
idstring (uuid)Call の一意の識別子
project_idstring (オプション)関連付けられた Project の識別子
op_namestring操作の名前 (参照である場合もあります)
display_namestring (オプション)Call のわかりやすい表示名
trace_idstring (uuid)この Call が属する Trace の識別子
parent_idstring (uuid)親 Call の識別子
started_atdatetimeCall が開始されたタイムスタンプ
attributesDict[str, Any]ユーザー定義の Call に関するメタデータ (実行中は読み取り専用)
inputsDict[str, Any]Call の入力引数
ended_atdatetime (オプション)Call が終了したタイムスタンプ
exceptionstring (オプション)Call が失敗した場合のエラーメッセージ
outputAny (オプション)Call の結果
summaryOptional[SummaryMap]実行後のサマリー情報。実行中にこれを変更してカスタムメトリクスを記録できます。
wb_user_idOptional[str]関連付けられた Weights & Biases の User ID
wb_run_idOptional[str]関連付けられた Weights & Biases の Run ID
deleted_atdatetime (オプション)Call が削除されたタイムスタンプ(該当する場合)
上の表は、Weave における Call の主要なプロパティをまとめたものです。各プロパティは、関数呼び出しのトラッキングと管理において重要な役割を果たします。
  • idtrace_id、および parent_id フィールドは、システム内での Call の整理と関連付けに役立ちます。
  • タイミング情報 (started_at, ended_at) により、パフォーマンス分析が可能になります。
  • attributesinputs フィールドは Call のコンテキストを提供します。属性は Call 開始時に固定されるため、呼び出し前に weave.attributes で設定してください。outputsummary は結果を取得し、実行中に summary を更新して追加のメトリクスを記録できます。
  • Weights & Biases とのインテグレーションは wb_user_id および wb_run_id を通じて行われます。
これらの包括的なプロパティセットにより、プロジェクト全体の関数呼び出しの詳細なトラッキングと分析が可能になります。 計算フィールド:
  • Cost (コスト)
  • Duration (期間)
  • Status (ステータス)

Saved views

Trace テーブルの設定、フィルタ、ソートを saved views として保存し、好みのセットアップに素早くアクセスできます。UI および Python SDK を通じて saved views を構成・アクセスできます。詳細については、Saved Views を参照してください。

Traces テーブルでの W&B run の表示

Weave を使用すると、コード内の関数呼び出しをトレースし、それらが実行された W&B Runs に直接リンクできます。 @weave.op() で関数をトレースし、wandb.init() コンテキスト内で呼び出すと、Weave は自動的にそのトレースを W&B Run と関連付けます。 関連付けられた Run へのリンクは Traces テーブルに表示されます。
以下の Python コードは、wandb.init() コンテキスト内で実行されたときに、トレースされた操作が W&B Runs にどのようにリンクされるかを示しています。これらのトレースは Weave UI に表示され、対応する Run と関連付けられます。
import wandb
import weave

def example_wandb(projname):
    # projname を entity と project に分割
    entity, project = projname.split("/", 1)

    # トレース用の Weave コンテキストを初期化
    weave.init(projname)

    # トレース可能な操作を定義
    @weave.op()
    def say(message: str) -> str:
        return f"I said: {message}"

    # 1つ目の W&B run
    with wandb.init(
        entity=entity,
        project=project,
        notes="Experiment 1",
        tags=["baseline", "paper1"],
    ) as run:
        say("Hello, world!")
        say("How are you!")
        run.log({"messages": 2})

    # 2つ目の W&B run
    with wandb.init(
        entity=entity,
        project=project,
        notes="Experiment 2",
        tags=["baseline", "paper1"],
    ) as run:
        say("Hello, world from experiment 2!")
        say("How are you!")
        run.log({"messages": 2})

if __name__ == "__main__":
    # 実際の W&B ユーザー名/プロジェクトに置き換えてください
    example_wandb("your-username/your-project")
コードサンプルの使用方法:
  1. ターミナルで依存関係をインストールします。
pip install wandb weave
  1. W&B にログインします。
wandb login
  1. スクリプト内で your-username/your-project を実際の W&B エンティティ/プロジェクトに置き換えます。
  2. スクリプトを実行します。
python weave_trace_with_wandb.py
  1. https://weave.wandb.ai にアクセスし、プロジェクトを選択します。
  2. Traces タブでトレース出力を確認します。関連付けられた Run へのリンクが Traces テーブルに表示されます。

オートパッチの設定

デフォルトでは、Weave は openaianthropiccoheremistral などの一般的な LLM ライブラリへの呼び出しを自動的にパッチし、トラッキングします。
autopatch_settings 引数は非推奨です。暗黙的なパッチを無効にするには implicitly_patch_integrations=False を使用するか、インテグレーションごとに設定を行うには patch_openai(settings={...}) のような特定のパッチ関数を呼び出してください。

すべてのオートパッチを無効にする

weave.init(..., implicitly_patch_integrations=False)

特定のインテグレーションを有効にする

import weave

weave.init(..., implicitly_patch_integrations=False)

# その後、必要なインテグレーションのみを手動でパッチします
weave.integrations.patch_anthropic()
weave.integrations.patch_cohere()

入力と出力のポストプロセス

パッチ関数に設定を渡すことで、入力と出力の処理方法(PII データなど)をカスタマイズできます。
import weave.integrations

def redact_inputs(inputs: dict) -> dict:
    if "email" in inputs:
        inputs["email"] = "[REDACTED]"
    return inputs

weave.init(...)
weave.integrations.patch_openai(
    settings={
        "op_settings": {"postprocess_inputs": redact_inputs}
    }
)
詳細については、PII データでの Weave の使用方法 を参照してください。

FAQs

大きなトレースが切り捨てられないようにするにはどうすればよいですか?

詳細については、トラブルシューティングガイドトレースデータが切り捨てられる を参照してください。

トレースを無効にするにはどうすればよいですか?

環境変数

プログラム全体で無条件にトレースを無効にしたい場合は、環境変数 WEAVE_DISABLED=true を設定できます。

クライアントの初期化

特定の条件に基づいて初期化時にトレースを無効にしたい場合があります。その場合は、init 設定で disabled フラグを指定してクライアントを初期化できます。
import weave

# クライアントを初期化
client = weave.init(..., settings={"disabled": True})

コンテキストマネージャ

最後に、アプリケーションロジックに基づいて特定の関数のみトレースを無効にしたい場合があります。その場合は、weave.trace.context.call_context からインポートできるコンテキストマネージャ with set_tracing_enabled(False) を使用できます。
import weave
from weave.trace.context.call_context import set_tracing_enabled

client = weave.init(...)

@weave.op
def my_op():
    ...

with set_tracing_enabled(False):
    my_op()

Call に関する情報を取得するにはどうすればよいですか?

通常は op を直接呼び出します。
@weave.op
def my_op():
    ...

my_op()
しかし、op の call メソッドを呼び出すことで、Call オブジェクトに直接アクセスすることもできます。
@weave.op
def my_op():
    ...

output, call = my_op.call()
ここから、call オブジェクトには入力、出力、その他のメタデータを含む、Call に関するすべての情報が含まれています。