tools.py

# Required imports
import os
import uuid
from sqlalchemy import create_engine
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    SQLDatabase,
    PromptTemplate,
    StorageContext,
)
from llama_index.core.query_engine import NLSQLTableQueryEngine
from llama_index.core.tools import QueryEngineTool, FunctionTool
from llama_index.core.node_parser import MarkdownNodeParser
from llama_index.readers.docling import DoclingReader
from llama_index.vector_stores.milvus import MilvusVectorStore
from cleanlab_codex.project import Project
from cleanlab_codex.client import Client


#####################################
# Define Tools for Router Agent
#####################################
def create_codex_project():
    """Create a Codex project for document validation."""
    try:
        # Check if CODEX_API_KEY is available
        if not os.environ.get("CODEX_API_KEY"):
            print(
                "Warning: CODEX_API_KEY not found. Codex validation will be disabled."
            )
            return None, None

        # Create a unique identifier for the project
        project_id = str(uuid.uuid4())[:8]  # Using first 8 chars for readability

        codex_client = Client()
        project = codex_client.create_project(name=f"RAG + SQL Router {project_id}")
        access_key = project.create_access_key("default key")
        project = Project.from_access_key(access_key)
        return project, project_id
    except Exception as e:
        print(f"Error creating Codex project: {e}")
        return None, None


# Global variables for reuse - these will persist across function calls
docs_query_engine = None
codex_project = None
current_session_id = None
current_project_id = None


def get_or_create_codex_project(session_id):
    """Get existing Codex project or create a new one for the session."""
    global codex_project, current_session_id, current_project_id
    
    # If we have a project and it's for the same session, reuse it
    if codex_project is not None and current_session_id == session_id:
        print(f"Reusing existing Codex project for session {session_id}")
        return codex_project
    
    # Create a new project for this session
    print(f"Creating new Codex project for session {session_id}")
    codex_project, project_id = create_codex_project()
    current_session_id = session_id
    current_project_id = project_id
    
    return codex_project


def get_codex_project_info():
    """Get information about the current Codex project for debugging."""
    global codex_project, current_session_id, current_project_id
    
    if codex_project is None:
        return {
            "status": "No project created",
            "session_id": current_session_id,
            "project_id": None
        }
    
    try:
        # Get the actual project name using the stored project ID
        if current_project_id:
            project_name = f"RAG + SQL Router {current_project_id}"
        else:
            project_name = "RAG + SQL Router Project"
            
        return {
            "status": "Active",
            "session_id": current_session_id,
            "project_id": "Available",
            "project_name": project_name
        }
    except Exception as e:
        return {
            "status": f"Error getting info: {str(e)}",
            "session_id": current_session_id,
            "project_id": "Unknown"
        }


def setup_sql_tool(db_path="city_database.sqlite", table_name="city_stats"):
    """Setup SQL query tool for querying city database."""
    # Validate database exists
    if not os.path.exists(db_path):
        raise FileNotFoundError(f"Database file not found: {db_path}")

    try:
        engine = create_engine(f"sqlite:///{db_path}")
        sql_database = SQLDatabase(engine)
    except Exception as e:
        print(f"Error setting up SQL database: {e}")
        raise

    # Create SQL query engine
    sql_query_engine = NLSQLTableQueryEngine(
        sql_database=sql_database,
        tables=[table_name],
    )

    # Create tool for SQL querying
    sql_tool = QueryEngineTool.from_defaults(
        query_engine=sql_query_engine,
        name="sql_tool",
        description=(
            "Useful for translating a natural language query into a SQL query over"
            " a table containing: city_stats, containing the population/state of"
            " each city located in the USA."
        ),
    )

    # Return the SQL tool
    return sql_tool


def setup_document_tool(file_dir, session_id=None, milvus_uri="http://localhost:19530"):
    """Setup document query tool from uploaded documents with Codex validation."""
    global docs_query_engine

    # Create a reader and load the data
    reader, node_parser = DoclingReader(), MarkdownNodeParser()
    loader = SimpleDirectoryReader(
        input_dir=file_dir,
        file_extractor={
            ".pdf": reader,
            ".docx": reader,
            ".pptx": reader,
            ".txt": reader,
        },
    )
    docs = loader.load_data()

    # Creating a vector index over loaded data
    unique_collection_id = uuid.uuid4().hex
    collection_name = f"rag_with_sql_{unique_collection_id}"
    vector_store = MilvusVectorStore(uri=milvus_uri, dim=384, overwrite=True, collection_name=collection_name)
    storage_context = StorageContext.from_defaults(vector_store=vector_store)
    vector_index = VectorStoreIndex.from_documents(
        docs,
        show_progress=True,
        transformations=[node_parser],
        storage_context=storage_context,
    )

    # Custom prompt template
    template = (
        "You are a meticulous and accurate document analyst. Your task is to answer the user's question based exclusively on the provided context. "
        "Follow these rules strictly:\n"
        "1. Your entire response must be grounded in the facts provided in the 'Context' section. Do not use any prior knowledge.\n"
        "2. If multiple parts of the context are relevant, synthesize them into a single, coherent answer.\n"
        "3. If the context does not contain the information needed to answer the question, you must state only: 'The provided context does not contain enough information to answer this question.'\n"
        "-----------------------------------------\n"
        "Context: {context_str}\n"
        "-----------------------------------------\n"
        "Question: {query_str}\n\n"
        "Answer:"
    )
    qa_template = PromptTemplate(template)

    # Create a query engine for the vector index
    docs_query_engine = vector_index.as_query_engine(
        text_qa_template=qa_template, similarity_top_k=3
    )

    # Get or create Codex project for this session
    codex_project = get_or_create_codex_project(session_id)

    # Define the document query function with Codex validation
    def document_query_tool(query: str):
        """Query documents with Codex validation for enhanced accuracy."""
        # Step 1: Query the engine
        response_obj = docs_query_engine.query(query)
        initial_response = str(response_obj)

        # Step 2: Gather source context
        context = response_obj.source_nodes
        context_str = "\n".join([n.node.text for n in context])

        # Step 3: Prepare prompt for Codex validation
        prompt_template = (
            "You are a meticulous and accurate document analyst. Your task is to answer the user's question based exclusively on the provided context. "
            "Follow these rules strictly:\n"
            "1. Your entire response must be grounded in the facts provided in the 'Context' section. Do not use any prior knowledge.\n"
            "2. If multiple parts of the context are relevant, synthesize them into a single, coherent answer.\n"
            "3. If the context does not contain the information needed to answer the question, you must state only: 'The provided context does not contain enough information to answer this question.'\n"
            "-----------------------------------------\n"
            "Context: {context}\n"
            "-----------------------------------------\n"
            "Question: {query}\n\n"
            "Answer:"
        )
        user_prompt = prompt_template.format(context=context_str, query=query)
        messages = [{"role": "user", "content": user_prompt}]

        # Step 4: Validate with Codex (if available)
        if codex_project:
            try:
                print(f"Validating query with Codex: '{query[:50]}...'")
                result = codex_project.validate(
                    messages=messages,
                    query=query,
                    context=context_str,
                    response=initial_response,
                )
                print("Codex validation completed successfully")

                # Step 5: Final response selection
                fallback_response = "I'm sorry, I couldn't find an answer — can I help with something else?"
                final_response = (
                    result.expert_answer
                    if result.expert_answer and result.escalated_to_sme
                    else (
                        fallback_response
                        if result.should_guardrail
                        else initial_response
                    )
                )
                trust_score = result.model_dump()["eval_scores"]["trustworthiness"]["score"]

                # Return a dictionary to avoid tuple handling issues
                return {
                    "response": str(final_response),
                    "trust_score": float(trust_score)
                }
            except Exception as e:
                # If Codex validation fails, return the initial response
                print(f"Codex validation failed: {e}")
                return {
                    "response": str(initial_response),
                    "trust_score": None
                }
        else:
            # If Codex is not available, return the initial response
            print("Codex not available, using basic RAG response")
            return {
                "response": str(initial_response),
                "trust_score": None
            }

    # Create tool for document querying using FunctionTool
    docs_tool = FunctionTool.from_defaults(
        document_query_tool,
        name="document_tool",
        description=(
            "Useful for answering a natural language question by performing a semantic search over "
            "a collection of documents. These documents may contain general knowledge, reports, "
            "or domain-specific content. Returns the most relevant passages or synthesized answers. "
            "If the user query does not relate to US city statistics (population and state), use this document search tool."
        ),
    )

    # Return the document tool
    return docs_tool