Ask your Confluence: Application with Confluence data and LlamaIndex

On models.py , we will create a single class Service and implement the Singleton pattern on it.

from django.db import models
import os
from llama_index import VectorStoreIndex, ServiceContext, set_global_service_context
from llama_index.query_engine import RetrieverQueryEngine
from llama_hub.confluence.base import ConfluenceReader


# Create your models here.
class Service:
    _service = None

    def __init__(self):
        self.base_url = os.environ.get(
            "CONFLUENCE_URL", "https://myserver.com/confluence/"
        )

    @classmethod
    def get_singleton_instance(self):
        """
        Creates the Service object if it is not yet created, 
        otherwise uses the already created object.
        """
        if self._service is None:
            print("creating singleton service")
            self._service = self()
        return self._service

    def load_documents(self):
        """
        Reads the documents of each Confluence space key
        using the ConfluenceReader class from llama_hub .
        """
        space_keys = ["spacekey1", "spacekey2", "spacekey3"]
        reader = ConfluenceReader(base_url=self.base_url)
        all_documents = []
        for space_key in space_keys:
            documents = reader.load_data(
                space_key=space_key, 
                include_attachments=False, 
                page_status="current"
            )
            all_documents.extend(documents)

        return all_documents

    def create_index(self):
        """
        Creates a Vector index from the documents loaded from Confluence.
        """
        service_context = ServiceContext.from_defaults(chunk_size=1024)
        set_global_service_context(service_context)
        index = VectorStoreIndex.from_documents(self.load_documents())
        return index

    def load_query_engine(self):
        """
        Create the query engine from the index.
        """
        index = self.create_index()
        return index.as_query_engine()

Since to load all the documents takes some time, we want to do this process during the application startup process. As such, go to apps.py and add the following content.

from django.apps import AppConfig
from .models import Service


class ApiConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "api"
    Service.get_singleton_instance().load_query_engine()

Now, all we need to do is to create the POST endpoint that accepts a Question object in the request body. In views.py , add the following endpoint.

from rest_framework import status
from rest_framework.response import Response
from rest_framework.decorators import api_view
from .serializers import QuestionSerializer
from .models import Service


@api_view(["POST"])
def ask_view(request):
    serializer = QuestionSerializer(data=request.data)
    if serializer.is_valid():
        question = serializer.validated_data["data"]
        # get the query engine from the singleton
        query_engine = Service.get_singleton_instance().load_query_engine()
        # query the LLM and return the answer
        return Response({"answer": query_engine.query(question).response})
    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

Inside the urls.py , update the routers by adding the newest endpoint we created.

from django.urls import path, include
from api.views import ask_view

urlpatterns = [
    path("api/question", ask_view, name="ask-question-api")
]

We’re done with the API.

It’s very important to set the environment variables when running in local.

export CONFLUENCE_PASSWORD=password
export CONFLUENCE_USERNAME=username
export OPENAI_API_KEY=sk-...

Now, start your app and test it.

python manage.py runserver