app.py

from kognieLlama import Kognie
import os
from dotenv import load_dotenv
import datetime

import asyncio
from functools import partial

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.legacy.llms.types import ChatMessage
from llama_index.tools.bing_search import BingSearchToolSpec
from llama_index.llms.openai import OpenAI
from llama_index.llms.anthropic import Anthropic
from llama_index.llms.mistralai import MistralAI
import gradio as gr
import time
import base64

# Convert image to base64
with open("drop-down.png", "rb") as f:
    base64_img = base64.b64encode(f.read()).decode("utf-8")

load_dotenv()

KOGNIE_BASE_URL = os.getenv("KOGNIE_BASE_URL")
KOGNIE_API_KEY = os.getenv("KOGNIE_API_KEY")
BING_SUBSCRIPTION_KEY = os.getenv('BING_SUBSCRIPTION_KEY')
BING_SEARCH_URL = os.getenv('BING_SEARCH_URL')
ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
MISTRAL_API_KEY = os.getenv('MISTRAL_API_KEY')



async def async_llm_call(model, messages):
    """Wraps the synchronous chat method in a thread to make it non-blocking"""
    loop = asyncio.get_running_loop()
    chat_func = partial(model.chat, messages=messages)
    try:
        result = await loop.run_in_executor(None, chat_func)
        return result
    except Exception as e:
        print(f"Error occurred while invoking {model.model}: {e}")
        return None

async def multillm_verifier_tool(claim: str) -> str:
    """

    An async tool that runs multiple LLMs to check/verify the claim in parallel.

    """
    gpt = Kognie(
        api_key=KOGNIE_API_KEY,
        model="gpt-4o-mini"
    )
    gemini = Kognie(
        api_key=KOGNIE_API_KEY,
        model='gemini-2.0-flash'
    )
    mistral = Kognie(
        api_key=KOGNIE_API_KEY,
        model='open-mistral-nemo'
    )
    
    prompt = f"""you are a helpful assistant. Your task is to provide evidence for the claim to the extent there is any such evidence and provide evidence against the claim to the extent there is any such evidence. Under no circumstances fabricate evidence. You must list out all the evidence you can find.

    Claim: {claim}

    Evidences:

    """
    
    messages = [ChatMessage(role="system", content=prompt), ChatMessage(role="user", content=claim)]
    
    tasks = [
        async_llm_call(gpt, messages),
        async_llm_call(gemini, messages),
        async_llm_call(mistral, messages)
    ]
    
    results = await asyncio.gather(*tasks)
    print("multillm done")
    return {
        "gpt-4": results[0],
        "gemini-2.0-flash": results[1],
        "open-mistral-nemo": results[2]
    }


def web_evidence_retriever_tool(claim: str) -> str:
    """

    A tool that retrieves relevant evidence from the web to support or refute a claim.

    Uses Bing Search API to gather information, then analyzes it to provide evidence.

    """
    # Step 1: Generate search queries based on the claim
    search_query = f"{claim} evidence facts verification"
    
    # Step 2: Retrieve search results
    search = BingSearchToolSpec(
        api_key=BING_SUBSCRIPTION_KEY,
        results=5,
    )
    search_results = search.bing_news_search(search_query)  # Get more results for better coverage
    print("web done")
    # Return the structured analysis
    return search_results




multiLLMVerifier = FunctionAgent(
    tools=[multillm_verifier_tool],
    llm=Anthropic(model="claude-3-5-sonnet-20240620", api_key=ANTHROPIC_API_KEY),
    system_prompt=f"you are a helpful assistant. Your task is to provide evidence for the claim to the extent there is any such evidence and provide evidence against the claim to the extent there is any such evidence. Under no circumstances fabricate evidence. You must list out all the evidence you can find.",
)

webEvidenceRetriever = FunctionAgent(
    tools=[web_evidence_retriever_tool],
    llm=Anthropic(model="claude-3-5-sonnet-20240620", api_key=ANTHROPIC_API_KEY),
    system_prompt=f"you are a helpful assistant. Your task is to provide evidence for the claim to the extent there is any such evidence and provide evidence against the claim to the extent there is any such evidence. Under no circumstances fabricate evidence. You must list out all the evidence you can find.. Today's date is: {datetime.datetime.now().strftime('%Y-%m-%d')}",
)

AgentResponses = {}

async def multiagent_tool_run(user_input: str):
    """

    This function runs multiple agents to verify a claim using different tools.

    Each agent will provide its own analysis, and the coordinator will make a final decision.

    """
    # Run agents concurrently
    responses = {}
    tasks = [
        multiLLMVerifier.run(user_input),
        webEvidenceRetriever.run(user_input)
    ]

    start_time = time.time()
    
    results = await asyncio.gather(*tasks)

    end_time = time.time()
    elapsed_time = end_time - start_time
    print(f"Elapsed time for multi-agent run: {elapsed_time} seconds")
    
    responses["MultiLLMVerifier"] = results[0]
    responses["WebEvidenceRetriever"] = results[1]
    
    # print("\n===========Responses from MultiLLMVerifier:================\n", responses["MultiLLMVerifier"])
    # print("\n===========Responses from WebEvidenceRetriever:================\n", responses["WebEvidenceRetriever"])

    try:
        print("Responses from all agents received.", responses)
        AgentResponses['GPT'] = responses['MultiLLMVerifier']
        AgentResponses['Web Research'] = responses['WebEvidenceRetriever']
        # AgentResponses['GPT'] = results[0].blocks[0].text
        # AgentResponses['Web Research'] = results[1].blocks[0].text
    except Exception as e:
        print(f"Error processing agent responses: {e}")
        AgentResponses['GPT'] = "No response from GPT"
        AgentResponses['Web Research'] = "No response from Web Research"


    return {
        "individual_responses": responses,
    }




BossAgent = FunctionAgent(
    tools=[multiagent_tool_run],
    return_intermediate_steps=True,
    llm=OpenAI(
    api_key=OPENAI_API_KEY,
    model="gpt-4o"
    ),
    system_prompt=f"You are a coordinator that runs multiple agents to verify claims using different tools. You are the final decision maker.Your decision must be based on the evidence presented by the different agents. Please generate a very short decision in html format. The main decision should come at the top in bold and larger fonts and color green if TRUE or red is FALSE and followed by some small reasoning and evidence. Do not include the backticks html. Give priority to the Web agent if it conflicts with the other agents as it has the latest information. Today's date is : {datetime.datetime.now().strftime('%Y-%m-%d')}",
)

async def main(claim: str):

    # Run the agent
    print(f"Running claim verification for: {claim}")
    response = await BossAgent.run(claim)
    print(str(response))
    return str(response)



async def verify_claim(message: str, history):
    """

    Use this tool to verify a claim

    """

    print(f"Received message: {message}")

    # Start running the main task in background
    task = asyncio.create_task(main(message))

    response = await task
    yield [gr.ChatMessage(role="assistant",
                content=gr.HTML(str(response)),
            ), gr.ChatMessage(
    role="assistant",
    content=gr.HTML(
        f"""

        <style>

            .collapsible {{

                display: flex;

                align-items: center;

                justify-content: flex-start;

                background-color: #3498db;

                color: white;

                cursor: pointer;

                padding: 15px;

                width: 100%;

                border: none;

                text-align: left;

                outline: none;

                font-size: 14px !important;

                border-radius: 5px;

            }}



            .arrow {{

                transition: transform 0.3s ease;

                filter: invert(1);

                margin: 0 !important;

                margin-left: 5px !important;

            }}



            .collapsible.active .arrow {{

                transform: rotate(180deg);

            }}



            .content {{

                padding: 0 15px;

                max-height: 0;

                overflow: hidden;

                transition: max-height 0.3s ease-out;

                background-color: transparent;

                border-radius: 0 0 5px 5px;

            }}

        </style>

        <button class="collapsible" onclick="this.classList.toggle('active'); 

            const content = this.nextElementSibling;

            if (content.style.maxHeight){{

              content.style.maxHeight = null;

            }} else {{

              content.style.maxHeight = content.scrollHeight + 'px';

            }}">Show analysis<img src="data:image/png;base64,{base64_img}" class="arrow"></button>

        <div class="content">

    {f'<p style="font-size: 16px;">Generation Specialist : <span style="font-size: 14px;">{AgentResponses["GPT"]}</span></p>' if 'GPT' in AgentResponses and AgentResponses['GPT'] else ''}



    {f'<p style="font-size: 16px;">Web Research : <span style="font-size: 14px;">{AgentResponses["Web Research"]}</span></p>' if 'Web Research' in AgentResponses and AgentResponses['Web Research'] else ''}

</div>

        """
    ),
)
]





# chatbot = gr.Chatbot(history, type="messages")

demo = gr.ChatInterface(
    verify_claim,
    type="messages",
    flagging_mode="never",
    save_history=True,
    show_progress="full",
    title="Claim Verification System using Kognie API",
    textbox=gr.Textbox(
        placeholder="Enter a claim to verify",
        show_label=False,
        elem_classes=["claim-input"],
        submit_btn=True
    ),
    css = '''

.claim-input {

    border-width: 2px !important;

    border-style: solid !important;

    border-color: #EA580C !important;

    border-radius: 5px;

    font-size: 16px;

}



'''

)

# Launch the interface
if __name__ == "__main__":
    # Remove or comment out the existing test run
    # result = final_run("RCB are the 2025 IPL winners.")
    # print("Final Decision:\n", result['output'])
    
    # Launch Gradio interface
    demo.launch(  
        share=True,   # Default Gradio port
        mcp_server=True              # Enable debug mode
    )