app.py

import os
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))

import gradio as gr
from huggingface_hub import InferenceClient
from ragassistant.rag import RAGPipeline

MODEL = "Qwen/Qwen2.5-7B-Instruct"
MAX_QUESTION_CHARS = 300

client = InferenceClient()
_pipeline = None


def get_pipeline():
    global _pipeline
    if _pipeline is None:
        _pipeline = RAGPipeline()
    return _pipeline


def answer(question):
    question = (question or "").strip()
    if not question:
        return "Ask a question about the product documentation."
    if len(question) > MAX_QUESTION_CHARS:
        return f"Please keep the question under {MAX_QUESTION_CHARS} characters."
    try:
        contexts = get_pipeline().retriever.retrieve(question)
        context_block = "\n\n".join(
            f"[{chunk['source']}]\n{chunk['text']}" for chunk, _ in contexts
        )
        messages = [
            {"role": "system", "content": "Answer the question using only the provided context. If the answer is not in the context, say you don't know. Cite the source filename in square brackets after each fact you use."},
            {"role": "user", "content": f"Context:\n{context_block}\n\nQuestion: {question}"},
        ]
        last_error = None
        for attempt in range(3):
            try:
                response = client.chat_completion(messages=messages, model=MODEL, max_tokens=400)
                return response.choices[0].message.content.strip()
            except Exception as error:
                last_error = error
                time.sleep(1.5 * (attempt + 1))
        raise last_error
    except Exception:
        return "The service is busy or unavailable right now. Please try again in a moment."


demo = gr.Interface(
    fn=answer,
    inputs=gr.Textbox(lines=2, label="Ask a question about the product documentation",
                      placeholder="What is the refund policy?"),
    outputs=gr.Textbox(lines=8, label="Answer (with sources)"),
    title="Document Question Answering (RAG)",
    description=(
        "Ask a question about a small documentation set (overview, pricing, API, "
        "security, FAQ). The system retrieves the most relevant passages and answers "
        "from them, citing the source file for each fact."
    ),
    article="Code: https://github.com/delcenjo/rag-document-assistant",
    cache_examples=False,
    examples=[
        ["What is the refund policy?"],
        ["How do I authenticate with the API?"],
        ["Where is the data stored?"],
    ],
)

if __name__ == "__main__":
    demo.launch(ssr_mode=False)