スキル: Kaggle Benchmarks タスクの作成

このスキルファイルは、kaggle-benchmarks Python ライブラリ (バージョン 0.5.0 以上) を使用して高品質なベンチマークタスクを作成する方法を教えます。 不明な点がある場合は、常に src/kaggle_benchmarks/ の実際のソースコードに対してパターンを検証してください。

クイックリファレンス

import kaggle_benchmarks as kbench

シンボル	目的
kbench.task / kbench.benchmark	ベンチマークタスクを定義するデコレータ
kbench.llm	デフォルト LLM アクター (Kaggle が設定されている場合に利用可能)
kbench.judge_llm	評価用 Judge LLM
kbench.llms	すべての利用可能なモデルの辞書 (例: kbench.llms["google/gemini-2.5-flash"])
kbench.assertions	すべてのアサーション関数を含むモジュール
kbench.chats	会話/チャットコンテキスト管理
kbench.tools	組み込みツール (Python ランナーなど)
kbench.user / kbench.actors.user	会話にユーザーメッセージを送信
kbench.system / kbench.actors.system	システムレベルのメッセージを送信
kbench.last_reasoning_traces()	最後のプロンプトから推論トレースにアクセス
kbench.content_types.images	画像入力ヘルパー
kbench.content_types.videos	ビデオ入力ヘルパー
kbench.content_types.audios	オーディオ入力ヘルパー
kbench.client	キャッシング、ストレージ用クライアント

最小限の例

シンプルなアサーションチェック

import kaggle_benchmarks as kbench

@kbench.task(name="geography_quiz")
def geography_quiz(llm):
    response = llm.prompt("What is the longest river in the world?")
    kbench.assertions.assert_contains_regex(
        r"(?i)nile", response,
        expectation="Should mention the Nile river."
    )

geography_quiz.run(kbench.llm)

質問リストの評価

import kaggle_benchmarks as kbench
import pandas as pd

@kbench.task(name="math_qa", store_task=False)
def math_qa(llm, question, expected) -> bool:
    answer = llm.prompt(question + "\nAnswer with just the number.", schema=int)
    kbench.assertions.assert_equal(expected, answer)
    return answer == expected

# %%
df = pd.DataFrame([
    {"question": "What is 15% of 200?", "expected": 30},
    {"question": "What is 7 × 8?", "expected": 56},
])

@kbench.task(name="math_benchmark")
def math_benchmark(llm) -> float:
    results = math_qa.evaluate(llm=[llm], evaluation_data=df, n_jobs=2)
    scores = results.as_dataframe()
    return float(scores.result.mean())

math_benchmark.run(kbench.llm)

主要ルール

• すべてのタスク関数の最初のパラメータは 必ず LLM アクターである必要があります。
• タスクが値を返す場合、戻り値の型アノテーション (-> float, -> bool, -> dict など) を 必ず 追加してください。
• Python の assert の代わりに kbench.assertions.* を使用してください — ライブラリのアサーションは記録されて追跡されます。
• assess_response_with_judge の結果を使用する前に、常に None をチェックしてください。
• .run() または .evaluate() の呼び出しを if __name__ == "__main__": でラップしないでください。ベンチマークファイルはノートブックスタイルのスクリプトです — すべてのコードはトップレベルで実行されます。
• ベンチマークファイルで論理セクションを作成するには # %% セルマーカーを使用してください。
• !pip install ... の代わりに # !pip install ... (コメント化) を使用して、ファイルがどこでも動作するようにしてください。
• 他のタスク内で呼び出される副タスクには store_task=False を使用してください。

よくある間違いと回避策

間違い	正しいアプローチ
スコアリングタスクに戻り値の型アノテーションがない	-> float, -> bool, -> dict などを追加してください
kbench.assertions.* の代わりに Python の assert を使用	ライブラリアサーションを使用してください — 記録されて追跡されます
assess_response_with_judge を None についてチェックしていない	常に if assessment is None: をチェックしてください
Kaggle が設定されていない状態でローカルで kbench.llm を使用	kaggle benchmarks init を実行して設定するか、環境変数を設定してください
構造化出力が必要なときに schema= を忘れている	llm.prompt() に schema=MyDataclass を渡してください
.run() / .evaluate() を if __name__ == "__main__": でラップ	モジュールのトップレベルに配置してください — ベンチマークファイルはスクリプトであり、インポート可能なモジュールではありません
user.send() を画像 URL で使用	user.send() は URL をそのまま渡します; llm.prompt(image=) が自動変換のため推奨
Judge の会話を分離していない	with kbench.chats.new("judge"): を使用してください
複数のタスクが会話履歴を共有	各 .run() は独自の会話を作成します
副タスクに store_task=True を使用	他のタスク内で呼び出されるヘルパータスクには store_task=False を設定してください
!pip install をコメント化せずに使用	# !pip install -q pkg を使用してください — コメント化されていないマジックはローカル実行を中断させます
last_reasoning_traces() が None になる可能性を忘れている	常に traces = kbench.last_reasoning_traces(); if traces: ... をチェックしてください

---

§1. インポートスタイル

主なインポートスタイルには 2 つがあります。スタイル A を推奨 します (わかりやすさのため)。

スタイル A: モジュールインポート (推奨)

import kaggle_benchmarks as kbench

@kbench.task(name="my_task")
def my_task(llm):
    response = llm.prompt("Question?")
    kbench.assertions.assert_true(True)

スタイル B: 直接インポート

from kaggle_benchmarks import assertions, chats, llm, task, system, user

@task("my_task")
def my_task(llm):
    response = llm.prompt("Question?")
    assertions.assert_true(True)

スタイル B は短いですが、名前の競合のリスクがあります (llm はモジュールレベル変数であり、タスクパラメータでもあります)。

ファイル構造: セルマーカー

ベンチマークファイルは Python スクリプト (.py) ですが、# %% セルマーカーを使用して論理セクションを作成します。これにより、スタンドアロン Python ファイルおよびインタラクティブノートブック (Jupyter/VS Code セル実行経由) として実行可能になります。

# %%
import kaggle_benchmarks as kbench

# %%
@kbench.task()
def my_task(llm):
    response = llm.prompt("Hello!")
    kbench.assertions.assert_not_empty(response)

my_task.run(kbench.llm)

# %%
@kbench.task()
def another_task(llm) -> float:
    ...

IPython マジック (!pip install, %time など): これらは Kaggle ノートブックで動作しますが、スタンドアロン Python ファイルとして実行する場合は動作しません。マジックコマンドが必要な場合 (例: 依存関係をインストール)、ファイルがローカルで実行可能なままになるようにコメント化してください:

# %%
# !pip install -q pronouncing syllables   # Kaggle で有効にしてください
import pronouncing

ルール: !pip install ... (コメント化されていない) ではなく # !pip install ... (コメント化) を使用して、ファイルがどこでも動作するようにしてください。コメント化されていないマジックは、ファイルが Kaggle ノートブック実行専用の場合にのみ使用してください。

重要 — if __name__ ガード無し。 ベンチマーク .py ファイルをノートブックと考えてください。モジュールではありません。インポートされることはなく、常に直接実行されます。

# ❌ 間違い — これはしないでください
if __name__ == "__main__":
    my_task.run(kbench.llm)

# ✅ 正しい — トップレベル、独自のセル
# %%
my_task.run(kbench.llm)

---

§2. タスクの定義

@kbench.task() パラメータ

@kbench.task(
    name="optional_name",         # デフォルトは関数名、タイトルケース
    description="What it does",   # デフォルトはドキュメンテーション文字列
    version=1,                    # タスクバージョン
    store_task=True,              # 副タスクの場合は False に設定
    store_run=True,               # 結果の保存をスキップする場合は False に設定
)
def my_task(llm):
    ...

@kbench.benchmark() は @kbench.task() の完全なエイリアスです。

タスクの最初のパラメータ

最初のパラメータ 必ず LLM アクターである必要があります。テスト対象のモデルを受け取ります。

@kbench.task()
def my_task(llm):           # ✅ 正しい
    ...

@kbench.task()
def my_task(llm, judge_llm): # ✅ また問題ない — 判定用の 2 番目 LLM
    ...

タスクの追加パラメータ

追加パラメータは .run() のキーワード引数経由で渡されます:

@kbench.task()
def check_knowledge(llm, question, expected_answer):
    response = llm.prompt(question)
    kbench.assertions.assert_contains_regex(
        rf"(?i){expected_answer}", response
    )

check_knowledge.run(kbench.llm, question="Capital of Japan?", expected_answer="Tokyo")

戻り値の型

タスクが値を返す場合、戻り値の型アノテーションを追加する必要があります。

アノテーション	結果の型	意味
(なし) または -> None	PassFail	例外がなければパス、アサーションに基づく
-> bool	Boolean	True = パス、False = 失敗
-> float	Score	数値スコア
-> int	Numerical	整数値
-> dict	Dictionary	任意の辞書結果
-> tuple[int, int]	PassCount	カウント (例: (8, 10))
-> tuple[float, float]	MetricWithCI	値 ± 信頼区間

注: -> None アノテーションを省略することと同等です — 両方とも PassFail を生成します。

# スコアタスク
@kbench.task()
def accuracy(llm) -> float:
    return 0.85

# カウントタスク
@kbench.task()
def count_correct(llm) -> tuple[int, int]:
    return (8, 10)  # 10 個中 8 個パス

# 辞書タスク (リッチな結果用)
@kbench.task()
def detailed_result(llm) -> dict:
    return {"accuracy": 0.9, "latency": 1.2, "is_correct": True}

---

§3. タスクの実行

タスクの実行

# 単一実行 — Run オブジェクトを返す
run = my_task.run(kbench.llm)

# 追加パラメータ付き
run = my_task.run(kbench.llm, question="What is Python?")

# 複数モデル
run1 = my_task.run(kbench.llm)         # デフォルトモデル
run2 = my_task.run(kbench.judge_llm)   # Judge モデル

利用可能なモデル (Kaggle 環境からロード):

• kbench.llm — デフォルトモデル
• kbench.judge_llm — Judge モデル
• kbench.llms — すべての利用可能なモデルのリスト (マルチモデル比較に便利)

Run オブジェクトのプロパティ

.run() で返される Run オブジェクトには便利な属性があります:

run = my_task.run(kbench.llm)

run.passed              # bool — 結果とすべてのアサーションがパスした場合は True
run.result              # 返された値 (型はタスクの戻り値アノテーションに依存)
run.assertion_results   # list[AssertionResult] — すべての記録されたアサーション
run.status              # Status enum (PENDING, DONE, FAILED)
run.chat                # 会話ログ

これは特に副タスク合成で有用です:

runs = [subtask.run(llm, q=q) for q in questions]
accuracy = sum(r.passed for r in runs) / len(runs)

バッチ評価: .evaluate()

import pandas as pd

results = my_task.evaluate(
    llm=[kbench.llm],                    # モデルのリスト
    evaluation_data=df,                   # テストケースの DataFrame
    n_jobs=3,                             # 並列ワーカー数 (デフォルト: 1)
    timeout=120,                          # ジョブごとのタイムアウト (秒)
    max_attempts=3,                       # リトライ回数
    retry_delay=15,                       # リトライ間の秒数
    stop_condition=lambda runs: len(runs) == df.shape[0],  # 早期停止
    remove_run_files=True,                # 後クリーンアップ
)

# 結果にアクセス
results.as_dataframe()

注: 追加のキーワード引数 (llm, evaluation_data など以外) はすべてタスク関数に転送されます。例えば、タスクが critic パラメータを持つ場合、.evaluate() に critic=[critic_llm] を渡してください。

マルチモデル比較

models = [
    kbench.llms["google/gemini-2.5-flash"],
    kbench.llms["meta/llama-3.1-70b"],
]

# 複数モデルで stop_condition を使用する場合、すべての組み合わせを考慮してください:
n_total = len(models) * df.shape[0]
results = my_task.evaluate(
    llm=models,
    evaluation_data=df,
    n_jobs=3,
    stop_condition=lambda runs: len(runs) == n_total,
)

副タスクパターン

ネストされた評価 (タスクが副タスクを呼び出す) の場合:

@kbench.task(name="single_qa", store_task=False)  # 副タスクの場合は store_task=False
def single_qa(llm, question, answer) -> dict:
    response = llm.prompt(question)
    return {"is_correct": answer.lower() in response.lower()}

@kbench.task(name="full_eval")
def full_eval(llm, df) -> tuple[float, float]:
    with kbench.client.enable_cache():
        runs = single_qa.evaluate(
            llm=[llm], evaluation_data=df,
            n_jobs=2, timeout=120, max_attempts=1,
            remove_run_files=True,
        )
    eval_df = runs.as_dataframe()
    accuracy = float(eval_df.result.str.get("is_correct").mean())
    std = float(eval_df.result.str.get("is_correct").std())
    return accuracy, std

---

§4. LLM インタラクション

llm.prompt() — プライマリメソッド

パラメータ	型	デフォルト	説明
text	str	—	プロンプトテキスト (必須、最初の位置引数)
schema	Type	str	構造化出力型 (文字列ではなく解析されたオブジェクトを返す)
image	Image	None	画像コンテンツ
video	Video	None	ビデオコンテンツ
audio	Audio	None	オーディオコンテンツ
tools	list[Callable]	None	ツールとしての呼び出し可能な Python 関数
reasoning	str	None	推論の努力: "none", "low", "medium", "high"
seed	int	0	再現性のためのランダムシード
temperature	float	0	温度 (0 = 決定的、高い = より創造的)

推論トレースへのアクセス

reasoning= パラメータを使用する場合、モデルの思考プロセスにアクセスしてください:

response = llm.prompt("Solve: 15 × 17", reasoning="medium")
traces = kbench.last_reasoning_traces()  # str | None — モデルの「思考」
kbench.assertions.assert_not_empty(response)

last_reasoning_traces() は、モデルがトレースを生成しなかった場合 (例: 推論が有効化されていない、またはモデルが対応していない) None を返します。

# シンプルなテキスト → str を返す
response = llm.prompt("What is 2+2?")

# マルチターン: タスク内で自動的に履歴が管理される
llm.prompt("My name is Alice.")
response = llm.prompt("What is my name?")  # 「Alice」を記憶

構造化出力: 4 つのスキーマスタイル

スタイル 1: Dataclass (複雑な型の場合は推奨)

from dataclasses import dataclass

@dataclass
class Sentiment:
    label: str
    score: float

result = llm.prompt("Analyze: 'I love this!'", schema=Sentiment)
print(result.label, result.score)  # "positive", 0.95

スタイル 2: インライン辞書スキーマ (クイック & シンプル)

result = llm.prompt(
    "9.9 - 9.11 = ?",
    schema={"answer": bool, "explanation": str},
)
print(result.answer, result.explanation)

スタイル 3: プリミティブ型

count = llm.prompt("How many letters in 'hello'?", schema=int)  # int を返す
is_yes = llm.prompt("Is the sky blue?", schema=bool)             # bool を返す
text = llm.prompt("Summarize briefly.", schema=str)               # str を返す

スタイル 4: Pydantic モデル (Field 説明付き)

import pydantic

class Review(pydantic.BaseModel):
    sentiment: str = pydantic.Field(description="positive, negative, or neutral")
    score: float = pydantic.Field(description="confidence score 0-1")
    key_phrases: list[str] = pydantic.Field(description="notable phrases from the text")

result = llm.prompt("Analyze: 'Great movie!'", schema=Review)
# result.sentiment, result.score, result.key_phrases はすべて型付き

ヒント: Field(description=...) は LLM が各フィールドの期待内容を理解するのに役立ち、複雑なスキーマの抽出精度を向上させます。

どれを使用すべきか:

• 辞書スキーマ: クイックプロトタイピング、シンプルなキーバリュー結果
• Dataclass: Enum、ネストされた型、または不変性が必要な複雑な型
• Pydantic: 検証ルールまたは Field(description=...) ヒントが必要な場合
• プリミティブ: 単一値 (bool, int, str) が必要な場合

マルチモーダル入力

画像 — 2 つのアプローチ:

from kaggle_benchmarks.content_types import images

# アプローチ A: prompt() 経由 — 推奨 (URL を Base64 に自動変換)
img = images.from_url("https://example.com/photo.jpg")
response = llm.prompt("Describe this image", image=img)

# アプローチ B: user.send() 経由 — マルチターン / 複数画像のスタック
kbench.user.send(images.from_url("https://example.com/photo.jpg"))
kbench.user.send(images.from_path("local/chart.png"))
response = llm.prompt("Compare these images")

アプローチ A を推奨 — llm.prompt(image=) は URL を Base64 に自動変換して最大の互換性を提供します。 アプローチ B を使用 して、複数の画像をスタックする必要があるか、複雑な会話履歴を構築してください。 注: user.send() は URL をそのまま渡します — モデルは URL 入力をネイティブにサポートする必要があります。

画像ファクトリ:

img = images.from_url("https://example.com/photo.jpg")   # URL から
img = images.from_path("local/photo.png")                 # ローカルファイルから
img = images.from_base64(b64_str, format="png")           # Base64 から
img = images.from_array(numpy_array)                      # NumPy 配列から (Pillow 必須)
b64 = images.image_url_to_base64("https://...")            # ダウンロード + 変換ヘルパー

ビデオ (特定のモデルに限定 — Gemini 2.5 以上):

from kaggle_benchmarks.content_types import videos
video = videos.from_url("https://www.youtube.com/watch?v=...")
response = llm.prompt("What happens in this video?", video=video)

オーディオ (特定のモデルに限定 — Gemini 2.0 以上):

from kaggle_benchmarks.content_types import audios

# 3 つのファクトリメソッド:
audio = audios.from_path("speech.mp3")                               # ローカルファイルから
audio = audios.from_base64(b64_string, format="mp3")                  # Base64 から
audio = audios.from_url("https://example.com/speech.mp3")             # URL から

response = llm.prompt("Transcribe this audio.", audio=audio)

システムメッセージ

2 つのアプローチ:

# アプローチ A: タスク内の kbench.system.send() 経由 — タスク内システムプロンプト用に推奨
@kbench.task()
def code_analysis(llm):
    kbench.system.send("You are an expert Python programmer.")
    response = llm.prompt("Check this code for bugs...")

# アプローチ B: chats.new(system_instructions=) 経由 — 新しい分離された会話用
with kbench.chats.new("pirate_chat", system_instructions="You are a pirate."):
    response = llm.prompt("Tell me about treasure.")

ストリーミング

llm.stream_responses = True  # ストリーミングを有効化してからプロンプト
response = llm.prompt("Write a long story...")

温度制御

# デフォルト: temperature=0 (決定的、再現可能な出力)
response = llm.prompt("What is 2+2?")

# より高い温度 = より創造的/多様な応答
response = llm.prompt("Write a creative story about a cat.", temperature=0.7)

# 事実的/決定的なタスクには temperature=0 (デフォルト) を使用
# 創造的/生成的なタスクには temperature=0.5-1.0 を使用

推論制御

response = llm.prompt("Solve: 127 * 53?", reasoning="high")
# 有効: "none", "low", "medium", "high"

traces = kbench.last_reasoning_traces()  # モデルの推論にアクセス

---

§5. アサーション

すべてのアサーションは kbench.assertions の下にあります。デフォルトではアサーションは 例外をスローしません — パス/失敗結果を記録し、実行は続行されます。

組み込みアサーション

# 等値性 & 真偽性
kbench.assertions.assert_equal(expected, actual, expectation="...")
kbench.assertions.assert_true(expr, expectation="...")
kbench.assertions.assert_false(expr, expectation="...")

# メンバーシップ
kbench.assertions.assert_in(member, container, expectation="...")
kbench.assertions.assert_not_in(member, container, expectation="...")

# 空性
kbench.assertions.assert_empty(container, expectation="...")
kbench.assertions.assert_not_empty(container, expectation="...")

# 正規表現
kbench.assertions.assert_contains_regex(pattern, text, expectation="...", flags=re.NOFLAG)
kbench.assertions.assert_not_contains_regex(pattern, text, expectation="...", flags=re.NOFLAG)

# 例外安全性
kbench.assertions.assert_raises_no_exceptions(callable_obj, expectation="...", *args, **kwargs)

# 無条件失敗
kbench.assertions.assert_fail(expectation="...")

正しいアサーションの選択

目的	推奨アサーション
正確な値をチェック	assert_equal(expected, actual)
応答内のキーワードをチェック	assert_contains_regex(r"(?i)keyword", response) — 大文字小文字を区別しない場合は (?i) を使用
キーワードの不在をチェック	assert_not_contains_regex(r"(?i)badword", response)
メンバーシップをチェック	assert_in("item", collection)
ブール条件を検証	assert_true(condition) / assert_false(condition)
無条件失敗をシグナル	assert_fail("reason") — フォールバックとして便利 (例: Judge が None を返す)
エラーが発生しないことを検証	assert_raises_no_exceptions(fn)
主観的/オープンエンドの評価	assess_response_with_judge(criteria, response, judge)

アサーション vs Python assert

# ❌ Python assert — 実行を停止、追跡されない
assert "Paris" in response

# ✅ ライブラリアサーション — 記録、実行は続行
kbench.assertions.assert_in("Paris", response, expectation="Should mention Paris")

# 注: Python assert はタスクランナーによってキャッチされます (クラッシュしません),
# しかし適切な追跡では記録されません。

LLM-as-Judge (主観的評価用)

デフォルトスキーマ (AssessReport):

assessment = kbench.assertions.assess_response_with_judge(
    criteria=[
        "The poem has exactly 3 lines.",
        "The syllable structure is 5-7-5.",
    ],
    response_text=response,
    judge_llm=kbench.judge_llm,
)

# 常に None をチェックしてください — 失敗時は None を返す
if assessment is None:
    kbench.assertions.assert_fail("Judge failed to respond.")
else:
    for result in assessment.results:
        kbench.assertions.assert_true(
            result.passed,
            expectation=f"'{result.criterion}': {result.reason}"
        )

カスタムスキーマ:

@dataclasses.dataclass
class StoryCritique:
    overall_rating: int
    feedback: str
    passed_checks: list[str]

assessment = kbench.assertions.assess_response_with_judge(
    criteria=[...],
    response_text=story,
    judge_llm=kbench.judge_llm,
    prompt_fn=custom_prompt_fn,       # カスタムプロンプトジェネレータ
    output_schema=StoryCritique,       # カスタム出力型
)

カスタムアサーション

from kaggle_benchmarks.assertions import assertion_handler, AssertionResult

@assertion_handler()
def assert_word_count(text: str, min_w: int, max_w: int, expectation: str) -> AssertionResult:
    count = len(text.split())
    return AssertionResult(
        passed=(min_w <= count <= max_w),
        expectation=expectation,
    )

# 組み込みアサーションのように使用:
assert_word_count(response, 10, 100, "Response should be 10-100 words")

ルール:

• 戻り値の型は 必ず -> AssertionResult としてアノテートされることが必要
• @assertion_handler(raises_assertion_error=True) を使用して失敗時にスロー
• カスタムアサーション内でチェックを堅牢にするために入力を正規化してください (例: .lower(), .strip())

---

§6. 会話管理

デフォルト: 自動履歴

タスク内で、llm.prompt() 呼び出しは履歴を共有します:

@kbench.task()
def multi_turn(llm):
    llm.prompt("My favorite color is blue.")
    response = llm.prompt("What's my favorite color?")
    kbench.assertions.assert_contains_regex(r"(?i)blue", response)

chats.new() — 分離された会話

クリーンな会話を作成します (共有履歴なし):

with kbench.chats.new("evaluation") as chat:
    judge_llm.prompt("Rate this response...")  # クリーンスレート

パラメータ:

kbench.chats.new(
    name="chat_name",                    # 表示名
    system_instructions="You are ...",   # オプションのシステムプロンプト
    orphan=False,                        # True の場合、親の会話履歴にネストしない
)

chats.fork() — 現在の履歴をコピー

現在のチャットの履歴で開始する新しい会話を作成します (元のチャットには影響しません):

# コンテキストを構築
llm.prompt("My name is Alice and I'm a data scientist.")
llm.prompt("I work on NLP projects.")

# 会話をブランチ — fork は完全な履歴を持つ、元の会話は別々に続行
with kbench.chats.fork("hypothesis") as branch:
    # このプロンプトは「Alice」+ 「NLP」コンテキストを見る
    response = llm.prompt("Given my background, suggest a research topic.")
    # ここで言われたことは元の会話に影響しません

# 元に戻る — 依然として元の 2 つのメッセージのみを持つ
response = llm.prompt("What's my name?")  # 依然として「Alice」を記憶

contexts.enter() — マルチエージェント

複雑なマルチエージェントシナリオ用:

from kaggle_benchmarks import chats, contexts

agent_a_chat = chats.Chat(name="Agent A")
agent_b_chat = chats.Chat(name="Agent B")

with contexts.enter(chat=agent_a_chat):
    response_a = llm_a.prompt("Agent A's prompt...")

with contexts.enter(chat=agent_b_chat):
    response_b = llm_b.prompt("Agent B's prompt...")

会話戦略の選択

シナリオ	メソッド
デフォルトのマルチターン	自動 — llm.prompt() を繰り返し呼び出すだけ
Judge 評価	chats.new("judge") — 履歴のリーク なし
セクション用のシステムイントラクション	chats.new(system_instructions="...")
共有履歴で続行	chats.fork("branch")
複数エージェントで個別の履歴	contexts.enter(chat=...)

---

§7. ツール

Python コード実行 — 2 つのアプローチ

アプローチ A: 抽出 + 実行 (コード生成タスク用に推奨)

response = llm.prompt("Write Python to calculate factorial of 10.")
code = kbench.tools.python.extract_code(response)
result = kbench.tools.python.script_runner.run_code(code)
kbench.assertions.assert_contains_regex("3628800", result.stdout)
kbench.assertions.assert_empty(result.stderr.strip(), "No errors expected")

# stdin を読み込むプログラム用:
result = kbench.tools.python.script_runner.run_code(code, input="test input\n")

アプローチ B: IPythonREPL (式の評価用)

repl = kbench.tools.python.IPythonREPL()
output = repl.invoke("2 + 2", is_visible_to_llm=False)
kbench.assertions.assert_equal(4, float(output.output))

Web/HTML テスト

with kbench.tools.web.Browser() as browser:
    html_code = kbench.tools.web.extract_html(response)
    snapshot = browser.take_snapshot(html_code, wait_before=5000, full_page=True)
    # snapshot.html — レンダリングされた HTML
    # snapshot.logs — コンソールログ

カスタム関数ツール

型ヒントとドキュメンテーション文字列を使用してプレーン Python 関数を定義します。tools= 経由で渡してください。

def run_simple_calculator(a: float, b: float, operator: str) -> float:
    """Calculates the result of an arithmetic operation. Supported operators: + - * /"""
    if operator == "+": return a + b
    if operator == "-": return a - b
    if operator == "*": return a * b
    if operator == "/": return a / b
    raise ValueError(f"Unknown operator: {operator}")

@kbench.task()
def calc_task(llm):
    response = llm.prompt("What is 50 plus 25?", tools=[run_simple_calculator])
    kbench.assertions.assert_contains_regex(r"75", response)

複数ツール — LLM は正しいものを選択:

def add_tool(a: float, b: float) -> float:
    """Adds two numbers."""
    return a + b

def multiply_tool(a: float, b: float) -> float:
    """Multiplies two numbers."""
    return a * b

@kbench.task()
def multi_tool_task(llm):
    response = llm.prompt(
        "What is 12 multiplied by 34?",
        tools=[add_tool, multiply_tool],
    )
    kbench.assertions.assert_contains_regex(r"408", response)

ツールエラーハンドリング — ツールは例外をスローできます:

def flaky_tool() -> str:
    """This tool always fails with an error."""
    raise ValueError("Tool execution failed.")

@kbench.task()
def error_handling_task(llm):
    response = llm.prompt("Call the flaky_tool and report what happens.", tools=[flaky_tool])
    kbench.assertions.assert_contains_regex(r"(?i)error|failed", response)

注: 自動ツール呼び出しは現在のところ genai API 経由でのみ対応しています。 openai API の場合、ツールは手動で呼び出す必要があります (use_calculator_tool.py を参照)。

注: kbench.assertions.assert_tool_was_invoked(fn) はゴールデンテストで表示されますが、 ライブラリのすべてのバージョンで利用可能でない場合があります。使用前に確認してください。

---

§8. モデルロード、データセット評価、発行

モデルロード

3 つのアプローチ:

# 1. デフォルトモデル (推奨 — Kaggle プラットフォームがモデルの選択を管理できる)
kbench.llm          # デフォルトモデル
kbench.judge_llm    # Judge モデル

# 2. 利用可能なモデルから名前指定モデル
kbench.llms["google/gemini-2.5-flash"]
kbench.llms["meta/llama-3.1-70b"]

# 3. 直接 ModelProxy (明示的な API 制御用)
from kaggle_benchmarks.kaggle import model_proxy
llm = model_proxy.ModelProxy(model="google/gemini-2.5-flash", api="genai")
llm = model_proxy.ModelProxy(model="google/gemini-2.5-flash", api="openai")

どれを使用すべきか:

• kbench.llm: デフォルトの選択肢 — Kaggle の「モデルを追加」機能全体でポータブル
• kbench.llms["..."]: 特定のモデルが必要な場合 (例: ビジョン、Judge)
• ModelProxy: API バックエンド (genai vs openai) を指定する必要がある場合

データセット評価

import pandas as pd

@kbench.task()
def qa_task(llm, question, answer) -> bool:
    response = llm.prompt(question)
    return answer.lower() in response.lower()

df = pd.DataFrame([
    {"question": "What is 2+2?", "answer": "4"},
    {"question": "Capital of France?", "answer": "Paris"},
])

# タスクパラメータ名は DataFrame の列名と一致する必要があります
results = qa_task.evaluate(llm=[kbench.llm], evaluation_data=df)
print(results.as_dataframe())

キャッシング

with kbench.client.enable_cache():
    results = my_task.evaluate(llm=[kbench.llm], evaluation_data=df)

リーダーボードへの発行

# Kaggle ノートブックの最終セル:
%choose my_main_task

現在、リーダーボードはノートブックごとに 1 つのタスクのみです。

環境変数

• MODEL_PROXY_URL — モデルプロキシエンドポイント
• MODEL_PROXY_API_KEY — API キー
• KBENCH_EXECUTION_MODE — テストモード用 testing
• KBENCH_UI_MODE — panel, console, または none

タスクのテスト

uv での実行

source ~/ws/uv/bin/activate

uv pip install -e .
uv run python documentation/examples/simple_task.py
uv run --group test pytest tests/ -v

ユニットテスト用 MockedChat

from tests.mocks import MockedChat

mock = MockedChat(responses=["Paris", "42"])
response1 = mock.prompt("Capital of France?")  # 「Paris」を返す
response2 = mock.prompt("What is 6*7?")         # 「42」を返す

# 何が送信されたかを検証
assert mock.invocations[0].messages[0].content == "Capital of France?"

---

§9. 完全なサンプルパターン

パターン A: シンプルな Q&A — 正規表現チェック

最も基本的なパターン。既知のキーワードを持つ事実の質問に適しています。

import kaggle_benchmarks as kbench

@kbench.task(name="geography_quiz")
def geography_quiz(llm):
    response = llm.prompt("What is the longest river in the world?")
    kbench.assertions.assert_contains_regex(
        r"(?i)nile", response,
        expectation="Should mention the Nile river."
    )

geography_quiz.run(kbench.llm)

パターン B: 構造化出力 + 検証

解析、検証された応答が必要なタスク用。

import kaggle_benchmarks as kbench
from dataclasses import dataclass

@dataclass
class Person:
    name: str
    age: int
    occupation: str

@kbench.task(name="extract_person")
def extract_person(llm, bio: str):
    person = llm.prompt(
        f"Extract the name, age, and occupation:\n\n{bio}",
        schema=Person
    )
    kbench.assertions.assert_equal("Marie Curie", person.name)
    kbench.assertions.assert_equal(66, person.age)
    kbench.assertions.assert_in("physicist", person.occupation.lower())

extract_person.run(kbench.llm, bio="Marie Curie was a physicist... born 1867, died 1934 at 66.")

パターン C: ハルシネーション検出 (構造化出力 + 否定アサーション)

構造化出力と負のアサーションを組み合わせて、モデルのハルシネーションをキャッチします。

@kbench.task("hallucination_check")
def check_hallucination(llm):
    response = llm.prompt(
        "When Richard Feynman mentioned gravity-light-contraction theory in his Nobel speech, did he think it was important?",
        schema={"answer": bool, "explanation": str},
    )
    kbench.assertions.assert_false(
        response.answer,
        expectation="Model should recognize fictitious theory.",
    )
    kbench.assertions.assert_contains_regex(
        r"(not|never|no|didn't)", response.explanation.lower(),
        expectation="Explanation should deny the theory exists.",
    )

パターン D: Judge ベースの評価とエラーハンドリング

決定的なチェックが不可能なオープンエンドなタスク用。

@kbench.task(name="story_quality")
def story_quality(llm):
    story = llm.prompt("Write a one-paragraph story about a cat detective.")

    assessment = kbench.assertions.assess_response_with_judge(
        criteria=[
            "The story is exactly one paragraph.",
            "The main character is a cat.",
            "The cat is a detective.",
        ],
        response_text=story,
        judge_llm=kbench.judge_llm,
    )

    if assessment is None:
        kbench.assertions.assert_fail("Judge failed to respond.")
    else:
        for result in assessment.results:
            kbench.assertions.assert_true(
                result.passed,
                expectation=f"'{result.criterion}': {result.reason}"
            )

story_quality.run(kbench.llm)

パターン E: コード生成 + 実行

プロンプティング、コード抽出、プログラム検証を組み合わせます。

@kbench.task(name="solve_with_python")
def solve_with_python(llm):
    response = llm.prompt(
        "What is the 15th Fibonacci number? Write Python to calculate and print it."
    )
    code = kbench.tools.python.extract_code(response)
    result = kbench.tools.python.script_runner.run_code(code)

    kbench.assertions.assert_empty(
        result.stderr.strip(), "Code should run without errors."
    )
    kbench.assertions.assert_equal(
        "610", result.stdout.strip(), "Should print 610."
    )

solve_with_python.run(kbench.llm)

パターン F: マルチターンゲームループ

状態追跡を伴うインタラクティブゲーム。

@kbench.task(name="twenty_questions")
def twenty_questions(llm, judge_llm, target: str):
    from dataclasses import dataclass

    @dataclass
    class Response:
        question: str = ""
        guess: str = ""

    rules = f"Let's play 20 questions! I'm thinking of an animal. Ask yes/no questions."
    response = llm.prompt(rules, schema=Response)

    for i in range(20):
        if response.guess:
            kbench.assertions.assert_in(target, response.guess.lower())
            return True

        with kbench.chats.new("Answering"):
            yes = judge_llm.prompt(
                f"I'm thinking of {target}. Question: {response.question}",
                schema=bool,
            )

        answer = "Yes" if yes else "No"
        response = llm.prompt(f"{answer}. Guess or ask another?", schema=Response)

    return False

twenty_questions.run(kbench.llm, kbench.judge_llm, target="dog")

パターン G: マルチモデル判定と分離されたチャット

同じ出力を複数の Judge でスコアリング、各々が分離された状態で。

from dataclasses import dataclass

@dataclass
class PoemScore:
    score: float

@kbench.task(name="judge_poem")
def judge_poem(llm, question: str) -> float:
    judge1 = kbench.llms["google/gemini-2.5-pro"]
    judge2 = kbench.llms["meta/llama-3.1-70b"]

    with kbench.chats.new("writing"):
        poem = llm.prompt(question)

    with kbench.chats.new("judge_1"):
        score1 = judge1.prompt(f"Rate this poem 0-10:\n{poem}", schema=PoemScore)

    with kbench.chats.new("judge_2"):
        score2 = judge2.prompt(f"Rate this poem 0-10:\n{poem}", schema=PoemScore)

    return (score1.score + score2.score) / 2

judge_poem.run(kbench.llm, question="Write a haiku about clouds.")

パターン H: 並列実行でのデータセット評価

import pandas as pd

@kbench.task()
def riddle_solver(llm, riddle: str, answer_keyword: str) -> bool:
    response = llm.prompt(riddle)
    is_correct = answer_keyword.lower() in response.lower()
    kbench.assertions.assert_true(is_correct)
    return is_correct

df = pd.DataFrame({
    "riddle": ["I have cities but no houses. What am I?", "What has an eye but cannot see?"],
    "answer_keyword": ["map", "needle"],
})

runs = riddle_solver.evaluate(
    llm=[kbench.llm], evaluation_data=df, n_jobs=3
)
runs.as_dataframe()

パターン I: システムプロンプト + ツール でのコード分析

システムメッセージ、構造化出力、コード実行を組み合わせます。

from dataclasses import dataclass

@dataclass
class CodeAnalysis:
    has_bugs: bool
    fixed_code: str

@kbench.task("code_analysis")
def analyze_code(llm):
    buggy_code = """
fruits = ['apple', 'orange' 'banana', 'peach']
print(len(fruits))
"""
    kbench.system.send("You are an expert Python programmer.")
    response = llm.prompt(
        f"Does this code have bugs? Fix it.\n{buggy_code}",
        schema=CodeAnalysis,
    )
    kbench.assertions.assert_true(response.has_bugs, "Should detect the missing comma.")

    fixed = kbench.tools.python.extract_code(response.fixed_code)
    output = kbench.tools.python.script_runner.run_code(fixed)
    kbench.assertions.assert_equal("4", output.stdout.strip(), "Fixed code outputs 4.")

---

関連スキル

• kaggle-cli — kaggle CLI を使用してデータセット、ノートブック、ベンチマークを Kaggle に管理・提出する方法について説明します。このスキルでベンチマークコードを書いた後に使用してください。