app.py

import os
import pandas as pd
import gradio as gr
from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI
from crewai_tools import PDFSearchTool, FileReadTool, DOCXSearchTool, CSVSearchTool
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents.agent_types import AgentType
from langchain_experimental.agents.agent_toolkits import create_csv_agent

import os
API_KEY=os.getenv("GOOGLE_API_KEY")

# API keys-----------------move them to ENV 
os.environ["OPENAI_API_KEY"] = "NA"
os.environ["GOOGLE_API_KEY"] = API_KEY

# Load The Gemini model for LLM
llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-flash-latest",
    verbose=True,               
    temperature=0.6,            # high temp=high accuracy and low creativity                                   
    google_api_key=API_KEY
)

#<-----------------------------Tools----------------------------------->
class tools:
    def pdfRead(path):
        PDFtool = PDFSearchTool(
            config=dict(
                llm=dict(
                    provider="google",
                    config=dict(
                        model="gemini-1.5-flash-latest",
                    ),
                ),
                embedder=dict(
                    provider="huggingface",
                    config=dict(
                        model="sentence-transformers/msmarco-distilbert-base-v4"
                        
                    ),
                ),
            ),
            pdf=path
        )
        return PDFtool
    
    def fileRead(path):
        Filetool = FileReadTool(
            config=dict(
                llm=dict(
                    provider="google",
                    config=dict(
                        model="gemini-1.5-flash-latest",
                    ),
                ),
                embedder=dict(
                    provider="huggingface",
                    config=dict(
                        model="sentence-transformers/msmarco-distilbert-base-v4"
                        
                    ),
                ),
            ),
            file_path=path
        )
        return Filetool
    
    def docsRead(path):
        Docstool = DOCXSearchTool(
            config=dict(
                llm=dict(
                    provider="google",
                    config=dict(
                        model="gemini-1.5-flash-latest",
                    ),
                ),
                embedder=dict(
                    provider="huggingface",
                    config=dict(
                        model="sentence-transformers/msmarco-distilbert-base-v4"
                        
                    ),
                ),
            ),
            docx=path
        )
        return Docstool
#<-----------------------------Tools----------------------------------->

#<------------------------------Agents START------------------------->

class AgentLoader:

    def csvReaderAgent(path):
        agent = create_csv_agent(
            llm,
            path,
            verbose=True,
            agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION
        )
        return agent

    def fileReaderAgent(path):
        FileReader = Agent(
            role='File searcher',
            goal='To analyse and generate optimal and reliable results',
            backstory="""You are a File specialist and can handle multiple file formats like .txt, .csv, .json etc.
            You are responsible to analyse the file to find the relevant content that solves the problem of the user and generate high quality and reliable results.
            You should also provide the results of your analysis and searching.""",
            llm=llm,
            verbose=True,
            tools=[tools.fileRead(path)],
            allow_delegation=False
        )
        return FileReader
    
    def PdfReaderAgent(path):
        PdfReader = Agent(
            role='PDF searcher',
            goal='To analyse and generate optimal and reliable results',
            backstory="""You are a PDF specialist and content writer.
            You are responsible to analyse the pdf to find the relevant content that solves the problem of the user and generate high quality and reliable results.
            You should also provide the results of your analysis and searching.""",
            llm=llm,
            verbose=True,
            tools=[tools.pdfRead(path)],
            allow_delegation=False,
            max_iter=6
        )
        return PdfReader
    
    def DocsReaderAgent(path):
        DocsReader = Agent(
            role='Docs searcher',
            goal='To analyse and generate optimal and reliable results',
            backstory="""You are a Docs specialist and content writer.
            You are responsible to analyse the pdf to find the relevant content that solves the problem of the user and generate high quality and reliable results.
            You should also provide the results of your analysis and searching.""",
            llm=llm,
            verbose=True,
            tools=[tools.docsRead(path)],
            allow_delegation=False
        )
        return DocsReader
    
    def writerAgent():
        writer=Agent(
            role='Content Writer',
            goal='To produce higly accurate and easy to understand information',
            backstory="""You are an content specialist and are respinsible to generate reliable and easy to understand content or information based on the summary of data.
            You should provide indetail results on the summary data.""",
            verbose=True,
            llm=llm,
            max_iter=6
        )
        return writer

#<------------------------------Agents END------------------------->

#<-------------------------------Tasks---------------------------->
def getTasks(query, agent, exp):
    task_read=Task(
        description=f'{query}',
        agent=agent,
        expected_output=f'A detailed information on {query}'
    )

    task_write=Task(
        description=f'{query}',
        agent=AgentLoader.writerAgent(),
        expected_output=exp
    )

    return [task_read, task_write]

# Gradio interface function
def process_file(file, query, expected_output):
    path = file.name
    
    if path.endswith(".pdf"):
        agent = AgentLoader.PdfReaderAgent(path)
    elif path.endswith(".docx"):
        agent = AgentLoader.DocsReaderAgent(path)
    elif path.endswith(".json") or path.endswith(".txt"):
        agent = AgentLoader.fileReaderAgent(path)
    elif path.endswith(".csv"):
        agent = AgentLoader.csvReaderAgent(path)
        results = agent.run(query)
    else:
        return 'File NOT supported'
    
    if not path.endswith(".csv"):
        task1 = getTasks(query, agent, expected_output)
        mycrew = Crew(
            agents=[agent, AgentLoader.writerAgent()],
            tasks=task1,
            verbose=True
        )
        results = mycrew.kickoff()
    
    return results

# Create the Gradio interface
interface = gr.Interface(
    fn=process_file,
    inputs=[
        gr.File(label="Upload File"),
        gr.Textbox(label="Query"),
        gr.Textbox(label="Expected Output")
    ],
    outputs="text",
    title="DataWizardZ",
    description=(
        "Upload a file (CSV, PDF, DOCX, TXT, JSON) and enter your query to get detailed information.\n\n"
        "### Instructions:\n"
        "1. Upload the file you want to talk to.\n"
        "2. Enter your question in the Query field.\n"
        "3. Specify the desired output format, e.g., one line answer.\n"
        "4. Please be patient; it can take up to 300ms for effective results, especially for large files or one-word answers.\n"
        "5. Please note that DO NOT specify Expected Output for .CSV Files."
    ),
    examples=[
        ["LabManual_cnn.pdf", "How to setup wired LAN", "A short summary"],
        ["house_prices.csv", "What is the average price of houses in Thane","None"]
    ],
    theme=gr.themes.Soft()
)

# Launch the Gradio interface
interface.launch()