Build a RAG Customer-Support Agent

export AIWAVE_API_KEY="sk-your-api-key"
python rag_support_bot.py "How do I reset my password?"

#!/usr/bin/env python3
import json
import math
import os
import sys
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

BASE_URL = os.environ.get("AIWAVE_BASE_URL", "https://aiwave.live/v1")
EMBEDDING_MODEL = "qwen3.7-text-embedding"
CHAT_MODEL = "deepseek-v4-flash"

KNOWLEDGE_BASE = [
    {
        "title": "Example Help Center | Password reset",
        "text": "To reset a password, open Settings, choose Security, select Reset password, and follow the emailed link. The link expires after 30 minutes.",
    },
    {
        "title": "Example Help Center | Refund request",
        "text": "For a refund request, open Billing, choose the relevant invoice, and select Request review. Include the invoice ID and a short reason. Support replies by email.",
    },
    {
        "title": "Example Help Center | API timeout",
        "text": "For an API timeout, record the request ID, retry once with a bounded delay, and check the service status page. Do not retry indefinitely.",
    },
]


def request_json(path: str, payload: dict) -> dict:
    api_key = os.environ.get("AIWAVE_API_KEY")
    if not api_key:
        raise SystemExit("Set AIWAVE_API_KEY before running this script")

    request = Request(
        f"{BASE_URL}{path}",
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    try:
        with urlopen(request, timeout=30) as response:
            return json.load(response)
    except HTTPError as error:
        body = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {error.code}: {body[:500]}") from error
    except URLError as error:
        raise RuntimeError(f"Network error: {error.reason}") from error


def embed(texts: list[str]) -> list[list[float]]:
    response = request_json("/embeddings", {
        "model": EMBEDDING_MODEL,
        "input": texts,
    })
    rows = sorted(response["data"], key=lambda row: row.get("index", 0))
    vectors = [row["embedding"] for row in rows]
    if len(vectors) != len(texts):
        raise RuntimeError("The embedding response length did not match the input")
    return vectors


def cosine(left: list[float], right: list[float]) -> float:
    dot = sum(a * b for a, b in zip(left, right))
    left_norm = math.sqrt(sum(value * value for value in left))
    right_norm = math.sqrt(sum(value * value for value in right))
    if not left_norm or not right_norm:
        return 0.0
    return dot / (left_norm * right_norm)


def retrieve(question: str, documents: list[dict], top_k: int = 2) -> list[dict]:
    passages = [f"{doc['title']}\n{doc['text']}" for doc in documents]
    passage_vectors = embed(passages)
    question_vector = embed([question])[0]
    ranked = sorted(
        zip(documents, passage_vectors),
        key=lambda item: cosine(question_vector, item[1]),
        reverse=True,
    )
    return [doc for doc, _ in ranked[:top_k]]


def answer(question: str, sources: list[dict]) -> str:
    context = "\n\n".join(
        f"[{index}] {doc['title']}\n{doc['text']}"
        for index, doc in enumerate(sources, start=1)
    )
    response = request_json("/chat/completions", {
        "model": CHAT_MODEL,
        "messages": [
            {
                "role": "system",
                "content": "Answer only from the supplied support context. If it does not contain the answer, say that the help center does not provide it. Cite sources as [1], [2].",
            },
            {
                "role": "user",
                "content": f"Question: {question}\n\nSupport context:\n{context}",
            },
        ],
    })
    return response["choices"][0]["message"]["content"]


def main() -> None:
    question = " ".join(sys.argv[1:]).strip()
    if not question:
        raise SystemExit("Usage: python rag_support_bot.py \"your support question\"")
    sources = retrieve(question, KNOWLEDGE_BASE)
    print("Retrieved sources:")
    for source in sources:
        print(f"- {source['title']}")
    print("\nAnswer:")
    print(answer(question, sources))


if __name__ == "__main__":
    main()

from rag_support_bot import KNOWLEDGE_BASE

assert len(KNOWLEDGE_BASE) == 3
assert all(doc["title"].startswith("Example Help Center") for doc in KNOWLEDGE_BASE)
assert all("customer" not in doc["text"].lower() for doc in KNOWLEDGE_BASE)
print("Local corpus checks passed")