EmbeddingGemma 2 发布：多模态本地检索，先验证内存口径与召回

[
  {
    "id": "d1",
    "title": "导出筛选结果",
    "text": "导出 CSV 时只包含当前筛选后的记录，保留界面排序。"
  },
  {
    "id": "d2",
    "title": "导出完整备份",
    "text": "完整备份包含所有记录，与当前筛选和排序无关。"
  },
  {
    "id": "d3",
    "title": "删除恢复",
    "text": "误删记录可在回收站保留期内恢复；清空后不能恢复。"
  },
  {
    "id": "d4",
    "title": "离线搜索",
    "text": "已建立本地索引的文档可在断网后搜索；新文件需先完成索引。"
  },
  {
    "id": "d5",
    "title": "切换外观",
    "text": "深色和浅色主题可以手动切换，跟随系统是另一选项。"
  },
  {
    "id": "d6",
    "title": "附件限制",
    "text": "上传附件之前检查大小与文件格式；上传失败不代表原文件已删除。"
  }
]

[
  {
    "id": "q1",
    "text": "我只想下载现在看到的几条记录",
    "relevant": [
      "d1"
    ]
  },
  {
    "id": "q2",
    "text": "筛选以后做全量备份会漏数据吗",
    "relevant": [
      "d2"
    ]
  },
  {
    "id": "q3",
    "text": "刚刚误删的内容还能找回来吗",
    "relevant": [
      "d3"
    ]
  },
  {
    "id": "q4",
    "text": "没有网络还能查已经收录的文件吗",
    "relevant": [
      "d4"
    ]
  },
  {
    "id": "q5",
    "text": "下载当前筛选结果，但不是完整备份",
    "relevant": [
      "d1"
    ]
  },
  {
    "id": "q6",
    "text": "如何开启多人视频会议",
    "relevant": []
  }
]

{
  "state": "NOT_RUN",
  "model_id": "google/embeddinggemma-2",
  "model_revision": null,
  "runtime_versions": {},
  "device": null,
  "os": null,
  "precision": null,
  "active_modalities": [],
  "dimension": null,
  "normalized": null,
  "query_prompt": null,
  "document_format": null,
  "corpus_sha256": null,
  "queries_sha256": null,
  "actual_download_bytes": null,
  "weight_files_bytes": null,
  "process_peak_memory_bytes": null,
  "accelerator_peak_memory_bytes": null,
  "cold_start_ms": null,
  "query_e2e_ms": [],
  "index_build_ms": null,
  "top3_by_query": {},
  "errors": [],
  "notes": "所有 null 均未测，不是 0；请记录采样工具、边界、批量和缓存状态。"
}

"""Standard-library validation only. Does not load or evaluate any model."""
import argparse
import json
import math
from pathlib import Path


def validate(data, dim, tolerance=1e-4):
    if not isinstance(data, dict):
        raise ValueError("root must be an object")
    count = 0
    for name in ("query_vectors", "document_vectors"):
        rows = data.get(name)
        if not isinstance(rows, list) or not rows:
            raise ValueError(name + " must be a non-empty array")
        for i, row in enumerate(rows):
            label = f"{name}[{i}]"
            if not isinstance(row, list) or len(row) != dim:
                raise ValueError(label + " has wrong dimension")
            if any(type(x) not in (int, float) for x in row):
                raise ValueError(label + " contains non-numeric values")
            try:
                finite = all(math.isfinite(x) for x in row)
                norm = math.hypot(*row)
            except (OverflowError, TypeError):
                raise ValueError(label + " has unsupported numeric range")
            if not finite or not math.isfinite(norm):
                raise ValueError(label + " contains non-finite values")
            if abs(norm - 1.0) > tolerance:
                raise ValueError(label + " is not L2 normalized")
            count += 1
    return count


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("file", type=Path)
    parser.add_argument("--dim", required=True, type=int)
    parser.add_argument("--synthetic", action="store_true")
    args = parser.parse_args()
    if args.dim <= 0:
        parser.error("dimension must be positive")
    data = json.loads(args.file.read_text(encoding="utf-8"))
    if args.synthetic:
        if data.get("kind") != "synthetic":
            parser.error("synthetic input must have kind=synthetic")
    elif args.dim not in (128, 256, 512, 768):
        parser.error("model output requires 128/256/512/768 dimensions")
    try:
        count = validate(data, args.dim)
    except ValueError as exc:
        parser.exit(1, "FAIL: " + str(exc) + "\n")
    print(f"PASS: {count} vectors; dim={args.dim}; numeric/finite/unit-norm")
    print("NOT TESTED: model inference, retrieval quality, memory, latency")

{"kind":"synthetic","query_vectors":[[1,0,0,0]],"document_vectors":[[0,1,0,0],[0.6,0.8,0,0]]}

python3 verify_vectors.py synthetic.json --dim 4 --synthetic

{
  "kind": "synthetic-not-model-output",
  "source_dim": 768,
  "positions": [0, 1, 128],
  "query": [0.6, 0, 0.8],
  "documents": [
    {"id": "d-front", "values": [0.8, 0.6, 0]},
    {"id": "d-tail", "values": [0.48, 0.64, 0.6]},
    {"id": "d-decoy", "values": [0.6, 0, -0.8]}
  ]
}

"""Offline synthetic arithmetic. No model, network or approximate index."""
import json
import math
from pathlib import Path

DIMS = (128, 256, 512, 768)


def expand(values, positions, dim):
    if type(dim) is not int or dim != 768:
        raise ValueError("source_dim")
    if (not isinstance(positions, list) or not positions
            or len(positions) != len(values)
            or any(type(p) is not int or not 0 <= p < dim for p in positions)
            or len(set(positions)) != len(positions)):
        raise ValueError("positions")
    row = [0.0] * dim
    for p, v in zip(positions, values):
        row[p] = v
    return row


def prepare(row, dim):
    if type(dim) is not int or dim not in DIMS:
        raise ValueError("target_dim")
    if not isinstance(row, list) or len(row) != 768:
        raise ValueError("source_shape")
    if any(type(x) not in (int, float) for x in row):
        raise ValueError("numeric_type")
    try:
        if not all(math.isfinite(x) for x in row):
            raise ValueError("non_finite")
        norm = math.hypot(*row)
    except OverflowError:
        raise ValueError("numeric_range") from None
    if not math.isfinite(norm) or abs(norm - 1) > 1e-9:
        raise ValueError("source_not_unit")
    prefix = row[:dim]
    norm = math.hypot(*prefix)
    if norm < 1e-12:
        raise ValueError("prefix_too_small")
    return [x / norm for x in prefix]


def dot(a, b):
    if len(a) != len(b):
        raise ValueError("score_shape")
    return math.fsum(x * y for x, y in zip(a, b))


def score_rows(q, docs):
    return sorted(((name, dot(q, v)) for name, v in docs),
                  key=lambda pair: (-pair[1], pair[0]))


def rank(q, docs, dim):
    ids = [name for name, _ in docs]
    if not ids or any(type(n) is not str or not n for n in ids):
        raise ValueError("document_id")
    if len(set(ids)) != len(ids):
        raise ValueError("duplicate_id")
    # Prepare the whole batch first; no silently dropped bad document.
    nq = prepare(q, dim)
    nd = [(name, prepare(v, dim)) for name, v in docs]
    return score_rows(nq, nd)


def load_fixture():
    f = json.loads(Path(__file__).with_name("fixture.json").read_text())
    if f["kind"] != "synthetic-not-model-output":
        raise ValueError("fixture_kind")
    def unpack(v):
        return expand(v, f["positions"], f["source_dim"])
    return unpack(f["query"]), [(d["id"], unpack(d["values"]))
                                for d in f["documents"]]


def shown(rows):
    # Round only for reporting; ranking used unrounded scores.
    return [[name, round(score, 6)] for name, score in rows]


def experiment():
    q, docs = load_fixture()
    full = rank(q, docs, 768)
    short = rank(q, docs, 128)
    # Explicitly wrong branch: the validated unit vectors are sliced,
    # but their prefixes are NOT normalized before dot-product scoring.
    wrong = score_rows(q[:128], [(n, v[:128]) for n, v in docs])
    reranked = {}
    for budget in (1, 2, 3):
        selected = {name for name, _ in short[:budget]}
        reranked[str(budget)] = shown(rank(
            q, [(n, v) for n, v in docs if n in selected], 768))
    return {"full_768": shown(full), "cosine_128": shown(short),
            "wrong_raw_128": shown(wrong), "reranked": reranked}


if __name__ == "__main__":
    print(json.dumps(experiment(), indent=2))

{
  "full_768": [["d-tail", 0.768], ["d-front", 0.48], ["d-decoy", -0.28]],
  "cosine_128": [["d-decoy", 1.0], ["d-front", 0.8], ["d-tail", 0.6]],
  "wrong_raw_128": [["d-front", 0.48], ["d-decoy", 0.36], ["d-tail", 0.288]],
  "reranked": {
    "1": [["d-decoy", -0.28]],
    "2": [["d-front", 0.48], ["d-decoy", -0.28]],
    "3": [["d-tail", 0.768], ["d-front", 0.48], ["d-decoy", -0.28]]
  }
}

import json
import math
from fractions import Fraction as F
from pathlib import Path
from rank_lab import (dot, experiment, expand, load_fixture, prepare,
                      rank, shown, score_rows)

checks = 0

def check(condition, label):
    global checks
    if not condition:
        raise AssertionError(label)
    checks += 1


def rejected(code, fn):
    try:
        fn()
    except ValueError as e:
        check(str(e) == code, (code, str(e)))
    else:
        raise AssertionError("accepted: " + code)


expected = json.loads(Path(__file__).with_name("expected.json").read_text())
# Independent analytic oracle: exact rationals, no rank/dot/prepare call.
# Coordinates are hand-selected; these are not sampled embeddings.
check(F(3,5)**2 + F(4,5)**2 == 1, "query unit identity")
check(F(12,25)**2 + F(16,25)**2 + F(3,5)**2 == 1,
      "tail unit identity")
full_scores = {"d-tail": F(3,5)*F(12,25) + F(4,5)*F(3,5),
               "d-front": F(3,5)*F(4,5),
               "d-decoy": F(3,5)*F(3,5) - F(4,5)*F(4,5)}
short_scores = {"d-decoy": F(1), "d-front": F(4,5), "d-tail": F(3,5)}
raw_scores = {"d-front": F(12,25), "d-decoy": F(9,25), "d-tail": F(36,125)}
for key, exact in (("full_768", full_scores), ("cosine_128", short_scores),
                   ("wrong_raw_128", raw_scores)):
    check({name: F(str(score)) for name, score in expected[key]} == exact,
          "independent rational oracle: " + key)
q_raw, docs_raw = load_fixture()
for rows, oracle in ((rank(q_raw, docs_raw, 768), full_scores),
                     (rank(q_raw, docs_raw, 128), short_scores),
                     (score_rows(q_raw[:128], [(n, v[:128]) for n, v in docs_raw]), raw_scores)):
    for name, score in rows:
        check(math.isclose(score, float(oracle[name]), rel_tol=0, abs_tol=1e-12),
              "unrounded rational score: " + name)
actual = experiment()
check(actual == expected, "fixed hand-computed oracle")
q, docs = load_fixture()
check(actual["wrong_raw_128"] != expected["cosine_128"], "negative control")
check(actual["reranked"]["2"][0][0] != "d-tail", "missing candidate remains missing")
check(actual["reranked"]["3"] == expected["full_768"], "wider shortlist recovers")
check(shown(rank(q, list(reversed(docs)), 128)) == expected["cosine_128"], "row order")
check(shown(rank(q, docs, 256)) == expected["full_768"], "retain informative coordinate")
check(shown(rank(q, docs, 512)) == expected["full_768"], "512 arithmetic")
check(rank(q, [("z", docs[0][1]), ("a", docs[0][1])], 128)[0][0] == "a", "tie break")
rejected("duplicate_id", lambda: rank(q, [docs[0], docs[0]], 128))
rejected("source_shape", lambda: prepare(q[:-1], 128))
rejected("target_dim", lambda: prepare(q, True))
rejected("target_dim", lambda: prepare(q, 127))
rejected("numeric_type", lambda: prepare([True] + q[1:], 128))
rejected("non_finite", lambda: prepare([float("nan")] + q[1:], 128))
rejected("non_finite", lambda: prepare([float("inf")] + q[1:], 128))
rejected("source_not_unit", lambda: prepare([2.0] + [0.0] * 767, 128))
rejected("source_not_unit", lambda: prepare([0.0] * 768, 128))
rejected("score_shape", lambda: dot([1], [1, 0]))
rejected("positions", lambda: expand([1, 0], [0, 0], 768))
zero_prefix = expand([0, 0, 1], [0, 1, 128], 768)
rejected("prefix_too_small", lambda: rank(q, docs + [("d-zero", zero_prefix)], 128))
rejected("prefix_too_small", lambda: rank(zero_prefix, docs, 128))
recovered = shown(rank(q, docs + [("d-zero", zero_prefix)], 256))
check(recovered == [["d-zero", 0.8], ["d-tail", 0.768],
                    ["d-front", 0.48], ["d-decoy", -0.28]], "whole-batch dimension recovery")
# Numerical-policy boundaries: this lab accepts unit source vectors only.
rejected("numeric_range", lambda: prepare([10**1000] + q[1:], 128))
rejected("source_not_unit", lambda: prepare([1e308] + q[1:], 128))
rejected("source_not_unit", lambda: prepare([1.7e308, 1.7e308] + [0.0]*766, 128))
rejected("non_finite", lambda: prepare(q[:128] + [float("nan")] + q[129:], 128))
rejected("non_finite", lambda: prepare(q[:128] + [float("-inf")] + q[129:], 128))
rejected("prefix_too_small", lambda: prepare(expand([5e-324, 0, 1], [0, 1, 128], 768), 128))
rejected("prefix_too_small", lambda: prepare(expand([0.5e-12, 0, 1], [0, 1, 128], 768), 128))
check(prepare(expand([1e-12, 0, 1], [0, 1, 128], 768), 128)[0] == 1,
      "exact prefix threshold accepted")
check(rank(q, [("a", docs[0][1]), ("z", docs[0][1])], 128)[0][0] == "a",
      "tie reverse-order independence")
near_docs = [(name, expand([score, math.sqrt(1-score*score), 0], [0,1,128], 768))
             for name, score in (("a", 0.8), ("b", 0.80000001))]
check(shown(rank(expand([1,0,0], [0,1,128], 768), near_docs, 128)) ==
      [["b", 0.8], ["a", 0.8]], "rank before reporting round")
print(f"PASS {checks} checks; fixed oracle, negative control, rejection and recovery")
print("NOT TESTED: model inference, semantic relevance, ANN, latency, memory")

"""Run real source mutants in isolated directories, with and without -O."""
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent
SOURCE = (ROOT / "rank_lab.py").read_text()
MUTANTS = {
    "hidden_score_drift": (
        "return math.fsum(x * y for x, y in zip(a, b))",
        "return math.fsum(x * y for x, y in zip(a, b)) + 1e-8",
        "unrounded rational score"),
    "round_before_ranking": (
        "key=lambda pair: (-pair[1], pair[0])",
        "key=lambda pair: (-round(pair[1], 6), pair[0])", "rank before reporting round"),
    "missing_renormalization": (
        "return [x / norm for x in prefix]", "return prefix",
        "unrounded rational score"),
    "truncation_ignored": (
        "prefix = row[:dim]", "prefix = row[:]",
        "unrounded rational score"),
    "reranker_sees_excluded_documents": (
        "for n, v in docs if n in selected", "for n, v in docs",
        "fixed hand-computed oracle"),
    "unstable_ties": (
        "key=lambda pair: (-pair[1], pair[0])", "key=lambda pair: -pair[1]",
        "tie break"),
    "silent_shape_mismatch": (
        'raise ValueError("score_shape")', 'pass', "accepted: score_shape"),
    "silent_zero_prefix": (
        'raise ValueError("prefix_too_small")', 'return prefix',
        "accepted: prefix_too_small"),
    "nonfinite_silent_zero": (
        'raise ValueError("non_finite")', 'return [0.0] * dim',
        "accepted: non_finite"),
}


def run(directory, optimized):
    command = [sys.executable] + (["-O"] if optimized else []) + ["check_rank.py"]
    return subprocess.run(command, cwd=directory, capture_output=True, text=True)


def main():
    killed = 0
    for optimized in (False, True):
        mode = "-O" if optimized else "normal"
        baseline = run(ROOT, optimized)
        if baseline.returncode != 0:
            raise RuntimeError("baseline failed: " + baseline.stderr)
        print(f"BASELINE {mode}: " + baseline.stdout.splitlines()[0])
        for name, (old, new, diagnostic) in MUTANTS.items():
            if SOURCE.count(old) != 1:
                raise RuntimeError("mutation anchor is not unique: " + name)
            with tempfile.TemporaryDirectory() as tmp:
                directory = Path(tmp)
                for filename in ("fixture.json", "expected.json", "check_rank.py"):
                    shutil.copyfile(ROOT / filename, directory / filename)
                (directory / "rank_lab.py").write_text(SOURCE.replace(old, new))
                result = run(directory, optimized)
                # A syntax/import/crash failure is not counted as a killed mutant.
                final_line = result.stderr.strip().splitlines()[-1] if result.stderr else ""
                if (result.returncode == 0 or
                        not final_line.startswith("AssertionError:") or
                        diagnostic not in final_line):
                    raise RuntimeError(f"mutant not caught by intended check: {name} {mode}\n"
                                       + result.stdout + result.stderr)
                print(f"KILLED {name} {mode}: {final_line}")
                killed += 1
    print(f"PASS {killed} mutation runs; nine actual source changes rejected in both modes")


if __name__ == "__main__":
    main()