Documentation

Open WebUI

Open WebUI is a an AI interface which is self hosted and can be extended to your needs. Your RheinInsights Retrieval Suite deployment can be integrated into Open WebUI as a search provider, for instance as a tool.

This way, your users will gain relevant answers through our query pipelines. In particular, our query pipelines can be queried in a way so that security trimming takes place. In turn, answers will always be only generated based on documents the users are allowed to see in the underlying knowledge sources.

Configuration

You can include the RheinInsights Retrieval Suite in Open WebUI as follows:

As an administrator open the admin panel
Click on Functions
Click on new function
Click on new function (yes, a second time)
Give this function a reasonable name
Give the function a description
Copy and paste the python code below into the field.
Click on save
Click on confirm
Click on the function’s gear wheel
Fill in the following parameters
1. Rheininsightsurl: this should be the FQDN towards your RheinInsights Retrieval Suite deployment
2. Authenticationtoken: leave empty if you did not define a passkey. Otherwise provide this passkey here
3. The id of the query pipeline which should be used. P
  For more information on how to configure query pipelines, see Query Pipelines .
  Please note that it must be configured for public access, see Query Pipeline - Access Configuration . Also, if you want to use secure search, you need to enable “Allow for ingesting userPrincipalName as part of the query” for this pipeline
Afterwards, activate the function by clicking on the checkbox
Click on settings
Open models
Open the menu of the new “model” by clicking on the three dots
Click on access
Adjust the accessibility to the user group which should use this model
Close the dialog and click on save & update
Afterwards the permitted users can use this model

Pipeline Description

"""
title: Enterprise Knowledge
author: RheinInsights
author_url: https://www.rheininsights.com
version: 1.0.0
required_open_webui_version: 0.3.30

"""

from typing import List, Union, Generator, Iterator
from pydantic import BaseModel, Field
import requests
from bs4 import BeautifulSoup
import urllib.parse


class Pipe:
    class Valves(BaseModel):
        RheininsightsUrl: str = Field(default="https://localhost")
        AuthenticationToken: str = Field(default="secret")
        QueryPipelineId: int = Field(default=0)

    def __init__(self):
        self.type = "manifold"
        self.id = "engine_search"
        self.name = "engines/"
        self.valves = self.Valves()

    def pipes(self) -> List[dict]:
        enabled_pipes = []
        enabled_pipes.append({"id": "knowledge", "name": "Knowledge Search"})
        return enabled_pipes

    def pipe(
        self, body: dict, __user__: dict, results=None
    ) -> Union[str, Generator, Iterator]:
        user_input = self._extract_user_input(body)
        if not user_input:
            return "No search query provided"

        model = body.get("model", "")
        print(f"Received model: {model}")  # Debug print

        if isinstance(results, str):
            try:
                results = int(results)
            except ValueError:
                return f"Invalid number of results '{results}'"

        return self._search_knowledge(
            user_input, body.get("messages", []), __user__["email"], results
        )

    def _extract_user_input(self, body: dict) -> str:
        messages = body.get("messages", [])
        if messages:
            last_message = messages[-1]
            if isinstance(last_message.get("content"), list):
                for item in last_message["content"]:
                    if item["type"] == "text":
                        return item["text"]
            else:
                return last_message.get("content", "")
        return ""

    def _search_knowledge(
        self, query: str, context: object, mail: str, results=None
    ) -> str:
        print("Searching with knowledge")
        try:
            url = f"{self.valves.RheininsightsUrl}/api/v1/querypipelines/search?queryPipelineId={self.valves.QueryPipelineId}"
            headers = {"Authorization": self.valves.AuthenticationToken}

            queryObject = {
                "query": query,
                "userPrincipalName": mail,
                "context": context,
            }

            response = requests.post(
                url, headers=headers, json=queryObject, verify=False
            )
            response.raise_for_status()

            data = response.json()
            print(f"Got response {data}")

            if not data:
                return f"No results found for: {query}"

            formatted_results = ""
            for i, result in enumerate(data["results"]):
                answer = self._process_references(
                    result["teaser"], result["references"]
                )
                formatted_results += f"{answer}\n\n"

            return formatted_results

        except Exception as e:
            return f"An error occurred while searching knowledge: {str(e)}"

    def _process_references(self, answer: str, references: dict) -> str:
        try:
            print(f"references are: {references}")
            if not references:
                return answer
            newAnswer = answer
            referencesString = ""

            for k in references:
                title = references[k]["title"]
                url = references[k]["url"]
                id = references[k]["id"]

                newAnswer = newAnswer.replace(f"[{id}]", f"")

            for k in references:
                title = references[k]["title"]
                url = references[k]["url"]
                id = references[k]["id"]

                referencesString += f"[{title}]({url})\n"

            if references != "":
                newAnswer += f"\n\n\nReferences\n\n{referencesString}"

            newAnswer = newAnswer.replace(f" .", f".")

            print(f"Got response:\n{newAnswer}")
            return newAnswer

        except Exception as e:
            return f"An error occurred while searching knowledge: {str(e)}"