Obiettivi

In questo tutorial, dovrai svolgere le seguenti attività:

• Effettua l'autenticazione all'API " Red Hat AI Inference ".
• Incorporare una knowledge base utilizzando il modello " granite-embedding-278m-multilingual ".
• Memorizza le rappresentazioni vettoriali in un indice vettoriale in memoria ed effettua la ricerca utilizzando la similarità coseno.
• Prova il modello di chat " gpt-oss-120b " sia senza che con il contesto recuperato per constatare di persona il miglioramento in termini di precisione.

L'applicazione che stai sviluppando presenta la seguente architettura:

1. Ogni dato contenuto nella vostra base di conoscenza viene convertito in un vettore di embedding e aggiunto a un MemoryStore.
2. Una domanda dell'utente viene incorporata utilizzando lo stesso modello.
3. La similarità coseno ordina i vettori memorizzati in base al vettore della domanda e vengono recuperati i primi 3 fatti.
4. Tali dati vengono inseriti nel messaggio di prompt inviato al modello di chat.
5. Il modello fornisce una risposta ben fondata.

Prima di iniziare

Assicurati di avere quanto segue:

• Un account “Pay-As-You-Go” o in abbonamento su IBM Cloud. Non sono supportati gli account di prova. Per ulteriori informazioni o per passare a un account di livello superiore, consulta la sezione " Tipi di account ".

• Un progetto di " Red Hat AI Inference ".

• Ruolo "Writer" o superiore nel servizio Red Hat AI Inference. Per ulteriori informazioni, vedi Gestione dell'accesso IAM.

• Python 3.9 o una versione successiva installata sulla propria workstation.

• Il pacchetto " openai " Python. Per installarlo, esegui il seguente comando:

  pip install openai
  

Configurare l'autenticazione API

Prima di poter effettuare una chiamata all'API " Red Hat AI Inference ", è necessario autenticare le richieste utilizzando un ID servizio e una chiave API. Se hai già completato il tutorial introduttivo, puoi riutilizzare l'ID del servizio e la chiave API che hai creato in quella occasione e passare direttamente alla sezione " Ottieni l'ID del progetto e l'endpoint ".

Ottieni l'ID del tuo progetto e l'endpoint API

L'ID del progetto è obbligatorio per tutte le richieste API.

1. Vai su Red Hat AI Inference progetti.

2. Apri il tuo progetto.

3. Copia l'ID del tuo progetto e conservalo per i passaggi successivi.

L' URL e di base per tutte le richieste API in questo tutorial ha la seguente forma:

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

Sostituisci " <project_id> " con l'ID del progetto che hai copiato.

Imposta le variabili di ambiente

Imposta il tuo endpoint e salva la tua chiave API in una variabile:

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

Sostituisci <project_id> con l'ID del tuo progetto e <your-api-key> con la chiave API che hai creato in precedenza.

Non inserire mai credenziali in modo statico nei file sorgente. Utilizza le variabili d'ambiente o un gestore di segreti come IBM Cloud Secrets Manager.

Genera un token al portatore

L'API richiede un token bearer IAM. Sostituisci la tua chiave API con un token bearer IAM eseguendo il seguente comando:

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

Verifica che il token sia stato emesso correttamente. Un token valido inizia con eyJ:

echo $AUTH | cut -c1-4

I token IAM scadono dopo 1 ora. Se durante l'esecuzione dello script viene visualizzato un errore 401, rieseguire il comando di scambio dei token.

Creare l'archivio vettoriale

Il vettore di archiviazione è un indice in memoria che contiene il testo insieme al relativo vettore di embedding. La seguente implementazione di Python utilizza la similarità coseno per individuare gli elementi semanticamente più simili a un vettore di query.

Crea un file denominato “ store.py ” e inserisci il seguente codice:

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]

La similarità coseno misura l'angolo tra due vettori, non la loro grandezza. Ciò lo rende particolarmente adatto al confronto tra embedding di testo, poiché rileva la direzione semantica indipendentemente dalla lunghezza del testo.

Creare il client API

Il client API racchiude due endpoint dell' Red Hat AI Inference: l'endpoint degli embedding e l'endpoint dei completamenti della chat. Questa implementazione utilizza il pacchetto openai, che fornisce un client Python per qualsiasi API OpenAI-compatible.

Crea un file denominato “ client.py ” e inserisci il seguente codice:

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

L'helper _client() crea un'istanza di OpenAI collegata all'API Red Hat AI Inference utilizzando le variabili d'ambiente ENDPOINT e AUTH . La funzione embed_text chiama l'endpoint degli embedding con encoding_format="float" per garantire che la risposta venga restituita come un semplice elenco di valori float. La funzione chat_complete chiama l'endpoint di completamento della chat con lo streaming abilitato, quindi i token vengono visualizzati sul terminale man mano che arrivano.

Incaricare la memoria di lavoro ed eseguire la demo

Una volta implementati il vettore di memorizzazione e il client API, è ora possibile scrivere lo script principale che mette insieme tutti gli elementi: esso popola il vettore di memorizzazione, risponde a ciascuna domanda senza ricorrere alla memoria, quindi fornisce una risposta tramite RAG e visualizza entrambe le risposte affiancate per consentirne il confronto.

Crea un file denominato “ main.py ” e inserisci il seguente codice:

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

Esegui lo script “ Python ”:

python main.py

Il risultato mostra ogni domanda con due risposte. La prima risposta deriva esclusivamente dalle conoscenze acquisite durante l'addestramento generale del modello. La seconda risposta si basa sui dati recuperati dal tuo archivio vettoriale:

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.

La risposta del RAG è più specifica e cita direttamente i fatti recuperati, poiché tali fatti sono stati inseriti nel prompt del sistema prima che venisse richiamato il modello di chat.

Passi successivi

Ora che hai creato una pipeline RAG completa, ecco alcuni modi per ampliarla:

• Utilizza la tua base di conoscenze — Sostituisci l'elenco " KNOWLEDGE " presente in main.py con qualsiasi elenco di stringhe: documentazione dei prodotti, ticket di assistenza, wiki interni o dati specifici di un determinato settore.
• Mantenere l'archivio vettoriale — Serializza le voci dell'archivio su disco (ad esempio, in formato JSON) in modo da non doverle reinserire ad ogni esecuzione.
• Ottimizzazione dei risultati — Modificare il parametro “ TOP_K ” in “ main.py ” per ottenere un numero maggiore o minore di risultati per ogni query e osservare l’effetto sulla qualità della risposta.