Gemini 3.8を「使えるAI」にする10の実装判断――Live・Flash・Cyberを経営と現場の共通言語で設計する

"""要件から応答方式の候補を選ぶ最小ルーター。"""
from dataclasses import dataclass

@dataclass
class Request:
    audio: bool = False
    deep_reasoning: bool = False
    external_write: bool = False
    sensitivity_known: bool = True

def choose(req: Request) -> str:
    if req.external_write or not req.sensitivity_known:
        return "Human review"
    if req.audio and req.deep_reasoning:
        return "Live Extended Thinking"
    if req.audio:
        return "Live"
    return "Flash"

def main() -> None:
    fixtures = [
        (Request(audio=True), "Live"),
        (Request(audio=True, deep_reasoning=True), "Live Extended Thinking"),
        (Request(external_write=True), "Human review"),
        (Request(), "Flash"),
    ]
    for req, expected in fixtures:
        actual = choose(req)
        assert actual == expected, (req, actual)
        print(f"{req} -> {actual}")

if __name__ == "__main__":
    main()

"""Gemini Flash Interactions REST例。既定はdry-runで課金しない。"""
import json, os, sys, urllib.error, urllib.request

MODEL = "gemini-3.8-flash"
URL = "https://generativelanguage.googleapis.com/v1beta/interactions"

def payload(prompt: str) -> dict:
    return {
        "model": MODEL,
        "input": prompt,
        "generation_config": {"thinking_level": "medium"},
    }

def parse_response(data: dict) -> str:
    texts = [part.get("text", "") for out in data.get("outputs", [])
             if out.get("type") == "text" for part in [out]]
    text = "\n".join(filter(None, texts)).strip()
    if not text:
        raise RuntimeError("empty text output")
    return text

def main() -> None:
    body = json.dumps(payload("3行で要約して"), ensure_ascii=False).encode()
    assert payload("x")["generation_config"]["thinking_level"] == "medium"
    if "--execute" not in sys.argv:
        print(json.dumps({"url": URL, "body": json.loads(body)}, ensure_ascii=False, indent=2))
        return
    key = os.environ.get("GEMINI_API_KEY")
    if not key:
        raise SystemExit("GEMINI_API_KEY is required with --execute")
    req = urllib.request.Request(URL, body, {"Content-Type": "application/json", "x-goog-api-key": key})
    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            print(parse_response(json.loads(res.read())))
    except urllib.error.HTTPError as exc:
        raise SystemExit(f"HTTP error: {exc.code}") from None
    except urllib.error.URLError:
        raise SystemExit("network error") from None

if __name__ == "__main__":
    assert parse_response({"outputs": [{"type": "text", "text": "ok"}]}) == "ok"
    try: parse_response({"outputs": []})
    except RuntimeError: pass
    else: raise AssertionError("empty response must fail")
    main()

"""PCM WAVが16bit・mono・16kHzか検査する。"""
import io, struct, wave

def inspect(source) -> dict:
    with wave.open(source, "rb") as wav:
        result = {
            "channels": wav.getnchannels(),
            "sample_width": wav.getsampwidth(),
            "sample_rate": wav.getframerate(),
            "frames": wav.getnframes(),
            "compression": wav.getcomptype(),
        }
    result["valid"] = (result["channels"], result["sample_width"],
                       result["sample_rate"], result["compression"]) == (1, 2, 16000, "NONE")
    return result

def fixture(channels=1, rate=16000) -> io.BytesIO:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wav:
        wav.setparams((channels, 2, rate, 160, "NONE", "not compressed"))
        wav.writeframes(struct.pack(f"<{160 * channels}h", *([0] * 160 * channels)))
    buf.seek(0)
    return buf

if __name__ == "__main__":
    actual = inspect(fixture())
    assert actual["valid"] and actual["frames"] == 160
    assert inspect(fixture(channels=2))["valid"] is False
    assert inspect(fixture(rate=48000))["valid"] is False
    print(actual)

"""複数ツールをasyncioで並行実行するデモ。"""
import asyncio, time

async def tool(name: str, delay: float, version: int) -> dict:
    started = time.perf_counter()
    await asyncio.sleep(delay)
    return {"name": name, "version": version, "elapsed": time.perf_counter() - started}

async def run(current_version: int = 2) -> tuple[list[dict], float]:
    started = time.perf_counter()
    results = await asyncio.gather(
        tool("search", 0.08, 1),       # 訂正前の古い依頼
        tool("calendar", 0.05, 2),
        tool("database", 0.06, 2),
    )
    accepted = [r for r in results if r["version"] == current_version]
    return accepted, time.perf_counter() - started

def main() -> None:
    results, elapsed = asyncio.run(run())
    assert [r["name"] for r in results] == ["calendar", "database"]
    assert all(r["version"] == 2 for r in results)
    print({"wall_seconds": round(elapsed, 3), "results": results})

if __name__ == "__main__":
    main()

"""アプリ内部の応答とタスク状態を管理する例（架空APIではない）。"""
from dataclasses import dataclass

ALLOWED = {"queued": {"running"}, "running": {"done", "failed"}, "done": set(), "failed": set()}

@dataclass
class Task:
    state: str = "queued"
    response: str | None = None

    def transition(self, target: str) -> None:
        if target not in ALLOWED[self.state]:
            raise ValueError(f"invalid transition: {self.state} -> {target}")
        self.state = target

    def complete(self, response: str, verified: bool) -> None:
        if self.state != "running":
            raise ValueError("task must be running")
        if not response.strip():
            raise ValueError("response must not be empty")
        if verified is not True:
            raise ValueError("verification is required")
        self.transition("done")
        self.response = response

if __name__ == "__main__":
    task = Task(); task.transition("running")
    try: task.complete("未検証結果", verified=False)
    except ValueError: assert task.state == "running" and task.response is None
    task.complete("処理結果", verified=True)
    assert task.state == "done" and task.response == "処理結果"
    try: task.transition("running")
    except ValueError: pass
    else: raise AssertionError("terminal state must be immutable")
    print(task)

"""長文を段落単位で分割する。実トークン数は未計測。"""
from dataclasses import dataclass

@dataclass
class Chunk:
    text: str
    chars: int
    measured_tokens: None = None  # 実トークナイザー未使用を明示

def split(text: str, max_chars: int = 18) -> list[Chunk]:
    chunks, current = [], ""
    for para in filter(None, (p.strip() for p in text.split("\n\n"))):
        candidate = para if not current else current + "\n\n" + para
        if current and len(candidate) > max_chars:  # 単一段落が上限超過でも段落自体は分割しない
            chunks.append(Chunk(current, len(current)))
            current = para
        else:
            current = candidate
    if current: chunks.append(Chunk(current, len(current)))
    return chunks

if __name__ == "__main__":
    sample = "背景を整理する。\n\n根拠を確認する。\n\n結論と次の行動を書く。"
    chunks = split(sample)
    assert "".join(c.text.replace("\n\n", "") for c in chunks) == sample.replace("\n\n", "")
    assert all(c.measured_tokens is None for c in chunks)
    print(chunks)

"""API料金と人手費を合わせたTCO試算（利用数は架空）。"""
from dataclasses import dataclass

@dataclass
class Usage:
    input_tokens: int
    output_tokens: int
    human_hours: float
    hourly_usd: float
    tool_usd: float
    successes: int

RATES = {2026: (0.75, 3.75), 2027: (1.50, 7.50)}  # USD / 1M tokens

def cost(year: int, usage: Usage) -> dict:
    values = (usage.input_tokens, usage.output_tokens, usage.human_hours, usage.hourly_usd, usage.tool_usd)
    if usage.successes <= 0 or any(value < 0 for value in values):
        raise ValueError("usage values must be non-negative and successes positive")
    input_rate, output_rate = RATES[year]
    # output_tokens は思考トークン込み。思考分を別加算しない。
    api = usage.input_tokens / 1_000_000 * input_rate + usage.output_tokens / 1_000_000 * output_rate
    labor = usage.human_hours * usage.hourly_usd
    total = api + labor + usage.tool_usd
    return {"api_usd": api, "labor_usd": labor, "tool_usd": usage.tool_usd,
            "total_usd": total, "per_success_usd": total / usage.successes}

if __name__ == "__main__":
    fictional = Usage(2_000_000, 400_000, 6.0, 35.0, 12.0, 8)
    c26, c27 = cost(2026, fictional), cost(2027, fictional)
    assert round(c26["api_usd"], 2) == 3.00
    assert c27["api_usd"] == c26["api_usd"] * 2
    assert c26["per_success_usd"] == c26["total_usd"] / 8
    try: cost(2026, Usage(1, 1, 0, 0, 0, 0))
    except ValueError: pass
    else: raise AssertionError("zero successes must fail")
    print({"fictional_usage": fictional, "2026": c26, "2027+": c27})

"""固定fixtureから完了率と遅延p95を再現可能に計算する。"""
import math

FIXTURES = [
    {"id": "A", "done": True, "latency_ms": 120},
    {"id": "B", "done": True, "latency_ms": 180},
    {"id": "C", "done": False, "latency_ms": 500},
    {"id": "D", "done": True, "latency_ms": 210},
    {"id": "E", "done": True, "latency_ms": 260},
    {"id": "F", "done": True, "latency_ms": 310},
    {"id": "G", "done": True, "latency_ms": 150},
    {"id": "H", "done": True, "latency_ms": 230},
    {"id": "I", "done": False, "latency_ms": 450},
    {"id": "J", "done": True, "latency_ms": 280},
]

def percentile_nearest_rank(values: list[int], p: float) -> int:
    if not values or not 0 < p <= 1:
        raise ValueError("values required and p must be in (0, 1]")
    ordered = sorted(values)
    rank = max(1, math.ceil(len(ordered) * p))
    return ordered[rank - 1]

def evaluate(rows: list[dict]) -> dict:
    if not rows:
        raise ValueError("rows must not be empty")
    return {"completion_rate": sum(r["done"] for r in rows) / len(rows),
            "latency_p95_ms": percentile_nearest_rank([r["latency_ms"] for r in rows], .95)}

if __name__ == "__main__":
    result = evaluate(FIXTURES)
    assert result == {"completion_rate": 0.8, "latency_p95_ms": 500}
    try: evaluate([])
    except ValueError: pass
    else: raise AssertionError("empty fixture must fail")
    print(result)

"""防御パッチを適用せず、承認可否だけ判定するfail-closed例。"""
REQUIRED = {"tests_pass", "rollback_ready", "reviewed"}

def decide(evidence: dict) -> dict:
    missing = sorted(k for k in REQUIRED if evidence.get(k) is not True)
    diff = evidence.get("diff_lines")
    valid_diff = type(diff) is int and 0 <= diff <= 200
    risky = evidence.get("risk") not in {"low", "medium"} or not valid_diff
    approved = not missing and not risky
    return {"approved": approved,
            "reason": "all gates passed" if approved else f"blocked: missing={missing}, risky={risky}"}

def is_fail_closed(evidence: dict) -> bool:
    return decide(evidence)["approved"] is False

if __name__ == "__main__":
    incomplete = {"tests_pass": True, "rollback_ready": True, "risk": "low", "diff_lines": 20}
    safe = {"tests_pass": True, "rollback_ready": True, "reviewed": True, "risk": "medium", "diff_lines": 80}
    assert is_fail_closed(incomplete)
    assert decide({**safe, "risk": "unknown"})["approved"] is False
    assert all(not decide({**safe, "diff_lines": bad})["approved"] for bad in (-1, "10", None))
    assert decide(safe)["approved"] is True
    print({"incomplete": decide(incomplete), "complete": decide(safe),
           "note": "判定のみ。パッチ適用処理は含まない"})

"""導入KPIを閾値で判定し、次のアクションを返す。"""
TARGETS = {"weekly_active_rate": 0.60, "success_rate": 0.85,
           "hours_saved": 40.0, "critical_incidents": 0}

def judge(actual: dict) -> dict:
    checks = {
        "weekly_active_rate": actual["weekly_active_rate"] >= TARGETS["weekly_active_rate"],
        "success_rate": actual["success_rate"] >= TARGETS["success_rate"],
        "hours_saved": actual["hours_saved"] >= TARGETS["hours_saved"],
        "critical_incidents": actual["critical_incidents"] <= TARGETS["critical_incidents"],
    }
    passed = sum(checks.values())
    action = ("pause" if actual["critical_incidents"] > 0 else
              "scale" if passed == 4 else "improve" if passed >= 2 else "pause")
    return {"checks": checks, "passed": passed, "action": action}

if __name__ == "__main__":
    pilot = {"weekly_active_rate": .72, "success_rate": .88,
             "hours_saved": 46, "critical_incidents": 0}
    weak = {"weekly_active_rate": .30, "success_rate": .70,
            "hours_saved": 12, "critical_incidents": 1}
    assert judge(pilot)["action"] == "scale"
    assert judge({**pilot, "critical_incidents": 1})["action"] == "pause"
    assert judge(weak)["action"] == "pause"
    print({"pilot": judge(pilot), "weak": judge(weak)})