|
|
import gradio as gr |
|
|
import requests |
|
|
import json |
|
|
import os |
|
|
from typing import List, Tuple |
|
|
|
|
|
|
|
|
API_URL = "https://monocopter.net" |
|
|
API_TOKEN = os.getenv("API_TOKEN", "") |
|
|
|
|
|
HEADERS = { |
|
|
"Authorization": f"Bearer {API_TOKEN}", |
|
|
"Content-Type": "application/json" |
|
|
} |
|
|
|
|
|
|
|
|
def query_rag_system(query: str, max_cards: int = 10) -> dict: |
|
|
"""Query the RAG system with a historical question.""" |
|
|
try: |
|
|
payload = {"query": query, "max_cards": max_cards, "include_sources": True} |
|
|
response = requests.post(f"{API_URL}/query", json=payload, headers=HEADERS) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
except requests.exceptions.RequestException as e: |
|
|
return {"error": f"Network Error: {e}"} |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
def search_cards_semantic(query: str, max_cards: int = 10) -> dict: |
|
|
"""Perform semantic search on cards.""" |
|
|
try: |
|
|
payload = {"query": query, "max_cards": max_cards} |
|
|
response = requests.post(f"{API_URL}/search", json=payload, headers=HEADERS) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
except requests.exceptions.RequestException as e: |
|
|
return {"error": f"Network Error: {e}"} |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
|
|
|
def format_rag_response(result: dict) -> str: |
|
|
"""Format the RAG response for display in the chatbot.""" |
|
|
if "error" in result: |
|
|
return f"β **Error**: {result['error']}" |
|
|
|
|
|
answer = result.get("answer", "I couldn't generate an answer for your question.") |
|
|
cards_used = result.get("cards_used", []) |
|
|
sources = result.get("sources", []) |
|
|
processing_time = result.get("processing_time", 0) |
|
|
|
|
|
|
|
|
response = f"**Answer:**\n{answer}\n\n" |
|
|
|
|
|
if sources: |
|
|
response += f"**π Sources ({len(sources)} documents):**\n" |
|
|
|
|
|
for i, source in enumerate(sources, 1): |
|
|
response += f"{i}. {source}\n" |
|
|
response += "\n" |
|
|
|
|
|
if cards_used: |
|
|
response += f"**ποΈ Retrieved {len(cards_used)} relevant cards** | " |
|
|
response += f"**β±οΈ {processing_time:.2f}s**" |
|
|
|
|
|
return response |
|
|
|
|
|
def format_search_response(result: dict) -> str: |
|
|
"""Format the search response for display.""" |
|
|
if "error" in result: |
|
|
return f"β **Error**: {result['error']}" |
|
|
|
|
|
cards = result.get("cards", []) |
|
|
if not cards: |
|
|
return "π No relevant historical information found for your search." |
|
|
|
|
|
response = f"π **Found {len(cards)} relevant historical entries:**\n\n" |
|
|
|
|
|
|
|
|
for i, card in enumerate(cards, 1): |
|
|
title = card.get('title', 'Untitled') |
|
|
summary = card.get('summary', 'No summary available') |
|
|
relevance = card.get('relevance_score', 0) |
|
|
|
|
|
|
|
|
if len(summary) > 200: |
|
|
summary = summary[:200] + "..." |
|
|
|
|
|
response += f"**{i}. {title}** (Relevance: {relevance:.3f})\n{summary}\n\n" |
|
|
|
|
|
processing_time = result.get("processing_time", 0) |
|
|
response += f"β±οΈ Search completed in {processing_time:.2f}s" |
|
|
|
|
|
return response |
|
|
|
|
|
def chatbot_respond(message: str, history: List[Tuple[str, str]], search_mode: bool) -> Tuple[str, List[Tuple[str, str]]]: |
|
|
"""Main chatbot response function.""" |
|
|
if not message.strip(): |
|
|
return "", history |
|
|
|
|
|
|
|
|
max_cards = 10 |
|
|
|
|
|
if search_mode: |
|
|
|
|
|
result = search_cards_semantic(message, max_cards) |
|
|
response = format_search_response(result) |
|
|
else: |
|
|
|
|
|
result = query_rag_system(message, max_cards) |
|
|
response = format_rag_response(result) |
|
|
|
|
|
|
|
|
history.append((message, response)) |
|
|
return "", history |
|
|
|
|
|
def clear_chat(): |
|
|
"""Clear the chat history.""" |
|
|
return [], "" |
|
|
|
|
|
|
|
|
def create_interface(): |
|
|
with gr.Blocks( |
|
|
title="Historical Knowledge Assistant", |
|
|
theme=gr.themes.Soft(), |
|
|
css=""" |
|
|
.chat-container { max-height: 600px; overflow-y: auto; } |
|
|
.examples { margin: 10px 0; } |
|
|
""" |
|
|
) as demo: |
|
|
|
|
|
gr.Markdown(""" |
|
|
# ποΈ Historical Knowledge Assistant |
|
|
Ask questions about historical events, people, and concepts. Powered by your Rolodex RAG database. |
|
|
""") |
|
|
|
|
|
with gr.Row(): |
|
|
with gr.Column(scale=4): |
|
|
chatbot = gr.Chatbot( |
|
|
label="Historical Q&A", |
|
|
height=500, |
|
|
show_copy_button=True, |
|
|
container=True, |
|
|
elem_classes=["chat-container"] |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
msg_input = gr.Textbox( |
|
|
placeholder="Ask about historical events, people, or concepts...", |
|
|
label="Your Question", |
|
|
lines=2, |
|
|
scale=4 |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
send_btn = gr.Button("π¬ Ask", variant="primary", scale=1) |
|
|
clear_btn = gr.Button("ποΈ Clear", variant="secondary", scale=1) |
|
|
|
|
|
with gr.Column(scale=1): |
|
|
search_mode = gr.Checkbox( |
|
|
label="π Search Mode", |
|
|
value=False, |
|
|
info="Toggle between Q&A (off) and Search (on)" |
|
|
) |
|
|
|
|
|
gr.Markdown(""" |
|
|
### π‘ Example Questions |
|
|
**Question & Answer Mode:** |
|
|
- What caused the American Revolution? |
|
|
- How did colonial resistance evolve? |
|
|
- Who were key figures in Bacon's Rebellion? |
|
|
|
|
|
**Search Mode:** |
|
|
- colonial resistance |
|
|
- Boston Massacre |
|
|
- taxation without representation |
|
|
|
|
|
### βΉοΈ Tips |
|
|
- **Q&A Mode**: Ask complete questions for detailed answers |
|
|
- **Search Mode**: Use keywords to explore topics |
|
|
- Sources and processing time shown with each response |
|
|
""", elem_classes=["examples"]) |
|
|
|
|
|
|
|
|
msg_input.submit( |
|
|
chatbot_respond, |
|
|
inputs=[msg_input, chatbot, search_mode], |
|
|
outputs=[msg_input, chatbot] |
|
|
) |
|
|
|
|
|
send_btn.click( |
|
|
chatbot_respond, |
|
|
inputs=[msg_input, chatbot, search_mode], |
|
|
outputs=[msg_input, chatbot] |
|
|
) |
|
|
|
|
|
clear_btn.click( |
|
|
clear_chat, |
|
|
outputs=[chatbot, msg_input] |
|
|
) |
|
|
|
|
|
return demo |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
print("π Launching Historical Knowledge Assistant...") |
|
|
print(f"π API URL: {API_URL}") |
|
|
if API_TOKEN: |
|
|
print(f"π Using API token: {API_TOKEN[:4]}...{API_TOKEN[-4:]}") |
|
|
else: |
|
|
print("π API token not found. Please set the API_TOKEN secret in your Space settings.") |
|
|
|
|
|
demo = create_interface() |
|
|
|
|
|
|
|
|
demo.launch( |
|
|
share=False, |
|
|
server_name="0.0.0.0", |
|
|
server_port=7860, |
|
|
show_error=True |
|
|
) |