Ziele

In diesem Tutorial führen Sie die folgenden Aufgaben durch:

• Authentifizierung bei der Red Hat AI Inference-API.
• Eine Wissensdatenbank mithilfe des Modells granite-embedding-278m-multilingual einbinden.
• Speichern Sie Einbettungen in einem In-Memory-Vektorindex und durchsuchen Sie diesen anhand der Kosinusähnlichkeit.
• Stellen Sie dem gpt-oss-120b Chat-Modell Fragen sowohl ohne als auch mit abgerufenem Kontext, um die Verbesserung der Genauigkeit selbst zu erleben.

Die von Ihnen entwickelte Anwendung weist folgende Architektur auf:

1. Jede Information in Ihrer Wissensdatenbank wird in einen Einbettungsvektor umgewandelt und einer MemoryStore. hinzugefügt.
2. Eine Nutzerfrage wird nach demselben Muster eingebettet.
3. Anhand der Kosinusähnlichkeit werden die gespeicherten Vektoren mit dem Fragenvektor verglichen, und die drei relevantesten Fakten werden abgerufen.
4. Diese Fakten werden in die Systemaufforderung eingefügt, die an das Chat-Modell gesendet wird.
5. Das Modell liefert eine fundierte Antwort.

Vorbereitende Schritte

Stellen Sie sicher, dass Sie Folgendes haben:

• Ein Pay-As-You-Go- oder Abonnement- IBM Cloud-Konto. Testkonten werden nicht unterstützt. Weitere Informationen oder Anleitungen zum Upgrade Ihres Kontos finden Sie unter Kontotypen.

• Ein Projekt von Red Hat AI Inference.

• Die Rolle Writer oder höher im Dienst Red Hat AI Inference. Weitere Informationen finden Sie unter Verwalten des IAM-Zugriffs.

• Python 3.9 oder höher auf Ihrem Arbeitsplatzrechner installiert sein.

• Das openai Python-Paket. Führen Sie den folgenden Befehl aus, um die Installation durchzuführen:

  pip install openai
  

API-Authentifizierung einrichten

Bevor Sie die Red Hat AI Inference-API aufrufen können, müssen Sie Ihre Anfragen mithilfe einer Service-ID und eines API-Schlüssels authentifizieren. Wenn Sie das Einführungs-Tutorial bereits abgeschlossen haben, können Sie die dort erstellte Service-ID und den API-Schlüssel wiederverwenden und direkt zum Abschnitt Projekt-ID und Endpunkt abrufen springen.

Rufen Sie Ihre Projekt-ID und Ihren API-Endpunkt ab

Ihre Projekt-ID ist für alle API-Anfragen erforderlich.

1. Zu den Red Hat AI Inference-Projekten.

2. Öffnen Sie Ihr Projekt.

3. Kopieren Sie Ihre Projekt-ID und bewahren Sie sie für die nächsten Schritte auf.

Die Basis- URL für alle API-Anfragen in diesem Tutorial hat folgende Form:

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

Ersetzen Sie <project_id> durch die von Ihnen kopierte Projekt-ID.

Umgebungsvariablen festlegen

Legen Sie Ihren Endpunkt fest und speichern Sie Ihren API-Schlüssel in einer Variablen:

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

Ersetzen Sie durch <project_id> Ihre Projekt-ID und <your-api-key> durch den zuvor erstellten API-Schlüssel.

Zugangsdaten dürfen niemals fest in Quelldateien hinterlegt werden. Verwenden Sie Umgebungsvariablen oder einen Secrets-Manager wie IBM Cloud Secrets Manager.

Ein Bearer-Token generieren

Die API erfordert ein IAM-Bearer-Token. Tauschen Sie Ihren API-Schlüssel gegen ein IAM-Bearer-Token aus, indem Sie den folgenden Befehl ausführen:

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'])")

Überprüfen Sie, ob das Token erfolgreich ausgestellt wurde. Ein gültiges Token beginnt mit eyJ:

echo $AUTH | cut -c1-4

IAM-Token verfallen nach einer Stunde. Führen Sie den Befehl zum Token-Austausch erneut aus, wenn beim Ausführen des Skripts ein 401-Fehler auftritt.

Erstellen des Vektorspeichers

Der Vektorspeicher ist ein In-Memory-Index, der Text zusammen mit seinem Einbettungsvektor speichert. Die folgende Implementierung von Python nutzt die Kosinus-Ähnlichkeit, um die Einträge zu finden, die einem Abfragevektor semantisch am ähnlichsten sind.

Erstellen Sie eine Datei mit dem Namen und store.py fügen Sie den folgenden Code hinzu:

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]

Die Kosinusähnlichkeit misst den Winkel zwischen zwei Vektoren, nicht deren Größe. Dadurch eignet es sich gut für den Vergleich von Text-Embeddings, da es die semantische Ausrichtung unabhängig von der Textlänge erfasst.

API-Client erstellen

Der API-Client bündelt zwei Endpunkte von Red Hat AI Inference: den Endpunkt embeddings und den Endpunkt chat completions. Diese Implementierung nutzt das Paket openai, das einen Python-Client für jede OpenAI-compatible-API bereitstellt.

Erstellen Sie eine Datei mit dem Namen und client.py fügen Sie den folgenden Code hinzu:

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()

Der Helper _client() erstellt eine Instanz OpenAI, die auf die Red Hat AI Inference-API verweist, und nutzt dabei Ihre und ENDPOINT AUTH Umgebungsvariablen. Die Funktion embed_text ruft den Embeddings-Endpunkt mit auf, um encoding_format="float" sicherzustellen, dass die Antwort als einfache Liste von Float-Werten zurückgegeben wird. Die Funktion chat_complete ruft den Endpunkt für Chat-Vervollständigungen mit aktiviertem Streaming auf, sodass Tokens bei ihrem Eintreffen auf dem Terminal ausgegeben werden.

Den Speicherspeicher initialisieren und die Demo ausführen

Nachdem der Vektorspeicher und der API-Client eingerichtet sind, können Sie nun das Hauptskript schreiben, das alles miteinander verknüpft: Es füllt den Speicher mit Startdaten, beantwortet jede Frage ohne Speicher, gibt dann eine Antwort mit RAG und gibt beide Antworten nebeneinander zur Vergleichszwecken aus.

Erstellen Sie eine Datei mit dem Namen und main.py fügen Sie den folgenden Code hinzu:

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()

Führen Sie das Python-Script aus:

python main.py

In der Ausgabe wird jede Frage zweimal beantwortet. Die erste Antwort basiert ausschließlich auf dem allgemeinen Trainingswissen des Modells. Die zweite Antwort basiert auf den Fakten, die aus Ihrem Vektorspeicher abgerufen wurden:

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.

Die RAG-Antwort ist spezifischer und zitiert die abgerufenen Fakten direkt, da diese Fakten vor dem Aufruf des Chat-Modells in die Systemaufforderung eingefügt wurden.

Nächste Schritte

Nachdem Sie nun eine vollständige RAG-Pipeline aufgebaut haben, finden Sie hier einige Möglichkeiten, diese zu erweitern:

• Nutzen Sie Ihre eigene Wissensdatenbank – Ersetzen Sie die KNOWLEDGE Liste in main.py durch eine beliebige Liste von Zeichenfolgen: Produktdokumentation, Support-Tickets, interne Wikis oder domänenspezifische Daten.
• Den Vektorspeicher persistent machen – Die Einträge des Speichers auf die Festplatte serialisieren (z. B. als JSON), damit sie nicht bei jedem Durchlauf neu eingebettet werden müssen.
• Abfrageoptimierung – Passen Sie TOP_K die Einstellung an, main.py um pro Abfrage mehr oder weniger Fakten abzurufen, und beobachten Sie die Auswirkungen auf die Qualität der Ergebnisse.