Building an End-to-End RAG Application Using FastAPI on BTP Cloud Foundry

[YangYue01]

Introduction

Retrieval-Augmented Generation (RAG) applications significantly enhance the contextual relevance of large language model (LLM) outputs by integrating real-time database lookups. This tutorial provides a step-by-step guide to implementing an end-to-end RAG application using FastAPI and deploying it on SAP Business Technology Platform (BTP) Cloud Foundry. Special thanks to my colleague Yu Xuan @yxlee12345, who came up with the FastAPI template for BTP CF app.

Prerequisites

• Python 3.9
• Poetry for dependency management

Initial Setup

Poetry is used here to manage all Python dependencies. Follow the installation guide to install poetry if you haven’t already.

Create a pyproject.toml file to specify all required dependencies for the app:

[tool.poetry]
name = "end-to-end-rag-example-fastapi"
version = "0.1.0"
description = ""
authors = [
		"Yang Yue <[email protected]>",
		"Lee Yu Xuan <[email protected]>"
		]

[tool.poetry.dependencies]
python = "^3.9"
fastapi = "^0.92.0"
uvicorn = "^0.20.0"
cfenv = "^0.5.3"
sap-xssec = "^4.0.0"
python-multipart = "^0.0.5"
requests = "^2.28.2"
pandas = "^2.2.2"
hana-ml = "^2.20.24042601"
langchain = "^0.1.19"
shapely = "^2.0.4"
pyyaml = { version = "!=6.0.0,!=5.4.0,!=5.4.1" }
urllib3 = "<2"
python-dotenv = "^1.0.1"

[tool.poetry.group.dev.dependencies]
black = "^23.7.0"
isort = "^5.12.0"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

[tool.isort]
profile = "black"

[tool.black]
line-length = 120

Set up your development environment:

# Ensure you're using the correct Python version
poetry env use python3.9

# Install all required dependencies
poetry install

# Activate virtual environment
poetry shell

# Generate the requirements.txt file (run this if there are changes to Python dependencies), which will be used later to build the Docker image
poetry export -f requirements.txt -o requirements.txt --without-hashes

Implementation Details

Core Functionalities

• Run SAP Business Optical Character Recognition (BOCR) on a given PDF file
• Convert text into embeddings
• Store embeddings in HANA vector database
• Perform similarity search and retrieve relevant contexts
• Send prompts with retrieved contexts to LLMs for responses

Step-by-Step Implementation

Step 1: Initialize HANA Connection

The first step is to establish a secure connection to your SAP HANA instance. Ensure the credential object passed to this function has all required fields.

from hana_ml import ConnectionContext

def initialize_hana_connection_context(credentials: dict) -> ConnectionContext:
    cc = ConnectionContext(
        address=credentials["host"],
        port=credentials["port"],
        user=credentials["user"],
        password=credentials["password"],
        currentSchema=credentials["schema"],
        encrypt=True,
    )
    return cc

Step 2: Read and Process PDF

Use SAP BOCR to extract text content from a given PDF file. This text will then be converted into embeddings for further processing.

import requests
import pandas as pd
import os

def process_document(file_path: str, token: str, service_url: str) -> pd.DataFrame:
    """Process document using SAP BOCR service and return a DataFrame with page contents."""
    data = {'options': '{"lang": "en", "outputType": "json"}'}
    files = {'files': (os.path.basename(file_path), open(file_path, 'rb'))}
    headers = {'Authorization': f'Bearer {token}'}

    response = requests.post(service_url, headers=headers, files=files, data=data)
    if response.status_code != 200:
        raise Exception(f"OCR request failed with status code {response.status_code}")

    bocr_result = response.json()
    page_contents = {}

    for item in bocr_result["predictions"]:
        page = item["page"]
        content_str = " ".join(
            [" ".join([word_box["content"] for word_box in line_box["word_boxes"]]) for line_box in item["line_boxes"]]
        )
        page_contents.setdefault(page, []).append(content_str.strip())

    df = pd.DataFrame(page_contents.items(), columns=["page", "content"])
    df["content"] = df["content"].apply(lambda x: " ".join(x))
    return df

Step 3: Generate Embeddings and Store in HANA

Convert the extracted text into embeddings using an embedding model and store these embeddings in the HANA database.

import requests

def get_embedding(input: str, embedding_deployment_id: str, oauth_token: str) -> str:
    """Get embedding for a given input text."""
    test_input = {"model": "text-embedding-ada-002", "input": input}
    deployment_url_prefix = "<https://your-model-deployment-url-prefix>"  # Replace with your own model deployment URL prefix
    deployment_url = deployment_url_prefix + embedding_deployment_id
    endpoint = f"{deployment_url}/embeddings?api-version=2023-05-15"
    headers = {
        "Authorization": f"Bearer {oauth_token}",
        "ai-resource-group": embedding_deployment_id,
        "Content-Type": "application/json",
    }
    response = requests.post(endpoint, headers=headers, json=test_input)
    return response.json()["data"][0]["embedding"]

def generate_embedding_and_prepare_dataframe_for_hana(df: pd.DataFrame, oauth_token: str, embedding_deployment_id: str) -> pd.DataFrame:
    """Generate embeddings and prepare DataFrame for HANA."""
    df["vector_string"] = df["content"].apply(
        lambda x: str(get_embedding(x, embedding_deployment_id=embedding_deployment_id, oauth_token=oauth_token))
    )
    return df

def create_table_in_hana(connection_context: ConnectionContext, table_name: str = "end_to_end_rag_example_table"):
    """Create table in HANA database."""
    try:
        cursor = connection_context.connection.cursor()
        cursor.execute(
            f"""
            CREATE TABLE {table_name}(
                page NVARCHAR(50),
                content NCLOB,
                vector REAL_VECTOR(1536)
            );
        """)
        connection_context.connection.commit()
    except:
        pass
    finally:
        cursor.close()

def insert_data_into_hana(connection_context: ConnectionContext, df: pd.DataFrame, oauth_token: str, embedding_deployment_id: str, table_name: str = "end_to_end_rag_example_table"):
    """Insert data into HANA table."""
    cursor = connection_context.connection.cursor()
    insert_sql = f"INSERT INTO {table_name} (page, content, vector) VALUES (?, ?, TO_REAL_VECTOR(?))"for _, row in df.iterrows():
        vector_str = str(get_embedding(row["content"], embedding_deployment_id, oauth_token)).replace(" ", "")
        cursor.execute(insert_sql, (row["page"], row["content"], vector_str))
    connection_context.connection.commit()
    cursor.close()

Step 4: Perform Similarity Search

Once the embeddings are stored, perform a similarity search to retrieve the most relevant contexts for a given query.

def run_vector_search(oauth_token: str, connection_context: ConnectionContext, query: str, embedding_deployment_id: str) -> pd.DataFrame:
    query_vector = get_embedding(query, embedding_deployment_id, oauth_token)
    sql = f"SELECT TOP 4 PAGE, CONTENT FROM end_to_end_rag_example_table ORDER BY COSINE_SIMILARITY(VECTOR, TO_REAL_VECTOR('{query_vector}')) DESC"
    hdf = connection_context.sql(sql)
    return hdf.collect()

Step 5: Generate Prompt and Get Response from LLM

Generate a prompt using the retrieved contexts and get a response from the LLM.

from langchain.prompts import PromptTemplate

def generate_prompt(context_str: str, query: str) -> str:
    """Generate prompt for LLM."""
    prompt_template_fstring = """
    You are an SAP legal expert.
    You are provided multiple context items that are related to the prompt you have to answer.
    Use the following pieces of context to answer the question at the end.

    Context:
    {context}

    Question:
    {query}
    """
    prompt_template = PromptTemplate.from_template(prompt_template_fstring)
    return prompt_template.format(query=query, context=context_str)

def get_response(oauth_token: str, input: str, chat_deployment_id: str) -> str:
    """Get response from chat model."""
    test_chat_input = {
        "model": "gpt-35-turbo",
        "messages": [{"content": input, "role": "user"}],
        "temperature": 0.0,
        "max_tokens": 250,
    }

    deployment_url_prefix = "<https://your-model-deployment-url-prefix>"  # Replace with your own model deployment URL prefix
    deployment_url = deployment_url_prefix + chat_deployment_id
    endpoint = f"{deployment_url}/chat/completions?api-version=2023-05-15"
    headers = {
        "Authorization": f"Bearer {oauth_token}",
        "ai-resource-group": chat_deployment_id,
        "Content-Type": "application/json",
    }

    chat_response = requests.post(endpoint, headers=headers, json=test_chat_input)
    return chat_response.json()["choices"][0]["message"]["content"]

Step 6: Define ask_llm Function

Define the ask_llm function to connect all functionalities.

def ask_llm(query: str, embedding_deployment_id: str, chat_deployment_id: str) -> dict:
    # Fetch credentials using your preferred method
    resource_group_credentials = get_service_credentials("resource_group_service_key")
    oauth_token = get_oauth_token(resource_group_credentials)

    hana_credentials = get_service_credentials("hana_vector_db_service_key")
    connection_context = initialize_hana_connection_context(hana_credentials)

    bocr_service_credentials = get_service_credentials("bocr_credential")
    file_path = bocr_service_credentials['file_path']
    token = bocr_service_credentials['token']
    service_url = bocr_service_credentials['service_url']

    df = process_document(file_path, token, service_url)

    # Prepare data and insert into HANA
    df = generate_embedding_and_prepare_dataframe_for_hana(df, oauth_token, embedding_deployment_id)
    create_table_in_hana(connection_context)
    insert_data_into_hana(connection_context, df, oauth_token, embedding_deployment_id)

    # Perform similarity search
    context = run_vector_search(oauth_token, connection_context, query, embedding_deployment_id)
    context_str = " ".join(context["CONTENT"].astype("string"))

    # Generate prompt and get response
    prompt = generate_prompt(context_str, query)
    response = get_response(oauth_token, input=prompt, chat_deployment_id=chat_deployment_id)

    return {"query": query, "context": context.to_json(orient="records"), "response": response}

Step 7: FastAPI Endpoint to Handle RAG Requests

Combine all the above functionalities into a FastAPI endpoint to handle incoming RAG requests.

from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
import logging
from sap import xssec

logger = logging.getLogger(__name__)

def get_security_context(auth_header: str) -> xssec.SecurityContextXSUAA:
    from src.config import UAA_SERVICE

    access_token = auth_header.replace("Bearer ", "")
    return xssec.create_security_context(access_token, UAA_SERVICE)

def verify_request(request: Request):
    # Get Authorization header
    auth_header = request.headers.get("Authorization")
    if auth_header is None:
        error_msg = "Authorization token not found in request header"
        logger.error(error_msg)
        raise HTTPException(
            status_code=401,
            detail={
                "title": "Authorization Header Missing",
                "details": error_msg,
                "status": 401,
            },
        )
    # Validate token
    try:
        xsuaa_context = get_security_context(auth_header)
    except RuntimeError as e:
        if e.args == ("Audience Validation Failed",):
            error_msg = "Client credentials used to generate XSUAA access token are different from expected"
            logger.error(error_msg)
            raise HTTPException(
                status_code=401,
                detail={
                    "title": "Wrong Client Credentials",
                    "details": error_msg,
                    "status": 401,
                },
            )
        raise
    if not xsuaa_context:
        error_msg = "XSUAA is not configured correctly"
        logger.error(error_msg)
        raise HTTPException(
            status_code=401,
            detail={"title": "XSUAA misconfiguration", "details": error_msg, "status": 401},
        )

    # Only allow logged in users
    if not xsuaa_context.get_grant_type:
        error_msg = "Please input the correct grant type"
        logger.error(error_msg)
        raise HTTPException(
            status_code=403,
            detail={"title": "Grant type mismatch", "details": error_msg, "status": 403},
        )

    logger.info("User verification completed successfully")
 
class AskRequest(BaseModel):
    query: str
    embedding_deployment_id: str
    chat_deployment_id: str
    
class AskResponse(BaseModel):
    query: str
    context: str
    response: str

router = APIRouter()

@router.post("/ask", response_model=AskResponse, tags=["RAG"], summary="Send a query to LLMs.")
def ask_llm_endpoint(request: AskRequest, auth: None = Depends(verify_request)):
    try:
        result = ask_llm(
            query=request.query,
            embedding_deployment_id=request.embedding_deployment_id,
            chat_deployment_id=request.chat_deployment_id,
        )
        return AskResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Deployment to BTP Cloud Foundry

Step 1: Create Manifest File

Create a manifest.yml file for your Cloud Foundry deployment configuration:

---
applications:
  - name: rag-example-rest-api
    disk-quota: 4G
    memory: 512MB
    random-route: false
    instances: 1
    # Specify services to bind with this app on BTP
    services:
      - contract-intelligence-xsuaa # Name of the xsuaa service
      -	sapai_bocr_service # Name of the BOCR service 
      - resource_group_service_key # Name of the RG credential service (user-defined)
      - hana_vector_db_service_key # Name of the HANA credential service (user-defined)
    env:
      APP_NAME: End-to-end RAG REST API
      UAA_SERVICE_NAME: contract-intelligence-xsuaa
      API_MAJOR_VERSION_LATEST: "1"

Step 2: Create an Entrypoint Python Script to Run the Application

Create a run.py file to run your FastAPI application:

import uvicorn

from src.config import APP_HOST, APP_PORT, APP_WORKER, LOCAL_APP

if __name__ == "__main__":
    if LOCAL_APP:
        uvicorn.run("src.api.main:app", host=APP_HOST, port=APP_PORT, workers=APP_WORKER, reload=True)
    else:
        uvicorn.run("src.api.main:app", host=APP_HOST, port=APP_PORT, workers=APP_WORKER)

Step 3: Create Dockerfile for Containerization

Create a Dockerfile to containerize your FastAPI application:

FROM python:3.9.14-slim-buster

# Install tzdata for the missing timezone in ubuntu docker image; DEBIAN_FRONTEND for non-interactive installation
ENV DEBIAN_FRONTEND='noninteractive'

# Argument is passed in from docker-compose.yml file .e.g api
ARG SRC_PATH='.'

RUN apt-get -y update && \\
    apt-get install -y --no-install-recommends \\
    python3-dev \\
    curl \\
    build-essential \\
    libmagic-dev \\
    poppler-utils \\
    && apt-get clean \\
    && rm -rf /var/lib/apt/lists/*

# Setup permissions group and user under which the micro service will run
RUN groupadd -r nonroot &&\\
    useradd -r -g nonroot -b /home -m nonroot && \\
    usermod -aG sudo nonroot && \\
    echo "nonroot ALL=(ALL) NOPASSWD:ALL" >> /etc/sudoers
USER nonroot

# Copy requirements from app and shared
COPY ${SRC_PATH}/requirements.txt /tmp/

# Always installed shared requirements first, since app may have dependencies that overwrite that
RUN pip3 install --no-cache-dir -U pip setuptools && pip3 install wheel
RUN pip3 install --no-cache-dir --user -r /tmp/requirements.txt

# Copy source code from microservice folder and the shared folder
COPY ${SRC_PATH}/src /home/nonroot/src
COPY ${SRC_PATH}/run.py /home/nonroot/run.py

WORKDIR /home/nonroot
ENV PYTHONPATH /

CMD python3 run.py

Step 4: Install CF CLI Tool

Refer to the official documentation to install CF CLI v8.

Step 5: Build and Push Docker Image to Artifactory

Create a scripts/build_push_to_artifactory.py file to handle building and pushing the Docker image to a container registry (e.g. Docker Hub).

import argparse
import os
import subprocess

def main(args):
    # Build Docker image
    image_base_name = os.path.basename(os.getcwd()) if args.image_base_name is None else args.image_base_name
    full_image_name = f"{container_registry_url}/{image_base_name}:{args.image_tag}"
    returncode = subprocess.call(f"docker build -t {full_image_name} .", shell=True)
    if returncode != 0:
        print("Error occurred while building Docker image, exiting...")
        exit(returncode)

    # Push Docker image
    if args.no_push:
        print("Pushing to Artifactory has been disabled, exiting now...")
        exit(0)
    returncode = subprocess.call(f"docker login {container_registry_url}", shell=True)
    if returncode != 0:
        print("Error occurred while trying to login to Artifactory, exiting now...")
        exit(returncode)
    returncode = subprocess.call(f"docker push {full_image_name}", shell=True)
    if returncode != 0:
        print("Error occurred while trying to push image to Artifactory, exiting now...")
        exit(returncode)
    print(f"Push to Artifactory completed for image {full_image_name}, exiting now...")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Build Docker image API server and optionally push to ACRMLH")
    parser.add_argument("container_registry_url", type=str, help="Repository domain for the Docker image")
    parser.add_argument("image_tag", type=str, help="Docker image tag to be added")
    parser.add_argument(
        "--image-base-name",
        type=str,
        default=None,
        help="Docker image name without the domain. Will be set to name of directory if no input is given",
    )
    parser.add_argument(
        "--no-push",
        action="store_true",
        help="Flag to disable pushing of image to Artifactory, to test if Docker container starts from image as expected",
    )

    args = parser.parse_args()
    main(args)

Activate your virtual environment and run the Python script to build and push the Docker image.

Note that the container_registry_url should be the actual URL for the registry where you want to push the image.

poetry shell
python scripts/build_push_to_artifactory.py <container_registry_url> <image_tag>

Step 6: Deploy to CF Dev Space

Log in to your CF dev space and push the Docker container:

cf login
cf push --docker-image <container_registry_url>/<image-name>:<image-tag> --docker-username <docker_user_name>

Conclusion

This tutorial provides a clear pathway to implementing a RAG application using FastAPI and deploying it on BTP Cloud Foundry. By following these steps, you can enhance your LLM applications with real-time contextual data, ensuring more relevant and accurate responses.

For detailed interaction with HANA Vector DB, please check out one of my previous posts. Additionally, for building a RAG application in Python with LangChain, HANA Vector DB, and Generative AI Hub SDK, please refer to @yxlee12345's previous post. Feel free to customize the logic to fit your specific requirements.