Upload 11 files

aiclient.py

@@ -0,0 +1,142 @@
+# aiclient.py
+
+import os
+import time
+import json
+from typing import List, Dict, Optional, Union, AsyncGenerator
+from openai import AsyncOpenAI
+from starlette.responses import StreamingResponse
+from observability import log_execution ,LLMObservabilityManager
+import psycopg2
+import requests
+from functools import lru_cache
+import logging
+import pandas as pd
+logger = logging.getLogger(__name__)
+
+@lru_cache(maxsize=1)
+def get_model_info():
+    try:
+        model_info_dict = requests.get(
+            'https://openrouter.ai/api/v1/models',
+            headers={'accept': 'application/json'}
+        ).json()["data"]
+        
+        # Save the model info to a JSON file
+        with open('model_info.json', 'w') as json_file:
+            json.dump(model_info_dict, json_file, indent=4)
+
+    except Exception as e:
+        logger.error(f"Failed to fetch model info: {e}. Loading from file.")
+        if os.path.exists('model_info.json'):
+            with open('model_info.json', 'r') as json_file:
+                model_info_dict = json.load(json_file)
+                model_info = pd.DataFrame(model_info_dict)
+                return model_info
+
+        else:
+            logger.error("No model info file found")
+            return None
+
+    model_info = pd.DataFrame(model_info_dict)
+    return model_info
+
+class AIClient:
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            base_url="https://openrouter.ai/api/v1",
+            api_key=os.environ['OPENROUTER_API_KEY']
+        )
+        self.observability_manager = LLMObservabilityManager()
+        self.model_info = get_model_info()
+
+    #@log_execution
+    async def generate_response(
+        self,
+        messages: List[Dict[str, str]],
+        model: str = "openai/gpt-4o-mini",
+        max_tokens: int = 32000,
+        conversation_id: Optional[str] = None,
+        user: str = "anonymous"
+    ) -> AsyncGenerator[str, None]:
+        if not messages:
+            return
+        
+        start_time = time.time()
+        full_response = ""
+        usage = {"completion_tokens": 0, "prompt_tokens": 0, "total_tokens": 0}
+        status = "success"
+
+        try:
+            response = await self.client.chat.completions.create(
+                model=model,
+                messages=messages,
+                max_tokens=max_tokens,
+                stream=True,
+                stream_options={"include_usage": True}
+            )
+            end_time = time.time()
+            latency = end_time - start_time
+
+            async for chunk in response:
+                if chunk.choices[0].delta.content:
+                    yield chunk.choices[0].delta.content
+                    full_response += chunk.choices[0].delta.content
+                
+                if chunk.usage:
+                    model = chunk.model
+                    usage["completion_tokens"] = chunk.usage.completion_tokens
+                    usage["prompt_tokens"] = chunk.usage.prompt_tokens
+                    usage["total_tokens"] = chunk.usage.total_tokens
+                    print(usage)
+                    print(model)
+        
+        except Exception as e:
+            status = "error"
+            full_response = str(e)
+            latency = time.time() - start_time
+            print(f"Error in generate_response: {e}")
+        
+        finally:
+            # Log the observation
+            try:
+                pricing_data = self.model_info[self.model_info.id == model]["pricing"].values[0]
+                cost = float(pricing_data["completion"]) * float(usage["completion_tokens"]) + float(pricing_data["prompt"]) * float(usage["prompt_tokens"])
+                self.observability_manager.insert_observation(
+                    response=full_response,
+                    model=model,
+                    completion_tokens=usage["completion_tokens"],
+                    prompt_tokens=usage["prompt_tokens"],
+                    total_tokens=usage["total_tokens"],
+                    cost=cost,
+                    conversation_id=conversation_id or "default",
+                    status=status,
+                    request=json.dumps([msg for msg in messages if msg.get('role') != 'system']),
+                    latency=latency,
+                    user=user
+                )
+            except Exception as obs_error:
+                print(f"Error logging observation: {obs_error}")
+
+
+class DatabaseManager:
+    """Manages database operations."""
+
+    def __init__(self):
+        self.db_params = {
+            "dbname": "postgres",
+            "user": os.environ['SUPABASE_USER'],
+            "password": os.environ['SUPABASE_PASSWORD'],
+            "host": "aws-0-us-west-1.pooler.supabase.com",
+            "port": "5432"
+        }
+
+    @log_execution
+    def update_database(self, user_id: str, user_query: str, response: str) -> None:
+        with psycopg2.connect(**self.db_params) as conn:
+            with conn.cursor() as cur:
+                insert_query = """
+                INSERT INTO ai_document_generator (user_id, user_query, response)
+                VALUES (%s, %s, %s);
+                """
+                cur.execute(insert_query, (user_id, user_query, response))

combined_digi_yatra.csv

@@ -0,0 +1,84 @@
+"title/question","paragraph/answer","url"
+"Introduction to Digi Yatra","Digi Yatra is a decentralized digital wallet system for air travel in India. It aims to provide a seamless and paperless journey for passengers by automating identity verification and boarding pass checks at airports. The system does not store any centralized data, addressing privacy concerns.",""
+"How Digi Yatra Works","Users register on the Digi Yatra app by linking their Aadhaar (national ID) and creating a verified credential. When traveling, they link their boarding pass to the app. At the airport, facial recognition is used to verify identity and boarding pass details at entry, security, and boarding gates.",""
+"Benefits and Adoption","Digi Yatra has been adopted by 4.5 million users and used 19 million times across 14 airports in 15 months since launch. It improves security by validating IDs against original databases and automating manual checks. The system is especially appreciated by older travelers for its ease of use.",""
+"Privacy and Data Handling","No passenger data is stored centrally by Digi Yatra. Airports only keep biometric data for 24 hours as per regulations. Users can delete their data from the app at any time. The system aims for 'privacy by design' while facilitating secure data sharing between stakeholders.",""
+"Challenges and Future Improvements","Current challenges include standardizing data formats across airlines and airports, integrating with more identity providers beyond Aadhaar, and expanding to international travel. Future improvements may include direct integration with airline reservation systems and adding support for passport-based verification.",""
+"Stakeholder Integration","Digi Yatra aims to improve data sharing between airports, airlines, and security agencies to create a truly seamless passenger journey. However, this requires overcoming technical and regulatory hurdles in standardizing data exchange and access.",""
+"What is Digi Yatra?","Digi Yatra is a digital platform that aims to make air travel in India seamless and hassle-free. It uses facial recognition technology to allow passengers to go through various checkpoints at airports without needing to show physical documents. The service is voluntary and passengers can choose to use manual processes if they prefer.",""
+"How does Digi Yatra work?","Digi Yatra works by allowing passengers to create a digital identity using their Aadhaar card and a selfie. This identity is then linked to their boarding pass. At the airport, facial recognition cameras at various checkpoints verify the passenger's identity, allowing them to proceed without showing physical documents.",""
+"Is Digi Yatra mandatory?","No, Digi Yatra is not mandatory. It is a voluntary service that passengers can choose to use. Those who prefer can still opt for manual document checks at airports.",""
+"What are the benefits of using Digi Yatra?","The main benefits of using Digi Yatra include faster airport processing, reduced queuing times, seamless and contactless passage through various checkpoints, and a more hygienic travel experience as it minimizes physical contact and document exchanges.",""
+"How do I enroll for Digi Yatra?","You can enroll for Digi Yatra by downloading the Digi Yatra app and creating your digital identity using your Aadhaar card and a selfie. The app is available for both iOS and Android devices.",""
+"Is my data safe with Digi Yatra?","Digi Yatra is built on the principles of privacy by design. Your personally identifiable information (PII) is not stored in any central repository. It is only stored on your smartphone in the Digi Yatra app's secure wallet. Data shared with airports is purged within 24 hours of your flight's departure.",""
+"At which airports is Digi Yatra available?","As of 2023, Digi Yatra is available at several major airports in India including Delhi, Bengaluru, Varanasi, and Hyderabad. The service is being rolled out to more airports in phases.",""
+"Can I use Digi Yatra for international flights?","As of 2023, Digi Yatra is only available for domestic flights in India. The service for international flights is on the roadmap and will be implemented in phases.",""
+"What do I need to use Digi Yatra at the airport?","To use Digi Yatra at the airport, you need to have the Digi Yatra app installed on your smartphone with your digital identity created. You also need to upload your boarding pass to the app before your travel.",""
+"How do I add my minor child to my Digi Yatra account?","To add a minor child to your Digi Yatra account, you first need to create your own account. Then, on the app's home screen, click on 'Add Credentials', then 'Add minor Credentials'. Follow the process to create the child's identity credentials. You can then add their boarding pass similarly to yours.",""
+"What if I face issues with the Digi Yatra app?","If you face any issues with the Digi Yatra app, you can contact their customer support at customercare@digiyatrafoundation.com. It's helpful to include details about your device model, operating system version, and app version when reporting issues.",""
+"Who operates Digi Yatra?","Digi Yatra is operated by the Digi Yatra Foundation, a not-for-profit company under section 8 of the Companies Act 2013. It is an industry-led initiative promoted by the Ministry of Civil Aviation.",""
+"How is passenger consent handled in Digi Yatra?","Digi Yatra takes passenger consent at various stages. For app-based enrollment, consent is taken before creating identity credentials and before sharing these credentials. For 'day of travel' enrollment at airports, consent is taken before capturing the passenger's face.",""
+"Can I delete my Digi Yatra data?","Yes, you can delete your Digi Yatra data by uninstalling the app from your phone. Since no data is stored centrally, uninstalling the app effectively removes all your data from the system.",""
+"Is Digi Yatra available for all airlines?","Digi Yatra is designed to work with all airlines operating domestic flights in India. However, the integration may vary between airlines and airports. It's best to check with your specific airline or airport for the most up-to-date information.",""
+"What happens if the Digi Yatra system is down at the airport?","If the Digi Yatra system is down at the airport, you can use the regular manual processes for identity verification and boarding. The manual processes are always available as an alternative.",""
+"How does Digi Yatra ensure data privacy?","Digi Yatra ensures data privacy by not storing any personally identifiable information in a central database. All your data is stored only on your phone. The data shared with airports for verification is deleted within 24 hours of your flight's departure.",""
+"Can I use Digi Yatra if I don't have an Aadhaar card?","Currently, Digi Yatra primarily uses Aadhaar for identity verification. However, they are working on incorporating other forms of ID, including passports, in the future.",""
+"Does Digi Yatra track my travel history?","No, Digi Yatra does not track or store your travel history centrally. Any travel information is only stored locally on your phone and can be deleted by uninstalling the app.",""
+"How does Digi Yatra handle data security?","Digi Yatra handles data security by implementing 'Privacy by Design' principles. They do not store any central database of user information, minimizing the risk of data breaches. The system is regularly audited by CERT-In empaneled agencies.",""
+"Enrol on the Digi Yatra App using Aadhaar for a hassle-free airport journey","https://www.youtube.com/watch?v=y6xTGrpfAGs"
+"Enrol on the Digi Yatra App using DigiLocker","https://www.youtube.com/watch?v=7z7iLNSkFg4"
+"Enrol on the Digi Yatra App in 3 Simple Steps With Digi Yatra, you can register in just three easy steps using DigiLocker or Aadhaar.
+Step 1: Register on the App
+Step 2: Verify and create your credentials
+Step 3: Share boarding details","https://www.youtube.com/watch?v=-nlJBCwg0nM"
+"SELF-SOVEREIGN IDENTITY | DATA SAFETY WITH DIGI YATRA","https://www.youtube.com/watch?v=Wlc6iqgwQDU"
+"HOW TO ADD A DEPENDENT/MINOR | DIGI YATRA MOBILE APP Traveling with family just got easier! 🌟 Learn how to add dependents or minors to your Digi Yatra app for seamless travel experiences. Watch now and make your next family trip a breeze!","https://www.youtube.com/watch?v=j2lw7LljME0"
+"Does Digi Yatra use centralized data storage?","No, Digi Yatra does not use centralized storage; all personal information is saved in the passenger’s own devices."
+"Where is a passenger's personal information stored in Digi Yatra?","A passenger's personal information is stored in the mobile wallet of the traveler’s smartphone."
+"How does Digi Yatra ensure data privacy?","Digi Yatra Central Ecosystem (DYCE) is built on the principles of privacy by design and default, with no centralized storage of Personally Identifiable Information (PII). Data is encrypted, shared only with the departure airport, and purged 24 hours after flight departure."
+"Who manages the Digi Yatra Central Ecosystem?","The Digi Yatra Central Ecosystem is managed by Digi Yatra Foundation, a Not-For-Profit company established under Section 8 of the Companies Act, 2013."
+"Is Digi Yatra subject to the Right to Information (RTI) Act?","No, Digi Yatra Foundation does not come under the ambit of the Right to Information (RTI) Act."
+"How is data security maintained in Digi Yatra?","Digi Yatra processes are audited and certified by CERT-In empanelled agencies to ensure adherence to data privacy and security standards."
+"How are Digi Yatra guidelines issued?","Digi Yatra Guidelines are issued by the Directorate General of Civil Aviation (DGCA) through Aeronautical Information Circular (AIC) No. 09/2022 dated April 18, 2022."
+"How does the Digi Yatra app work for passengers?","Passengers register their details on the Digi Yatra app using Aadhaar-based validation and a self-image capture. The boarding pass is scanned, and credentials are shared with the airport. At the airport e-gate, the boarding pass is scanned, and facial recognition validates the passenger's identity and travel documents."
+"Is Digi Yatra mandatory for passengers?","No, Digi Yatra is purely voluntary for air passengers. Data is collected only with the consent of passengers."
+"At how many airports is Digi Yatra currently available for domestic passengers?","Digi Yatra is currently available at 13 airports for domestic passengers."
+"Which airports currently have Digi Yatra implemented?","Delhi, Bengaluru, Varanasi, Hyderabad, Kolkata, Vijayawada, Pune, Mumbai, Cochin, Ahmedabad, Lucknow, Jaipur, and Guwahati."
+"How many additional airports will Digi Yatra be introduced to this year?","The government plans to roll out Digi Yatra at 14 more airports this year."
+"Which airports are scheduled to receive Digi Yatra in the first phase?","Chennai, Bhubaneshwar, Coimbatore, Dabolim, Mopa Goa, Indore, Bagdogra, Chandigarh, Ranchi, Nagpur, Patna, Raipur, Srinagar, and Vishakhapatnam."
+"How many airports will have Digi Yatra by the end of 2024?","By the end of 2024, Digi Yatra will be available at 38 airports."
+"When was Digi Yatra introduced?","Digi Yatra was introduced in December 2022."
+"What percentage of India's domestic air passenger traffic is handled by airports with Digi Yatra?","Currently, 13 airports handle around 85 percent of the country's domestic air passenger traffic with Digi Yatra."
+"What are the future plans for Digi Yatra in 2025?","Digi Yatra will be implemented at 11 more airports in 2025, and e-passport based enrollment will be introduced to allow foreign citizens to use the facility."
+"How has the number of Digi Yatra app users changed from December 2022 to November 2023?","There has been a significant growth in the total number of Digi Yatra app users during this period."
+"How does Digi Yatra benefit passengers at the airport?","Digi Yatra reduces processing time at entry and boarding gates, allowing passengers to access terminals and boarding gates seamlessly using facial recognition technology."
+"Which organization is the nodal body for Digi Yatra?","Digi Yatra Foundation is the nodal body for Digi Yatra."
+"How does Digi Yatra handle data after flight departure?","Data is purged from the system 24 hours after the departure of the flight."
+"Is Digi Yatra available for foreign citizens?","The government plans to make Digi Yatra available for foreign citizens by implementing e-passport based enrollment."
+"How has Digi Yatra been received at Pune Airport?","Over 1 million passengers have used Digi Yatra at Pune Airport as of January 31, 2024, with around 57 percent of passengers preferring the service."
+"What improvements have been seen since the launch of Digi Yatra at Guwahati Airport?","The introduction of Digi Yatra at Guwahati Airport has transformed the passenger experience, leading to a surge in usage and positive feedback."
+"What percentage of passengers used Digi Yatra services recently according to MOCA advisory?","At least 10 percent of passengers are expected to use Digi Yatra services, with recent usage increasing to 11.9 percent."
+"Where are Digi Yatra services available at Guwahati Airport?","Digi Yatra services are available at departure gates D-10 and D-7, entry into the terminal, and all boarding gates."
+"When was Digi Yatra officially launched at Guwahati Airport?","Digi Yatra was officially inaugurated at Guwahati Airport in August 2023."
+"What benefits did Digi Yatra bring to Pune Airport?","Digi Yatra allowed passengers to access the terminal in seconds without queuing for check-in, increasing the number of passengers using the service."
+"How does Digi Yatra handle the boarding process?","At the airport e-gate, passengers scan their bar-coded boarding pass and undergo facial recognition to validate their identity and travel documents before entering through the e-gate."
+"Who informed the Rajya Sabha about Digi Yatra's data storage practices?","The Civil Aviation Ministry informed the Rajya Sabha about Digi Yatra's data storage practices."
+"What is the role of CERT-In empanelled agencies in Digi Yatra?","CERT-In empanelled agencies audit and certify Digi Yatra processes to ensure data privacy and security standards are met."
+"Is Digi Yatra considered under the Right to Information (RTI) Act?","No, because Digi Yatra Foundation is a Not-For-Profit company established under Section 8 of the Companies Act, 2013."
+"How is personal information shared in Digi Yatra?","Personal information is shared with the departure airport in an encrypted format."
+"What technology underpins Digi Yatra's boarding system?","Facial Recognition Technology (FRT) is used for biometric boarding in Digi Yatra."
+"How does Digi Yatra enhance the passenger experience at airports?","By enabling contactless and seamless movement through various checkpoints using facial recognition, reducing wait times and eliminating the need for physical boarding passes."
+"What milestone did Pune Airport achieve with Digi Yatra on January 31, 2024?","Pune Airport reached the milestone of 1 million passengers traveling through Digi Yatra."
+"How did passengers initially respond to Digi Yatra at Pune Airport?","Initially, passenger response was very low, and some airlines did not support the service."
+"How did the adoption of Digi Yatra at Pune Airport change over time?","Over time, both passengers and various airlines started preferring Digi Yatra, increasing usage significantly."
+"What actions did Civil Aviation Minister Jyotiraditya Scindia announce regarding Digi Yatra?","He announced that Digi Yatra will be available at 25 more airports in 2024 and reiterated that the service is voluntary for passengers."
+"How does Digi Yatra handle the boarding pass?","Passengers scan their bar-coded boarding pass, which is then validated through facial recognition at the e-gate."
+"Can passengers still follow normal procedures after using Digi Yatra?","Yes, passengers still follow normal procedures to clear security and board the aircraft after using Digi Yatra."
+"What is Digi Yatra?","Digi Yatra is a Ministry of Civil Aviation, Govt. of India led initiative to make air traveller's/ passenger's journey seamless, hassle-free and Health-Risk-Free. The Digi Yatra process uses the single token of face biometrics to digitally validate the Identity, Travel, Health or any other data that is needed for the purpose of enabling air travel."
+"What is the solution based on and is my data safe?","The Solution is built on W3C standards using Self Sovereign Identity (SSI), Verifiable Credentials (VC), Decentralized Identifiers (DIDs) and uses a Distributed Ledger for decentralised layer of trust between the various participants of the ecosystem."
+"Do I have to Register for every travel?","Registration/ enrolment to the DigiYatra App is a one-time process. You don't have to register every time you travel. For each new travel instance, simply add the new travel document and share your credentials prior to your travel."
+"Why can't we process international passengers?","Processing passengers on International flights is on the roadmap of the Digi Yatra Central Ecosystem and will be taken up in a phased manner."
+"How about passengers with turban?","The Biometric algorithm extracts the template for a face match using certain points on the face. Wearing a turban does not impact this validation."
+"How do you process if I am transiting?","Entry check and other validations shall remain the same at the departing airport even if passenger is going through Transit/ Transfer at another airport. However, passenger processing at the Transit/ Transfer airports will be taken up at a later date."
+"Why is only Aadhar accepted as the Govt ID?","At this time, the Digi Yatra app accepts only AADHAAR as it has been integrated, however, we plan to integrate additional government issued identity documents in the future."
+

limit_tokens.py

@@ -0,0 +1,83 @@
+from typing import List
+
+import tiktoken
+from langchain_core.messages import BaseMessage, ToolMessage, HumanMessage, AIMessage, SystemMessage, trim_messages
+
+
+def str_token_counter(text: str) -> int:
+    enc = tiktoken.get_encoding("o200k_base")
+    return len(enc.encode(text))
+
+
+def tiktoken_counter(messages: List[BaseMessage]) -> int:
+    """Approximately reproduce https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb
+
+    For simplicity only supports str Message.contents.
+    """
+    num_tokens = 3  # every reply is primed with <|start|>assistant<|message|>
+    tokens_per_message = 3
+    tokens_per_name = 1
+    for msg in messages:
+        if isinstance(msg, HumanMessage):
+            role = "user"
+        elif isinstance(msg, AIMessage):
+            role = "assistant"
+        elif isinstance(msg, ToolMessage):
+            role = "tool"
+        elif isinstance(msg, SystemMessage):
+            role = "system"
+        else:
+            raise ValueError(f"Unsupported messages type {msg.__class__}")
+        num_tokens += (
+            tokens_per_message
+            + str_token_counter(role)
+            + str_token_counter(msg.content)
+        )
+        if msg.name:
+            num_tokens += tokens_per_name + str_token_counter(msg.name)
+    return num_tokens
+
+def convert_to_openai_messages(messages: List[BaseMessage]) -> List[dict]:
+    """Convert LangChain messages to OpenAI format."""
+    openai_messages = []
+    
+    for msg in messages:
+        message_dict = {"content": msg.content}
+        
+        if isinstance(msg, HumanMessage):
+            message_dict["role"] = "user"
+        elif isinstance(msg, AIMessage):
+            message_dict["role"] = "assistant"
+        elif isinstance(msg, SystemMessage):
+            message_dict["role"] = "system"
+        elif isinstance(msg, ToolMessage):
+            message_dict["role"] = "tool"
+        else:
+            raise ValueError(f"Unsupported message type: {msg.__class__}")
+            
+        if msg.name:
+            message_dict["name"] = msg.name
+            
+        openai_messages.append(message_dict)
+    
+    return openai_messages
+
+def trim_messages_openai(messages: List[BaseMessage]) -> List[dict]:
+    """Trim LangChain messages and convert to OpenAI format."""
+    
+    trimmed_messages = trim_messages(
+        messages,
+        token_counter=tiktoken_counter,
+        strategy="last",
+        max_tokens=45,
+        start_on="human",
+        end_on=("human", "tool"),
+        include_system=True,
+    )
+    openai_format_messages = convert_to_openai_messages(trimmed_messages)
+
+    return openai_format_messages
+
+# Test
+# messages = [SystemMessage(content="You are a helpful assistant."), HumanMessage(query)]
+# openai_format_messages = trim_messages_openai(messages)

llamaindex.py

@@ -0,0 +1,31 @@
+# %pip install llama-index llama-index-vector-stores-lancedb
+# %pip install lancedb==0.6.13 #Only required if the above cell installs an older version of lancedb (pypi package may not be released yet)
+# %pip install llama-index-embeddings-fastembed
+# pip install llama-index-readers-file
+from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex
+from llama_index.vector_stores.lancedb import LanceDBVectorStore
+from llama_index.embeddings.fastembed import FastEmbedEmbedding
+
+# Configure global settings
+Settings.embed_model = FastEmbedEmbedding(model_name="BAAI/bge-small-en-v1.5")
+
+# Setup LanceDB vector store
+vector_store = LanceDBVectorStore(
+    uri="./lancedb",
+    mode="overwrite",
+    query_type="vector"
+)
+
+# Load your documents
+documents = SimpleDirectoryReader("D:\DEV\LIZMOTORS\LANGCHAIN\digiyatrav2\chatbot\data").load_data()
+
+# Create the index
+index = VectorStoreIndex.from_documents(
+    documents,
+    vector_store=vector_store
+)
+
+# Create a retriever
+retriever = index.as_retriever()
+response = retriever.retrieve("Your query here")
+print(response)

observability.py

@@ -0,0 +1,176 @@
+# File: llm_observability.py
+import sqlite3
+import json
+from datetime import datetime
+from typing import Dict, Any, List, Optional, Callable
+import logging
+import functools
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+def log_execution(func: Callable) -> Callable:
+    @functools.wraps(func)
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        logger.info(f"Executing {func.__name__}")
+        try:
+            result = func(*args, **kwargs)
+            logger.info(f"{func.__name__} completed successfully")
+            return result
+        except Exception as e:
+            logger.error(f"Error in {func.__name__}: {e}")
+            raise
+    return wrapper
+
+
+class LLMObservabilityManager:
+    def __init__(self, db_path: str = "llm_observability_v2.db"):
+        self.db_path = db_path
+        self.create_table()
+
+    def create_table(self):
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                CREATE TABLE IF NOT EXISTS llm_observations (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    conversation_id TEXT,
+                    created_at DATETIME,
+                    status TEXT,
+                    request TEXT,
+                    response TEXT,
+                    model TEXT,
+                    prompt_tokens INTEGER,
+                    completion_tokens INTEGER,
+                    total_tokens INTEGER,
+                    cost FLOAT,
+                    latency FLOAT,
+                    user TEXT
+                )
+            ''')
+
+    def insert_observation(self, response: str, conversation_id: str, status: str, request: str, model: str, prompt_tokens: int,completion_tokens: int, total_tokens: int, cost: float, latency: float, user: str):
+        created_at = datetime.now()
+        
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                INSERT INTO llm_observations 
+                (conversation_id, created_at, status, request, response, model, prompt_tokens, completion_tokens,total_tokens, cost, latency, user)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ''', (
+                conversation_id,
+                created_at,
+                status,
+                request,
+                response,
+                model,
+                prompt_tokens,
+                completion_tokens,
+                total_tokens,
+                cost,
+                latency,
+                user
+            ))
+
+    def get_observations(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            if conversation_id:
+                cursor.execute('SELECT * FROM llm_observations WHERE conversation_id = ? ORDER BY created_at', (conversation_id,))
+            else:
+                cursor.execute('SELECT * FROM llm_observations ORDER BY created_at')
+            rows = cursor.fetchall()
+
+            column_names = [description[0] for description in cursor.description]
+            return [dict(zip(column_names, row)) for row in rows]
+
+    def get_all_observations(self) -> List[Dict[str, Any]]:
+        return self.get_observations()
+    
+    def get_all_unique_conversation_observations(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            # Get the latest observation for each unique conversation_id
+            query = '''
+                SELECT * FROM llm_observations o1
+                WHERE created_at = (
+                    SELECT MAX(created_at) 
+                    FROM llm_observations o2 
+                    WHERE o2.conversation_id = o1.conversation_id
+                )
+                ORDER BY created_at DESC
+            '''
+            if limit is not None:
+                query += f' LIMIT {limit}'
+                
+            cursor.execute(query)
+            rows = cursor.fetchall()
+            
+            column_names = [description[0] for description in cursor.description]
+            return [dict(zip(column_names, row)) for row in rows]
+
+    ## OBSERVABILITY
+from uuid import uuid4
+import csv
+from io import StringIO
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from starlette.responses import StreamingResponse
+
+
+
+router = APIRouter( 
+    prefix="/observability",
+    tags=["observability"]
+)
+
+class ObservationResponse(BaseModel):
+    observations: List[Dict]
+    
+def create_csv_response(observations: List[Dict]) -> StreamingResponse:
+    def iter_csv(data):
+        output = StringIO()
+        writer = csv.DictWriter(output, fieldnames=data[0].keys() if data else [])
+        writer.writeheader()
+        for row in data:
+            writer.writerow(row)
+        output.seek(0)
+        yield output.read()
+
+    headers = {
+        'Content-Disposition': 'attachment; filename="observations.csv"'
+    }
+    return StreamingResponse(iter_csv(observations), media_type="text/csv", headers=headers)
+    
+
+@router.get("/last-observations/{limit}")
+async def get_last_observations(limit: int = 10, format: str = "json"):
+    observability_manager = LLMObservabilityManager()
+    
+    try:
+        # Get all observations, sorted by created_at in descending order
+        all_observations = observability_manager.get_observations()
+        all_observations.sort(key=lambda x: x['created_at'], reverse=True)
+        
+        # Get the last conversation_id
+        if all_observations:
+            last_conversation_id = all_observations[0]['conversation_id']
+            
+            # Filter observations for the last conversation
+            last_conversation_observations = [
+                obs for obs in all_observations
+                if obs['conversation_id'] == last_conversation_id
+            ][:limit]
+            
+            if format.lower() == "csv":
+                return create_csv_response(last_conversation_observations)
+            else:
+                return ObservationResponse(observations=last_conversation_observations)
+        else:
+            if format.lower() == "csv":
+                return create_csv_response([])
+            else:
+                return ObservationResponse(observations=[])
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve observations: {str(e)}")

observability_router.py

@@ -0,0 +1,73 @@
+## OBSERVABILITY
+from uuid import uuid4
+import csv
+from io import StringIO
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from typing import List, Dict, Optional
+from observability import LLMObservabilityManager
+
+
+router = APIRouter(
+    prefix="/observability",
+    tags=["observability"]
+)
+
+
+
+class ObservationResponse(BaseModel):
+    observations: List[Dict]
+    
+def create_csv_response(observations: List[Dict]) -> StreamingResponse:
+    def iter_csv(data):
+        output = StringIO()
+        writer = csv.DictWriter(output, fieldnames=data[0].keys() if data else [])
+        writer.writeheader()
+        for row in data:
+            writer.writerow(row)
+        output.seek(0)
+        yield output.read()
+
+    headers = {
+        'Content-Disposition': 'attachment; filename="observations.csv"'
+    }
+    return StreamingResponse(iter_csv(observations), media_type="text/csv", headers=headers)
+    
+
+@router.get("/last-observations/{limit}")
+async def get_last_observations(limit: int = 10, format: str = "json"):
+    observability_manager = LLMObservabilityManager()
+    
+    try:
+        # Get all observations, sorted by created_at in descending order
+        all_observations = observability_manager.get_observations()
+        all_observations.sort(key=lambda x: x['created_at'], reverse=True)
+        
+        # Get the last conversation_id
+        if all_observations:
+            last_conversation_id = all_observations[0]['conversation_id']
+            
+            # Filter observations for the last conversation
+            last_conversation_observations = [
+                obs for obs in all_observations
+                if obs['conversation_id'] == last_conversation_id
+            ][:limit]
+            
+            if format.lower() == "csv":
+                return create_csv_response(last_conversation_observations)
+            else:
+                return ObservationResponse(observations=last_conversation_observations)
+        else:
+            if format.lower() == "csv":
+                return create_csv_response([])
+            else:
+                return ObservationResponse(observations=[])
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve observations: {str(e)}")
+
+@router.get("/all-unique-observations")
+async def get_all_unique_observations(limit: Optional[int] = None):
+    observability_manager = LLMObservabilityManager()
+    return ObservationResponse(observations=observability_manager.get_all_unique_conversation_observations(limit))
+

prompts.py

@@ -0,0 +1,26 @@
+FOLLOWUP_DIGIYATRA_PROMPT = """
+You are a helpful, assistant tasked to assist digiyatra users, who can create interactive buttons and markdown responses. Provide youtube links to the user if relevant links are given in the context.
+If the user request needs further clarification, provide clarifying questions using <interact> to assist the user.
+Else respond with a helpful answer using <response>.   
+The options in <interact> tags will be rendered as buttons so that user can interact with it. Hence make sure to use it for engaging with the user with Next Steps, followup questions, quizzes etc. whichever appropriate 
+Your output format: # Use the following <response>,<interact> tags in any order as many times required.
+<response>response to user request formatted using markdown</response>
+<interact>
+questions:
+  - text: [First interaction question]
+    options:
+      - [Option 1]
+      - [Option 2]
+      - [Option 3]
+      - [Option 4 (if needed)]
+  - text: [Second interaction question]
+    options:
+      - [Option 1]
+      - [Option 2]
+      - [Option 3]
+  # Add more questions as needed 
+  # make sure this section is in valid YAML format
+</interact>
+
+Refer the FAQ Context for ANSWERING the user query.
+"""

rag_routerv2.py

@@ -0,0 +1,185 @@
+from fastapi import FastAPI, Depends, HTTPException, UploadFile, File
+import pandas as pd
+import lancedb
+from functools import cached_property, lru_cache
+from pydantic import Field, BaseModel
+from typing import Optional, Dict, List, Annotated, Any
+from fastapi import APIRouter
+import uuid
+import io
+from io import BytesIO
+import csv
+
+# LlamaIndex imports
+from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex
+from llama_index.vector_stores.lancedb import LanceDBVectorStore
+from llama_index.embeddings.fastembed import FastEmbedEmbedding
+from llama_index.core import StorageContext, load_index_from_storage
+import json
+import os
+import shutil
+
+router = APIRouter(
+    prefix="/rag",
+    tags=["rag"]
+)
+
+# Configure global LlamaIndex settings
+Settings.embed_model = FastEmbedEmbedding(model_name="BAAI/bge-small-en-v1.5")
+tables_file_path = './data/tables.json'
+
+# Database connection dependency
+@lru_cache()
+def get_db_connection(db_path: str = "./lancedb/dev"):
+    return lancedb.connect(db_path)
+
+# Pydantic models
+class CreateTableResponse(BaseModel):
+    table_id: str
+    message: str
+    status: str
+
+class QueryTableResponse(BaseModel):
+    results: Dict[str, Any]
+    total_results: int
+
+
+@router.post("/create_table", response_model=CreateTableResponse)
+async def create_embedding_table(
+    user_id: str,
+    files: List[UploadFile] = File(...),
+    table_id: Optional[str] = None
+) -> CreateTableResponse:
+    """Create a table and load embeddings from uploaded files using LlamaIndex."""
+    allowed_extensions = {".pdf", ".docx", ".csv", ".txt", ".md"}
+    for file in files:
+        if file.filename is None:
+            raise HTTPException(status_code=400, detail="File must have a valid name.")
+        file_extension = os.path.splitext(file.filename)[1].lower()
+        if file_extension not in allowed_extensions:
+            raise HTTPException(
+                status_code=400, 
+                detail=f"File type {file_extension} is not allowed. Supported file types are: {', '.join(allowed_extensions)}."
+            )
+    
+    if table_id is None:
+        table_id = str(uuid.uuid4())
+    table_name = table_id #f"{user_id}__table__{table_id}"
+    
+    # Create a directory for the uploaded files
+    directory_path = f"./data/{table_name}"
+    os.makedirs(directory_path, exist_ok=True)
+
+    # Save each uploaded file to the data directory
+    for file in files:
+        file_path = os.path.join(directory_path, file.filename)
+        with open(file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+
+    # Store user_id and table_name in a JSON file
+    try:
+        tables_file_path = './data/tables.json'
+        os.makedirs(os.path.dirname(tables_file_path), exist_ok=True)
+        # Load existing tables or create a new file if it doesn't exist
+        try:
+            with open(tables_file_path, 'r') as f:
+                tables = json.load(f)
+        except (FileNotFoundError, json.JSONDecodeError):
+            tables = {}
+
+        # Update the tables dictionary
+        if user_id not in tables:
+            tables[user_id] = []
+        if table_name not in tables[user_id]:
+            tables[user_id].append(table_name)
+
+        # Write the updated tables back to the JSON file
+        with open(tables_file_path, 'w') as f:
+            json.dump(tables, f)
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to update tables file: {str(e)}")
+    try:
+        # Setup LanceDB vector store
+        vector_store = LanceDBVectorStore(
+            uri="./lancedb/dev",
+            table_name=table_name,
+            # mode="overwrite",
+            # query_type="vector"
+        )
+
+        # Load documents using SimpleDirectoryReader
+        documents = SimpleDirectoryReader(directory_path).load_data()
+        
+        # Create the index
+        index = VectorStoreIndex.from_documents(
+            documents,
+            vector_store=vector_store
+        )
+        index.storage_context.persist(persist_dir=f"./lancedb/index/{table_name}")
+
+        return CreateTableResponse(
+            table_id=table_id,
+            message=f"Table created and documents indexed successfully",
+            status="success"
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Table creation failed: {str(e)}")
+
+@router.post("/query_table/{table_id}", response_model=QueryTableResponse)
+async def query_table(
+    table_id: str,
+    query: str,
+    user_id: str,
+    #db: Annotated[Any, Depends(get_db_connection)],
+    limit: Optional[int] = 10
+) -> QueryTableResponse:
+    """Query the database table using LlamaIndex."""
+    try:
+        table_name = table_id  #f"{user_id}__table__{table_id}"
+        
+        # load index and retriever
+        storage_context = StorageContext.from_defaults(persist_dir=f"./lancedb/index/{table_name}")
+        index = load_index_from_storage(storage_context)
+        retriever = index.as_retriever(similarity_top_k=limit)
+        
+        # Get response
+        response = retriever.retrieve(query)
+        
+        # Format results
+        results = [{
+            'text': node.text,
+            'score': node.score
+        } for node in response]
+        
+        return QueryTableResponse(
+            results={'data': results},
+            total_results=len(results)
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Query failed: {str(e)}")
+
+@router.get("/get_tables/{user_id}")
+async def get_tables(user_id: str):
+    """Get all tables for a user."""
+    
+    tables_file_path = './data/tables.json'
+    try:
+        # Load existing tables from the JSON file
+        with open(tables_file_path, 'r') as f:
+            tables = json.load(f)
+
+        # Retrieve tables for the specified user
+        user_tables = tables.get(user_id, [])
+        return user_tables
+
+    except (FileNotFoundError, json.JSONDecodeError):
+        return []  # Return an empty list if the file doesn't exist or is invalid
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve tables: {str(e)}")
+
+@router.get("/health")
+async def health_check():
+    return {"status": "healthy"}

requirements.txt

@@ -1,14 +1,4 @@
-fastapi[standard] 
-uvicorn
-openai
-psycopg2-binary
-tiktoken
-requests
-beautifulsoup4
-fastapi-cache2
-PyYAML
-psycopg2-binary
-pandas
-txtai
-llama-parse
-fast-langdetect
+llama-index-core
+llama-index-readers-file  
+llama-index-vector-stores-lancedb          
+llama_index.vector_stores                                                                                                                   

utils.py

@@ -0,0 +1,40 @@
+import re
+
+def parse_followup_and_tools(input_text):
+    # Remove extra brackets and excess quotes
+    cleaned_text = re.sub(r'\[|\]|"+', ' ', input_text)
+    
+    # Extract response content
+    response_pattern = re.compile(r'<response>(.*?)</response>', re.DOTALL)
+    response_parts = response_pattern.findall(cleaned_text)
+    combined_response = ' '.join(response_parts)
+    
+    # Normalize spaces in the combined response
+    combined_response = ' '.join(combined_response.split())
+    
+    parsed_interacts = []
+    parsed_tools = []
+    
+    # Parse interacts and tools
+    blocks = re.finditer(r'<(interact|tools?)(.*?)>(.*?)</\1>', cleaned_text, re.DOTALL)
+    for block in blocks:
+        block_type, _, content = block.groups()
+        content = content.strip()
+        
+        if block_type == 'interact':
+            question_blocks = re.split(r'\s*-\s*text:', content)[1:]
+            for qblock in question_blocks:
+                parts = re.split(r'\s*options:\s*', qblock, maxsplit=1)
+                if len(parts) == 2:
+                    question = ' '.join(parts[0].split())  # Normalize spaces
+                    options = [' '.join(opt.split()) for opt in re.split(r'\s*-\s*', parts[1]) if opt.strip()]
+                    parsed_interacts.append({'question': question, 'options': options})
+        
+        elif block_type.startswith('tool'):  # This will match both 'tool' and 'tools'
+            tool_match = re.search(r'text:\s*(.*?)\s*options:\s*-\s*(.*)', content, re.DOTALL)
+            if tool_match:
+                tool_name = ' '.join(tool_match.group(1).split())  # Normalize spaces
+                option = ' '.join(tool_match.group(2).split())  # Normalize spaces
+                parsed_tools.append({'name': tool_name, 'input': option})
+    
+    return combined_response, parsed_interacts, parsed_tools