| 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 |
| ) |