#!/usr/bin/env python3 import argparse, json, sys, time from urllib import request # 日本語環境に合わせたプロンプトの構成 PROMPTS = [ { "name": "code_python", "prompt": "メモ化を使用してn番目のフィボナッチ数を返すPython関数を作成してください。docstringも含めてください。" }, { "name": "code_cpp", "prompt": "xを[lo, hi]の範囲に収めて（クランプして）返すC++のテンプレート関数 `clamp(x, lo, hi)` を作成してください。`std::clamp` は使用しないでください。" }, { "name": "explain_concept", "prompt": "大規模言語モデルの推論において、投機的デコーディング（Speculative Decoding）がどのように機能するかを、短い3つの段落で説明してください。" }, { "name": "summarize", "prompt": "次の文章を2文で要約してください：『産業革命は18世紀後半に英国で始まり、機械化、蒸気機関、および工場システムを通じて製造業を変革しました。19世紀中にヨーロッパ大陸と北米に拡大しました。』" }, { "name": "qa_factual", "prompt": "Q: 物理学における4つの基本相互作用（基本の力）は何ですか？\nA:" }, { "name": "translation", "prompt": "次の日本語を英語に翻訳してください：『素早い茶色の狐が怠け者の犬を飛び越える』" }, { "name": "creative_short", "prompt": "古い灯台をテーマにした、味わい深い4行の詩を書いてください。" }, { "name": "stepwise_math", "prompt": "ステップバイステップで解いてください：ある列車が駅Aを時速60kmで出発しました。その2時間後、同じ駅から同じ線路上を別の快速列車が時速90kmで出発しました。快速列車が最初の列車に追いつくまでに何時間かかりますか？" }, { "name": "long_code_review", "prompt": ( "あなたは本番環境で断続的なレイテンシスパイクに悩まされているバックエンドサービスのコードレビューを行っています。 " "以下に関連するコードとシステムの説明があります。慎重に読み通した上で、次の3つのセクションで構成される構造化されたレビューを作成してください： " "(1) 確率の高い順に並べた考えられる根本原因、 " "(2) 最初に実施すべき具体的なコードまたは設定の変更、 " "(3) 診断を確定させるために追加すべきテレメトリ。\n\n" "システム説明：Postgres 15データベースの前段に配置された、PythonのFastAPIサービス。 " "nginxロードバランサの背後に4つのレプリカとしてデプロイされています。各リクエストは、ユーザーレコードを読み取り、 " "パーティション化されたeventsテーブルから最新の50件のイベントを取得し、集計スコアを計算し、そのスコアをユーザー行に書き戻し、 " "JSONレスポンスを返します。平均ペイロードは4KBです。p50レイテンシは35ミリ秒ですが、p99レイテンシは約90秒ごとに規則的なパターンで " "1.8秒にスパイクします。このスパイクはPostgresの接続数増加ではなく、PostgresのCPU高負荷と相関しています。 " "アプリケーションの接続プールサイズは、レプリカあたり20接続に設定されています。 " "Postgresの前段にはPgBouncerがトランザクションプールモードで配置されており、プールサイズは50です。\n\n" "コード抜粋（負荷の高いエンドポイント）：\n" "```python\n@app.post('/score/{user_id}')\nasync def score(user_id: int, payload: ScoreRequest):\n" " async with db.transaction() as tx:\n user = await tx.fetchrow(\n" " 'SELECT id, tier, last_score FROM users WHERE id = $1 FOR UPDATE',\n user_id,\n )\n" " if user is None:\n raise HTTPException(404)\n events = await tx.fetch(\n" " 'SELECT type, weight, ts FROM events '\n 'WHERE user_id = $1 ORDER BY ts DESC LIMIT 50',\n user_id,\n )\n" " new_score = compute_score(user['tier'], events, payload.signals)\n" " await tx.execute(\n 'UPDATE users SET last_score = $1, updated_at = now() WHERE id = $2',\n new_score, user_id,\n )\n" " await tx.execute(\n 'INSERT INTO score_history (user_id, score, ts) VALUES ($1, $2, now())',\n user_id, new_score,\n )\n" " await cache.set(f'score:{user_id}', new_score, ex=300)\n" " metrics.histogram('score.latency_ms').observe((time.time() - start) * 1000)\n" " return {'user_id': user_id, 'score': new_score}\n```\n\n" "スキーマノート：`users`は約5000万行、`events`は月ごとにパーティション分割されており、合計で約20億行あります。 " "また、`(user_id, ts DESC)`のB-treeインデックスが貼られています。`score_history`はパーティション分割されておらず、 " "約8億行あり、`user_id`に単一のインデックスが貼られています。Postgresのautovacuumはデフォルト設定のままです。 " "毎日UTC 02:00にマテリアライズドビューを再構築する夜間バッチジョブが実行されていますが、スパイクはバッチ実行時間帯に限らず、 " "1日中発生しています。接続プールメトリクスによると、スパイク発生時にPgBouncerの待機接続数が一時的に8?12に達することがありますが、 " "飽和することはありません。FastAPIレプリカのCPU使用率は、スパイク発生時も含めて常に30%未満に維持されています。 " "アプリケーションとPostgresの間のネットワーク往復時間（RTT）は常に0.4ミリ秒です。\n\n" "それでは、レビューを開始してください。" ) }, ] def post(url, payload): req = request.Request(url, data=json.dumps(payload).encode(), headers={"Content-Type":"application/json"}, method="POST") with request.urlopen(req, timeout=300) as r: return json.loads(r.read()) def run(args): out = {"results": []} for p in PROMPTS: t0 = time.time() r = post(f"{args.url}/v1/chat/completions", { "model": "llama", "messages": [{"role": "user", "content": p["prompt"]}], "max_tokens": 192, "seed": 42, }) wall = time.time() - t0 usage = r.get("usage", {}) or {} t = r.get("timings", {}) or {} predicted_n = usage.get("completion_tokens") or t.get("predicted_n") predicted_per_second = t.get("predicted_per_second") or (predicted_n / wall if wall > 0 else 0) rec = {"name": p["name"], "wall_s": round(wall,3), "predicted_n": predicted_n, "predicted_per_second": round(predicted_per_second, 2), "draft_n": t.get("draft_n",0), "draft_n_accepted": t.get("draft_n_accepted",0)} rec["accept_rate"] = round(rec["draft_n_accepted"]/rec["draft_n"],4) if rec["draft_n"] else None out["results"].append(rec) ar = f"{rec['accept_rate']:.3f}" if rec["accept_rate"] is not None else "n/a" print(f" {rec['name']:<18} pred={rec['predicted_n']:>4} draft={rec['draft_n']:>4} acc={rec['draft_n_accepted']:>4} rate={ar} tok/s={rec['predicted_per_second']:.1f}") td = sum(x["draft_n"] or 0 for x in out["results"]) ta = sum(x["draft_n_accepted"] or 0 for x in out["results"]) tp = sum(x["predicted_n"] or 0 for x in out["results"]) tw = sum(x["wall_s"] for x in out["results"]) out["aggregate"] = {"n_requests": len(out["results"]), "total_predicted": tp, "total_draft": td, "total_draft_accepted": ta, "aggregate_accept_rate": round(ta/td,4) if td else None, "wall_s_total": round(tw,2)} print("\nAggregate:", json.dumps(out["aggregate"], indent=2)) if args.out: json.dump(out, open(args.out,"w"), indent=2); print("Wrote", args.out) def diff(a, b): A, B = json.load(open(a)), json.load(open(b)) print(f"{'metric':<24} {'A':>14} {'B':>14} {'delta':>10}") for k in ("aggregate_accept_rate","total_predicted","total_draft","total_draft_accepted","wall_s_total"): va, vb = A["aggregate"].get(k), B["aggregate"].get(k) if va is None or vb is None: print(f"{k:<24} {str(va):>14} {str(vb):>14}"); continue d = vb - va s = f"{d:>+10.4f}" if isinstance(d,float) else f"{d:>+10}" print(f"{k:<24} {va:>14} {vb:>14} {s}") by_a = {x["name"]: x for x in A["results"]} print("\n{:<20} {:>8} {:>8} {:>8}".format("prompt","A","B","delta")) for rb in B["results"]: ra = by_a.get(rb["name"]) or {} ar = ra.get("accept_rate") or 0; br = rb.get("accept_rate") or 0 print(f"{rb['name']:<20} {ar:>8.3f} {br:>8.3f} {br-ar:>+8.3f}") ap = argparse.ArgumentParser() ap.add_argument("--url", default="http://127.0.0.1:8080") ap.add_argument("--out") ap.add_argument("--diff", nargs=2) a = ap.parse_args() if a.diff: diff(*a.diff) else: run(a)