tools.py

import os
import re
from langchain.tools import DuckDuckGoSearchRun
from langchain.chains import RetrievalQA
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.prompts import PromptTemplate
from datasets import load_dataset
from smolagents import CodeAgent, DuckDuckGoSearchTool, InferenceClientModel

# System prompt for formatting answers
SYSTEM_PROMPT = """
You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
"""

# Initialize web search tool
search_tool = DuckDuckGoSearchRun()

# Create custom prompt template with system instructions
prompt_template = SYSTEM_PROMPT + "\n\nContext: {context}\nQuestion: {question}\n"
PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# Load GAIA dataset and setup RAG components
def load_gaia_and_setup_rag():
    try:
        # Load GAIA dataset (requires HUGGINGFACE_HUB_TOKEN)
        dataset = load_dataset("GAIA", split="train")
        texts = [item['text'] for item in dataset if 'text' in item]
        
        # Create embeddings and vector store
        embeddings = OpenAIEmbeddings()
        vectorstore = FAISS.from_texts(texts, embeddings)
        
        # Create retriever and QA chain with custom prompt
        retriever = vectorstore.as_retriever()
        qa_chain = RetrievalQA.from_chain_type(
            llm=SmoalAgent(),
            chain_type="stuff",
            retriever=retriever,
            chain_type_kwargs={"prompt": PROMPT}
        )
        return qa_chain
    except Exception as e:
        print(f"RAG initialization error: {str(e)}")
        return None

# Extract final answer from model response
def extract_final_answer(response):
    """Extracts the final answer using the specified template format"""
    match = re.search(r"FINAL ANSWER: (.*)", response, re.IGNORECASE)
    if match:
        return match.group(1).strip()
    # Fallback to return full response if pattern not found
    return response

# Initialize RAG chain
global rag_chain
rag_chain = load_gaia_and_setup_rag()

# Initialize search tool
search_tool = DuckDuckGoSearchTool()

# Load GAIA dataset and setup RAG
rag_chain = None

def load_gaia_and_setup_rag():
    try:
        from datasets import load_dataset
        # Load GAIA dataset (test split)
        dataset = load_dataset("gaia-benchmark/gaia", split="test")
        
        # Extract contexts from dataset
        contexts = [item["context"] for item in dataset if "context" in item and item["context"]]
        
        # Create embeddings and vector store
        embeddings = OpenAIEmbeddings()
        vector_store = FAISS.from_texts(contexts, embeddings)
        
        # Create retriever
        retriever = vector_store.as_retriever(search_kwargs={"k": 3})
        
        # Define prompt template
        SYSTEM_PROMPT = """
        You are a precise QA system. Answer ONLY with the exact answer, no explanations.
        Answers must be in one of these formats:
        - A single number
        - A single string
        - A comma-separated list of numbers or strings
        Do not include any additional text, explanations, or formatting.
        """
        
        prompt_template = PromptTemplate(
            template=SYSTEM_PROMPT + "\nContext: {context}\nQuestion: {question}\nAnswer:",
            input_variables=["context", "question"]
        )
        
        # Create RAG chain
        global rag_chain
        rag_chain = RetrievalQA.from_chain_type(
            llm=OpenAI(temperature=0),
            chain_type="stuff",
            retriever=retriever,
            chain_type_kwargs={"prompt": prompt_template}
        )
        
        print(f"Successfully loaded GAIA dataset and created RAG chain with {len(contexts)} contexts")
        return True
    except Exception as e:
        print(f"Error setting up RAG: {e}")
        return False

# Initialize RAG when the module is loaded
load_gaia_and_setup_rag()

# Initialize CodeAgent
def initialize_code_agent():
    try:
        # Initialize model with environment variables
        model = InferenceClientModel(
            api_key=os.getenv("OPENAI_API_KEY"),
            model_name="gpt-3.5-turbo"
        )
        
        # Create agent with search tool
        agent = CodeAgent(
            tools=[search_tool],
            model=model
        )
        
        print("CodeAgent initialized successfully")
        return agent
    except Exception as e:
        print(f"Error initializing CodeAgent: {e}")
        return None

# Final answer extraction
def extract_final_answer(text):
    # Use regex to find the final answer pattern
    match = re.search(r'FINAL ANSWER: (.*)', text, re.IGNORECASE)
    if match:
        return match.group(1).strip()
    # If no pattern found, return the text as is (with cleanup)
    return text.strip()