Created first Gradio Space (MVP)

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* 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
+*.faiss filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,6 @@
+venv/
+.env
+__pycache__
+.qodo
+.DS_Store
+*nogit*

README.md

@@ -1,12 +1,16 @@
 ---
 title: Aina RAG
-emoji: 🚀
-colorFrom: indigo
-colorTo: purple
+emoji: 🐡
+colorFrom: purple
+colorTo: blue
 sdk: gradio
-sdk_version: 5.45.0
+sdk_version: 5.28.0
 app_file: app.py
 pinned: false
+license: apache-2.0
+short_description: Conversational space enhanced for Aina Challenge
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Aina Challenge
+
+An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).

app.py

@@ -0,0 +1,142 @@
+from dotenv import load_dotenv
+
+
+import gradio as gr
+from gradio import ChatMessage
+
+import json
+from openai import OpenAI
+from datetime import datetime
+import os
+import re
+
+from termcolor import cprint
+import logging
+logging.basicConfig(level=logging.INFO, format='[%(asctime)s][%(name)s][%(levelname)s] - %(message)s')
+log = logging.getLogger(__name__)
+
+
+
+from omegaconf import OmegaConf
+from src.tools import tools, oitools
+
+
+# Load the configuration file
+# ===========================================================================
+# Environment variables
+load_dotenv(".env", override=True) 
+HF_TOKEN = os.environ.get("HF_TOKEN")
+LLM_BASE_URL = os.environ.get("LLM_BASE_URL")
+
+log.info(f"Using HF_TOKEN: {HF_TOKEN[:4]}...{HF_TOKEN[-4:]}")
+log.info(f"Using LLM_BASE_URL: {LLM_BASE_URL[:10]}...{LLM_BASE_URL[-10:]}")
+
+# Configuration file
+config_file = "config.yaml"
+cfg = OmegaConf.load(config_file)
+
+# OpenAI API parameters
+chat_params = cfg.openai.chat_params
+client = OpenAI(
+    base_url=f"{LLM_BASE_URL}",
+    api_key=HF_TOKEN
+)
+logging.info(f"Client initialized: {client}")
+# ===========================================================================
+
+
+def today_date():
+    return datetime.today().strftime('%A, %B %d, %Y, %I:%M %p')
+
+
+def clean_json_string(json_str):
+    return re.sub(r'[ ,}\s]+$', '', json_str) + '}'
+
+
+def completion(history, model, system_prompt: str, tools=None, chat_params=chat_params):
+    messages = [{"role": "system", "content": system_prompt.format(date=today_date())}]
+    for msg in history:
+        if isinstance(msg, dict):  
+            msg = ChatMessage(**msg)
+        if msg.role == "assistant" and hasattr(msg, "metadata") and msg.metadata:  
+            tools_calls = json.loads(msg.metadata.get("title", "[]")) 
+            messages.append({"role": "assistant", "tool_calls": tools_calls, "content": ""})
+            messages.append({"role": "tool", "content": msg.content})
+        else:
+            messages.append({"role": msg.role, "content": msg.content})
+    
+    request_params = {
+        "model": model,
+        "messages": messages,
+        **chat_params
+    }
+    if tools:
+        request_params.update({"tool_choice": "auto", "tools": tools})
+    
+
+    return client.chat.completions.create(**request_params)  
+
+
+def llm_in_loop(history, system_prompt, recursive):  
+    
+    try:   
+        models = client.models.list()
+        model = models.data[0].id
+    except Exception as err:
+        gr.Warning("The model is initializing. Please wait; this may take 5 to 10 minutes ⏳.", duration=20)
+        raise err
+    
+    arguments = ""
+    name = ""
+    chat_completion = completion(history=history, tools=oitools, model=model, system_prompt=system_prompt)  
+    appended = False
+
+
+    for chunk in chat_completion:
+        if chunk.choices and chunk.choices[0].delta.tool_calls:
+            call = chunk.choices[0].delta.tool_calls[0]
+            if hasattr(call.function, "name") and call.function.name:
+                name = call.function.name
+            if hasattr(call.function, "arguments") and call.function.arguments:
+                arguments += call.function.arguments
+
+        elif chunk.choices[0].delta.content:
+            if not appended:
+                history.append(ChatMessage(role="assistant", content=""))
+                appended = True
+            history[-1].content += chunk.choices[0].delta.content
+            yield history[recursive:]
+    
+    arguments = clean_json_string(arguments) if arguments else "{}"
+    arguments = json.loads(arguments)
+
+
+    if appended:
+        recursive -= 1
+    if name:
+        try:
+            result = str(tools[name].invoke(input=arguments))
+
+        except Exception as err:
+            result = f"💥 Error: {err}"
+
+        history.append(ChatMessage(
+            role="assistant", 
+            content=result, 
+            metadata={"title": json.dumps([{"id": "call_id", "function": {"arguments": json.dumps(arguments, ensure_ascii=False), "name": name}, "type": "function"}], ensure_ascii=False)}))
+        
+        yield history[recursive:]
+        yield from llm_in_loop(history, system_prompt, recursive - 1)
+
+
+def respond(message, history, additional_inputs):  
+    history.append(ChatMessage(role="user", content=message))
+    yield from llm_in_loop(history, additional_inputs, -1)
+
+
+
+if __name__ == "__main__":
+
+    system_prompt = gr.Textbox(label="System prompt", value=cfg.system_prompt_template, lines=10)
+    demo = gr.ChatInterface(respond, type="messages", additional_inputs=[system_prompt])
+    demo.launch()

config.yaml

@@ -0,0 +1,26 @@
+
+# Embeddings configuration
+# ================================================================================
+vdb:
+  embeddings_model: BAAI/bge-m3
+  number_of_contexts: 5
+  vs_local_path: data/vdb
+  embedding_score_threshold: 0.4
+
+  # Context formatting (join retrieved chunks with this string)
+  join_str: "\n\n"
+
+
+# LLM client configuration
+# ================================================================================
+llm_generation: true
+system_prompt_template: |
+  You are an AI assistant designed to answer user questions using externally retrieved information. You must detect the user's language, **translate the query into Spanish**, and **respond to the user in their original language**.
+  All retrieved content is available **only in Spanish**.
+
+openai:
+  chat_params:
+    stream: True
+    max_tokens: 1000
+    temperature: 0.0
+    top_p: 0.9

data/vdb/index.faiss

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

data/vdb/index.pkl

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

requirements.txt

@@ -0,0 +1,9 @@
+gradio==5.23.0
+openai==1.68.2
+python-dotenv==1.1.0
+langchain-community==0.3.20
+langchain-core==0.3.48
+faiss-cpu==1.10.0
+faiss-gpu==1.7.2
+sentence-transformers==3.4.1
+termcolor

src/tools.py

@@ -0,0 +1,123 @@
+
+from abc import ABC, abstractmethod
+from typing import Dict, Union, get_origin, get_args
+from pydantic import BaseModel, Field
+from types import UnionType
+import logging
+from src.vectorstore import VectorStore
+from omegaconf import OmegaConf
+
+
+class ToolBase(BaseModel, ABC):
+    @abstractmethod
+    def invoke(cls, input: Dict):
+        pass
+
+    @classmethod
+    def to_openai_tool(cls):
+        """
+        Extracts function metadata from a Pydantic class, including function name, parameters, and descriptions.
+        Formats it into a structure similar to OpenAI's function metadata.
+        """
+        function_metadata = {
+            "type": "function",
+            "function": {
+                "name": cls.__name__,  # Function name is same as the class name, in lowercase
+                "description": cls.__doc__.strip(),
+                "parameters": {
+                    "type": "object",
+                    "properties": {},
+                    "required": [],
+                },
+            },
+        }
+
+        # Iterate over the fields to add them to the parameters
+        for field_name, field_info in cls.model_fields.items():
+            
+            # Field properties
+            field_type = "string"  # Default to string, will adjust if it's a different type
+            annotation = field_info.annotation.__args__[0] if getattr(field_info.annotation, "__origin__", None) is Union else field_info.annotation
+            
+            has_none = False
+            if get_origin(annotation) is UnionType:  # Check if it's a Union type
+                args = get_args(annotation)
+                if type(None) in args:
+                    has_none = True
+                args = [arg for arg in args if type(None) != arg]
+                if len(args) > 1:
+                    raise TypeError("It can be union of only a valid type (str, int, bool, etc) and None")
+                elif len(args) == 0:
+                    raise TypeError("There must be a valid type (str, int, bool, etc) not only None")
+                else:
+                    annotation = args[0]
+            
+            if annotation == int:
+                field_type = "integer"
+            elif annotation == bool:
+                field_type = "boolean"
+            
+            # Add the field's description and type to the properties
+            function_metadata["function"]["parameters"]["properties"][field_name] = {
+                "type": field_type,
+                "description": field_info.description,
+            }
+
+            # Determine if the field is required (not Optional or None)
+            if field_info.is_required():
+                function_metadata["function"]["parameters"]["required"].append(field_name)
+                has_none = True
+
+            # If there's an enum (like for `unit`), add it to the properties
+            if hasattr(field_info, 'default') and field_info.default is not None and isinstance(field_info.default, list):
+                function_metadata["function"]["parameters"]["properties"][field_name]["enum"] = field_info.default
+                if not has_none:
+                    function_metadata["function"]["parameters"]["required"].append(field_name)
+
+
+
+        return function_metadata
+
+
+
+# Load the configuration file
+# ===========================================================================
+config_file = "config.yaml"
+cfg = OmegaConf.load(config_file)
+
+
+# Initialize VectorStore, tools and oitools
+# ===========================================================================
+vdb = VectorStore(**cfg.vdb)
+tools: Dict[str, ToolBase] = {}
+oitools = []
+
+
+
+def tool_register(cls: BaseModel):
+    oaitool = cls.to_openai_tool()
+    
+    oitools.append(oaitool)
+    tools[oaitool["function"]["name"]] = cls
+
+
+@tool_register
+class retrieve_aina_data(ToolBase):
+    """Retrieves relevant information from Aina Challenge vectorstore, based on the user's query."""
+    logging.info("@tool_register: retrieve_aina_data()")
+
+    query: str = Field(description="The user's input or question, used to search in Aina Challenge vectorstore.")
+    logging.info(f"query: {query}")
+
+    @classmethod
+    def invoke(cls, input: Dict) -> str:
+        logging.info(f"retrieve_aina_data.invoke() input: {input}")
+        
+        # Check if the input is a dictionary
+        query = input.get("query", None)
+        if not query:
+            return "Missing required argument: query."
+        
+        return vdb.get_context(query)
+
+

src/vectorstore.py

@@ -0,0 +1,83 @@
+from langchain_community.vectorstores import FAISS
+# from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain_huggingface import HuggingFaceEmbeddings
+from huggingface_hub import snapshot_download
+import logging
+
+from termcolor import cprint
+
+
+class VectorStore:
+    def __init__(self, 
+                 embeddings_model: str, 
+                 vs_local_path: str = None, 
+                 vs_hf_path: str = None, 
+                 number_of_contexts: int = 2,
+                 context_template: str = "{}",
+                 embedding_score_threshold: float = None, 
+                 join_str: str = "\n\n"
+                 ):
+        
+        logging.info("Loading vectorstore...")
+
+
+        self.number_of_contexts = number_of_contexts
+        self.context_template = context_template
+        self.join_str = join_str
+        self.embedding_score_threshold = embedding_score_threshold
+
+        embeddings = HuggingFaceEmbeddings(model_name=embeddings_model)
+        logging.info(f"Loaded embeddings model: {embeddings_model}")
+
+
+        if vs_hf_path:
+            hf_vectorstore = snapshot_download(repo_id=vs_hf_path)
+            self.vdb = FAISS.load_local(hf_vectorstore, embeddings, allow_dangerous_deserialization=True)
+            logging.info(f"Loaded vectorstore from {vs_hf_path}")
+        else:
+            self.vdb = FAISS.load_local(vs_local_path, embeddings, allow_dangerous_deserialization=True)
+            logging.info(f"Loaded vectorstore from {vs_local_path}")
+
+    
+    def get_context(self, query,):
+
+
+        # Retrieve documents
+        results = self.vdb.similarity_search_with_relevance_scores(query=query, k=self.number_of_contexts, score_threshold=self.embedding_score_threshold)
+        logging.info(f"Retrieved {len(results)} documents from the vectorstore.")
+
+        # Return formatted context
+        return self._beautiful_context(results)
+
+
+    def _beautiful_context(self, docs):
+        
+        logging.info(f"Formatting {len(docs)} contexts...")
+
+
+        contexts = []
+        
+        for i, doc in enumerate(docs):
+
+            print()    
+            cprint("-"*150, "yellow")
+            cprint(f"Document {i}:", "yellow")
+            cprint(f"Score: {doc[1]}", "yellow")
+            cprint("-"*150, "yellow")
+            print(doc[0].page_content)
+            
+            contexts.append(doc[0].page_content)
+
+        
+        context = self.join_str.join(contexts)
+        
+        
+        print()
+        cprint("-"*150, "green")
+        cprint(f"Final formatted context:", "green")
+        cprint("-"*150, "green")
+        print(context)
+
+
+        return context
+