from pathlib import Path

from chainlit.cli import run_chainlit


def run():
    """Python entrypoint to run chainlit server.

    Used to provide a script via poetry.

    """
    run_chainlit(Path(Path(__file__).parent, "server.py").as_posix())

qa_rag/cli.py
sebastianschramm/german-qa-rag
import logging
from typing import List

import pydantic
from haystack import Pipeline
from haystack.document_stores import InMemoryDocumentStore
from haystack.nodes import AnswerParser, PromptNode, PromptTemplate, TopPSampler
from haystack.nodes.ranker import LostInTheMiddleRanker
from haystack.nodes.retriever.web import WebRetriever
from haystack.nodes.translator import TransformersTranslator

from qa_rag.prompts import prompt_answer
from qa_rag.settings import settings


def get_prompt_node(prompt: str, settings: pydantic.BaseSettings) -> PromptNode:
    return PromptNode(
        model_name_or_path=settings.openai_model,
        default_prompt_template=PromptTemplate(prompt, output_parser=AnswerParser()),
        api_key=settings.openai_api_key,
        max_length=768,
        model_kwargs={"stream": False},
    )


def get_web_retriever(
    settings: pydantic.BaseSettings, allowed_domains: List[str]
) -> WebRetriever:
    return WebRetriever(
        api_key=settings.serper_api_key,
        allowed_domains=allowed_domains,
        top_search_results=10,
        mode="preprocessed_documents",
        top_k=50,
        cache_document_store=InMemoryDocumentStore(),
    )


def get_pipeline():
    pipeline = Pipeline()
    pipeline.add_node(
        component=TransformersTranslator("Helsinki-NLP/opus-mt-en-de"),
        name="Translator_en_de",
        inputs=["Query"],
    )
    pipeline.add_node(
        component=get_web_retriever(
            settings=settings, allowed_domains=["stadt-koeln.de"]
        ),
        name="Retriever",
        inputs=["Translator_en_de"],
    )
    pipeline.add_node(
        component=TopPSampler(top_p=0.90), name="Sampler", inputs=["Retriever"]
    )
    pipeline.add_node(
        component=LostInTheMiddleRanker(1024),
        name="LostInTheMiddleRanker",
        inputs=["Sampler"],
    )
    pipeline.add_node(
        component=get_prompt_node(prompt=prompt_answer, settings=settings),
        name="PromptNode",
        inputs=["LostInTheMiddleRanker"],
    )
    pipeline.add_node(
        component=TransformersTranslator("Helsinki-NLP/opus-mt-de-en"),
        name="Translator_de_en",
        inputs=["PromptNode"],
    )

    logging.disable(logging.CRITICAL)
    return pipeline

qa_rag/pipeline.py
sebastianschramm/german-qa-rag
prompt_answer = """
Synthesize a comprehensive answer from the provided paragraphs and the given question.\n
Focus on the question and avoid unnecessary information in your answer.\n
\n\n Paragraphs: {join(documents)} \n\n Question: {query} \n\n Answer:
"""

qa_rag/prompts.py
sebastianschramm/german-qa-rag
import chainlit as cl

from qa_rag.pipeline import get_pipeline

pipeline = get_pipeline()


@cl.on_message
async def main(message: str):
    response = await cl.make_async(pipeline.run)(message)
    await cl.Message(author="Haystack", content=response["answers"][0].answer).send()

qa_rag/server.py
sebastianschramm/german-qa-rag
from pydantic import BaseSettings


class Settings(BaseSettings):
    serper_api_key: str
    openai_api_key: str
    openai_model: str


settings = Settings()

qa_rag/settings.py
sebastianschramm/german-qa-rag