AIP-C01 03: RAG・埋め込み・検索

03_rag_vector_search.livemd

Ryo Wakabayashi

@RyoWakabayashi

elixir-learning

Share to X

Share to Bluesky

More notebooks

AIP-C01 03: RAG・埋め込み・検索

Mix.install([
  {:aws, "~> 1.0.15"},
  {:hackney, "~> 4.7"},
  {:kino, "~> 0.19"},
  {:kino_vega_lite, "~> 0.1"}
])

概要

最終確認日: 2026-09-22

対応範囲: 1.3, 1.4, 1.5, 3.1, 4.2, 5.1, 5.2

このノートで身につけること

チャンク、埋め込み、ベクトル検索、リランキング、生成の流れを実装する
fixed / hierarchical / semantic chunking を使い分ける
semantic / keyword / hybrid search とメタデータフィルターを比較する
検索品質と生成品質を分離して評価する

接続

access_key_input = Kino.Input.password("AWS_ACCESS_KEY_ID")
secret_key_input = Kino.Input.password("AWS_SECRET_ACCESS_KEY")
session_token_input = Kino.Input.password("AWS_SESSION_TOKEN（一時認証の場合）")
region_input = Kino.Input.text("AWS_REGION", default: "ap-northeast-1")

Kino.Layout.grid([access_key_input, secret_key_input, session_token_input, region_input], columns: 2)

access_key = Kino.Input.read(access_key_input)
secret_key = Kino.Input.read(secret_key_input)
session_token = Kino.Input.read(session_token_input)
region = Kino.Input.read(region_input)

client =
  if session_token in [nil, ""] do
    AWS.Client.create(access_key, secret_key, region)
  else
    AWS.Client.create(access_key, secret_key, session_token, region)
  end

ハンズオン1: チャンクの粒度を観察する

document = """
Amazon Bedrock は基盤モデルを利用するためのマネージドサービスです。
Converse API は対応モデル間で共通のメッセージ形式を提供します。
Knowledge Bases はデータの取り込み、埋め込み、検索、生成を支援します。
Guardrails は入力と出力に安全ポリシーを適用します。
CloudWatch では呼び出し数、レイテンシー、トークン、エラーを監視できます。
"""

chunk_words = fn text, size, overlap ->
  words = String.split(text)
  step = max(size - overlap, 1)

  words
  |> Enum.chunk_every(size, step, :discard)
  |> Enum.with_index()
  |> Enum.map(fn {chunk, index} -> %{id: index, text: Enum.join(chunk, " ")} end)
end

chunk_words.(document, 12, 3)
|> Kino.DataTable.new()

実際のトークン境界はモデルの tokenizer に依存します。このセルは「小さいチャンクは精密だが文脈が減る」「overlap は再現率を上げ得るが保存量と重複を増やす」ことを観察するための簡略版です。

ハンズオン2: Titan Text Embeddings V2 でベクトル化する

embed = fn text ->
  {:ok, body, _} =
    AWS.BedrockRuntime.invoke_model(
      client,
      "amazon.titan-embed-text-v2:0",
      %{
        "accept" => "application/json",
        "contentType" => "application/json",
        "inputText" => text,
        "dimensions" => 256,
        "normalize" => true
      },
      recv_timeout: 60_000
    )

  body["embedding"] || get_in(body, ["embeddingsByType", "float"])
end

documents = [
  %{id: "bedrock", text: "Amazon Bedrock は基盤モデルを API で利用するマネージドサービスです。", team: "platform"},
  %{id: "guardrails", text: "Bedrock Guardrails は有害コンテンツや PII を検出・制御します。", team: "security"},
  %{id: "cloudwatch", text: "CloudWatch はメトリクス、ログ、アラームを提供します。", team: "operations"}
]

indexed_documents =
  Enum.map(documents, fn doc -> Map.put(doc, :embedding, embed.(doc.text)) end)

ハンズオン3: cosine similarity で検索する

正規化済みベクトルの cosine similarity は内積で求められますが、ここでは一般形を実装します。

cosine_similarity = fn left, right ->
  dot = Enum.zip_with(left, right, &Kernel.*/2) |> Enum.sum()
  left_norm = :math.sqrt(Enum.map(left, &(&1 * &1)) |> Enum.sum())
  right_norm = :math.sqrt(Enum.map(right, &(&1 * &1)) |> Enum.sum())
  dot / (left_norm * right_norm)
end

search = fn query, filters, top_k ->
  query_embedding = embed.(query)

  indexed_documents
  |> Enum.filter(fn doc ->
    Enum.all?(filters, fn {key, value} -> Map.fetch!(doc, key) == value end)
  end)
  |> Enum.map(fn doc ->
    Map.put(doc, :score, cosine_similarity.(query_embedding, doc.embedding))
  end)
  |> Enum.sort_by(& &1.score, :desc)
  |> Enum.take(top_k)
  |> Enum.map(&Map.drop(&1, [:embedding]))
end

search_results = search.("個人情報を保護するには？", %{}, 2)

similarity_chart =
  VegaLite.new(width: 620, height: 240, title: "検索結果の cosine similarity")
  |> VegaLite.data_from_values(search_results)
  |> VegaLite.mark(:bar, tooltip: true)
  |> VegaLite.encode_field(:y, "id", type: :nominal, title: "document", sort: "-x")
  |> VegaLite.encode_field(:x, "score", type: :quantitative, title: "cosine similarity")
  |> Kino.VegaLite.new()

Kino.Layout.tabs([
  {"類似度", similarity_chart},
  {"検索結果", Kino.DataTable.new(search_results)}
])

search.("監視方法は？", %{team: "operations"}, 2)
|> Kino.DataTable.new()

メタデータフィルターは検索後の表示制御ではなく、原則として検索時の認可境界にも組み込みます。取得後に隠すだけでは、権限外データがモデルのコンテキストへ入る恐れがあります。

ハンズオン4: Retrieval と Generation を分離する

query = "生成 AI の入出力を安全にする AWS 機能は？"
contexts = search.(query, %{}, 2)

grounded_prompt = """
次の参照情報だけを使って質問に答えてください。
参照情報に答えがなければ「情報不足」と答えてください。
回答の末尾に参照した id を列挙してください。

<context>
#{Enum.map_join(contexts, "\n", &"[#{&1.id}] #{&1.text}")}
</context>

<question>#{query}</question>
"""

Kino.Text.new(grounded_prompt)

ここで検索結果が誤っていれば、強い生成モデルに変えても根拠は改善しません。検索失敗（recall / relevance）と生成失敗（faithfulness / correctness）を分離して診断します。

ハンズオン5: semantic / keyword / hybrid を評価セットで比較する

商品コード、エラーコード、固有名詞は keyword 検索が強く、言い換えは semantic 検索が強い傾向があります。どちらかを直感で選ばず、同一クエリに対する順位を Reciprocal Rank Fusion (RRF) で統合し、ACL フィルター後の Recall@k を比較します。

retrieval_fixture = [
  %{id: "d1", tenant: "alpha", text: "API の TooManyRequestsException は 429 を返す", semantic: 0.73, keyword: 1.00},
  %{id: "d2", tenant: "alpha", text: "呼び出し過多では指数バックオフを使う", semantic: 0.94, keyword: 0.25},
  %{id: "d3", tenant: "beta", text: "429 の上限緩和を申請する", semantic: 0.88, keyword: 0.80},
  %{id: "d4", tenant: "alpha", text: "認証失敗は再試行しない", semantic: 0.31, keyword: 0.10}
]

rank = fn rows, score_key ->
  rows
  |> Enum.sort_by(&Map.fetch!(&1, score_key), :desc)
  |> Enum.with_index(1)
  |> Map.new(fn {row, position} -> {row.id, position} end)
end

authorized = Enum.filter(retrieval_fixture, &(&1.tenant == "alpha"))
semantic_rank = rank.(authorized, :semantic)
keyword_rank = rank.(authorized, :keyword)

hybrid_results =
  authorized
  |> Enum.map(fn row ->
    rrf = 1 / (60 + semantic_rank[row.id]) + 1 / (60 + keyword_rank[row.id])
    Map.merge(row, %{semantic_rank: semantic_rank[row.id], keyword_rank: keyword_rank[row.id], rrf: rrf})
  end)
  |> Enum.sort_by(& &1.rrf, :desc)
  |> Enum.with_index(1)
  |> Enum.map(fn {row, position} -> Map.put(row, :hybrid_rank, position) end)

Kino.DataTable.new(hybrid_results)

relevant = MapSet.new(["d1", "d2"])

recall_at_2 = fn ranked_ids ->
  ranked_ids
  |> Enum.take(2)
  |> MapSet.new()
  |> MapSet.intersection(relevant)
  |> MapSet.size()
  |> Kernel./(MapSet.size(relevant))
end

comparison = [
  %{strategy: "semantic", recall_at_2: recall_at_2.(semantic_rank |> Enum.sort_by(&elem(&1, 1)) |> Enum.map(&elem(&1, 0)))},
  %{strategy: "keyword", recall_at_2: recall_at_2.(keyword_rank |> Enum.sort_by(&elem(&1, 1)) |> Enum.map(&elem(&1, 0)))},
  %{strategy: "hybrid RRF", recall_at_2: recall_at_2.(Enum.map(hybrid_results, & &1.id))}
]

VegaLite.new(width: 600, height: 230, title: "検索方式ごとの Recall@2")
|> VegaLite.data_from_values(comparison)
|> VegaLite.mark(:bar, tooltip: true)
|> VegaLite.encode_field(:x, "strategy", type: :nominal, title: "検索方式")
|> VegaLite.encode_field(:y, "recall_at_2", type: :quantitative, title: "Recall@2", scale: [domain: [0, 1]])
|> Kino.VegaLite.new()

この fixture では tenant=beta の文書が高得点でも、ランキング前に除外しています。検索後に ACL を適用すると、権限外の文書がモデル入力へ入り得るだけでなく、top-k の枠を消費して Recall も下がります。本番比較では、同じ評価セットで latency、Recall@k、MRR、フィルター後の候補数、rerank gain を記録します。

実務ケース: 8 TB / 4,000万ベクトルをどう選ぶか

次の順序で判断を書き出してください。

hard constraint: データ所在地、暗号化、tenant 分離、復旧目標
workload: ベクトル数、増分更新、QPS、p95 latency、hybrid / facet の要否
economics: 保存・検索・取り込み・運用人件費
experiment: 代表クエリと adversarial query で recall / latency / cost を比較

低頻度で保存コスト重視なら S3 Vectors、高 QPS・低レイテンシー・高度な hybrid search なら OpenSearch が候補になりますが、規模だけで自動的に決まりません。Aurora PostgreSQL は既存の関係データとの整合性、Neptune Analytics は関係探索が主要要件のときに比較します。

設計比較

要件	第一候補	理由
低頻度・大量ベクトル・コスト重視	S3 Vectors	永続ストレージコストを重視
高 QPS、低レイテンシー、hybrid / facet	OpenSearch	高度な検索と運用機能
既存 PostgreSQL とトランザクション統合	Aurora PostgreSQL + pgvector	既存データと近接
関係探索とベクトル検索	Neptune Analytics	グラフ関係を利用