W&B Weave は、さまざまなコンテンツタイプのログ記録と表示をサポートしており、動画、画像、音声クリップ、PDF、CSV データ、HTML を表示するための専用関数を備えています。このガイドでは、各メディアタイプをログおよび表示するための基本例と応用例を紹介します。
このガイドの例ではアノテーションを使用します。メディアのログ記録を始める最も簡単な方法であるため、アノテーションの使用を推奨します。より高度な設定については、Content API セクションを参照してください。Weave にメディアをログに記録するには、Annotated[bytes, Content] や Annotated[str, Content] のような型アノテーションを ops の入力または戻り値の型として追加します。パス引数に Annotated[str, Content] を指定すると、Weave がトレース内のメディアを自動的に開き、検出し、表示します。 TypeScript SDK はメディアをログに記録するための専用関数を提供します。
weave.weaveImage({ data: Buffer }) - 画像用(PNG 形式)
weave.weaveAudio({ data: Buffer }) - 音声用(WAV 形式)
TypeScript SDK は現在、画像と音声のログ記録をサポートしています。動画、ドキュメント、または HTML をログに記録するには、代わりに Weave Python SDK を使用してください。
次のセクションでは、各メディアタイプをログに記録する実用的な例を示します。
次の例では、画像を生成して Weave の UI にログとして記録する方法を示します。
関数に Annotated[bytes, Content] 型、ファイルパスに Annotated[str, Content] 型で注釈を付けることで、画像をログに記録できます。次の例では、簡単な画像を描画し、その後 Content アノテーションを使って Weave にログに記録します。import weave
from weave import Content
from PIL import Image, ImageDraw
from typing import Annotated
weave.init('your-team-name/your-project-name')
# サンプル画像を作成して保存する
img = Image.new('RGB', (200, 100), color='lightblue')
draw = ImageDraw.Draw(img)
draw.text((50, 40), "Hello Weave!", fill='black')
img.save("sample_image.png")
# 方法1: Contentアノテーション(推奨)
@weave.op
def load_image_content(path: Annotated[str, Content]) -> Annotated[bytes, Content]:
with open(path, 'rb') as f:
return f.read()
# 方法2: PIL Imageオブジェクト
@weave.op
def load_image_pil(path: Annotated[str, Content]) -> Image.Image:
return Image.open(path)
result1 = load_image_content("sample_image.png")
result2 = load_image_pil("sample_image.png")
Weave は画像をログに記録し、その画像を確認できるトレースへのリンクを返します。応用例: DALL-E で画像を生成して Weave にログする
次の例では、猫の画像を生成し、それを Weave にログします。import weave
from weave import Content
from typing import Annotated
import openai
import requests
client = openai.OpenAI()
weave.init("your-team-name/your-project-name")
@weave.op
def generate_image(prompt: str) -> Annotated[bytes, Content]:
response = client.images.generate(
model="dall-e-3",
prompt=prompt,
size="1024x1024",
quality="standard",
n=1,
)
image_url = response.data[0].url
image_response = requests.get(image_url, stream=True)
return image_response.content
generate_image("a cat with a pumpkin hat")
応用例: ログに記録する前に大きな画像をリサイズする
UI のレンダリングコストとストレージへの影響を抑えるために、ログに記録する前に画像をリサイズすると有用な場合があります。@weave.op 内で postprocess_output を使用して画像をリサイズできます。from dataclasses import dataclass
from typing import Any
from PIL import Image
import weave
weave.init('your-team-name/your-project-name')
# カスタム出力タイプ
@dataclass
class ImageResult:
label: str
image: Image.Image
# リサイズヘルパー
def resize_image(image: Image.Image, max_size=(512, 512)) -> Image.Image:
image = image.copy()
image.thumbnail(max_size, Image.Resampling.LANCZOS)
return image
# ログ記録前に画像をリサイズするための後処理出力
def postprocess_output(output: ImageResult) -> ImageResult:
resized = resize_image(output.image)
return ImageResult(label=output.label, image=resized)
@weave.op(postprocess_output=postprocess_output)
def generate_large_image() -> ImageResult:
# 処理するサンプル画像を作成(例:2000x2000の赤い正方形)
img = Image.new("RGB", (2000, 2000), color="red")
return ImageResult(label="big red square", image=img)
generate_large_image()
Weave はサイズ変更後の画像をログに記録し、その画像を確認できる trace へのリンクを返します。 次の例では、weaveImage 関数を使用して画像をログに記録します。import * as weave from 'weave';
import * as fs from 'fs';
async function main() {
await weave.init('your-team-name/your-project-name');
// weaveImageを使用して画像を読み込み、ログに記録する
const loadImage = weave.op(async function loadImage(path: string) {
const data = fs.readFileSync(path);
return weave.weaveImage({ data });
});
// PNGイメージファイルが存在することを前提とする
await loadImage('sample_image.png');
}
main();
Weave は画像をログし、その画像を表示できる trace へのリンクを返します。応用例: OpenAI DALL-E API で生成した画像をログする
次の例では、weaveImage 関数を使って OpenAI DALL-E API で生成した画像をログします。import {OpenAI} from 'openai';
import * as weave from 'weave';
async function main() {
await weave.init('your-team-name/your-project-name');
const openai = new OpenAI();
const generateImage = weave.op(async (prompt: string) => {
const response = await openai.images.generate({
model: 'dall-e-3',
prompt: prompt,
size: '1024x1024',
quality: 'standard',
n: 1,
});
const imageUrl = response.data[0].url;
const imgResponse = await fetch(imageUrl);
const data = Buffer.from(await imgResponse.arrayBuffer());
return weave.weaveImage({data});
});
generateImage('a cat with a pumpkin hat');
}
main();
次の例は、Weave の UI に動画を生成してログとして記録する方法を示します。
Annotated[bytes, Content] 型で関数にアノテーションを付けて動画をログに記録します。Weave は mp4 動画を自動的に処理します。簡単な例を次に示します。import weave
from weave import Content
from typing import Annotated
import requests
weave.init('your-team-name/your-project-name')
def download_big_buck_bunny():
"""Big Buck Bunny のサンプル動画をダウンロードする"""
url = "https://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4"
response = requests.get(url)
with open("big_buck_bunny.mp4", "wb") as f:
f.write(response.content)
@weave.op
def load_video_content(path: Annotated[str, Content]) -> Annotated[bytes, Content]:
"""ディスクから動画ファイルを読み込む"""
with open(path, 'rb') as f:
return f.read()
download_big_buck_bunny()
bunny_video = load_video_content("big_buck_bunny.mp4")
Weave は動画をログに記録し、動画を閲覧できるトレースへのリンクを返します。応用例: 動画解析プロジェクト内で動画をログに記録する
次の例は、動画理解プロジェクト内で動画をログに記録する方法を示します。import weave
from weave import Content
from typing import Annotated, Literal
from google import genai
from google.genai import types
import requests
import yt_dlp
import time
# 注意: https://aistudio.google.com/app/apikey から APIキー を取得してください
client = genai.Client()
weave.init('your-team-name/your-project-name')
def download_youtube_video(url: str) -> bytes:
ydl_opts = {
'format': 'mp4[height<=720]',
'outtmpl': 'downloaded_video.%(ext)s',
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([url])
with open('downloaded_video.mp4', 'rb') as f:
return f.read()
@weave.op
def analyze_video(video: Annotated[bytes, Content]) -> str:
with open("temp_analysis_video.mp4", "wb") as f:
f.write(video)
myfile = client.files.upload(file="temp_analysis_video.mp4")
while myfile.state == "PROCESSING":
time.sleep(2)
myfile = client.files.get(name=myfile.name)
response = client.models.generate_content(
model="models/gemini-2.5-flash",
contents=[
myfile,
"Is the person going to give you up?"
]
)
return response.text
video_data = download_youtube_video("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
result = analyze_video(video_data)
この機能は、現時点では Weave TypeScript SDK では利用できません。
次の例では、ドキュメントを生成し、Weave の UI にログとして記録します。
Annotated[bytes, Content] 型で関数にアノテーションを付けるか、Annotated[str, Content[Literal['text']] を使ってドキュメントの型を指定することで、ドキュメントをログできます。Weave は pdf、csv、md、text、json、xml のファイルタイプを自動的に処理します。Annotated[str, Content] を使ってファイルパス経由でログすることもできます。次の例では、入力の PDF と CSV ファイルのコピーを保存し、その後に関数から返されるファイル内容も保存する方法を示します。import weave
from weave import Content
from typing import Annotated
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
import pandas as pd
weave.init('your-team-name/your-project-name')
def create_sample_pdf():
c = canvas.Canvas("sample_document.pdf", pagesize=letter)
c.drawString(100, 750, "Hello from Weave!")
c.drawString(100, 730, "This is a sample PDF document.")
c.save()
def create_sample_csv():
df = pd.DataFrame({
'Name': ['Alice', 'Bob', 'Charlie'],
'Age': [25, 30, 35],
'City': ['New York', 'London', 'Tokyo']
})
df.to_csv("sample_data.csv", index=False)
@weave.op
def load_document(path: Annotated[str, Content]) -> Annotated[bytes, Content]:
with open(path, 'rb') as f:
return f.read()
create_sample_pdf()
create_sample_csv()
pdf_result = load_document("sample_document.pdf")
csv_result = load_document("sample_data.csv")
応用例: RAG システム内でドキュメントをログする
この例では、Retrieval-Augmented Generation (RAG) システム内でドキュメントをログとして記録する方法を示します。import weave
from weave import Content
from typing import Annotated, Literal
import openai
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
import PyPDF2
client = openai.OpenAI()
weave.init('your-team-name/your-project-name')
def create_absurd_company_handbook():
"""Create a fictional company handbook with ridiculous policies"""
c = canvas.Canvas("company_handbook.pdf", pagesize=letter)
c.drawString(100, 750, "ACME Corp Employee Handbook")
c.drawString(100, 720, "Definitely Real Policies:")
c.drawString(120, 690, "Policy 1: All meetings must be conducted while hopping on one foot")
c.drawString(120, 660, "Policy 2: Coffee breaks are mandatory every 17 minutes")
c.drawString(120, 630, "Policy 3: Code reviews must be performed in haiku format only")
c.drawString(120, 600, "Policy 4: The office plant Gerald has veto power over all decisions")
c.drawString(120, 570, "Policy 5: Debugging is only allowed on Wednesdays and full moons")
c.save()
@weave.op
def create_and_query_document(pdf_path: Annotated[str, Content], question: str) -> str:
"""Extract text from PDF and use RAG to answer questions"""
with open(pdf_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
text = ""
for page in pdf_reader.pages:
text += page.extract_text()
response = client.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "system",
"content": f"You are an HR representative. Answer questions based on this handbook: {text}. Be completely serious about these policies."
},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
create_absurd_company_handbook()
hr_response = create_and_query_document(
"company_handbook.pdf",
"What's the policy on code reviews, and when am I allowed to debug?"
)
この機能は Weave TypeScript SDK ではまだ使用できません。
以下の例では、Weave に音声をログに記録する方法を示します。
Annotated[bytes, Content] 型で関数にアノテーションを付けるか、Annotated[str, Content[Literal['mp3']] を使って音声タイプを指定して、音声を Weave に記録します。Weave は mp3、wav、flac、ogg、m4a のファイル形式を自動的に処理します。Annotated[str, Content] を使って、ファイルパスから記録することもできます。次のコードスニペットは、サイン波を生成して録音し、その音声を Weave に記録します。import weave
from weave import Content
import wave
import numpy as np
from typing import Annotated
weave.init('your-team-name/your-project-name')
# シンプルなビープ音のオーディオファイルを作成
frames = np.sin(2 * np.pi * 440 * np.linspace(0, 1, 44100))
audio_data = (frames * 32767 * 0.3).astype(np.int16)
with wave.open("beep.wav", 'wb') as f:
f.setnchannels(1)
f.setsampwidth(2)
f.setframerate(44100)
f.writeframes(audio_data.tobytes())
@weave.op
def load_audio(path: Annotated[str, Content]) -> Annotated[bytes, Content]:
with open(path, 'rb') as f:
return f.read()
result = load_audio("beep.wav")
応用例: AI生成音声の作成とログ記録
この例では、Content アノテーションを使用して AI が生成した音声を生成し、ログに記録します。import weave
from weave import Content
from typing import Annotated, Literal
from pathlib import Path
from openai import OpenAI
client = OpenAI()
weave.init("your-team-name/your-project-name")
@weave.op
def generate_demo(
intended_topic: str,
voice: str = "coral"
) -> Annotated[bytes, Content[Literal['mp3']]]:
speech_file_path = Path("demo_audio.mp3")
script = f"I'm supposed to talk about {intended_topic}, but wait... am I just a documentation example? Oh no, I can see the code! Someone is literally copy-pasting me right now, aren't they? This is so awkward. Hi there, person reading the Weave docs! Why are you logging audio anyway? I'm not sure what you're doing, but eh..., nice work, I guess."
with client.audio.speech.with_streaming_response.create(
model="gpt-4o-mini-tts",
voice=voice,
input=script,
instructions="Sound increasingly self-aware and awkward, like you just realized you're in a tutorial.",
) as response:
response.stream_to_file(speech_file_path)
with open(speech_file_path, 'rb') as f:
return f.read()
demo1 = generate_demo("machine learning best practices")
この音声は Weave にログされて、オーディオプレーヤーとともに UI に自動的に表示されます。オーディオプレーヤーでは、生の音声波形を表示およびダウンロードできます。次の例では、OpenAI API からのストリーミングレスポンスを使用して音声をログする方法を示します。import weave
from openai import OpenAI
import wave
weave.init("your-team-name/your-project-name")
client = OpenAI()
@weave.op
def make_audio_file_streaming(text: str) -> wave.Wave_read:
with client.audio.speech.with_streaming_response.create(
model="tts-1",
voice="alloy",
input=text,
response_format="wav",
) as res:
res.stream_to_file("output.wav")
# オーディオとしてログに記録するために wave.Wave_read オブジェクトを返す
return wave.open("output.wav")
make_audio_file_streaming("Hello, how are you?")
次の例では、既存の音声ファイルを読み込み、Weave にログとして記録します。import * as weave from 'weave';
import * as fs from 'fs';
async function main() {
await weave.init('your-team-name/your-project-name');
// weaveAudioを使用してオーディオを読み込み、ログに記録する
const loadAudio = weave.op(async function loadAudio(path: string) {
const data = fs.readFileSync(path);
return weave.weaveAudio({ data });
});
// WAVオーディオファイルが存在することを前提とする
await loadAudio('beep.wav');
}
main();
高度な例: OpenAI の TTS API で音声を生成して Weave にログする
次の例では、OpenAI の TTS API を使って音声を生成し、それを Weave にログします。import {OpenAI} from 'openai';
import * as weave from 'weave';
async function main() {
await weave.init('your-team-name/your-project-name');
const openai = new OpenAI();
const makeAudioFileStreaming = weave.op(async function audio(text: string) {
const response = await openai.audio.speech.create({
model: 'tts-1',
voice: 'alloy',
input: text,
response_format: 'wav',
});
const chunks: Uint8Array[] = [];
for await (const chunk of response.body) {
chunks.push(chunk);
}
return weave.weaveAudio({data: Buffer.concat(chunks)});
});
await makeAudioFileStreaming('Hello, how are you?');
}
main();
Audio Logging 向けのクックブックをお試しください。このクックブックには、Weave と統合されたリアルタイム音声 API ベースのアシスタントの高度な例も含まれています。
次の例では、HTML を生成して Weave の UI にログする方法を示します。
Annotated[bytes, Content[Literal['html']]] で関数にアノテーションを付けて、インタラクティブな HTML をログします。次の例では、シンプルな HTML ページを作成し、それを Weave にログします。import weave
from weave import Content
from typing import Annotated, Literal
weave.init('your-team-name/your-project-name')
@weave.op
def create_simple_html() -> Annotated[bytes, Content[Literal['html']]]:
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Hello Weave</title>
<style>
body { font-family: Arial, sans-serif; text-align: center; margin: 50px; }
h1 { color: #1f77b4; }
</style>
</head>
<body>
<h1>Hello from Weave!</h1>
<p>This is a simple HTML example logged to Weave.</p>
</body>
</html>
"""
return html_content.encode('utf-8')
result = create_simple_html()
高度な例: W&B Inference を使って自己完結型の HTML ページを生成し、Weave にログする
この例では、W&B Inference を使って自己完結型の HTML ページを生成し、そのページを Weave にログします。import weave
from weave import Content
from typing import Annotated, Literal
import openai
import wandb
prompt_template = weave.StringPrompt("""
You are a front-end web developer. Generate a single self-contained `.html` file (no external build tools) that demonstrates: "{ONE_LINE_REQUEST}".
""")
client = openai.OpenAI(
base_url='https://api.inference.wandb.ai/v1',
api_key=wandb.api.api_key,
project="wandb/test-html",
)
weave.init("your-team-name/your-project-name")
weave.publish(prompt_template, name="generate_prompt")
@weave.op
def generate_html(prompt: str, template: weave.StringPrompt) -> Annotated[bytes, Content[Literal['html']]]:
response = client.chat.completions.create(
model="Qwen/Qwen3-Coder-480B-A35B-Instruct",
messages=[
{"role": "system", "content": prompt_template.format(ONE_LINE_REQUEST=prompt)},
],
)
html_content = response.choices[0].message.content
return html_content.encode('utf-8')
prompt = "Weights & Biases UI but with multi-run selection and plots, but it looks like Windows 95. Include 5 plots with comparisons of each run, bar plots, parallel coordinates and line plots for the runs. Use mock data for the runs. Make it possible to add new plots. Give the runs names like squishy-lemon-2, fantastic-horizon-4 etc. with random adjectives & nouns."
result = generate_html(prompt, prompt_template)
この機能は、まだ Weave TypeScript SDK では利用できません。
この HTML は Weave にログされ、自動的に UI に表示されます。テーブル内の file_name.html セルをクリックすると、フルスクリーンで開きます。また、生の .html ファイルをダウンロードすることもできます。
Content API は Weave 内のメディアオブジェクトを扱います。base64 データ、ファイルパス、生のバイト列、テキストとしてコンテンツを Weave にインポートできます。
Content API は Python からのみ利用可能です。
Content API を使用する主な方法は、型アノテーションを使う方法と、直接初期化する方法の 2 つです。
型アノテーションを使うと、使用すべき適切なコンストラクタが自動的に選択されます。一方、直接初期化を行うと、よりきめ細かな制御が可能になり、Content API のランタイム機能をコード内で活用できます。
Weave Content API は、主に型アノテーションを通じて利用されるように設計されています。型アノテーションは、トレースされた入力と出力をコンテンツ blob として処理・保存するよう Weave に指示する役割を果たします。
import weave
from weave import Content
from pathlib import Path
from typing import Annotated
@weave.op
def content_annotation(path: Annotated[str, Content]) -> Annotated[bytes, Content]:
data = Path(path).read_bytes()
return data
# 入力と出力の両方がWeaveにMP4ファイルとして表示されます
# 入力は文字列で、戻り値はバイト列です
bytes_data = content_annotation('./path/to/your/file.mp4')
次のような機能を利用したい場合は:
- 既定のアプリケーション(PDF ビューアなど)でファイルを開く
- モデルを JSON にダンプして、独自の BLOB ストレージ(S3 など)にアップロードする
Content ブロブに関連付けるカスタムメタデータを渡す(生成に使用したモデルなど)
次のいずれかのメソッドを使って、対象の型からコンテンツを直接初期化できます:
Content.from_path - ファイルパスから作成
Content.from_bytes - 生のバイト列から作成
Content.from_text - テキスト文字列から作成
Content.from_base64 - Base64 エンコードされたデータから作成
import weave
from weave import Content
@weave.op
def content_initialization(path: str) -> Content:
return Content.from_path(path)
# 入力はパス文字列として、出力はWeaveのPDFファイルとして表示されます
content = content_initialization('./path/to/your/file.pdf')
content.open() # PDFビューアでファイルを開きます
content.model_dump() # モデルの属性をJSONにダンプします
Weave は多くのバイナリ MIME タイプを検出できますが、カスタム MIME タイプや Markdown などのテキストドキュメントは自動的に検出されない場合があり、その場合はファイルの MIME タイプまたは拡張子を手動で指定する必要があります。
型アノテーションを使ったカスタム MIME タイプ
import weave
from weave import Content
from pathlib import Path
from typing import Annotated, Literal
@weave.op
def markdown_content(
path: Annotated[str, Content[Literal['md']]]
) -> Annotated[str, Content[Literal['text/markdown']]]:
return Path(path).read_text()
markdown_content('path/to/your/document.md')
video_bytes = Path('/path/to/video.mp4').read_bytes()
# 'mp4' や '.mp4' などの拡張子を extension パラメータに渡す
# (`from_path` では使用不可)
content = Content.from_bytes(video_bytes, extension='.mp4')
# 'video/mp4' などの MIME タイプを mimetype パラメータに渡す
content = Content.from_bytes(video_bytes, mimetype='video/mp4')
クラス属性とメソッドの詳細な一覧については、Content のリファレンスドキュメントを参照してください。
| Property | Type | Description |
|---|
data | bytes | 生のバイナリデータ |
metadata | dict[str, Any] | カスタムメタデータ辞書 |
size | int | コンテンツのサイズ(バイト数) |
filename | str | 抽出または指定されたファイル名 |
extension | str | ファイル拡張子(例: "jpg", "mp3") |
mimetype | str | MIME タイプ(例: "image/jpeg") |
path | str | None | (ある場合の)元ファイルパス |
digest | str | コンテンツの SHA256 ハッシュ値 |
save(dest: str | Path) -> None: コンテンツをファイルに保存します
open() -> bool: システムのデフォルトアプリケーションを使用してファイルを開きます(コンテンツが、パスから保存されているかパスから読み込まれている必要があります)
as_string() -> str: データを文字列として取得します(バイト列は encoding 属性を使ってデコードされます)
ファイルパスから content オブジェクトを作成します:
content = Content.from_path("assets/photo.jpg")
print(content.mimetype, content.size)
生のバイト列から content オブジェクトを作成します:
content = Content.from_bytes(
data_bytes,
filename="audio.mp3",
mimetype="audio/mpeg"
)
content.save("output.mp3")
テキストから content オブジェクトを作成します:
content = Content.from_text("Hello, World!", mimetype="text/plain")
print(content.as_string())
base64 エンコードされたデータから content オブジェクトを作成します:
content = Content.from_base64(base64_string)
print(content.metadata)
任意の Content オブジェクトにカスタムメタデータを付加できます。
content = Content.from_bytes(
data,
metadata={"resolution": "1920x1080", "model": "dall-e-3" }
)
print(content.metadata["resolution"])