Merge remote-tracking branch 'origin' into feature/talk_to_data

app.py

@@ -81,7 +81,7 @@ llm = get_llm(provider="openai", max_tokens=1024, temperature=0.0)
 if os.environ["GRADIO_ENV"] == "local":
     reranker = get_reranker("nano")
 else:
-    reranker = get_reranker("large")
+    reranker = get_reranker("nano")
 
 agent = make_graph_agent(
     llm=llm,

climateqa/chat.py

@@ -14,7 +14,7 @@ from .handle_stream_events import (
     handle_retrieved_owid_graphs,
 )
 from .logging import (
-    log_interaction_to_huggingface
+    log_interaction
 )
         
 # Chat functions
@@ -189,6 +189,6 @@ async def chat_stream(
         raise gr.Error(str(e))
 
     # Call the function to log interaction
-    log_interaction_to_huggingface(history, output_query, sources, docs, share_client, user_id)
+    log_interaction(history, output_query, sources, docs, share_client, user_id)
 
     yield history, docs_html, output_query, output_language, related_contents, graphs_html, follow_up_examples#, vanna_data

climateqa/constants.py

@@ -65,4 +65,39 @@ OWID_CATEGORIES = ['Access to Energy', 'Agricultural Production',
        'Oil Spills', 'Outdoor Air Pollution', 'Ozone Layer', 'Pandemics',
        'Pesticides', 'Plastic Pollution', 'Renewable Energy', 'Soil',
        'Transport', 'Urbanization', 'Waste Management', 'Water Pollution',
-       'Water Use & Stress', 'Wildfires']
+       'Water Use & Stress', 'Wildfires']
+
+
+DOCUMENT_METADATA_DEFAULT_VALUES = {
+    "chunk_type": "",
+    "document_id": "",
+    "document_number": 0.0,
+    "element_id": "",
+    "figure_code": "",
+    "file_size": "",
+    "image_path": "",
+    "n_pages": 0.0,
+    "name": "",
+    "num_characters": 0.0,
+    "num_tokens": 0.0,
+    "num_tokens_approx": 0.0,
+    "num_words": 0.0,
+    "page_number": 0,
+    "release_date": 0.0,
+    "report_type": "",
+    "section_header": "",
+    "short_name": "",
+    "source": "",
+    "toc_level0": "",
+    "toc_level1": "",
+    "toc_level2": "",
+    "toc_level3": "",
+    "url": "",
+    "similarity_score": 0.0,
+    "content": "",
+    "reranking_score": 0.0,
+    "query_used_for_retrieval": "",
+    "sources_used": [""],
+    "question_used": "",
+    "index_used": ""
+}

climateqa/logging.py

@@ -4,111 +4,157 @@ import json
 from huggingface_hub import HfApi
 import gradio as gr
 import csv
+import pandas as pd
+import io
+from typing import TypedDict, List
+from climateqa.constants import DOCUMENT_METADATA_DEFAULT_VALUES
+from langchain_core.documents import Document
 
-def serialize_docs(docs:list)->list:
+def serialize_docs(docs:list[Document])->list:
+    """Convert document objects to a simplified format compatible with Hugging Face datasets.
+    
+    This function processes document objects by extracting their page content and metadata,
+    normalizing the metadata structure to ensure consistency. It applies default values 
+    from DOCUMENT_METADATA_DEFAULT_VALUES for any missing metadata fields.
+    
+    Args:
+        docs (list): List of document objects, each with page_content and metadata attributes
+        
+    Returns:
+        list: List of dictionaries with standardized "page_content" and "metadata" fields
+    """
     new_docs = []
     for doc in docs:
-        new_doc = {}
-        new_doc["page_content"] = doc.page_content
-        new_doc["metadata"] = doc.metadata
+        # Make sure we have a clean doc format
+        new_doc = {
+            "page_content": doc.page_content,
+            "metadata": {}
+        }
+        
+        # Ensure all metadata fields exist with defaults if missing
+        for field, default_value in DOCUMENT_METADATA_DEFAULT_VALUES.items():
+            new_value =  doc.metadata.get(field, default_value)
+            try:
+                new_doc["metadata"][field] = type(default_value)(new_value)
+            except:
+                new_doc["metadata"][field] = default_value
+
         new_docs.append(new_doc)
+        
+    if new_docs == []:
+        new_docs = [{"page_content": "No documents found", "metadata": DOCUMENT_METADATA_DEFAULT_VALUES}]
     return new_docs
 
 ## AZURE LOGGING - DEPRECATED
 
-# def log_on_azure(file, logs, share_client):
-#     """Log data to Azure Blob Storage.
+def log_on_azure(file, logs, share_client):
+    """Log data to Azure Blob Storage.
     
-#     Args:
-#         file (str): Name of the file to store logs
-#         logs (dict): Log data to store
-#         share_client: Azure share client instance
-#     """
-#     logs = json.dumps(logs)
-#     file_client = share_client.get_file_client(file)
-#     file_client.upload_file(logs)
+    Args:
+        file (str): Name of the file to store logs
+        logs (dict): Log data to store
+        share_client: Azure share client instance
+    """
+    logs = json.dumps(logs)
+    file_client = share_client.get_file_client(file)
+    file_client.upload_file(logs)
 
 
-# def log_interaction_to_azure(history, output_query, sources, docs, share_client, user_id):
-#     """Log chat interaction to Azure and Hugging Face.
+def log_interaction_to_azure(history, output_query, sources, docs, share_client, user_id):
+    """Log chat interaction to Azure and Hugging Face.
     
-#     Args:
-#         history (list): Chat message history
-#         output_query (str): Processed query
-#         sources (list): Knowledge base sources used
-#         docs (list): Retrieved documents
-#         share_client: Azure share client instance
-#         user_id (str): User identifier
-#     """
-#     try:
-#         # Log interaction to Azure if not in local environment
-#         if os.getenv("GRADIO_ENV") != "local":
-#             timestamp = str(datetime.now().timestamp())
-#             prompt = history[1]["content"]
-#             logs = {
-#                 "user_id": str(user_id),
-#                 "prompt": prompt,
-#                 "query": prompt,
-#                 "question": output_query,
-#                 "sources": sources,
-#                 "docs": serialize_docs(docs),
-#                 "answer": history[-1].content,
-#                 "time": timestamp,
-#             }
-#             # Log to Azure
-#             log_on_azure(f"{timestamp}.json", logs, share_client)
-#     except Exception as e:
-#         print(f"Error logging on Azure Blob Storage: {e}")
-#         error_msg = f"ClimateQ&A Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
-#         raise gr.Error(error_msg)
+    Args:
+        history (list): Chat message history
+        output_query (str): Processed query
+        sources (list): Knowledge base sources used
+        docs (list): Retrieved documents
+        share_client: Azure share client instance
+        user_id (str): User identifier
+    """
+    try:
+        # Log interaction to Azure if not in local environment
+        if os.getenv("GRADIO_ENV") != "local":
+            timestamp = str(datetime.now().timestamp())
+            prompt = history[1]["content"]
+            logs = {
+                "user_id": str(user_id),
+                "prompt": prompt,
+                "query": prompt,
+                "question": output_query,
+                "sources": sources,
+                "docs": serialize_docs(docs),
+                "answer": history[-1].content,
+                "time": timestamp,
+            }
+            # Log to Azure
+            log_on_azure(f"{timestamp}.json", logs, share_client)
+    except Exception as e:
+        print(f"Error logging on Azure Blob Storage: {e}")
+        error_msg = f"ClimateQ&A Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
+        raise gr.Error(error_msg)
     
-# def log_drias_interaction_to_azure(query, sql_query, data, share_client, user_id):
-#     """Log Drias data interaction to Azure and Hugging Face.
+def log_drias_interaction_to_azure(query, sql_query, data, share_client, user_id):
+    """Log Drias data interaction to Azure and Hugging Face.
     
-#     Args:
-#         query (str): User query
-#         sql_query (str): SQL query used
-#         data: Retrieved data
-#         share_client: Azure share client instance
-#         user_id (str): User identifier
-#     """
-#     try:
-#         # Log interaction to Azure if not in local environment
-#         if os.getenv("GRADIO_ENV") != "local":
-#             timestamp = str(datetime.now().timestamp())
-#             logs = {
-#                 "user_id": str(user_id),
-#                 "query": query,
-#                 "sql_query": sql_query,
-#                 "time": timestamp,
-#             }
-#             log_on_azure(f"drias_{timestamp}.json", logs, share_client)
-#             print(f"Logged Drias interaction to Azure Blob Storage: {logs}")
-#         else:
-#             print("share_client or user_id is None, or GRADIO_ENV is local")
-#     except Exception as e:
-#         print(f"Error logging Drias interaction on Azure Blob Storage: {e}")
-#         error_msg = f"Drias Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
-#         raise gr.Error(error_msg)    
+    Args:
+        query (str): User query
+        sql_query (str): SQL query used
+        data: Retrieved data
+        share_client: Azure share client instance
+        user_id (str): User identifier
+    """
+    try:
+        # Log interaction to Azure if not in local environment
+        if os.getenv("GRADIO_ENV") != "local":
+            timestamp = str(datetime.now().timestamp())
+            logs = {
+                "user_id": str(user_id),
+                "query": query,
+                "sql_query": sql_query,
+                "time": timestamp,
+            }
+            log_on_azure(f"drias_{timestamp}.json", logs, share_client)
+            print(f"Logged Drias interaction to Azure Blob Storage: {logs}")
+        else:
+            print("share_client or user_id is None, or GRADIO_ENV is local")
+    except Exception as e:
+        print(f"Error logging Drias interaction on Azure Blob Storage: {e}")
+        error_msg = f"Drias Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
+        raise gr.Error(error_msg)    
     
 ## HUGGING FACE LOGGING
 
-def log_on_huggingface(log_filename, logs):
+def log_on_huggingface(log_filename, logs, log_type="chat"):
     """Log data to Hugging Face dataset repository.
     
     Args:
         log_filename (str): Name of the file to store logs
         logs (dict): Log data to store
+        log_type (str): Type of log to store
     """
     try:
-        # Get Hugging Face token from environment
-        hf_token = os.getenv("HF_LOGS_TOKEN")
-        if not hf_token:
-            print("HF_LOGS_TOKEN not found in environment variables")
-            return
+        if log_type =="chat":   
+            # Get Hugging Face token from environment
+            hf_token = os.getenv("HF_LOGS_TOKEN")
+            if not hf_token:
+                print("HF_LOGS_TOKEN not found in environment variables")
+                return
 
-        # Get repository name from environment or use default
-        repo_id = os.getenv("HF_DATASET_REPO", "timeki/climateqa_logs")
+            # Get repository name from environment or use default
+            repo_id = os.getenv("HF_DATASET_REPO", "Ekimetrics/climateqa_logs")
+        
+        elif log_type =="drias":
+            # Get Hugging Face token from environment
+            hf_token = os.getenv("HF_LOGS_DRIAS_TOKEN")
+            if not hf_token:
+                print("HF_LOGS_DRIAS_TOKEN not found in environment variables")
+                return
+
+            # Get repository name from environment or use default
+            repo_id = os.getenv("HF_DATASET_REPO_DRIAS", "Ekimetrics/climateqa_logs_talk_to_data")
+
+        else:
+            raise ValueError(f"Invalid log type: {log_type}")
         
         # Initialize HfApi
         api = HfApi(token=hf_token)
@@ -158,10 +204,13 @@ def log_interaction_to_huggingface(history, output_query, sources, docs, share_c
                 "time": timestamp,
             }
             # Log to Hugging Face
-            log_on_huggingface(f"chat/{timestamp}.json", logs)
+            log_on_huggingface(f"chat/{timestamp}.json", logs, log_type="chat")
+            print(f"Logged interaction to Hugging Face")
+        else:
+            print("Did not log to Hugging Face because GRADIO_ENV is local")
     except Exception as e:
         print(f"Error logging to Hugging Face: {e}")
-        error_msg = f"ClimateQ&A Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
+        error_msg = f"ClimateQ&A Error: {str(e)[:100]})"
         raise gr.Error(error_msg)
 
 def log_drias_interaction_to_huggingface(query, sql_query, user_id):
@@ -182,7 +231,7 @@ def log_drias_interaction_to_huggingface(query, sql_query, user_id):
                 "sql_query": sql_query,
                 "time": timestamp,
             }
-            log_on_huggingface(f"drias/drias_{timestamp}.json", logs)
+            log_on_huggingface(f"drias/drias_{timestamp}.json", logs, log_type="drias")
             print(f"Logged Drias interaction to Hugging Face: {logs}")
         else:
             print("share_client or user_id is None, or GRADIO_ENV is local")
@@ -191,4 +240,44 @@ def log_drias_interaction_to_huggingface(query, sql_query, user_id):
         error_msg = f"Drias Error: {str(e)[:100]} - The error has been noted, try another question and if the error remains, you can contact us :)"
         raise gr.Error(error_msg)
 
+def log_interaction(history, output_query, sources, docs, share_client, user_id):
+    """Log chat interaction to Hugging Face, and fall back to Azure if that fails.
+    
+    Args:
+        history (list): Chat message history
+        output_query (str): Processed query
+        sources (list): Knowledge base sources used
+        docs (list): Retrieved documents
+        share_client: Azure share client instance
+        user_id (str): User identifier
+    """
+    try:
+        # First try to log to Hugging Face
+        log_interaction_to_huggingface(history, output_query, sources, docs, share_client, user_id)
+    except Exception as e:
+        print(f"Failed to log to Hugging Face, falling back to Azure: {e}")
+        try:
+            # Fall back to Azure logging
+            if os.getenv("GRADIO_ENV") != "local":
+                timestamp = str(datetime.now().timestamp())
+                prompt = history[1]["content"]
+                logs = {
+                    "user_id": str(user_id),
+                    "prompt": prompt,
+                    "query": prompt,
+                    "question": output_query,
+                    "sources": sources,
+                    "docs": serialize_docs(docs),
+                    "answer": history[-1].content,
+                    "time": timestamp,
+                }
+                # Log to Azure
+                log_on_azure(f"{timestamp}.json", logs, share_client)
+                print("Successfully logged to Azure as fallback")
+        except Exception as azure_error:
+            print(f"Error in Azure fallback logging: {azure_error}")
+            error_msg = f"ClimateQ&A Logging Error: {str(azure_error)[:100]})"
+            # Don't raise error to avoid disrupting user experience
+            print(error_msg)
+