inti

.gitattributes

@@ -0,0 +1,36 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.index filter=lfs diff=lfs merge=lfs -text
+*.db filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+.vscode
+huggingface/
+.env
+.streamlit

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.7.4-stretch
+
+WORKDIR /home/user
+
+RUN apt-get update && apt-get install -y curl git pkg-config cmake
+
+# copy code
+COPY setup.py /home/user
+COPY utils.py /home/user
+COPY app.py /home/user
+COPY requirements.txt /home/user
+
+# install as a package
+RUN pip install --upgrade pip
+RUN pip install -r requirements.txt
+RUN python3 -c "from utils import get_pipelines;get_pipelines()" 
+
+EXPOSE 8501
+
+# cmd for running the API
+CMD ["python", "-m", "streamlit", "run", "app.py"]

README.md

@@ -0,0 +1,12 @@
+---
+title: Ai Advisor
+emoji: 📈
+colorFrom: yellow
+colorTo: red
+sdk: streamlit
+sdk_version: 1.17.0
+app_file: app.py
+pinned: false
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,207 @@
+import os
+
+cwd = os.getcwd()
+os.environ['PYTORCH_TRANSFORMERS_CACHE'] = os.path.join(cwd, 'huggingface/transformers/')
+os.environ['TRANSFORMERS_CACHE'] = os.path.join(cwd, 'huggingface/transformers/')
+os.environ['HF_HOME'] = os.path.join(cwd, 'huggingface/')
+# import sys
+import logging
+from json import JSONDecodeError
+from pathlib import Path
+
+# import zipfile
+import pandas as pd
+import streamlit as st
+from markdown import markdown
+
+from utils import get_backlink, get_pipelines, query, send_feedback, upload_doc
+
+# Adjust to a question that you would like users to see in the search bar when they load the UI:
+DEFAULT_QUESTION_AT_STARTUP = os.getenv(
+    "DEFAULT_QUESTION_AT_STARTUP", "How to get TPS?")
+DEFAULT_ANSWER_AT_STARTUP = os.getenv(
+    "DEFAULT_ANSWER_AT_STARTUP", "You must file a Form I-765")
+
+# Sliders
+DEFAULT_DOCS_FROM_RETRIEVER = int(
+    os.getenv("DEFAULT_DOCS_FROM_RETRIEVER", "5"))
+DEFAULT_NUMBER_OF_ANSWERS = int(os.getenv("DEFAULT_NUMBER_OF_ANSWERS", "1"))
+
+
+# Whether the file upload should be enabled or not
+DISABLE_FILE_UPLOAD = bool(os.getenv("DISABLE_FILE_UPLOAD", "True"))
+
+LANG_MAP = {"English": "English", "Ukrainian": "Ukrainian", "russian": "russian"}
+
+
+pipelines = get_pipelines()
+
+
+def set_state_if_absent(key, value):
+    if key not in st.session_state:
+        st.session_state[key] = value
+
+
+def main():
+
+    st.set_page_config(page_title="AI advisor")
+
+    # Persistent state
+    set_state_if_absent("question", DEFAULT_QUESTION_AT_STARTUP)
+    set_state_if_absent("answer", DEFAULT_ANSWER_AT_STARTUP)
+    set_state_if_absent("results", None)
+    set_state_if_absent("raw_json", None)
+    set_state_if_absent("random_question_requested", False)
+
+    # Small callback to reset the interface in case the text of the question changes
+    def reset_results(*args):
+        st.session_state.answer = None
+        st.session_state.results = None
+        st.session_state.raw_json = None
+
+    # Title
+    st.write("# AI Immigration advisor")
+
+    # Sidebar
+    st.sidebar.header("Options")
+    language = LANG_MAP[st.sidebar.selectbox(
+        "Select language: ", ("English", "Ukrainian", "russian"))]
+    debug = False
+    # debug = st.sidebar.checkbox("Show debug info")
+    if debug:
+        top_k_reader = st.sidebar.slider(
+            "Max. number of answers",
+            min_value=1,
+            max_value=100,
+            value=DEFAULT_NUMBER_OF_ANSWERS,
+            step=1,
+            on_change=reset_results,
+        )
+
+        top_k_retriever = st.sidebar.slider(
+            "Max. number of documents from retriever",
+            min_value=1,
+            max_value=100,
+            value=DEFAULT_DOCS_FROM_RETRIEVER,
+            step=1,
+            on_change=reset_results,
+        )
+    else:
+        top_k_reader = DEFAULT_NUMBER_OF_ANSWERS
+        top_k_retriever = DEFAULT_DOCS_FROM_RETRIEVER
+    # File upload block
+    if not DISABLE_FILE_UPLOAD:
+        st.sidebar.write("## File Upload:")
+        data_files = st.sidebar.file_uploader(
+            "", type=["pdf", "txt", "docx"], accept_multiple_files=True)
+        for data_file in data_files:
+            # Upload file
+            if data_file:
+                raw_json = upload_doc(data_file)
+                st.sidebar.write(str(data_file.name) + "    ✅ ")
+                if debug:
+                    st.subheader("REST API JSON response")
+                    st.sidebar.write(raw_json)
+
+    # st.sidebar.markdown(
+    #     f"""
+    # <style>
+    #     a {{
+    #         text-decoration: none;
+    #     }}
+    #     .haystack-footer {{
+    #         text-align: center;
+    #     }}
+    #     .haystack-footer h4 {{
+    #         margin: 0.1rem;
+    #         padding:0;
+    #     }}
+    #     footer {{
+    #         opacity: 0;
+    #     }}
+    # </style>
+    # <div class="haystack-footer">
+    #     <hr />
+    #     <h4>Debug parameters</h4>
+    #     <small>Data crawled from <a href="https://www.uscis.gov">USCIS</a></small></div>
+    # """,
+    #     unsafe_allow_html=True,
+    # )
+
+    # Search bar
+    question = st.text_input(
+        "", value=st.session_state.question, max_chars=100, on_change=reset_results)
+    col1, col2 = st.columns(2)
+    col1.markdown(
+        "<style>.stButton button {width:100%;}</style>", unsafe_allow_html=True)
+    col2.markdown(
+        "<style>.stButton button {width:100%;}</style>", unsafe_allow_html=True)
+
+    # Run button
+    run_pressed = col1.button("Run")
+
+    run_query = (
+        run_pressed or question != st.session_state.question
+    ) and not st.session_state.random_question_requested
+
+    # Get results for query
+    if run_query and question:
+        reset_results()
+        st.session_state.question = question
+
+        with st.spinner("🧠 &nbsp;&nbsp; Performing neural search on documents... \n "):
+            try:
+                st.session_state.results, st.session_state.raw_json = query(
+                    pipelines, question, top_k_reader=top_k_reader, top_k_retriever=top_k_retriever, language=language
+                )
+            except JSONDecodeError as je:
+                st.error(
+                    "👓 &nbsp;&nbsp; An error occurred reading the results. Is the document store working?")
+                return
+            except Exception as e:
+                logging.exception(e)
+                if "The server is busy processing requests" in str(e) or "503" in str(e):
+                    st.error(
+                        "🧑‍🌾 &nbsp;&nbsp; All our workers are busy! Try again later.")
+                else:
+                    st.error(
+                        "🐞 &nbsp;&nbsp; An error occurred during the request.")
+                return
+
+    if st.session_state.results:
+
+        st.write("## Results:")
+
+        for count, result in enumerate(st.session_state.results):
+            if result["answer"]:
+                answer, context = result["answer"], result["context"]
+                start_idx = context.find(answer)
+                end_idx = start_idx + len(answer)
+                # Hack due to this bug: https://github.com/streamlit/streamlit/issues/3190
+                st.write(
+                    markdown(f"**Answer:** {answer}"), unsafe_allow_html=True)
+                # st.write(
+                #     markdown(context[:start_idx] + str(annotation(answer, "ANSWER", "#8ef")) + context[end_idx:]),
+                #     unsafe_allow_html=True,
+                # )
+                source = ""
+                url, title = get_backlink(result)
+                if url and title:
+                    source = f"[{result['document']['meta']['title']}]({result['document']['meta']['url']})"
+                else:
+                    source = f"{result['source']}"
+                st.markdown(f"**Source:** {source}")
+
+            else:
+                st.info(
+                    "🤔 &nbsp;&nbsp; Unsure whether any of the documents contain an answer to your question. Try to reformulate it!"
+                )
+
+            st.write("___")
+
+        if debug:
+            st.subheader("REST API JSON response")
+            st.write(st.session_state.raw_json)
+
+
+main()

config.py

@@ -0,0 +1,19 @@
+import os
+from pathlib import Path
+
+
+PIPELINE_YAML_PATH = os.getenv(
+    "PIPELINE_YAML_PATH", str((Path(__file__).parent / "pipeline" / "pipelines.haystack-pipeline.yml").absolute())
+)
+FAISS_INDEX_PATH = os.getenv(
+    "FAISS_INDEX_PATH", str((Path(__file__).parent / "faiss.index").absolute())
+)
+QUERY_PIPELINE_NAME = os.getenv("QUERY_PIPELINE_NAME", "query")
+INDEXING_PIPELINE_NAME = os.getenv("INDEXING_PIPELINE_NAME", "indexing")
+
+FILE_UPLOAD_PATH = os.getenv("FILE_UPLOAD_PATH", str((Path(__file__).parent / "file-upload").absolute()))
+
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
+ROOT_PATH = os.getenv("ROOT_PATH", "/")
+
+CONCURRENT_REQUEST_PER_WORKER = int(os.getenv("CONCURRENT_REQUEST_PER_WORKER", "4"))

faiss.index

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:fa1c641d84f7bf7e58968610c31e913a3d2c44e341fde639c0c8a0ce48b16d19
+size 24035373

faiss.json

@@ -0,0 +1 @@
+{"sql_url": "sqlite:///openai_faiss_document_store.db", "embedding_dim": 1024, "faiss_index_factory_str": "Flat", "similarity": "cosine"}

openai_faiss_document_store.db

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:532271b901d948f42992d4ccc4e9e3b5b9b7ebb9ea1547ce6d7adf09c3b81bda
+size 17989632

pipeline/pipelines.haystack-pipeline.yml

@@ -0,0 +1,56 @@
+# To allow your IDE to autocomplete and validate your YAML pipelines, name them as <name of your choice>.haystack-pipeline.yml
+
+version: 1.7.0
+
+components:    # define all the building-blocks for Pipeline
+  - name: DocumentStore
+    type: FAISSDocumentStore  # consider using MilvusDocumentStore or WeaviateDocumentStore for scaling to large number of documents
+    params:
+      faiss_index_path: faiss.index
+      # faiss_config_path: rest_api/faiss.json
+      # sql_url: sqlite:///rest_api/faiss_document_store.db
+  - name: Retriever
+    type: DensePassageRetriever
+    params:
+      document_store: DocumentStore    # params can reference other components defined in the YAML
+      passage_embedding_model: vblagoje/dpr-ctx_encoder-single-lfqa-wiki
+      query_embedding_model: vblagoje/dpr-question_encoder-single-lfqa-wiki
+  - name: Generator       # custom-name for the component; helpful for visualization & debugging
+    type: Seq2SeqGenerator    # Haystack Class name for the component
+    params:
+      model_name_or_path: vblagoje/bart_lfqa
+      max_length: 300
+      min_length: 10
+  # - name: TextFileConverter
+  #   type: TextConverter
+  # - name: PDFFileConverter
+  #   type: PDFToTextConverter
+  # - name: Preprocessor
+  #   type: PreProcessor
+  #   params:
+  #     split_by: word
+  #     split_length: 300
+  # - name: FileTypeClassifier
+  #   type: FileTypeClassifier
+
+pipelines:
+  - name: query    # generative-qa Pipeline
+    nodes:
+      - name: Retriever
+        inputs: [Query]
+      - name: Generator
+        inputs: [Retriever]
+  # - name: indexing
+  #   nodes:
+  #     - name: FileTypeClassifier
+  #       inputs: [File]
+  #     - name: TextFileConverter
+  #       inputs: [FileTypeClassifier.output_1]
+  #     - name: PDFFileConverter
+  #       inputs: [FileTypeClassifier.output_2]
+  #     - name: Preprocessor
+  #       inputs: [PDFFileConverter, TextFileConverter]
+  #     - name: Retriever
+  #       inputs: [Preprocessor]
+  #     - name: DocumentStore
+  #       inputs: [Retriever]

requirements.txt

@@ -0,0 +1,6 @@
+farm-haystack==1.13.2
+streamlit>=1.2.0, <2 
+st-annotated-text>=2.0.0, <3
+markdown>=3.3.4, <4 
+faiss-cpu==1.7.2
+openai==0.27.2

setup.py

@@ -0,0 +1,40 @@
+import logging
+from pathlib import Path
+
+from setuptools import setup, find_packages
+
+
+VERSION = "0.0.0"
+try:
+    # After git clone, VERSION.txt is in the root folder
+    VERSION = open(Path(__file__).parent.parent / "VERSION.txt", "r").read()
+except Exception:
+    try:
+        # In Docker, VERSION.txt is in the same folder
+        VERSION = open(Path(__file__).parent / "VERSION.txt", "r").read()
+    except Exception as e:
+        logging.exception("No VERSION.txt found!")
+
+setup(
+    name="farm-haystack-ui",
+    version=VERSION,
+    description="Demo UI for Haystack (https://github.com/deepset-ai/haystack)",
+    author="deepset.ai",
+    author_email="[email protected]",
+    url=" https://github.com/deepset-ai/haystack/tree/master/ui",
+    classifiers=[
+        "Development Status :: 5 - Production/Stable",
+        "Intended Audience :: Science/Research",
+        "Topic :: Scientific/Engineering :: Artificial Intelligence",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+    ],
+    packages=find_packages(),
+    python_requires=">=3.7, <4",
+    install_requires=["streamlit>=1.2.0, <2", "st-annotated-text>=2.0.0, <3", "markdown>=3.3.4, <4", "farm-haystack==1.7.0"],
+)

test/test_ui_utils.py

@@ -0,0 +1,15 @@
+from unittest.mock import patch
+
+from ui.utils import haystack_is_ready
+
+
+def test_haystack_is_ready():
+    with patch("requests.get") as mocked_get:
+        mocked_get.return_value.status_code = 200
+        assert haystack_is_ready()
+
+
+def test_haystack_is_ready_fail():
+    with patch("requests.get") as mocked_get:
+        mocked_get.return_value.status_code = 400
+        assert not haystack_is_ready()

utils.py

@@ -0,0 +1,239 @@
+from typing import List, Dict, Any, Tuple, Optional
+import yaml
+import os
+
+cwd = os.getcwd()
+os.environ["PYTORCH_TRANSFORMERS_CACHE"] = os.path.join(
+    cwd, "huggingface/transformers/"
+)
+os.environ["TRANSFORMERS_CACHE"] = os.path.join(cwd, "huggingface/transformers/")
+os.environ["HF_HOME"] = os.path.join(cwd, "huggingface/")
+import logging
+import json
+from time import sleep, time
+
+import requests
+import streamlit as st
+
+from pathlib import Path
+
+from haystack.pipelines.base import Pipeline
+from haystack.nodes import EmbeddingRetriever, Shaper
+from haystack.document_stores import FAISSDocumentStore, InMemoryDocumentStore
+from haystack.errors import PipelineConfigError
+# from googletrans import Translator
+import openai
+
+openai.api_key = st.secrets["gpt35_api_key"]
+
+logger = logging.getLogger(__name__)
+pipelines = None
+
+system_prompt_template = """You are an US experienced immigration attorney.  Your answer should be in your own words, be detailed and be no longer than 350 words. 
+You should ask for more information or clarifications to give a more precise answer for an each client's case. 
+Synthesize a comprehensive answer from your knowledge and the following topk most relevant paragraphs and the given question. 
+Give an answer in the {0} language.
+"""
+users_prompt_template = """
+Paragraphs: {0}
+
+Question: {1}
+"""
+# Since each instance of FAISSDocumentStore creates an in-memory FAISS index, the Indexing & Query Pipelines would
+# end up with different indices. The same applies for InMemoryDocumentStore.
+UNSUPPORTED_DOC_STORES = (FAISSDocumentStore, InMemoryDocumentStore)
+
+API_ENDPOINT = os.getenv("API_ENDPOINT", "http://localhost:8000")
+STATUS = "initialized"
+HS_VERSION = "hs_version"
+DOC_REQUEST = "query"
+DOC_FEEDBACK = "feedback"
+DOC_UPLOAD = "file-upload"
+
+# translator = Translator()
+
+
+def query(
+    pipelines, query, filters={}, language="en", top_k_reader=3, top_k_retriever=5
+) -> Tuple[List[Dict[str, Any]], Dict[str, str]]:
+    """
+    Send a query to the REST API and parse the answer.
+    Returns both a ready-to-use representation of the results and the raw JSON.
+    """
+    query_pipeline = pipelines.get("query_pipeline", None)
+    start_time = time()
+
+    params = {
+        "retriever": {"top_k": top_k_retriever},
+    }
+
+    lang = language.lower() or "english"
+
+    response = query_pipeline.run(
+        query=query,
+        params=params,
+    )
+    context = ""
+    sources = []
+    for doc in response["documents"]:
+        doc = doc.to_dict()
+        doc_name = doc["meta"].get("name")
+        doc_url = doc["meta"].get("url")
+        source = (
+            "https://www.uscis.gov/sites/default/files/document/forms/" + doc_name
+            if doc_name
+            else doc_url
+        )
+        if not source.endswith('.txt'):
+            sources.append(source)
+        if len(context)<top_k_reader:
+            context += " " + doc.get("content")
+    # Ensure answers and documents exist, even if they're empty lists
+    if not "documents" in response:
+        response["documents"] = []
+
+    # prepare openAI api call
+    messages = []
+    system_prompt = system_prompt_template.format(lang)
+    user_prompt = users_prompt_template.format(context, response["query"])
+    messages.append({"role": "system", "content": system_prompt})
+    messages.append({"role": "user", "content": user_prompt})
+
+    openai_response = openai.ChatCompletion.create(
+        model="gpt-3.5-turbo", messages=messages
+    )
+    bot_response = openai_response["choices"][0]["message"]["content"]
+    response["answers"] = [bot_response]
+    logger.info(
+        json.dumps(
+            {
+                "request": query,
+                "response": response,
+                "time": f"{(time() - start_time):.2f}",
+            },
+            default=str,
+        )
+    )
+
+    # Format response
+    results = []
+    answers = response["answers"]
+    documents = response["documents"]
+    for answer, doc in zip(answers, documents):
+        doc = doc.to_dict()
+        if answer:
+            context = doc.get("content")
+            results.append(
+                {
+                    "context": "..." + context if context else "",
+                    "answer": answer,
+                    "source": "\n".join(sources),
+                    "_raw": answer,
+                }
+            )
+        else:
+            results.append({"context": None, "answer": None, "_raw": answer})
+    return results, response
+
+
+def send_feedback(
+    query, answer_obj, is_correct_answer, is_correct_document, document
+) -> None:
+    """
+    Send a feedback (label) to the REST API
+    """
+    url = f"{API_ENDPOINT}/{DOC_FEEDBACK}"
+    req = {
+        "query": query,
+        "document": document,
+        "is_correct_answer": is_correct_answer,
+        "is_correct_document": is_correct_document,
+        "origin": "user-feedback",
+        "answer": answer_obj,
+    }
+    response_raw = requests.post(url, json=req)
+    if response_raw.status_code >= 400:
+        raise ValueError(
+            f"An error was returned [code {response_raw.status_code}]: {response_raw.json()}"
+        )
+
+
+def upload_doc(file):
+    url = f"{API_ENDPOINT}/{DOC_UPLOAD}"
+    files = [("files", file)]
+    response = requests.post(url, files=files).json()
+    return response
+
+
+def get_backlink(result) -> Tuple[Optional[str], Optional[str]]:
+    if result.get("document", None):
+        doc = result["document"]
+        if isinstance(doc, dict):
+            if doc.get("meta", None):
+                if isinstance(doc["meta"], dict):
+                    if doc["meta"].get("url", None) and doc["meta"].get("title", None):
+                        return doc["meta"]["url"], doc["meta"]["title"]
+    return None, None
+
+
+def setup_pipelines() -> Dict[str, Any]:
+    # Re-import the configuration variables
+    import config  # pylint: disable=reimported
+
+    pipelines = {}
+    document_store = FAISSDocumentStore(
+        faiss_config_path="faiss.json", faiss_index_path="faiss.index"
+    )
+    retriever = EmbeddingRetriever(
+        document_store=document_store,
+        batch_size=128,
+        embedding_model="ada",
+        api_key=st.secrets["api_key"],
+        max_seq_len=1024,
+    )
+
+    shaper = Shaper(
+        func="join_documents", inputs={"documents": "documents"}, outputs=["documents"]
+    )
+
+    pipe = Pipeline()
+    pipe.add_node(component=retriever, name="retriever", inputs=["Query"])
+
+    logging.info(f"Loaded pipeline nodes: {pipe.graph.nodes.keys()}")
+    pipelines["query_pipeline"] = pipe
+
+    # Find document store
+
+    logging.info(f"Loaded docstore: {document_store}")
+    pipelines["document_store"] = document_store
+
+    # Load indexing pipeline (if available)
+    try:
+        indexing_pipeline = Pipeline.load_from_yaml(
+            Path(config.PIPELINE_YAML_PATH), pipeline_name=config.INDEXING_PIPELINE_NAME
+        )
+        docstore = indexing_pipeline.get_document_store()
+        if isinstance(docstore, UNSUPPORTED_DOC_STORES):
+            indexing_pipeline = None
+            raise PipelineConfigError(
+                "Indexing pipelines with FAISSDocumentStore or InMemoryDocumentStore are not supported by the REST APIs."
+            )
+
+    except PipelineConfigError as e:
+        indexing_pipeline = None
+        logger.error(f"{e.message}\nFile Upload API will not be available.")
+
+    finally:
+        pipelines["indexing_pipeline"] = indexing_pipeline
+
+    # Create directory for uploaded files
+    os.makedirs(config.FILE_UPLOAD_PATH, exist_ok=True)
+
+    return pipelines
+
+
+def get_pipelines():
+    global pipelines  # pylint: disable=global-statement
+    if not pipelines:
+        pipelines = setup_pipelines()
+    return pipelines