목표

이 튜토리얼에서는 다음 작업을 수행하게 됩니다:

• Red Hat AI Inference API에 인증합니다.
• granite-embedding-278m-multilingual 모델을 사용하여 지식 기반을 삽입합니다.
• 임베딩을 메모리 내 벡터 인덱스에 저장하고 코사인 유사도를 사용하여 검색합니다.
• gpt-oss-120b 채팅 모델에 검색된 컨텍스트를 포함하거나 포함하지 않은 상태로 쿼리를 실행하여 정확도 향상을 직접 확인해 보세요.

구축하는 애플리케이션의 아키텍처는 다음과 같습니다:

1. 지식베이스에 포함된 각 사실은 임베딩 벡터로 변환되어 벡터 맵( MemoryStore)에 추가됩니다.
2. 사용자의 질문은 동일한 모델을 사용하여 삽입됩니다.
3. 코사인 유사도를 통해 저장된 벡터들을 질문 벡터와 비교하여 순위를 매기고, 상위 3개의 사실을 검색합니다.
4. 이러한 사실들은 채팅 모델로 전송되는 시스템 프롬프트에 반영됩니다.
5. 모델은 현실에 입각한 답변을 내놓습니다.

시작하기 전에

다음 사항이 준비되어 있는지 확인하십시오:

• 종량제 또는 구독형 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와 API 엔드포인트를 확인하세요

모든 API 요청 시 프로젝트 ID가 필요합니다.

1. Red Hat AI Inference 프로젝트 로 이동하세요.

2. 프로젝트를 열어주세요.

3. 프로젝트 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]

코사인 유사도는 두 벡터의 크기 대신 두 벡터 사이의 각도를 측정합니다. 이는 텍스트 길이에 관계없이 의미적 방향을 포착하기 때문에 텍스트 임베딩을 비교하는 데 매우 적합합니다.

API 클라이언트 생성하기

이 API 클라이언트는 ‘ Red Hat AI Inference ’의 두 가지 엔드포인트, 즉 임베딩 엔드포인트와 채팅 자동 완성 엔드포인트를 래핑합니다. 이 구현은 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" 매개변수를 사용하여 임베딩 엔드포인트를 호출합니다. 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

출력 결과에는 각 질문에 대한 답변이 두 번씩 표시됩니다. 첫 번째 응답은 모델의 일반 훈련을 통해 습득한 지식만을 바탕으로 도출됩니다. 두 번째 응답은 귀하의 벡터 저장소에서 검색된 사실에 근거합니다:

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 ’ 목록을 제품 설명서, 지원 티켓, 내부 위키 또는 도메인별 데이터 등 원하는 문자열 목록으로 대체할 수 있습니다.
• 벡터 저장소를 영구 저장 — 저장소의 항목을 디스크에 직렬화(예: JSON 형식)하여 매 실행 시마다 다시 삽입하지 않도록 합니다.
• 조절 — main.py 에서 ‘ TOP_K ’을 조정하여 쿼리당 검색되는 사실의 수를 늘리거나 줄이고, 응답 품질에 미치는 영향을 확인해 보세요.