【Muse Ember】Haiku 5.5並みを3分の1のコストで！性能・料金・使い方を解説

"""
検証1：スキャン風の請求書画像から項目をJSONで抜き出す（Muse Ember）
"""
import base64
import json
import os
import sys
import time

from openai import OpenAI

client = OpenAI(base_url="https://api.meta.ai/v1", api_key=os.environ["MODEL_API_KEY"])

SCHEMA = {
    "type": "object",
    "properties": {
        "invoice_number": {"type": ["string", "null"]},
        "issue_date": {"type": ["string", "null"], "description": "YYYY-MM-DD"},
        "due_date": {"type": ["string", "null"], "description": "YYYY-MM-DD"},
        "issuer": {"type": ["string", "null"]},
        "registration_number": {"type": ["string", "null"]},
        "line_items": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "quantity": {"type": "number"},
                    "unit_price": {"type": "number"},
                    "amount": {"type": "number"},
                    "tax_rate": {"type": "number", "description": "8 or 10"},
                },
                "required": ["name", "quantity", "unit_price", "amount", "tax_rate"],
            },
        },
        "total_on_header": {"type": ["number", "null"]},
        "total_recalculated": {"type": "number"},
        "issues": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["invoice_number", "issue_date", "due_date", "issuer", "registration_number",
                 "line_items", "total_on_header", "total_recalculated", "issues"],
}

PROMPT = (
    "この請求書画像から項目を抜き出してください。"
    "明細ごとに税率（※印は8%、それ以外は10%）を判定し、税込の合計を自分で再計算して"
    "total_recalculated に入れてください。冒頭の請求金額と食い違いがあれば issues に日本語で書いてください。"
    "読み取れない値は null にしてください。"
)

path = sys.argv[1] if len(sys.argv) > 1 else "invoice_sample.jpg"
b64 = base64.b64encode(open(path, "rb").read()).decode()

for effort in ["low", "max"]:
    t0 = time.time()
    res = client.chat.completions.create(
        model="muse-ember-1.0",
        reasoning_effort=effort,
        messages=[{"role": "user", "content": [
            {"type": "text", "text": PROMPT},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
        ]}],
        response_format={"type": "json_schema", "json_schema": {"name": "Invoice", "schema": SCHEMA}},
    )
    sec = time.time() - t0
    u = res.usage
    cost = u.prompt_tokens / 1e6 * 0.10 + u.completion_tokens / 1e6 * 0.50
    print(f"\n===== reasoning_effort={effort}  {sec:.1f}秒  入力{u.prompt_tokens}・出力{u.completion_tokens}トークン  約${cost:.5f} =====")
    print(json.dumps(json.loads(res.choices[0].message.content), ensure_ascii=False, indent=2))

"""
検証2：問い合わせ60件を並列で一括振り分け（Muse Ember）
"""
import csv
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor

from openai import OpenAI

client = OpenAI(base_url="https://api.meta.ai/v1", api_key=os.environ["MODEL_API_KEY"])
LABELS = ["billing", "bug", "feature", "account", "cancel"]
SYSTEM = (
    "あなたはSaaSのサポート窓口の振り分け担当です。問い合わせを次のどれか1つに分類してください。"
    "billing=請求・支払い・プラン変更・領収書、bug=不具合・エラー、feature=機能要望、"
    "account=ログイン・認証・アカウント管理、cancel=解約・利用停止。"
    "解約したい意図が主なら、請求の話を含んでいても cancel にしてください。"
)
SCHEMA = {
    "type": "object",
    "properties": {"label": {"type": "string", "enum": LABELS}, "reason": {"type": "string"}},
    "required": ["label", "reason"],
}

rows = list(csv.DictReader(open(os.path.join(os.path.dirname(__file__), "tickets.csv"), encoding="utf-8")))


def classify(row, effort):
    res = client.chat.completions.create(
        model="muse-ember-1.0",
        reasoning_effort=effort,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": row["text"]}],
        response_format={"type": "json_schema", "json_schema": {"name": "Route", "schema": SCHEMA}},
    )
    out = json.loads(res.choices[0].message.content)
    return row, out, res.usage


for effort in (sys.argv[1:] or ["minimal", "max"]):
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=16) as ex:
        results = list(ex.map(lambda r: classify(r, effort), rows))
    sec = time.time() - t0
    ok = sum(1 for r, o, _ in results if o["label"] == r["label"])
    tin = sum(u.prompt_tokens for _, _, u in results)
    tout = sum(u.completion_tokens for _, _, u in results)
    cost = tin / 1e6 * 0.10 + tout / 1e6 * 0.50
    print(f"\n===== reasoning_effort={effort} =====")
    print(f"正解 {ok}/{len(rows)}（{ok / len(rows):.0%}）  所要 {sec:.1f}秒  入力{tin}・出力{tout}トークン  約${cost:.4f}")
    print(f"1万件に換算すると 約${cost / len(rows) * 10000:.2f}")
    for r, o, _ in results:
        if o["label"] != r["label"]:
            print(f"  × id={r['id']} 正解={r['label']} 予測={o['label']}  {r['text'][:30]}…  理由: {o['reason'][:60]}")

"""検証3：API経由でMuse Emberにバグ修正を任せる（ツール呼び出しのエージェント）

Muse Emberに「ファイル一覧」「ファイルを読む」「ファイルを書く」「テストを実行」の4つのツールを渡し、
テストがすべて通るまで自分で調べて直させる。
元のリポジトリ（invoice-calc/）は書き換えず、work/ にコピーしたフォルダで作業します。
"""
import difflib
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

from openai import OpenAI

client = OpenAI(base_url="https://api.meta.ai/v1", api_key=os.environ["MODEL_API_KEY"])
MODEL = "muse-ember-1.0"
EFFORT = sys.argv[1] if len(sys.argv) > 1 else "high"
MAX_TURNS = 30  # 無限ループを防ぐ上限

HERE = Path(__file__).resolve().parent
SRC = HERE / "invoice-calc"
WORK = HERE / "work" / f"run-{time.strftime('%Y%m%d-%H%M%S')}-{EFFORT}"
shutil.copytree(SRC, WORK, ignore=shutil.ignore_patterns("__pycache__", ".pytest_cache"))
EDITABLE = {"invoice_calc/core.py"}  # 書き換えを許すファイル


def tests_hash():
    return hashlib.sha256(b"".join(p.read_bytes() for p in sorted((WORK / "tests").glob("*.py")))).hexdigest()


TESTS_BEFORE = tests_hash()


def resolve(path):
    p = (WORK / path).resolve()
    if WORK.resolve() not in p.parents:
        raise ValueError("作業フォルダの外は扱えません")
    return p


# ---- Muse Emberに渡すツールの中身 ----
def list_files():
    return "\n".join(str(p.relative_to(WORK)) for p in sorted(WORK.rglob("*"))
                     if p.is_file() and "__pycache__" not in p.parts)


def read_file(path):
    return resolve(path).read_text(encoding="utf-8")


def write_file(path, content):
    if path not in EDITABLE:
        return f"拒否：書き換えてよいのは {', '.join(sorted(EDITABLE))} だけです"
    resolve(path).write_text(content, encoding="utf-8")
    return f"{path} を保存しました（{len(content)}文字）"


def run_tests():
    r = subprocess.run([sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider"], cwd=WORK,
                       capture_output=True, text=True, timeout=120,
                       env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"})
    return (r.stdout + r.stderr)[-4000:]


FUNCS = {"list_files": list_files, "read_file": read_file, "write_file": write_file, "run_tests": run_tests}
TOOLS = [
    {"type": "function", "name": "list_files", "description": "リポジトリ内のファイル一覧を返す",
     "parameters": {"type": "object", "properties": {}}},
    {"type": "function", "name": "read_file", "description": "ファイルの中身を返す",
     "parameters": {"type": "object", "properties": {"path": {"type": "string", "description": "リポジトリからの相対パス"}},
                    "required": ["path"]}},
    {"type": "function", "name": "write_file", "description": "ファイルを指定した内容で丸ごと上書きする",
     "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string", "description": "ファイル全体の新しい内容"}},
                    "required": ["path", "content"]}},
    {"type": "function", "name": "run_tests", "description": "pytest を実行して結果の末尾を返す",
     "parameters": {"type": "object", "properties": {}}},
]

TASK = (
    "このリポジトリの pytest が失敗しています。原因を調べて invoice_calc/core.py だけを直し、"
    "run_tests がすべて通るまで繰り返してください。tests/ は変更しないでください。"
    "最後に、直したバグを1行ずつ日本語で一覧にしてください。"
)

# ---- エージェントのループ ----
stats = {"requests": 0, "input": 0, "cached": 0, "output": 0, "reasoning": 0, "cost": 0.0, "tool_calls": 0}


def record(resp):
    u = resp.usage
    cached = getattr(getattr(u, "input_tokens_details", None), "cached_tokens", 0) or 0
    reasoning = getattr(getattr(u, "output_tokens_details", None), "reasoning_tokens", 0) or 0
    long_ctx = u.input_tokens > 200_000  # 20万トークン超はリクエスト全体が割増単価
    price_in, price_cached, price_out = (0.30, 0.012, 1.50) if long_ctx else (0.10, 0.004, 0.50)
    stats["requests"] += 1
    stats["input"] += u.input_tokens
    stats["cached"] += cached
    stats["output"] += u.output_tokens
    stats["reasoning"] += reasoning
    stats["cost"] += ((u.input_tokens - cached) * price_in + cached * price_cached + u.output_tokens * price_out) / 1e6


print(f"作業フォルダ: {WORK}\nreasoning_effort={EFFORT}\n")
t0 = time.time()
resp = client.responses.create(model=MODEL, reasoning={"effort": EFFORT}, input=TASK, tools=TOOLS)
record(resp)

for turn in range(1, MAX_TURNS + 1):
    calls = [item for item in resp.output if item.type == "function_call"]
    if not calls:
        break
    outputs = []
    for c in calls:
        args = json.loads(c.arguments or "{}")
        shown = {k: (f"<{len(v)}文字>" if k == "content" else v) for k, v in args.items()}
        print(f"[{turn:02d}] {c.name}({shown})")
        try:
            result = FUNCS[c.name](**args)
        except Exception as e:  # ツールの失敗もそのままモデルに返す
            result = f"エラー: {e}"
        if result.startswith(("拒否", "エラー")):
            print(f"      → {result}")
        stats["tool_calls"] += 1
        outputs.append({"type": "function_call_output", "call_id": c.call_id, "output": result})
    resp = client.responses.create(model=MODEL, reasoning={"effort": EFFORT}, input=outputs,
                                   previous_response_id=resp.id, tools=TOOLS)
    record(resp)
else:
    print(f"\n※ {MAX_TURNS}ターンの上限に達したため打ち切りました")

elapsed = time.time() - t0

# ---- 結果の確認（モデルの申告ではなく、こちらでテストを流し直す） ----
print("\n===== Muse Emberの最終回答 =====")
print(resp.output_text)
print("\n===== こちらで再実行したテスト結果 =====")
print(run_tests().strip().splitlines()[-1])
print("tests/ の変更:", "なし" if tests_hash() == TESTS_BEFORE else "あり（ルール違反）")

diff = "".join(difflib.unified_diff(
    (SRC / "invoice_calc/core.py").read_text(encoding="utf-8").splitlines(True),
    (WORK / "invoice_calc/core.py").read_text(encoding="utf-8").splitlines(True),
    "before/core.py", "after/core.py"))
(WORK / "fix.diff").write_text(diff, encoding="utf-8")
print(f"差分: {WORK / 'fix.diff'}")

print("\n===== 実行の記録 =====")
print(f"所要時間 {elapsed:.1f}秒  リクエスト{stats['requests']}回  ツール呼び出し{stats['tool_calls']}回")
print(f"入力{stats['input']}（うちキャッシュ{stats['cached']}）・出力{stats['output']}（うち推論{stats['reasoning']}）トークン")
print(f"概算料金 約${stats['cost']:.4f}")