main.py

import os
import gradio as gr
import autogen
from openai import OpenAI
import json
from src.mapper.e5map import E5Mapper
from src.mapper.scimap import scimap
from src.mapper.parser import MapperParser
from src.datatonic.dataloader import DataLoader
from src.agentics.agents import AgentsFactory

title = """# Welcome to 👩🏻‍🔬🧪SciTonic
this is a highly adaptive technical operator that will listen to your query and load datasets and multi-agent teams based on those. Simply describe your problem in detail, ask a question and provide a reasoning method to get started:
"""

def update_config_file(api_key):
    config_path = "./src/config/OAI_CONFIG_LIST.json"
    with open(config_path, "r") as file:
        config = json.load(file)

    for item in config:
        item["api_key"] = api_key

    with open(config_path, "w") as file:
        json.dump(config, file, indent=4)


def process_audio_image_input(input_type, input_data, MODEL_ID):
    PAT = os.getenv("CLARIFAI_PAT")
    if not PAT:
        raise ValueError("Clarifai Personal Access Token not set in environment variables")

    channel = ClarifaiChannel.get_grpc_channel()
    stub = service_pb2_grpc.V2Stub(channel)
    metadata = (("authorization", "Key " + PAT),)

    if input_type == "audio":
        file_bytes = input_data
    elif input_type == "image":
        file_bytes = base64.b64encode(input_data).decode("utf-8")

    post_model_outputs_response = stub.PostModelOutputs(
        service_pb2.PostModelOutputsRequest(
            model_id=MODEL_ID,
            inputs=[
                resources_pb2.Input(
                    data=resources_pb2.Data(
                        audio=resources_pb2.Audio(base64=file_bytes) if input_type == "audio" else None,
                        image=resources_pb2.Image(base64=file_bytes) if input_type == "image" else None
                    )
                )
            ],
        ),
        metadata=metadata,
    )

    if post_model_outputs_response.status.code != status_code_pb2.SUCCESS:
        print(post_model_outputs_response.status)
        raise Exception(
            "Post model outputs failed, status: "
            + post_model_outputs_response.status.description
        )

    output = post_model_outputs_response.outputs[0]
    return output.data.text.raw


def process_query(oai_key, query, max_auto_reply):
    update_config_file(oai_key)
    os.environ['OAI_KEY'] = oai_key
    llm_config = autogen.config_list_from_json(
        env_or_file="./src/config/OAI_CONFIG_LIST.json",
        filter_dict={"model": {"gpt-4", "gpt-3.5-turbo-16k", "gpt-4-1106-preview"}}
    )

    # Initialize mappers
    taskmapper = E5Mapper(oai_key)
    teammapper = scimap(oai_key)

    # Get responses from mappers
    taskmap_response = taskmapper.get_completion(query)
    teammap_response = teammapper.get_completion(query)

    # Parse responses
    task = MapperParser.parse_taskmapper_response(taskmap_response)
    team = MapperParser.parse_teammapper_response(teammap_response)

    # Load dataset based on task
    data_loader = DataLoader()
    dataset = data_loader.load_and_process(task.lower())

    # Save dataset to a JSON file and get the file path
    json_file_name = "dataset.json"  # Provide a suitable file name
    json_file_path = os.path.join("./src/datatonic/", json_file_name)  # Define the complete file path
    data_loader.save_to_json(dataset, json_file_path)

    # Initialize AgentsFactory with the path to the JSON file
    agents_factory = AgentsFactory(llm_config, json_file_path)

    # Retrieve the Boss Assistant agent
    boss_aid = agents_factory.scitonic()

    def _reset_agents():
        boss_aid.reset()

    # Define functions for each team
    def codingteam():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, coder, pm, reviewer],
            messages=[],
            max_round=12,
            speaker_selection_method="round_robin"
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, problem=PROBLEM, n_results=3)

    def covid19team():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, covid19_scientist, healthcare_expert, finance_analyst],
            messages=[],
            max_round=12
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, covid19_problem=COVID19_PROBLEM, n_results=3)

    def financeteam():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, finance_analyst, pm, reviewer, finance_expert],
            messages=[],
            max_round=12,
            speaker_selection_method="round_robin"
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, finance_problem=FINANCE_PROBLEM, n_results=3)

    def debateteam():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, debate_expert, pm, reviewer, debate_champion],
            messages=[],
            max_round=12,
            speaker_selection_method="round_robin"
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, debate_problem=DEBATE_PROBLEM, n_results=3)

    def homeworkteam():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, academic_expert, pm, reviewer, academic_whiz],
            messages=[],
            max_round=12,
            speaker_selection_method="round_robin"
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, homework_problem=HOMEWORK_PROBLEM, n_results=3)

    def consultingteam():
        _reset_agents()
        team = autogen.GroupChat(
            agents=[boss_aid, consultant, pm, reviewer, consulting_pro],
            messages=[],
            max_round=12,
            speaker_selection_method="round_robin"
        )

        manager = autogen.GroupChatManager(groupchat=team, llm_config=llm_config)
        boss_aid.initiate_chat(manager, consulting_problem=CONSULTING_PROBLEM, n_results=3)

    # Select and initiate team based on team mapping
    team_function = {
        "CodingTeam": codingteam,
        "Covid19Team": covid19team,
        "FinanceTeam": financeteam,
        "DebateTeam": debateteam,
        "HomeworkTeam": homeworkteam,
        "ConsultingTeam": consultingteam
    }

    team_action = team_function.get(team, lambda: "No appropriate team found for the given input.")
    return team_action()



def main():
    with gr.Blocks() as demo:
        gr.Markdown(title)
        with gr.Row():
            txt_oai_key = gr.Textbox(label="OpenAI API Key", type="password")
            txt_pat = gr.Textbox(label="Clarifai PAT", type="password", placeholder="Enter Clarifai PAT here")
            txt_query = gr.Textbox(label="Describe your problem in detail:")
            txt_max_auto_reply = gr.Number(label="Max Auto Replies", value=50)
            audio_input = gr.Audio(label="Or speak your problem here:", type="numpy",)
            image_input = gr.Image(label="Or upload an image related to your problem:", type="numpy", )
        btn_submit = gr.Button("Submit")
        output = gr.Textbox(label="Output",)

        def process_and_submit(oai_key, pat, query, max_auto_reply, audio, image):
            os.environ['CLARIFAI_PAT'] = pat
            os.environ['OAI_KEY'] = oai_key

            if audio is not None:
                query = process_audio_image_input("audio", audio, "asr-wav2vec2-base-960h-english")
            elif image is not None:
                query = process_audio_image_input("image", image, "general-english-image-caption-blip")
            return process_query(oai_key, query, max_auto_reply)

        btn_submit.click(
            process_and_submit,
            inputs=[txt_oai_key, txt_pat, txt_query, txt_max_auto_reply, audio_input, image_input],
            outputs=output
        )

    demo.launch()

if __name__ == "__main__":
    main()