Update app.py

app.py

@@ -0,0 +1,295 @@
+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
+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')
+
+
+
+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-4.1"
+    ),
+    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. 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(      # Default Gradio port
+        mcp_server=True              # Enable debug mode
+    )