📖
EmbeddingGemmaでベクトル検索をやってみる

2025/09/07に公開
 はじめにEmbeddingGemma が発表されました。
テキストを固定長ベクトルに変換できる埋め込みモデルで、類似度検索やレコメンド、クラスタリングなどに向いています。詳細はブログ記事や公式の解説をご参照ください。
https://developers.googleblog.com/en/introducing-embeddinggemma/
https://qiita.com/youtoy/items/20928c270fe2fc0b2793

 試してみることEmbeddingGemmaを使って、Zennに投稿している自分の記事一覧をRSSで取得し、簡素なベクトル検索を試します。

 機能概要検索ワードを指定
Zennの自分のアカウントから「タイトル／URL／サマリー」を取得
タイトル60%＋RSSサマリ40%の重みで一致度を計算
上位3件を表示

 実行Hugging FaceでEmbeddingGemmaを使うため、アクセストークン（API Key）を準備してください。
https://zenn.dev/protoout/articles/73-hugging-face-setup
Google Colab で実行できるノートブックを用意しました。
https://colab.research.google.com/drive/1Du2ZtBX-5VfNSkoBSz5t9O4pkE9eGCOX?usp=sharing
コード!pip install sentence-transformers faiss-cpu beautifulsoup4 feedparser tqdm
# hugging faceのAPIKEY
from huggingface_hub import login

# 自分のAPIKEYを入力（環境変数での設定を推奨）
login(token="xxxxxxxxxxxxxx") 
print("HF login ok")

from __future__ import annotations

# ===== 依存・型 =====
from sentence_transformers import SentenceTransformer
import feedparser, requests, bs4, re, numpy as np, torch
from typing import List, Dict, Tuple

# ===== ユーザー設定 =====
USER = "hikaruy"
FEED_URL = f"https://zenn.dev/{USER}/feed"
QUERY = "causal impactについて"

TOP_K = 3                       # 上位何件を返すか
WEIGHT_VALUE = 0.6              # 本文ベクトルの重み
WEIGHT_SUMMARY = 0.4            # サマリベクトルの重み
REQUEST_TIMEOUT = 20            # HTTPタイムアウト（秒）
USER_AGENT = (                  # シンプルなUA（ブロック回避用）
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
)

# ===== モデルのロード =====
# - GPUがあれば自動でCUDAを利用
# - SentenceTransformerでEmbeddingGemmaを読み込み
model_id = "google/embeddinggemma-300m"
device = "cuda" if torch.cuda.is_available() else "cpu"
model = SentenceTransformer(model_id).to(device)
print("Embedding dim:", model.get_sentence_embedding_dimension(), "| device:", device)

# ===== ユーティリティ群 =====
def _normalize(x: np.ndarray) -> np.ndarray:
    """L2正規化（数値安定性のための微小値付与）"""
    norm = np.linalg.norm(x, axis=-1, keepdims=True) + 1e-12
    return x / norm

def reweight_pairwise(w_value: float, w_summary: float) -> Tuple[float, float]:
    """2つの重みを和が1になるよう正規化（両方0ならvalue=1,summary=0）"""
    s = max(w_value + w_summary, 0.0)
    return (1.0, 0.0) if s == 0 else (w_value / s, w_summary / s)

def encode_texts(model: SentenceTransformer, texts: List[str]) -> np.ndarray:
    """テキスト群を埋め込みベクトル（float32, 正規化済み）へ変換"""
    with torch.inference_mode():
        emb = model.encode(
            texts,
            convert_to_numpy=True,
            normalize_embeddings=True,  # ここでL2正規化
            show_progress_bar=False,
        )
    return emb.astype("float32")

def fetch_clean_text(url: str) -> str:
    """URLを取得→HTMLをプレーンテキストへ整形（script/style除去・空白圧縮）"""
    r = requests.get(url, timeout=REQUEST_TIMEOUT, headers={"User-Agent": USER_AGENT})
    r.raise_for_status()
    soup = bs4.BeautifulSoup(r.text, "html.parser")
    for bad in soup(["script", "style", "noscript"]):
        bad.decompose()
    text = soup.get_text("\n")
    text = re.sub(r"\n{2,}", "\n", text)
    text = re.sub(r"[ \t]{2,}", " ", text)
    return text.strip()

def make_summary(text: str, limit: int = 200) -> str:
    """1行要約（全角・半角空白をまとめ、指定文字数で省略）"""
    t = re.sub(r"\s+", " ", text or "").strip()
    return (t[:limit] + "…") if len(t) > limit else t

# ===== データ生成 =====
def build_items_from_feed(entries) -> List[Dict[str, str]]:
    """
    RSSエントリから検索対象レコードを作成
    - base   : タイトルとURL（表示用）
    - value  : 記事本文（HTML除去後のプレーンテキスト）
    - summary: RSS要約（最大800文字）
    """
    items: List[Dict[str, str]] = []
    for e in entries:
        title = getattr(e, "title", "").strip()
        link = getattr(e, "link", "").strip()
        summ = (getattr(e, "summary", "") or getattr(e, "description", "")).strip()
        base = f"{title} | {link}".strip()

        # 本文取得（失敗時はsummary→titleの順でフォールバック）
        body = ""
        if link:
            try:
                body = fetch_clean_text(link)
            except Exception as ex:
                print("skip:", link, ex)
        if not body:
            body = summ or title

        items.append({"base": base, "value": body, "summary": summ[:800]})
    return items

# ===== 検索本体 =====
def search_by_value_and_summary(
    items: List[Dict[str, str]],
    query: str,
    model: SentenceTransformer,
    top_k: int = 1,
    w_value: float = 0.6,
    w_summary: float = 0.4,
    return_score: bool = True,
):
    """
    value（本文）とsummary（RSS要約）を重み付きで合成した埋め込みで近似検索
    - 手順
      1) valueとsummaryを埋め込み
      2) 正規化された重みで加算し、再度正規化
      3) クエリ埋め込みと内積（コサイン類似度と等価）でランキング
    - 返り値
      return_score=Trueなら (base, value, summary, score) のタプルを上位top_k件
    """
    bases = [str(it.get("base", "")) for it in items]
    values = [str(it.get("value", "")) for it in items]
    summaries = [str(it.get("summary", "")) for it in items]

    wv, ws = reweight_pairwise(w_value, w_summary)
    emb_value = encode_texts(model, values)
    emb_summary = encode_texts(model, summaries)
    emb_items = _normalize(wv * emb_value + ws * emb_summary).astype("float32")
    emb_query = encode_texts(model, [query])[0]

    scores = emb_items @ emb_query
    order = np.argsort(-scores)[:max(1, top_k)]

    results = []
    for i in order:
        tup = (bases[i], values[i], summaries[i])
        results.append((*tup, float(scores[i])) if return_score else tup)
    return results

# ===== 実行（エントリポイント） =====
# 実行時の流れ：
#  1) RSSを取得 → エントリ一覧を得る
#  2) 各記事の本文を取得してrecordsを構築
#  3) クエリに対して類似度検索
#  4) 上位K件を出力（要約は本文から生成）
if __name__ == "__main__":
    # 1) RSS取得
    feed = feedparser.parse(FEED_URL)
    entries = getattr(feed, "entries", []) or []
    print(f"記事数: {len(entries)}")

    # 2) レコード構築
    items = build_items_from_feed(entries)
    print(f"作成レコード数: {len(items)}")

    # 3) 重みの正規化 → 検索実行
    W_VALUE, W_SUMMARY = reweight_pairwise(WEIGHT_VALUE, WEIGHT_SUMMARY)
    hits = search_by_value_and_summary(
        items,
        QUERY,
        model=model,
        top_k=TOP_K,
        w_value=W_VALUE,
        w_summary=W_SUMMARY,
        return_score=True,
    )

    # 4) 結果表示
    print("=== Hits ===")
    for rank, (base, value, summary, score) in enumerate(hits, 1):
        print(f"{rank}番目")
        print("Base     :", base)
        print("Score    :", round(score, 3))
        print("[Summary]", make_summary(value, 200))
        print("--------")



 検索ワードQUERY = "causal impactについて"

 結果Embedding dim: 768 | device: cpu
記事数: 9
作成レコード数: 9
=== Hits ===
1番目
Base     : Causal impactを調べてみた。 | https://zenn.dev/hikaruy/articles/4502ebadea0d09
Score    : 0.408
[Summary] Causal impactを調べてみた。 Zenn ヤマシタヒカル 👿 Causal impactを調べてみた。 2024/10/27 に公開 2024/10/28 did ベイズ 因果推論 causalimact tech はじめに これまで手持ちデータでPythonライブラリでCausal imactを使っていましたが、モデルの中身を調べず使っていたので、今回、論文や参考サイトなどで調べてみま…
--------
2番目
Base     : もう一度、強化学習を理解する | https://zenn.dev/hikaruy/articles/09cfeb1a166cd6
Score    : 0.384
[Summary] もう一度、強化学習を理解する Zenn ヤマシタヒカル 🖖 もう一度、強化学習を理解する 2025/01/18 に公開 2025/01/26 強化学習 LLM ppo tech はじめに 強化学習はLLMのRLHFや、ロボット制御、自動運転技術などで活用されています。 以前、私が書いた DPOの記事 で触れたPPOの実装を試そうと思いましたが、強化学習は機械学習や深層学習と比べて抽象的で分かりづら…
--------
3番目
Base     : Delta Learning Hypothesis（デルタ学習仮説）をやってみた | https://zenn.dev/hikaruy/articles/667d3dad4209bc
Score    : 0.38
[Summary] Delta Learning Hypothesis（デルタ学習仮説）をやってみた Zenn ヤマシタヒカル 📌 Delta Learning Hypothesis（デルタ学習仮説）をやってみた 2025/07/25 に公開 2025/07/27 LLM dpo Gemma 3 orpo deltalearning tech はじめに 「デルタ学習仮説（Delta Learning Hypothes…
--------

 まとめ狙った用語で検索はできましたが、スコア差はそこまで大きくありませんでした。モデルが軽量なので、対象件数が少なければローカルでも試せます。重み付けや文分割（チャンク）の工夫、サマリー生成の質などを詰めるともう少し精度が上がりそうです。

 おわりに今回は Zenn の自分の記事一覧を取得して簡単な検索を行いました。スクレイピングやクローリングを禁止しているサイトもあります。robots.txt や利用規約を必ず確認し、サーバーに負荷をかけないよう配慮しましょう。
はじめに

試してみること

機能概要

実行

検索ワード

結果

まとめ

おわりに

Discussion
