ベクトル埋め込みとRAGを活用したチャット補完機能の改善

Red Hat AI Inference on IBM Cloud API のベクトル埋め込みを活用して、チャットモデルが独自のナレッジベースから正確で事実に基づいた回答を生成できる、検索強化生成（RAG）パイプラインを構築する方法をご紹介します。

大規模言語モデルは強力ですが、学習データ以外の事実については何も知りません。製品カタログ、お客様履歴、ドメインコーパスなど、特定の知識ベースに基づいて推論を行うモデルが必要な場合、クエリ実行時に適切な情報を抽出する手段が必要となります。このチュートリアルでは、RAGループの全プロセスを解説します。具体的には、ナレッジベースを埋め込み、それをベクトルインデックスに保存し、最も関連性の高い事実をチャット補完リクエストに組み込むことで、モデルの回答がデータに基づいたものになるようにします。

検索、クラスタリング、分類など、エンベディングのユースケースに関する詳細については、「エンベディングのユースケース」を参照してください。

目標

このチュートリアルでは、以下のタスクを実行します：

Red Hat AI Inference API への認証を行います。
granite-embedding-278m-multilingual モデルを使用して、ナレッジベースを埋め込みます。
埋め込みベクトルをメモリ内のベクトルインデックスに格納し、コサイン類似度を用いて検索する。
gpt-oss-120b のチャットモデルに対し、取得したコンテキストの有無の両方でクエリを実行し、精度の向上が実際にどの程度あるかを直接確認してみてください。

構築するアプリケーションのアーキテクチャは以下の通りです：

ナレッジベース内の各ファクトは、埋め込みベクトルに変換され、 MemoryStore に追加されます。
ユーザーの質問は、同じモデルを使用して埋め込まれています。
コサイン類似度を用いて、保存されているベクトルを質問ベクトルに対して順位付けし、上位3つの事実が抽出されます。
これらの事実は、チャットモデルに送信されるプロンプトに組み込まれます。
そのモデルは、地に足のついた答えを返す。

開始前に

以下のものが揃っていることを確認してください：

従量課金型またはサブスクリプション型の IBM Cloud アカウント。体験版アカウントご利用いただけません。詳細情報やアカウントのアップグレードについては、アカウントの種類をご覧ください。
Red Hat AI Inference のプロジェクトです。
Red Hat AI Inference サービスにおいて、「Writer」ロール以上。詳細については、「 IAM アクセスの管理」を参照してください。
Python 3.9 またはそれ以降のバージョンがワークステーションにインストールされていること。
openai の Python パッケージ。次のコマンドを実行してインストールしてください：
```
pip install openai
```

API認証の設定

Red Hat AI Inference API を呼び出すには、まずサービス ID と API キーを使用してリクエストの認証を行う必要があります。すでに「はじめに」のチュートリアルを完了している場合は、そこで作成したサービス ID と API キーを再利用して、「プロジェクト ID とエンドポイントの取得」のセクションに進むことができます。

サービスIDを作成し、アクセス権を割り当てる

サービス ID は、 Red Hat AI Inference プロジェクトへのアクセスを管理・割り当てる上で便利な手段です。サービス ID を作成し、そのサービス ID にプロジェクトへのアクセス権を割り当ててください。

IBM Cloud コンソールで、[管理]>[アクセス(IAM)]>[サービス ID ] の順に選択し、 [作成] をクリックします。
サービス ID の名前と説明を入力し、「作成」をクリックしてください。
サービスIDページで、「 アクセス権の割り当て 」をクリックします。
サービスとして Red Hat AI Inference をサービスとして選択します。
「リソース 」内で、「 特定のリソース 」を選択し、プロジェクトを選んでください。
「 ロールとアクション 」内で、サービスアクセスロールとして「 ライター 」を選択します。

APIへのアクセスには、プラットフォームアクセス権限は必要ありません。
アクセス概要を確認し、「 割り当て 」をクリックします。

API キーの作成

サービス ID がプロジェクトにアクセスできるようになったので、サービス ID の API キーを作成してください。

サービスIDページで、「 APIキー 」をクリックします。
[作成 ] をクリックし、API キーの名前を入力してください。
「漏洩したキーの処理」については、「 漏洩したキーを無効にする 」を選択すると、キーが漏洩したと検出された場合に自動的に無効化されます。
キーの有効期限を設定します。セキュリティ上の理由から、定期的な鍵の交換をお勧めします。
「作成」 をクリックします。
APIキーをコピーし、安全な場所に保存してください。このキーは再度表示することはできません。

プロジェクトIDとAPIエンドポイントを取得する

すべてのAPIリクエストには、プロジェクトIDの指定が必要です。

Red Hat AI Inference のプロジェクトページへ移動してください。
プロジェクトを開いてください。
プロジェクトIDをコピーし、次の手順のために控えておいてください。

このチュートリアルにおけるすべての API リクエストの基本となる URL は、以下の形式になります：

https://us-east.rhai.ibm.com/v1/projects/<project_id>

「 <project_id> 」を、コピーしたプロジェクトIDに置き換えてください。

環境変数の設定

エンドポイントを設定し、API キーを変数に格納します：

export ENDPOINT="https://us-east.rhai.ibm.com/v1/projects/<project_id>"
export API_KEY="<your-api-key>"

<project_id> を自分のプロジェクト ID に、 <your-api-key> を先ほど作成した API キーに置き換えてください。

ソースファイルに認証情報をハードコーディングしてはいけません。環境変数、または次のようなシークレットマネージャーを使用してください IBM Cloud Secrets Manager.

ベアラー・トークンを生成する

このAPIでは、IAMベアラートークンが必要です。次のコマンドを実行して、API キーを IAM ベアラートークンに交換してください：

export AUTH=$(curl -s -X POST "https://iam.cloud.ibm.com/identity/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=urn:ibm:params:oauth:grant-type:apikey" \
  --data-urlencode "apikey=$API_KEY" \
  | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

トークンが正常に発行されたことを確認してください。有効なトークンは「 eyJ 」で始まります：

echo $AUTH | cut -c1-4

IAMトークンの有効期限は1時間です。スクリプトの実行中に401エラーが発生した場合は、トークン交換コマンドを再実行してください。

ベクトルストアを構築する

ベクトルストアは、テキストとその埋め込みベクトルを一緒に保持するインメモリインデックスです。以下の Python の実装では、コサイン類似度を用いて、クエリベクトルと意味的に最も類似したエントリを検索します。

「 store.py 」という名前のファイルを作成し、以下のコードを追加してください：

import math
def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Return the cosine similarity between two vectors."""
    dot = sum(x * y for x, y in zip(a, b))
    mag_a = math.sqrt(sum(x * x for x in a))
    mag_b = math.sqrt(sum(x * x for x in b))
    if mag_a == 0 or mag_b == 0:
        return 0.0
    return dot / (mag_a * mag_b)
class MemoryStore:
    """In-memory vector store with cosine similarity search."""
    def __init__(self) -> None:
        self._entries: list[dict] = []
    def add(self, text: str, vector: list[float]) -> None:
        """Store a text/vector pair."""
        self._entries.append({"text": text, "vector": vector})
    def search(self, query: list[float], top_k: int) -> list[dict]:
        """Return the top_k entries most similar to the query vector."""
        scored = [
            {"text": e["text"], "score": cosine_similarity(query, e["vector"])}
            for e in self._entries
        ]
        scored.sort(key=lambda x: x["score"], reverse=True)
        return scored[:top_k]

コサイン類似度は、2つのベクトルの大きさではなく、それらのなす角度を測定するものです。これにより、テキストの長さに関係なく意味的な方向性を捉えることができるため、テキストの埋め込みの比較に非常に適しています。

APIクライアントを作成する

このAPIクライアントは、 Red Hat AI Inference の2つのエンドポイント、すなわち「embeddings」エンドポイントと「chat completions」エンドポイントをラップしています。この実装では、 openai パッケージを使用しています。このパッケージは、 OpenAI-compatible のあらゆるAPIに対応した Python クライアントを提供します。

「 client.py 」という名前のファイルを作成し、以下のコードを追加してください：

import os
from openai import OpenAI
def _client() -> OpenAI:
    """Return an OpenAI client pointed at the InstructLab endpoint."""
    return OpenAI(
        api_key=os.environ["AUTH"],
        base_url=f"{os.environ['ENDPOINT']}/inference/v1",
    )
def embed_text(text: str) -> list[float]:
    """Call the embeddings endpoint and return the vector for the given text."""
    response = _client().embeddings.create(
        model="granite-embedding-278m-multilingual",
        input=text,
        encoding_format="float",
    )
    return response.data[0].embedding
def chat_complete(system_prompt: str, user_message: str, prefix: str = "") -> None:
    """Call the chat completions endpoint with streaming and print tokens as they arrive."""
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": user_message})
    stream = _client().chat.completions.create(
        model="gpt-oss-120b",
        messages=messages,
        stream=True,
    )
    print(prefix, end="", flush=True)
    for chunk in stream:
        content = chunk.choices[0].delta.content if chunk.choices else None
        if content:
            print(content.replace("\n", f"\n{prefix}"), end="", flush=True)
    print()

_client() ヘルパーは、環境変数 ENDPOINT および AUTH を使用して、 Red Hat AI Inference APIを指す OpenAI インスタンスを生成します。 embed_text 関数は、 encoding_format="float" を指定してembeddingsエンドポイントを呼び出し、レスポンスが単純なfloat型のリストとして返されるようにします。 chat_complete 関数は、ストリーミングを有効にした状態でチャット補完エンドポイントを呼び出すため、トークンが到着するたびにターミナルに出力されます。

メモリストアにデータを格納し、デモを実行する

ベクトルストアとAPIクライアントの準備が整ったところで、これらすべてを統合するメインスクリプトを作成できます。このスクリプトは、ストアに初期データを設定し、メモリを使用せずに各質問に回答した後、RAGを用いて回答し、両方の回答を並べて表示して比較できるようにします。

「 main.py 」という名前のファイルを作成し、以下のコードを追加してください：

import os
import sys
from client import chat_complete, embed_text
from store import MemoryStore
# knowledge is the set of facts to embed into the vector store.
# This example uses IBM Cloud support FAQs as the knowledge base.
KNOWLEDGE = [
    "To reset your IBM Cloud account password, go to the IBM Cloud login page and click 'Forgot password' to receive a reset email.",
    "IBM Cloud Pay-As-You-Go accounts are charged monthly based on resource usage, with no upfront commitment required.",
    "IBM Cloud resource groups are used to organize account resources and manage access with IAM policies.",
    "You can monitor IBM Cloud service health and view planned maintenance on the IBM Cloud Status page at cloud.ibm.com/status.",
    "IBM Cloud Lite accounts are free and include a selection of always-free services, but do not support all paid features.",
    "To create an IBM Cloud API key, go to Manage > Access (IAM) > API keys in the IBM Cloud console.",
    "IBM Cloud support plans range from Basic (included with all accounts) to Advanced and Premium tiers with faster response times.",
    "IBM Cloud Object Storage stores data as objects across multiple regions for high availability and durability.",
]
QUESTIONS = [
    "How do I reset my IBM Cloud password?",
    "What is the difference between a Lite account and a Pay-As-You-Go account?",
]
TOP_K = 3
def build_system_prompt(entries: list[dict]) -> str:
    facts = "\n".join(f"{i + 1}. {e['text']}" for i, e in enumerate(entries))
    return (
        "You are a helpful assistant with access to the following verified facts.\n"
        "Use these facts to answer the user's question accurately and concisely.\n\n"
        f"Relevant facts:\n{facts}\n"
    )
def main() -> None:
    if not os.environ.get("ENDPOINT") or not os.environ.get("AUTH"):
        print("ERROR: ENDPOINT and AUTH environment variables must be set.", file=sys.stderr)
        sys.exit(1)
    # Seed the vector store.
    print("Embedding knowledge base...")
    store = MemoryStore()
    for i, fact in enumerate(KNOWLEDGE):
        print(f"  [{i + 1}/{len(KNOWLEDGE)}] {fact[:60]}")
        store.add(fact, embed_text(fact))
    print(f"\nMemory store ready with {len(KNOWLEDGE)} entries.\n")
    # Run each question without memory, then with RAG.
    for question in QUESTIONS:
        print(f"Question: {question}\n")
        print("  WITHOUT memory:")
        chat_complete("", question, prefix="    ")
        print("\n  WITH memory (RAG):")
        query_vec = embed_text(question)
        relevant = store.search(query_vec, TOP_K)
        print("  Retrieved facts:")
        for i, entry in enumerate(relevant):
            print(f"    {i + 1}. {entry['text']}")
        print()
        chat_complete(build_system_prompt(relevant), question, prefix="    ")
        print("\n" + "─" * 70)
if __name__ == "__main__":
    main()

Python スクリプトを実行します。

python main.py

出力には、各質問に対する回答が2回ずつ表示されています。最初の応答は、モデルの汎用学習で得られた知識のみに基づいて生成されます。 2つ目の応答は、ベクトルストアから取得した事実に基づいています：

Question: How do I reset my IBM Cloud password?
  WITHOUT memory:
    To reset your password, visit the account login page and look for a
    'Forgot password' or 'Reset password' option ...
  WITH memory (RAG):
  Retrieved facts:
    1. To reset your IBM Cloud account password, go to the IBM Cloud login
       page and click 'Forgot password' to receive a reset email.
    2. To create an IBM Cloud API key, go to Manage > Access (IAM) > API keys
       in the IBM Cloud console.
    3. IBM Cloud support plans range from Basic to Advanced and Premium tiers.
    To reset your IBM Cloud account password, go to the IBM Cloud login page
    and click 'Forgot password'. You will receive an email with instructions
    to set a new password.

RAGの応答はより具体的であり、検索された事実を直接引用しています。これは、チャットモデルが呼び出される前に、それらの事実がシステムのプロンプトに組み込まれていたためです。

次のステップ

RAGパイプラインを完全に構築できたところで、これを拡張する方法をいくつかご紹介します：

独自のナレッジベースを活用しましょう — main.py にある「 KNOWLEDGE 」リストを、製品ドキュメント、サポートチケット、社内Wiki、あるいはドメイン固有のデータなど、任意の文字列リストに置き換えることができます。
ベクトルストアを永続化する — ストアのエントリをディスクに（たとえばJSON形式で）シリアライズし、実行のたびに再埋め込みが行われないようにする。
曲の検索 — main.py 内の TOP_K を調整して、クエリごとのファクトの取得数を増減させ、応答の品質にどのような影響があるかを確認してください。