Objetivos

En este tutorial, realizarás las siguientes tareas:

• Autentícate en la API de Red Hat AI Inference.
• Incorporar una base de conocimientos utilizando el granite-embedding-278m-multilingual modelo.
• Almacena las representaciones vectoriales en un índice vectorial en memoria y realza la búsqueda mediante la similitud coseno.
• Realiza consultas al modelo de chat gpt-oss-120b tanto sin contexto como con contexto recuperado para comprobar de primera mano la mejora en la precisión.

La aplicación que estás desarrollando tiene la siguiente arquitectura:

1. Cada dato de tu base de conocimientos se convierte en un vector de incrustación y se añade a un MemoryStore.
2. Se incluye una pregunta de un usuario siguiendo el mismo modelo.
3. La similitud coseno clasifica los vectores almacenados en función del vector de la pregunta y se recuperan los tres datos principales.
4. Esos datos se introducen en la indicación del sistema que se envía al modelo de chat.
5. El modelo ofrece una respuesta fundamentada.

Antes de empezar

Asegúrate de que dispones de lo siguiente:

• Una cuenta de IBM Cloud de pago por uso o por suscripción. No se admiten cuentas de prueba. Para obtener más información o para ampliar tu cuenta, consulta Tipos de cuenta.

• Un proyecto de Red Hat AI Inference.

• Rol de Writer o superior en el servicio Red Hat AI Inference. Para obtener más información, consulta Gestión del acceso a IAM.

• Python 3.9 o una versión posterior instalada en su estación de trabajo.

• El openai paquete Python. Instálalo ejecutando el siguiente comando :

  pip install openai
  

Configurar autenticación de API

Antes de poder llamar a la API de Red Hat AI Inference, debes autenticar tus solicitudes utilizando un ID de servicio y una clave de API. Si ya has completado el tutorial de introducción, puedes reutilizar el ID de servicio y la clave de API que creaste allí y pasar directamente a Obtener tu ID de proyecto y punto final.

Consigue tu ID de proyecto y tu punto final de la API

Es necesario indicar el ID de tu proyecto en todas las solicitudes de API.

1. Ir a los proyectos de Red Hat AI Inference.

2. Abre tu proyecto.

3. Copia tu ID de proyecto y guárdalo para los siguientes pasos.

La URL base URL para todas las solicitudes de API de este tutorial tiene el siguiente formato:

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

Sustituye <project_id> por el ID del proyecto que has copiado.

Establecer variables de entorno

Configura tu punto final y guarda tu clave de API en una variable:

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

Sustituye <project_id> por tu ID de proyecto y <your-api-key> por la clave de API que has creado anteriormente.

Nunca incluyas credenciales de forma estática en los archivos fuente. Utilice variables de entorno o un gestor de secretos como IBM Cloud Secrets Manager.

Generar un token de portador

La API requiere un token de portador de IAM. Cambia tu clave API por un token de portador de IAM ejecutando el siguiente 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'])")

Comprueba que el token se haya emitido correctamente. Un token válido comienza por eyJ:

echo $AUTH | cut -c1-4

Los tokens de IAM caducan al cabo de 1 hora. Vuelve a ejecutar el comando de intercambio de tokens si recibes un error 401 al ejecutar el script.

Crear el almacén vectorial

El almacén de vectores es un índice en memoria que contiene texto junto con su vector de incrustación. La siguiente implementación de Python utiliza la similitud coseno para encontrar las entradas más similares semánticamente a un vector de consulta.

Crea un archivo llamado store.py y añade el siguiente código:

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 similitud coseno mide el ángulo entre dos vectores, no su magnitud. Esto lo hace muy adecuado para comparar representaciones de texto, ya que capta la dirección semántica independientemente de la longitud del texto.

Crear el cliente de la API

El cliente de la API engloba dos puntos de conexión de Red Hat AI Inference: el punto de conexión de incrustaciones y el punto de conexión de autocompletado de chat. Esta implementación utiliza el paquete openai, que proporciona un cliente Python para cualquier API de OpenAI-compatible.

Crea un archivo llamado client.py y añade el siguiente código:

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

El helper _client() crea una instancia OpenAI que apunta a la API de Red Hat AI Inference utilizando tus variables de entorno AUTH ENDPOINT y. La función embed_text llama al punto final de incrustaciones con encoding_format="float" para garantizar que la respuesta se devuelva como una lista simple de números flotantes. La función chat_complete llama al punto final de autocompletado del chat con la transmisión activada, por lo que los tokens se muestran en la terminal a medida que llegan.

Inicializa el almacén de memoria y ejecuta la demostración

Una vez que el almacén vectorial y el cliente API estén listos, ya puedes escribir el script principal que lo integra todo: este inicializa el almacén, responde a cada pregunta sin utilizar la memoria, a continuación responde con RAG y muestra ambas respuestas una al lado de la otra para compararlas.

Crea un archivo llamado main.py y añade el siguiente código:

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

Ejecute el script Python:

python main.py

El resultado muestra cada pregunta respondida dos veces. La primera respuesta se basa únicamente en los conocimientos generales adquiridos durante el entrenamiento del modelo. La segunda respuesta se basa en los datos extraídos de tu almacén vectorial:

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 respuesta del RAG es más específica y cita directamente los datos recuperados, ya que dichos datos se introdujeron en la indicación del sistema antes de que se activara el modelo de chat.

Próximos pasos

Ahora que ya ha creado un proceso RAG completo, aquí tiene algunas formas de ampliarlo:

• Aporta tu propia base de conocimientos: sustituye la lista KNOWLEDGE de main.py por cualquier lista de cadenas: documentación de productos, tickets de asistencia, wikis internos o datos específicos de un ámbito.
• Persistir el almacén vectorial: serializar las entradas del almacén en el disco (por ejemplo, como JSON) para no tener que volver a incrustarlas en cada ejecución.
• Ajuste la recuperación de resultados: modifique el valor de in TOP_K main.py para recuperar más o menos datos por consulta y observe el efecto en la calidad de la respuesta.