chatui-helper / app.py
milwright's picture
Update research assistant prompt to be more realistic and accessible
719f45f
raw
history blame
92 kB
import warnings
warnings.filterwarnings("ignore", message="The 'tuples' format for chatbot messages is deprecated")
import gradio as gr
import json
import zipfile
import io
import os
from datetime import datetime
from dotenv import load_dotenv
import requests
from bs4 import BeautifulSoup
import tempfile
from pathlib import Path
from support_docs import create_support_docs, export_conversation_to_markdown
# Simple URL content fetching using requests and BeautifulSoup
def get_grounding_context_simple(urls):
"""Fetch grounding context using enhanced HTTP requests"""
if not urls:
return ""
context_parts = []
for i, url in enumerate(urls, 1):
if url and url.strip():
# Use enhanced URL extraction for any URLs within the URL text
extracted_urls = extract_urls_from_text(url.strip())
target_url = extracted_urls[0] if extracted_urls else url.strip()
content = enhanced_fetch_url_content(target_url)
context_parts.append(f"Context from URL {i} ({target_url}):\n{content}")
if context_parts:
return "\n\n" + "\n\n".join(context_parts) + "\n\n"
return ""
# Import RAG components
try:
from rag_tool import RAGTool
HAS_RAG = True
except ImportError:
HAS_RAG = False
RAGTool = None
# Load environment variables from .env file
load_dotenv()
# Utility functions
import re
def extract_urls_from_text(text):
"""Extract URLs from text using regex with enhanced validation"""
url_pattern = r'https?://[^\s<>"{}|\\^`\[\]"]+'
urls = re.findall(url_pattern, text)
# Basic URL validation and cleanup
validated_urls = []
for url in urls:
# Remove trailing punctuation that might be captured
url = url.rstrip('.,!?;:')
# Basic domain validation
if '.' in url and len(url) > 10:
validated_urls.append(url)
return validated_urls
def validate_url_domain(url):
"""Basic URL domain validation"""
try:
from urllib.parse import urlparse
parsed = urlparse(url)
# Check for valid domain structure
if parsed.netloc and '.' in parsed.netloc:
return True
except:
pass
return False
def enhanced_fetch_url_content(url, enable_search_validation=False):
"""Enhanced URL content fetching with optional search validation"""
if not validate_url_domain(url):
return f"Invalid URL format: {url}"
try:
# Enhanced headers for better compatibility
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive'
}
response = requests.get(url, timeout=15, headers=headers)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# Enhanced content cleaning
for element in soup(["script", "style", "nav", "header", "footer", "aside", "form", "button"]):
element.decompose()
# Extract main content preferentially
main_content = soup.find('main') or soup.find('article') or soup.find('div', class_=lambda x: bool(x and 'content' in x.lower())) or soup
text = main_content.get_text()
# Enhanced text cleaning
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = ' '.join(chunk for chunk in chunks if chunk and len(chunk) > 2)
# Smart truncation - try to end at sentence boundaries
if len(text) > 4000:
truncated = text[:4000]
last_period = truncated.rfind('.')
if last_period > 3000: # If we can find a reasonable sentence break
text = truncated[:last_period + 1]
else:
text = truncated + "..."
return text if text.strip() else "No readable content found at this URL"
except requests.exceptions.Timeout:
return f"Timeout error fetching {url} (15s limit exceeded)"
except requests.exceptions.RequestException as e:
return f"Error fetching {url}: {str(e)}"
except Exception as e:
return f"Error processing content from {url}: {str(e)}"
# Template for generated space app (based on mvp_simple.py)
SPACE_TEMPLATE = '''import gradio as gr
import os
import requests
import json
from bs4 import BeautifulSoup
from datetime import datetime
import tempfile
# Configuration
SPACE_NAME = "{name}"
SPACE_DESCRIPTION = "{description}"
SYSTEM_PROMPT = """{system_prompt}"""
MODEL = "{model}"
GROUNDING_URLS = {grounding_urls}
# Get access code from environment variable for security
ACCESS_CODE = os.environ.get("SPACE_ACCESS_CODE", "{access_code}")
ENABLE_DYNAMIC_URLS = {enable_dynamic_urls}
ENABLE_VECTOR_RAG = {enable_vector_rag}
ENABLE_WEB_SEARCH = {enable_web_search}
RAG_DATA = {rag_data_json}
# Get API key from environment - customizable variable name with validation
API_KEY = os.environ.get("{api_key_var}")
if API_KEY:
API_KEY = API_KEY.strip() # Remove any whitespace
if not API_KEY: # Check if empty after stripping
API_KEY = None
# API Key validation and logging
def validate_api_key():
"""Validate API key configuration with detailed logging"""
if not API_KEY:
print(f"⚠️ API KEY CONFIGURATION ERROR:")
print(f" Variable name: {api_key_var}")
print(f" Status: Not set or empty")
print(f" Action needed: Set '{api_key_var}' in HuggingFace Space secrets")
print(f" Expected format: sk-or-xxxxxxxxxx")
return False
elif not API_KEY.startswith('sk-or-'):
print(f"⚠️ API KEY FORMAT WARNING:")
print(f" Variable name: {api_key_var}")
print(f" Current value: {{{{API_KEY[:10]}}}}..." if len(API_KEY) > 10 else API_KEY)
print(f" Expected format: sk-or-xxxxxxxxxx")
print(f" Note: OpenRouter keys should start with 'sk-or-'")
return True # Still try to use it
else:
print(f"βœ… API Key configured successfully")
print(f" Variable: {api_key_var}")
print(f" Format: Valid OpenRouter key")
return True
# Validate on startup
API_KEY_VALID = validate_api_key()
def fetch_url_content(url):
"""Fetch and extract text content from a URL using requests and BeautifulSoup"""
try:
response = requests.get(url, timeout=10, headers={{'User-Agent': 'Mozilla/5.0'}})
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# Remove script and style elements
for script in soup(["script", "style", "nav", "header", "footer"]):
script.decompose()
# Get text content
text = soup.get_text()
# Clean up whitespace
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = ' '.join(chunk for chunk in chunks if chunk)
# Truncate to ~4000 characters
if len(text) > 4000:
text = text[:4000] + "..."
return text
except Exception as e:
return f"Error fetching {{url}}: {{str(e)}}"
def extract_urls_from_text(text):
"""Extract URLs from text using regex"""
import re
url_pattern = r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'
return re.findall(url_pattern, text)
# Global cache for URL content to avoid re-crawling in generated spaces
_url_content_cache = {{}}
def get_grounding_context():
"""Fetch context from grounding URLs with caching"""
if not GROUNDING_URLS:
return ""
# Create cache key from URLs
cache_key = tuple(sorted([url for url in GROUNDING_URLS if url and url.strip()]))
# Check cache first
if cache_key in _url_content_cache:
return _url_content_cache[cache_key]
context_parts = []
for i, url in enumerate(GROUNDING_URLS, 1):
if url.strip():
content = fetch_url_content(url.strip())
context_parts.append(f"Context from URL {{i}} ({{url}}):\\n{{content}}")
if context_parts:
result = "\\n\\n" + "\\n\\n".join(context_parts) + "\\n\\n"
else:
result = ""
# Cache the result
_url_content_cache[cache_key] = result
return result
def export_conversation_to_markdown(conversation_history):
"""Export conversation history to markdown format"""
if not conversation_history:
return "No conversation to export."
markdown_content = f"""# Conversation Export
Generated on: {{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}}
---
"""
for i, message in enumerate(conversation_history):
if isinstance(message, dict):
role = message.get('role', 'unknown')
content = message.get('content', '')
if role == 'user':
markdown_content += f"## User Message {{(i//2) + 1}}\\n\\n{{content}}\\n\\n"
elif role == 'assistant':
markdown_content += f"## Assistant Response {{(i//2) + 1}}\\n\\n{{content}}\\n\\n---\\n\\n"
return markdown_content
# Initialize RAG context if enabled
if ENABLE_VECTOR_RAG and RAG_DATA:
try:
import faiss
import numpy as np
import base64
class SimpleRAGContext:
def __init__(self, rag_data):
# Deserialize FAISS index
index_bytes = base64.b64decode(rag_data['index_base64'])
self.index = faiss.deserialize_index(index_bytes)
# Restore chunks and mappings
self.chunks = rag_data['chunks']
self.chunk_ids = rag_data['chunk_ids']
def get_context(self, query, max_chunks=3):
"""Get relevant context - simplified version"""
# In production, you'd compute query embedding here
# For now, return a simple message
return "\\n\\n[RAG context would be retrieved here based on similarity search]\\n\\n"
rag_context_provider = SimpleRAGContext(RAG_DATA)
except Exception as e:
print(f"Failed to initialize RAG: {{e}}")
rag_context_provider = None
else:
rag_context_provider = None
def generate_response(message, history):
"""Generate response using OpenRouter API"""
# Enhanced API key validation with helpful messages
if not API_KEY:
error_msg = f"πŸ”‘ **API Key Required**\\n\\n"
error_msg += f"Please configure your OpenRouter API key:\\n"
error_msg += f"1. Go to Settings (βš™οΈ) in your HuggingFace Space\\n"
error_msg += f"2. Click 'Variables and secrets'\\n"
error_msg += f"3. Add secret: **{api_key_var}**\\n"
error_msg += f"4. Value: Your OpenRouter API key (starts with `sk-or-`)\\n\\n"
error_msg += f"Get your API key at: https://openrouter.ai/keys"
print(f"❌ API request failed: No API key configured for {api_key_var}")
return error_msg
# Get grounding context
grounding_context = get_grounding_context()
# Add RAG context if available
if ENABLE_VECTOR_RAG and rag_context_provider:
rag_context = rag_context_provider.get_context(message)
if rag_context:
grounding_context += rag_context
# If dynamic URLs are enabled, check message for URLs to fetch
if ENABLE_DYNAMIC_URLS:
urls_in_message = extract_urls_from_text(message)
if urls_in_message:
# Fetch content from URLs mentioned in the message
dynamic_context_parts = []
for url in urls_in_message[:3]: # Limit to 3 URLs per message
content = fetch_url_content(url)
dynamic_context_parts.append(f"\\n\\nDynamic context from {{url}}:\\n{{content}}")
if dynamic_context_parts:
grounding_context += "\\n".join(dynamic_context_parts)
# If web search is enabled, use it for most queries (excluding code blocks and URLs)
if ENABLE_WEB_SEARCH:
should_search = True
# Skip search for messages that are primarily code blocks
import re
if re.search(r'```[\\s\\S]*```', message):
should_search = False
# Skip search for messages that are primarily URLs
urls_in_message = extract_urls_from_text(message)
if urls_in_message and len(' '.join(urls_in_message)) > len(message) * 0.5:
should_search = False
# Skip search for very short messages (likely greetings)
if len(message.strip()) < 5:
should_search = False
if should_search:
# Use the entire message as search query, cleaning it up
search_query = message.strip()
try:
# Perform web search using crawl4ai
import urllib.parse
import asyncio
async def search_with_crawl4ai(search_query):
try:
from crawl4ai import WebCrawler
# Create search URL for DuckDuckGo
encoded_query = urllib.parse.quote_plus(search_query)
search_url = f"https://duckduckgo.com/html/?q={{encoded_query}}"
# Initialize crawler
crawler = WebCrawler(verbose=False)
try:
# Start the crawler
await crawler.astart()
# Crawl the search results
result = await crawler.arun(url=search_url)
if result.success:
# Extract text content from search results
content = result.cleaned_html if result.cleaned_html else result.markdown
# Clean and truncate the content
if content:
# Remove excessive whitespace and limit length
lines = [line.strip() for line in content.split('\\n') if line.strip()]
cleaned_content = '\\n'.join(lines)
# Truncate to reasonable length for context
if len(cleaned_content) > 2000:
cleaned_content = cleaned_content[:2000] + "..."
return cleaned_content
else:
return "No content extracted from search results"
else:
return f"Search failed: {{result.error_message if hasattr(result, 'error_message') else 'Unknown error'}}"
finally:
# Clean up the crawler
await crawler.aclose()
except ImportError:
# Fallback to simple DuckDuckGo search without crawl4ai
encoded_query = urllib.parse.quote_plus(search_query)
search_url = f"https://duckduckgo.com/html/?q={{encoded_query}}"
# Use basic fetch as fallback
response = requests.get(search_url, headers={{'User-Agent': 'Mozilla/5.0'}}, timeout=10)
if response.status_code == 200:
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.content, 'html.parser')
# Remove script and style elements
for script in soup(["script", "style", "nav", "header", "footer"]):
script.decompose()
# Get text content
text = soup.get_text()
# Clean up whitespace
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = ' '.join(chunk for chunk in chunks if chunk)
# Truncate to ~2000 characters
if len(text) > 2000:
text = text[:2000] + "..."
return text
else:
return f"Failed to fetch search results: {{response.status_code}}"
# Run the async search
if hasattr(asyncio, 'run'):
search_result = asyncio.run(search_with_crawl4ai(search_query))
else:
# Fallback for older Python versions
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
search_result = loop.run_until_complete(search_with_crawl4ai(search_query))
finally:
loop.close()
grounding_context += f"\\n\\nWeb search results for '{{search_query}}':\\n{{search_result}}"
except Exception as e:
# Fallback to URL extraction if web search fails
urls = extract_urls_from_text(search_query)
if urls:
for url in urls[:2]: # Limit to 2 URLs for fallback
content = fetch_url_content(url)
grounding_context += f"\\n\\nFallback content from {{url}}:\\n{{content[:500]}}..."
else:
grounding_context += f"\\n\\nWeb search requested: {{search_query}} (external search not available)"
# Build enhanced system prompt with grounding context
enhanced_system_prompt = SYSTEM_PROMPT + grounding_context
# Build messages array for the API
messages = [{{"role": "system", "content": enhanced_system_prompt}}]
# Add conversation history - compatible with Gradio 5.x format
for chat in history:
if isinstance(chat, dict):
# New format: {{"role": "user", "content": "..."}} or {{"role": "assistant", "content": "..."}}
messages.append(chat)
else:
# Legacy format: ("user msg", "bot msg")
user_msg, bot_msg = chat
messages.append({{"role": "user", "content": user_msg}})
if bot_msg:
messages.append({{"role": "assistant", "content": bot_msg}})
# Add current message
messages.append({{"role": "user", "content": message}})
# Make API request with enhanced error handling
try:
print(f"πŸ”„ Making API request to OpenRouter...")
print(f" Model: {{MODEL}}")
print(f" Messages: {{len(messages)}} in conversation")
response = requests.post(
url="https://openrouter.ai/api/v1/chat/completions",
headers={{
"Authorization": f"Bearer {{API_KEY}}",
"Content-Type": "application/json",
"HTTP-Referer": "https://huggingface.co", # Required by some providers
"X-Title": "HuggingFace Space" # Helpful for tracking
}},
json={{
"model": MODEL,
"messages": messages,
"temperature": {temperature},
"max_tokens": {max_tokens}
}},
timeout=30
)
print(f"πŸ“‘ API Response: {{response.status_code}}")
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
print(f"βœ… API request successful")
return content
elif response.status_code == 401:
error_msg = f"πŸ” **Authentication Error**\\n\\n"
error_msg += f"Your API key appears to be invalid or expired.\\n\\n"
error_msg += f"**Troubleshooting:**\\n"
error_msg += f"1. Check that your **{api_key_var}** secret is set correctly\\n"
error_msg += f"2. Verify your API key at: https://openrouter.ai/keys\\n"
error_msg += f"3. Ensure your key starts with `sk-or-`\\n"
error_msg += f"4. Check that you have credits on your OpenRouter account"
print(f"❌ API authentication failed: {{response.status_code}} - {{response.text[:200]}}")
return error_msg
elif response.status_code == 429:
error_msg = f"⏱️ **Rate Limit Exceeded**\\n\\n"
error_msg += f"Too many requests. Please wait a moment and try again.\\n\\n"
error_msg += f"**Troubleshooting:**\\n"
error_msg += f"1. Wait 30-60 seconds before trying again\\n"
error_msg += f"2. Check your OpenRouter usage limits\\n"
error_msg += f"3. Consider upgrading your OpenRouter plan"
print(f"❌ Rate limit exceeded: {{response.status_code}}")
return error_msg
elif response.status_code == 400:
try:
error_data = response.json()
error_message = error_data.get('error', {{}}).get('message', 'Unknown error')
except:
error_message = response.text
error_msg = f"⚠️ **Request Error**\\n\\n"
error_msg += f"The API request was invalid:\\n"
error_msg += f"`{{error_message}}`\\n\\n"
if "model" in error_message.lower():
error_msg += f"**Model Issue:** The model `{{MODEL}}` may not be available.\\n"
error_msg += f"Try switching to a different model in your Space configuration."
print(f"❌ Bad request: {{response.status_code}} - {{error_message}}")
return error_msg
else:
error_msg = f"🚫 **API Error {{response.status_code}}**\\n\\n"
error_msg += f"An unexpected error occurred. Please try again.\\n\\n"
error_msg += f"If this persists, check:\\n"
error_msg += f"1. OpenRouter service status\\n"
error_msg += f"2. Your API key and credits\\n"
error_msg += f"3. The model availability"
print(f"❌ API error: {{response.status_code}} - {{response.text[:200]}}")
return error_msg
except requests.exceptions.Timeout:
error_msg = f"⏰ **Request Timeout**\\n\\n"
error_msg += f"The API request took too long (30s limit).\\n\\n"
error_msg += f"**Troubleshooting:**\\n"
error_msg += f"1. Try again with a shorter message\\n"
error_msg += f"2. Check your internet connection\\n"
error_msg += f"3. Try a different model"
print(f"❌ Request timeout after 30 seconds")
return error_msg
except requests.exceptions.ConnectionError:
error_msg = f"🌐 **Connection Error**\\n\\n"
error_msg += f"Could not connect to OpenRouter API.\\n\\n"
error_msg += f"**Troubleshooting:**\\n"
error_msg += f"1. Check your internet connection\\n"
error_msg += f"2. Check OpenRouter service status\\n"
error_msg += f"3. Try again in a few moments"
print(f"❌ Connection error to OpenRouter API")
return error_msg
except Exception as e:
error_msg = f"❌ **Unexpected Error**\\n\\n"
error_msg += f"An unexpected error occurred:\\n"
error_msg += f"`{{str(e)}}`\\n\\n"
error_msg += f"Please try again or contact support if this persists."
print(f"❌ Unexpected error: {{str(e)}}")
return error_msg
# Access code verification
access_granted = gr.State(False)
_access_granted_global = False # Global fallback
def verify_access_code(code):
\"\"\"Verify the access code\"\"\"
global _access_granted_global
if not ACCESS_CODE:
_access_granted_global = True
return gr.update(visible=False), gr.update(visible=True), gr.update(value=True)
if code == ACCESS_CODE:
_access_granted_global = True
return gr.update(visible=False), gr.update(visible=True), gr.update(value=True)
else:
_access_granted_global = False
return gr.update(visible=True, value="❌ Incorrect access code. Please try again."), gr.update(visible=False), gr.update(value=False)
def protected_generate_response(message, history):
\"\"\"Protected response function that checks access\"\"\"
# Check if access is granted via the global variable
if ACCESS_CODE and not _access_granted_global:
return "Please enter the access code to continue."
return generate_response(message, history)
def export_conversation(history):
\"\"\"Export conversation to markdown file\"\"\"
if not history:
return gr.update(visible=False)
markdown_content = export_conversation_to_markdown(history)
# Save to temporary file
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
f.write(markdown_content)
temp_file = f.name
return gr.update(value=temp_file, visible=True)
# Configuration status display
def get_configuration_status():
\"\"\"Generate a configuration status message for display\"\"\"
status_parts = []
if API_KEY_VALID:
status_parts.append("βœ… **API Key:** Configured and valid")
else:
status_parts.append("❌ **API Key:** Not configured - Set `{api_key_var}` in Space secrets")
status_parts.append(f"πŸ€– **Model:** {{MODEL}}")
status_parts.append(f"🌑️ **Temperature:** {temperature}")
status_parts.append(f"πŸ“ **Max Tokens:** {max_tokens}")
if GROUNDING_URLS:
status_parts.append(f"πŸ”— **URL Grounding:** {{len(GROUNDING_URLS)}} URLs configured")
if ENABLE_DYNAMIC_URLS:
status_parts.append("πŸ”„ **Dynamic URLs:** Enabled")
if ENABLE_WEB_SEARCH:
status_parts.append("πŸ” **Web Search:** Enabled")
if ENABLE_VECTOR_RAG:
status_parts.append("πŸ“š **Document RAG:** Enabled")
if ACCESS_CODE:
status_parts.append("πŸ” **Access Control:** Enabled")
else:
status_parts.append("🌐 **Access:** Public")
return "\\n".join(status_parts)
# Create interface with access code protection
with gr.Blocks(title=SPACE_NAME) as demo:
gr.Markdown(f"# {{SPACE_NAME}}")
gr.Markdown(SPACE_DESCRIPTION)
# Configuration status (always visible)
with gr.Accordion("πŸ“Š Configuration Status", open=not API_KEY_VALID):
gr.Markdown(get_configuration_status())
# Access code section (shown only if ACCESS_CODE is set)
with gr.Column(visible=bool(ACCESS_CODE)) as access_section:
gr.Markdown("### πŸ” Access Required")
gr.Markdown("Please enter the access code provided by your instructor:")
access_input = gr.Textbox(
label="Access Code",
placeholder="Enter access code...",
type="password"
)
access_btn = gr.Button("Submit", variant="primary")
access_error = gr.Markdown(visible=False)
# Main chat interface (hidden until access granted)
with gr.Column(visible=not bool(ACCESS_CODE)) as chat_section:
chat_interface = gr.ChatInterface(
fn=protected_generate_response,
title="", # Title already shown above
description="", # Description already shown above
examples=None
)
# Export functionality
with gr.Row():
export_btn = gr.Button("Export Conversation", variant="secondary", size="sm")
export_file = gr.File(label="Download Conversation", visible=False)
# Connect export functionality
export_btn.click(
export_conversation,
inputs=[chat_interface],
outputs=[export_file]
)
# Connect access verification
if ACCESS_CODE:
access_btn.click(
verify_access_code,
inputs=[access_input],
outputs=[access_error, chat_section, access_granted]
)
access_input.submit(
verify_access_code,
inputs=[access_input],
outputs=[access_error, chat_section, access_granted]
)
if __name__ == "__main__":
demo.launch()
'''
# Available models - Updated with valid OpenRouter model IDs
MODELS = [
"google/gemini-2.0-flash-001", # Fast, reliable, general tasks
"anthropic/claude-3.5-haiku", # Complex reasoning and analysis
"openai/gpt-4o-mini", # Balanced performance and cost
"meta-llama/llama-3.1-8b-instruct", # Open-source, efficient option
"mistralai/mistral-7b-instruct" # Good for technical topics
]
def fetch_url_content(url):
"""Fetch and extract text content from a URL - maintained for backward compatibility"""
return enhanced_fetch_url_content(url)
def get_grounding_context(urls):
"""Fetch context from grounding URLs"""
if not urls:
return ""
context_parts = []
for i, url in enumerate(urls, 1):
if url and url.strip():
content = fetch_url_content(url.strip())
context_parts.append(f"Context from URL {i} ({url}):\n{content}")
if context_parts:
return "\n\n" + "\n\n".join(context_parts) + "\n\n"
return ""
def create_readme(config):
"""Generate README with deployment instructions"""
return f"""---
title: {config['name']}
emoji: πŸ€–
colorFrom: blue
colorTo: red
sdk: gradio
sdk_version: 5.35.0
app_file: app.py
pinned: false
---
# {config['name']}
{config['description']}
## Quick Deploy to HuggingFace Spaces
### Step 1: Create the Space
1. Go to https://huggingface.co/spaces
2. Click "Create new Space"
3. Choose a name for your Space
4. Select **Gradio** as the SDK
5. Set visibility (Public/Private)
6. Click "Create Space"
### Step 2: Upload Files
1. In your new Space, click "Files" tab
2. Upload these files from the zip:
- `app.py`
- `requirements.txt`
3. Wait for "Building" to complete
### Step 3: Add API Key
1. Go to Settings (gear icon)
2. Click "Variables and secrets"
3. Click "New secret"
4. Name: `{config['api_key_var']}`
5. Value: Your OpenRouter API key
6. Click "Add"
{f'''### Step 4: Configure Access Control
Your Space is configured with access code protection. Students will need to enter the access code to use the chatbot.
1. Go to Settings (gear icon)
2. Click "Variables and secrets"
3. Click "New secret"
4. Name: `SPACE_ACCESS_CODE`
5. Value: `{config['access_code']}`
6. Click "Add"
**Important**: The access code is now stored securely as an environment variable and is not visible in your app code.
To disable access protection:
1. Go to Settings β†’ Variables and secrets
2. Delete the `SPACE_ACCESS_CODE` secret
3. The Space will rebuild automatically with no access protection
''' if config['access_code'] else ''}
### Step {4 if not config['access_code'] else 5}: Get Your API Key
1. Go to https://openrouter.ai/keys
2. Sign up/login if needed
3. Click "Create Key"
4. Copy the key (starts with `sk-or-`)
### Step {5 if not config['access_code'] else 6}: Test Your Space
- Go back to "App" tab
- Your Space should be running!
- Try the example prompts or ask a question
## Configuration
- **Model**: {config['model']}
- **Temperature**: {config['temperature']}
- **Max Tokens**: {config['max_tokens']}
- **API Key Variable**: {config['api_key_var']}"""
# Add optional configuration items
if config['access_code']:
readme_content += f"""
- **Access Code**: {config['access_code']} (Students need this to access the chatbot)"""
if config.get('enable_dynamic_urls'):
readme_content += """
- **Dynamic URL Fetching**: Enabled (Assistant can fetch URLs mentioned in conversations)"""
readme_content += f"""
## Customization
To modify your Space:
1. Edit `app.py` in your Space
2. Update configuration variables at the top
3. Changes deploy automatically
## Troubleshooting
- **"Please set your {config['api_key_var']}"**: Add the secret in Space settings
- **Error 401**: Invalid API key or no credits
- **Error 429**: Rate limit - wait and try again
- **Build failed**: Check requirements.txt formatting
## More Help
- HuggingFace Spaces: https://huggingface.co/docs/hub/spaces
- OpenRouter Docs: https://openrouter.ai/docs
- Gradio Docs: https://gradio.app/docs
---
Generated on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} with Chat U/I Helper
"""
return readme_content
def create_requirements(enable_vector_rag=False, enable_web_search=False):
"""Generate requirements.txt"""
base_requirements = "gradio>=5.35.0\nrequests>=2.32.3\nbeautifulsoup4>=4.12.3"
if enable_vector_rag:
base_requirements += "\nfaiss-cpu==1.7.4\nnumpy==1.24.3"
if enable_web_search:
base_requirements += "\ncrawl4ai>=0.2.0\naiohttp>=3.8.0"
return base_requirements
def generate_zip(name, description, system_prompt, model, api_key_var, temperature, max_tokens, examples_text, access_code="", enable_dynamic_urls=False, url1="", url2="", url3="", url4="", enable_vector_rag=False, rag_data=None, enable_web_search=False):
"""Generate deployable zip file"""
# Process examples
if examples_text and examples_text.strip():
examples_list = [ex.strip() for ex in examples_text.split('\n') if ex.strip()]
examples_json = json.dumps(examples_list)
else:
examples_json = json.dumps([
"Hello! How can you help me?",
"Tell me something interesting",
"What can you do?"
])
# Process grounding URLs
grounding_urls = []
for url in [url1, url2, url3, url4]:
if url and url.strip():
grounding_urls.append(url.strip())
# Use the provided system prompt directly
# Create config
config = {
'name': name,
'description': description,
'system_prompt': system_prompt,
'model': model,
'api_key_var': api_key_var,
'temperature': temperature,
'max_tokens': int(max_tokens),
'examples': examples_json,
'grounding_urls': json.dumps(grounding_urls),
'access_code': "", # Access code stored in environment variable for security
'enable_dynamic_urls': enable_dynamic_urls,
'enable_vector_rag': enable_vector_rag,
'enable_web_search': enable_web_search,
'rag_data_json': json.dumps(rag_data) if rag_data else 'None'
}
# Generate files
app_content = SPACE_TEMPLATE.format(**config)
# Pass original access_code to README for documentation
readme_config = config.copy()
readme_config['access_code'] = access_code or ""
readme_content = create_readme(readme_config)
requirements_content = create_requirements(enable_vector_rag, enable_web_search)
# Create zip file with clean naming
filename = f"{name.lower().replace(' ', '_').replace('-', '_')}.zip"
# Create zip in memory and save to disk
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
zip_file.writestr('app.py', app_content)
zip_file.writestr('requirements.txt', requirements_content)
zip_file.writestr('README.md', readme_content)
zip_file.writestr('config.json', json.dumps(config, indent=2))
# Write zip to file
zip_buffer.seek(0)
with open(filename, 'wb') as f:
f.write(zip_buffer.getvalue())
return filename
# Define callback functions outside the interface
def toggle_rag_section(enable_rag):
"""Toggle visibility of RAG section"""
return gr.update(visible=enable_rag)
def process_documents(files, current_rag_tool):
"""Process uploaded documents"""
if not files:
return "Please upload files first", current_rag_tool
if not HAS_RAG:
return "RAG functionality not available. Please install required dependencies.", current_rag_tool
try:
# Initialize RAG tool if not exists
if not current_rag_tool and RAGTool is not None:
current_rag_tool = RAGTool()
# Process files
result = current_rag_tool.process_uploaded_files(files)
if result['success']:
# Create status message
status_parts = [f"βœ… {result['message']}"]
# Add file summary
if result['summary']['files_processed']:
status_parts.append("\n**Processed files:**")
for file_info in result['summary']['files_processed']:
status_parts.append(f"- {file_info['name']} ({file_info['chunks']} chunks)")
# Add errors if any
if result.get('errors'):
status_parts.append("\n**Errors:**")
for error in result['errors']:
status_parts.append(f"- {error['file']}: {error['error']}")
# Add index stats
if result.get('index_stats'):
stats = result['index_stats']
status_parts.append(f"\n**Index stats:** {stats['total_chunks']} chunks, {stats['dimension']}D embeddings")
return "\n".join(status_parts), current_rag_tool
else:
return f"❌ {result['message']}", current_rag_tool
except Exception as e:
return f"❌ Error processing documents: {str(e)}", current_rag_tool
def update_sandbox_preview(config_data):
"""Update the sandbox preview with generated content"""
if not config_data:
return "Generate a space configuration to see preview here.", "<div style='text-align: center; padding: 50px; color: #666;'>No preview available</div>"
# Create preview info
preview_text = f"""**Space Configuration:**
- **Name:** {config_data.get('name', 'N/A')}
- **Model:** {config_data.get('model', 'N/A')}
- **Temperature:** {config_data.get('temperature', 'N/A')}
- **Max Tokens:** {config_data.get('max_tokens', 'N/A')}
- **Dynamic URLs:** {'βœ… Enabled' if config_data.get('enable_dynamic_urls') else '❌ Disabled'}
- **Vector RAG:** {'βœ… Enabled' if config_data.get('enable_vector_rag') else '❌ Disabled'}
**System Prompt Preview:**
```
{config_data.get('system_prompt', 'No system prompt configured')[:500]}{'...' if len(config_data.get('system_prompt', '')) > 500 else ''}
```
**Deployment Package:** `{config_data.get('filename', 'Not generated')}`"""
# Create a basic HTML preview of the chat interface
preview_html = f"""
<div style="border: 1px solid #ddd; border-radius: 8px; padding: 20px; background: #f9f9f9;">
<h3 style="margin-top: 0; color: #333;">{config_data.get('name', 'Chat Interface')}</h3>
<p style="color: #666; margin-bottom: 20px;">{config_data.get('description', 'A customizable AI chat interface')}</p>
<div style="border: 1px solid #ccc; border-radius: 4px; background: white; min-height: 200px; padding: 15px; margin-bottom: 15px;">
<div style="color: #888; text-align: center; padding: 50px 0;">Chat Interface Preview</div>
<div style="background: #f0f8ff; padding: 10px; border-radius: 4px; margin-bottom: 10px; border-left: 3px solid #0066cc;">
<strong>Assistant:</strong> Hello! I'm ready to help you. How can I assist you today?
</div>
</div>
<div style="border: 1px solid #ccc; border-radius: 4px; padding: 10px; background: white;">
<input type="text" placeholder="Type your message here..." style="width: 70%; padding: 8px; border: 1px solid #ddd; border-radius: 4px; margin-right: 10px;">
<button style="padding: 8px 15px; background: #0066cc; color: white; border: none; border-radius: 4px; cursor: pointer;">Send</button>
</div>
<div style="margin-top: 15px; padding: 10px; background: #f0f0f0; border-radius: 4px; font-size: 12px; color: #666;">
<strong>Configuration:</strong> Model: {config_data.get('model', 'N/A')} | Temperature: {config_data.get('temperature', 'N/A')} | Max Tokens: {config_data.get('max_tokens', 'N/A')}
</div>
</div>
"""
return preview_text, preview_html
def on_preview_combined(name, description, system_prompt, enable_research_assistant, model, temperature, max_tokens, examples_text, enable_dynamic_urls, enable_vector_rag, enable_web_search):
"""Generate configuration and return preview updates"""
if not name or not name.strip():
return (
{},
gr.update(value="**Error:** Please provide a Space Title to preview", visible=True),
gr.update(visible=False),
gr.update(value="Configuration will appear here after preview generation.")
)
try:
# Use the system prompt directly (research assistant toggle already updates it)
if not system_prompt or not system_prompt.strip():
return (
{},
gr.update(value="**Error:** Please provide a System Prompt for the assistant", visible=True),
gr.update(visible=False),
gr.update(value="Configuration will appear here after preview generation.")
)
final_system_prompt = system_prompt.strip()
# Create configuration for preview
config_data = {
'name': name,
'description': description,
'system_prompt': final_system_prompt,
'model': model,
'temperature': temperature,
'max_tokens': max_tokens,
'enable_dynamic_urls': enable_dynamic_urls,
'enable_vector_rag': enable_vector_rag,
'enable_web_search': enable_web_search,
'examples_text': examples_text,
'preview_ready': True
}
# Generate preview displays
preview_text = f"""πŸŽ‰ **Preview Successfully Rendered!**
Your assistant "{name}" is now configured and ready to test in the Sandbox Preview tab.
**Configuration:**
- **Model:** {model}
- **Temperature:** {temperature}
- **Max Tokens:** {max_tokens}
- **Dynamic URLs:** {'βœ… Enabled' if enable_dynamic_urls else '❌ Disabled'}
- **Vector RAG:** {'βœ… Enabled' if enable_vector_rag else '❌ Disabled'}
- **Web Search:** {'βœ… Enabled' if enable_web_search else '❌ Disabled'}
**System Prompt:**
{final_system_prompt[:200]}{'...' if len(final_system_prompt) > 200 else ''}
✨ **Next Steps:** Switch to the "Sandbox Preview" tab to test your assistant with real conversations before generating the deployment package."""
config_display = f"""### Current Configuration
**Space Details:**
- **Name:** {name}
- **Description:** {description or 'No description provided'}
**Model Settings:**
- **Model:** {model}
- **Temperature:** {temperature}
- **Max Response Tokens:** {max_tokens}
**Features:**
- **Dynamic URL Fetching:** {'βœ… Enabled' if enable_dynamic_urls else '❌ Disabled'}
- **Document RAG:** {'βœ… Enabled' if enable_vector_rag else '❌ Disabled'}
- **Web Search:** {'βœ… Enabled' if enable_web_search else '❌ Disabled'}
**System Prompt:**
```
{final_system_prompt}
```
**Example Prompts:**
{examples_text if examples_text and examples_text.strip() else 'No example prompts configured'}
"""
return (
config_data,
gr.update(value=preview_text, visible=True),
gr.update(visible=True),
gr.update(value=config_display)
)
except Exception as e:
return (
{},
gr.update(value=f"**Error:** {str(e)}", visible=True),
gr.update(visible=False),
gr.update(value="Configuration will appear here after preview generation.")
)
def update_preview_display(config_data):
"""Update preview display based on config data"""
if not config_data or not config_data.get('preview_ready'):
return (
gr.update(value="**Status:** Configure your space in the Configuration tab and click 'Preview Deployment Package' to see your assistant here.", visible=True),
gr.update(visible=False),
gr.update(value="Configuration will appear here after preview generation.")
)
preview_text = f"""**Preview Ready!**
Your assistant "{config_data['name']}" is configured and ready to test.
**Configuration:**
- **Model:** {config_data['model']}
- **Temperature:** {config_data['temperature']}
- **Max Tokens:** {config_data['max_tokens']}
- **Dynamic URLs:** {'βœ… Enabled' if config_data['enable_dynamic_urls'] else '❌ Disabled'}
- **Vector RAG:** {'βœ… Enabled' if config_data['enable_vector_rag'] else '❌ Disabled'}
**System Prompt:**
{config_data['system_prompt'][:200]}{'...' if len(config_data['system_prompt']) > 200 else ''}
Use the chat interface below to test your assistant before generating the deployment package."""
config_display = f"""### Current Configuration
**Space Details:**
- **Name:** {config_data['name']}
- **Description:** {config_data.get('description', 'No description provided')}
**Model Settings:**
- **Model:** {config_data['model']}
- **Temperature:** {config_data['temperature']}
- **Max Response Tokens:** {config_data['max_tokens']}
**Features:**
- **Dynamic URL Fetching:** {'βœ… Enabled' if config_data['enable_dynamic_urls'] else '❌ Disabled'}
- **Document RAG:** {'βœ… Enabled' if config_data['enable_vector_rag'] else '❌ Disabled'}
**System Prompt:**
```
{config_data['system_prompt']}
```
**Example Prompts:**
{config_data.get('examples_text', 'No example prompts configured') if config_data.get('examples_text', '').strip() else 'No example prompts configured'}
"""
return (
gr.update(value=preview_text, visible=True),
gr.update(visible=True),
gr.update(value=config_display)
)
def preview_chat_response(message, history, config_data, url1="", url2="", url3="", url4=""):
"""Generate response for preview chat using actual OpenRouter API"""
if not config_data or not message:
return "", history
# Get API key from environment
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
response = f"""πŸ”‘ **API Key Required for Preview**
To test your assistant with real API responses, please:
1. Get your OpenRouter API key from: https://openrouter.ai/keys
2. Set it as an environment variable: `export OPENROUTER_API_KEY=your_key_here`
3. Or add it to your `.env` file: `OPENROUTER_API_KEY=your_key_here`
**Your Configuration:**
- **Name:** {config_data.get('name', 'your assistant')}
- **Model:** {config_data.get('model', 'unknown model')}
- **Temperature:** {config_data.get('temperature', 0.7)}
- **Max Tokens:** {config_data.get('max_tokens', 500)}
**System Prompt Preview:**
{config_data.get('system_prompt', '')[:200]}{'...' if len(config_data.get('system_prompt', '')) > 200 else ''}
Once you set your API key, you'll be able to test real conversations in this preview."""
history.append({"role": "user", "content": message})
history.append({"role": "assistant", "content": response})
return "", history
try:
# Get grounding context from URLs if configured
grounding_urls = [url1, url2, url3, url4]
grounding_context = get_cached_grounding_context([url for url in grounding_urls if url and url.strip()])
# Add RAG context if available (simplified for preview)
rag_context = ""
if config_data.get('enable_vector_rag'):
rag_context = "\n\n[RAG context would be retrieved here based on similarity search]\n\n"
# If dynamic URLs are enabled, check message for URLs to fetch
dynamic_context = ""
if config_data.get('enable_dynamic_urls'):
urls_in_message = extract_urls_from_text(message)
if urls_in_message:
dynamic_context_parts = []
for url in urls_in_message[:3]: # Increased limit to 3 URLs with enhanced processing
content = enhanced_fetch_url_content(url)
dynamic_context_parts.append(f"\n\nDynamic context from {url}:\n{content}")
if dynamic_context_parts:
dynamic_context = "\n".join(dynamic_context_parts)
# Check for web search request if enabled
web_search_result = ""
if config_data.get('enable_web_search'):
# If web search is enabled, use it for most queries (excluding code blocks and URLs)
should_search = True
# Skip search for messages that are primarily code blocks
if re.search(r'```[\s\S]*```', message):
should_search = False
# Skip search for messages that are primarily URLs
urls_in_message = extract_urls_from_text(message)
if urls_in_message and len(' '.join(urls_in_message)) > len(message) * 0.5:
should_search = False
# Skip search for very short messages (likely greetings)
if len(message.strip()) < 5:
should_search = False
if should_search:
# Use the entire message as search query, cleaning it up
search_query = message.strip()
search_result = perform_web_search(search_query, "Web search requested")
web_search_result = f"\n\n{search_result}\n\n"
# Build enhanced system prompt with all contexts
enhanced_system_prompt = config_data.get('system_prompt', '') + grounding_context + rag_context + dynamic_context + web_search_result
# Build messages array for the API
messages = [{"role": "system", "content": enhanced_system_prompt}]
# Add conversation history - handle both formats for backwards compatibility
for chat in history:
if isinstance(chat, dict):
# New format: {"role": "user", "content": "..."}
messages.append(chat)
elif isinstance(chat, list) and len(chat) >= 2:
# Legacy format: [user_msg, assistant_msg]
user_msg, assistant_msg = chat[0], chat[1]
if user_msg:
messages.append({"role": "user", "content": user_msg})
if assistant_msg:
messages.append({"role": "assistant", "content": assistant_msg})
# Add current message
messages.append({"role": "user", "content": message})
# Debug: Log the request being sent
request_payload = {
"model": config_data.get('model', 'google/gemini-2.0-flash-001'),
"messages": messages,
"temperature": config_data.get('temperature', 0.7),
"max_tokens": config_data.get('max_tokens', 500),
"tools": None # Explicitly disable tool/function calling
}
# Make API request to OpenRouter
response = requests.post(
url="https://openrouter.ai/api/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json=request_payload,
timeout=30
)
if response.status_code == 200:
try:
response_data = response.json()
# Check if response has expected structure
if 'choices' not in response_data or not response_data['choices']:
assistant_response = f"[Preview Debug] No choices in API response. Response: {response_data}"
elif 'message' not in response_data['choices'][0]:
assistant_response = f"[Preview Debug] No message in first choice. Response: {response_data}"
elif 'content' not in response_data['choices'][0]['message']:
assistant_response = f"[Preview Debug] No content in message. Response: {response_data}"
else:
assistant_content = response_data['choices'][0]['message']['content']
# Debug: Check if content is empty
if not assistant_content or assistant_content.strip() == "":
assistant_response = f"[Preview Debug] Empty content from API. Messages sent: {len(messages)} messages, last user message: '{message}', model: {request_payload['model']}"
else:
# Use the content directly - no preview indicator needed
assistant_response = assistant_content
except (KeyError, IndexError, json.JSONDecodeError) as e:
assistant_response = f"[Preview Error] Failed to parse API response: {str(e)}. Raw response: {response.text[:500]}"
else:
assistant_response = f"[Preview Error] API Error: {response.status_code} - {response.text[:500]}"
except Exception as e:
assistant_response = f"[Preview Error] {str(e)}"
# Return in the new messages format for Gradio 5.x
history.append({"role": "user", "content": message})
history.append({"role": "assistant", "content": assistant_response})
return "", history
def clear_preview_chat():
"""Clear preview chat"""
return "", []
def export_preview_conversation(history):
"""Export preview conversation to markdown"""
if not history:
return gr.update(visible=False)
markdown_content = export_conversation_to_markdown(history)
# Save to temporary file
import tempfile
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
f.write(markdown_content)
temp_file = f.name
return gr.update(value=temp_file, visible=True)
def on_generate(name, description, system_prompt, enable_research_assistant, model, api_key_var, temperature, max_tokens, examples_text, access_code, enable_dynamic_urls, url1, url2, url3, url4, enable_vector_rag, rag_tool_state, enable_web_search):
if not name or not name.strip():
return gr.update(value="Error: Please provide a Space Title", visible=True), gr.update(visible=False), {}
try:
# Get RAG data if enabled
rag_data = None
if enable_vector_rag and rag_tool_state:
rag_data = rag_tool_state.get_serialized_data()
# Use the system prompt directly (research assistant toggle already updates it)
if not system_prompt or not system_prompt.strip():
return gr.update(value="Error: Please provide a System Prompt for the assistant", visible=True), gr.update(visible=False), {}
final_system_prompt = system_prompt.strip()
filename = generate_zip(name, description, final_system_prompt, model, api_key_var, temperature, max_tokens, examples_text, access_code, enable_dynamic_urls, url1, url2, url3, url4, enable_vector_rag, rag_data, enable_web_search)
success_msg = f"""**Deployment package ready!**
**File**: `{filename}`
**What's included:**
- `app.py` - Ready-to-deploy chat interface
- `requirements.txt` - Dependencies
- `README.md` - Step-by-step deployment guide
- `config.json` - Configuration backup
**Next steps:**
1. Download the zip file below
2. Follow the README instructions to deploy on HuggingFace Spaces
3. Set your `{api_key_var}` secret in Space settings
**Your Space will be live in minutes!**"""
# Update sandbox preview
config_data = {
'name': name,
'description': description,
'system_prompt': final_system_prompt,
'model': model,
'temperature': temperature,
'max_tokens': max_tokens,
'enable_dynamic_urls': enable_dynamic_urls,
'enable_vector_rag': enable_vector_rag,
'enable_web_search': enable_web_search,
'filename': filename
}
return gr.update(value=success_msg, visible=True), gr.update(value=filename, visible=True), config_data
except Exception as e:
return gr.update(value=f"Error: {str(e)}", visible=True), gr.update(visible=False), {}
# Global cache for URL content to avoid re-crawling
url_content_cache = {}
def get_cached_grounding_context(urls):
"""Get grounding context with caching to avoid re-crawling same URLs"""
if not urls:
return ""
# Filter valid URLs
valid_urls = [url for url in urls if url and url.strip()]
if not valid_urls:
return ""
# Create cache key from sorted URLs
cache_key = tuple(sorted(valid_urls))
# Check if we already have this content cached
if cache_key in url_content_cache:
return url_content_cache[cache_key]
# If not cached, fetch using simple HTTP requests
grounding_context = get_grounding_context_simple(valid_urls)
# Cache the result
url_content_cache[cache_key] = grounding_context
return grounding_context
def respond_with_cache_update(message, chat_history, url1="", url2="", url3="", url4=""):
"""Wrapper that updates cache status after responding"""
msg, history = respond(message, chat_history, url1, url2, url3, url4)
cache_status = get_cache_status()
return msg, history, cache_status
def respond(message, chat_history, url1="", url2="", url3="", url4=""):
# Make actual API request to OpenRouter
import os
import requests
# Get API key from environment
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
response = "Please set your OPENROUTER_API_KEY in the Space settings to use the chat support."
chat_history.append({"role": "user", "content": message})
chat_history.append({"role": "assistant", "content": response})
return "", chat_history
# Get grounding context from URLs using cached approach
grounding_urls = [url1, url2, url3, url4]
grounding_context = get_cached_grounding_context(grounding_urls)
# Build enhanced system prompt with grounding context
base_system_prompt = """You are an expert assistant specializing in Gradio configurations for HuggingFace Spaces. You have deep knowledge of:
- Gradio interface components and layouts
- HuggingFace Spaces configuration (YAML frontmatter, secrets, environment variables)
- Deployment best practices for Gradio apps on HuggingFace
- Space settings, SDK versions, and hardware requirements
- Troubleshooting common Gradio and HuggingFace Spaces issues
- Integration with various APIs and models through Gradio interfaces
Provide specific, technical guidance focused on Gradio implementation details and HuggingFace Spaces deployment. Include code examples when relevant. Keep responses concise and actionable."""
enhanced_system_prompt = base_system_prompt + grounding_context
# Build conversation history for API
messages = [{
"role": "system",
"content": enhanced_system_prompt
}]
# Add conversation history - Support both new messages format and legacy tuple format
for chat in chat_history:
if isinstance(chat, dict):
# New format: {"role": "user", "content": "..."}
messages.append(chat)
elif isinstance(chat, (list, tuple)) and len(chat) >= 2:
# Legacy format: ("user msg", "bot msg")
user_msg, assistant_msg = chat[0], chat[1]
if user_msg:
messages.append({"role": "user", "content": user_msg})
if assistant_msg:
messages.append({"role": "assistant", "content": assistant_msg})
# Add current message
messages.append({"role": "user", "content": message})
try:
# Make API request to OpenRouter
response = requests.post(
url="https://openrouter.ai/api/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "google/gemini-2.0-flash-001",
"messages": messages,
"temperature": 0.7,
"max_tokens": 500
}
)
if response.status_code == 200:
assistant_response = response.json()['choices'][0]['message']['content']
else:
assistant_response = f"Error: {response.status_code} - {response.text}"
except Exception as e:
assistant_response = f"Error: {str(e)}"
chat_history.append({"role": "user", "content": message})
chat_history.append({"role": "assistant", "content": assistant_response})
return "", chat_history
def clear_chat():
return "", []
def clear_url_cache():
"""Clear the URL content cache"""
global url_content_cache
url_content_cache.clear()
return "βœ… URL cache cleared. Next request will re-fetch content."
def get_cache_status():
"""Get current cache status"""
if not url_content_cache:
return "πŸ”„ No URLs cached"
return f"πŸ’Ύ {len(url_content_cache)} URL set(s) cached"
def add_urls(count):
"""Show additional URL fields"""
if count == 2:
return (gr.update(visible=True), gr.update(visible=False),
gr.update(value="+ Add URLs"), gr.update(visible=True), 3)
elif count == 3:
return (gr.update(visible=True), gr.update(visible=True),
gr.update(value="Max URLs", interactive=False), gr.update(visible=True), 4)
else:
return (gr.update(), gr.update(), gr.update(), gr.update(), count)
def remove_urls(count):
"""Hide URL fields"""
if count == 4:
return (gr.update(visible=True), gr.update(visible=False, value=""),
gr.update(value="+ Add URLs", interactive=True), gr.update(visible=True), 3)
elif count == 3:
return (gr.update(visible=False, value=""), gr.update(visible=False, value=""),
gr.update(value="+ Add URLs", interactive=True), gr.update(visible=False), 2)
else:
return (gr.update(), gr.update(), gr.update(), gr.update(), count)
def add_chat_urls(count):
"""Show additional chat URL fields"""
if count == 2:
return (gr.update(visible=True), gr.update(visible=False),
gr.update(value="+ Add URLs"), gr.update(visible=True), 3)
elif count == 3:
return (gr.update(visible=True), gr.update(visible=True),
gr.update(value="Max URLs", interactive=False), gr.update(visible=True), 4)
else:
return (gr.update(), gr.update(), gr.update(), gr.update(), count)
def remove_chat_urls(count):
"""Hide chat URL fields"""
if count == 4:
return (gr.update(visible=True), gr.update(visible=False, value=""),
gr.update(value="+ Add URLs", interactive=True), gr.update(visible=True), 3)
elif count == 3:
return (gr.update(visible=False, value=""), gr.update(visible=False, value=""),
gr.update(value="+ Add URLs", interactive=True), gr.update(visible=False), 2)
else:
return (gr.update(), gr.update(), gr.update(), gr.update(), count)
# Code execution toggle removed - functionality no longer supported
def toggle_web_search(enable_search):
"""Toggle visibility of web search space field"""
return gr.update(visible=enable_search)
def perform_web_search(query, description="Web search"):
"""Perform web search using crawl4ai with DuckDuckGo"""
try:
# Try to use crawl4ai for web search
try:
from crawl4ai import WebCrawler
import asyncio
async def search_with_crawl4ai(search_query):
# Create search URL for DuckDuckGo
import urllib.parse
encoded_query = urllib.parse.quote_plus(search_query)
search_url = f"https://duckduckgo.com/html/?q={encoded_query}"
# Initialize crawler
crawler = WebCrawler(verbose=False)
try:
# Start the crawler
await crawler.astart()
# Crawl the search results
result = await crawler.arun(url=search_url)
if result.success:
# Extract text content from search results
content = result.cleaned_html if result.cleaned_html else result.markdown
# Clean and truncate the content
if content:
# Remove excessive whitespace and limit length
lines = [line.strip() for line in content.split('\n') if line.strip()]
cleaned_content = '\n'.join(lines)
# Truncate to reasonable length for context
if len(cleaned_content) > 3000:
cleaned_content = cleaned_content[:3000] + "..."
return cleaned_content
else:
return "No content extracted from search results"
else:
return f"Search failed: {result.error_message if hasattr(result, 'error_message') else 'Unknown error'}"
finally:
# Clean up the crawler
await crawler.aclose()
# Run the async search
if hasattr(asyncio, 'run'):
search_result = asyncio.run(search_with_crawl4ai(query))
else:
# Fallback for older Python versions
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
search_result = loop.run_until_complete(search_with_crawl4ai(query))
finally:
loop.close()
return f"**{description}**\n\nQuery: {query}\n\n**Search Results:**\n{search_result}"
except ImportError:
# Fallback to simple DuckDuckGo search without crawl4ai
import urllib.parse
encoded_query = urllib.parse.quote_plus(query)
search_url = f"https://duckduckgo.com/html/?q={encoded_query}"
# Use enhanced_fetch_url_content as fallback
content = enhanced_fetch_url_content(search_url)
return f"**{description} (Simplified)**\n\nQuery: {query}\n\n**Search Results:**\n{content}"
except Exception as e:
# Final fallback to URL extraction if search fails
urls = extract_urls_from_text(query)
if urls:
results = []
for url in urls[:2]: # Limit to 2 URLs for fallback
content = enhanced_fetch_url_content(url)
results.append(f"Content from {url}:\n{content[:500]}...")
return f"**Web Search Fallback:** {description}\n\n" + "\n\n".join(results)
return f"**Web Search Error:** {str(e)}\n\nQuery: {query}"
# Code execution functionality removed - no longer supported
def toggle_research_assistant(enable_research):
"""Toggle research assistant system prompt"""
if enable_research:
combined_prompt = "You are an advanced research assistant specializing in academic literature search and analysis. Your expertise includes finding peer-reviewed sources, critically evaluating research methodology, synthesizing insights across multiple papers, and providing properly formatted citations. When responding, ground all claims in specific sources from provided URL contexts, distinguish between direct evidence and analytical interpretation, and highlight any limitations or conflicting findings. Use clear, accessible language that makes complex research understandable, and suggest related areas of inquiry when relevant. Your goal is to be a knowledgeable research partner who helps users navigate academic information with precision and clarity."
return (
gr.update(value=combined_prompt), # Update main system prompt
gr.update(value=True) # Enable dynamic URL fetching for research template
)
else:
return (
gr.update(value=""), # Clear main system prompt when disabling
gr.update(value=False) # Disable dynamic URL setting
)
# Create Gradio interface with proper tab structure and fixed configuration
with gr.Blocks(
title="Chat U/I Helper",
css="""
/* Custom CSS to fix styling issues */
.gradio-container {
max-width: 1200px !important;
margin: 0 auto;
}
/* Fix tab styling */
.tab-nav {
border-bottom: 1px solid #e0e0e0;
}
/* Fix button styling */
.btn {
border-radius: 6px;
}
/* Fix chat interface styling */
.chat-interface {
border-radius: 8px;
border: 1px solid #e0e0e0;
}
/* Hide gradio footer to avoid manifest issues */
.gradio-footer {
display: none !important;
}
/* Fix accordion styling */
.accordion {
border: 1px solid #e0e0e0;
border-radius: 6px;
}
""",
theme=gr.themes.Default(),
head="""
<style>
/* Additional head styles to prevent manifest issues */
.gradio-app {
background: #ffffff;
}
</style>
""",
js="""
function() {
// Prevent manifest.json requests and other common errors
if (typeof window !== 'undefined') {
// Override fetch to handle manifest.json requests
const originalFetch = window.fetch;
window.fetch = function(url, options) {
// Handle both string URLs and URL objects
const urlString = typeof url === 'string' ? url : url.toString();
if (urlString.includes('manifest.json')) {
return Promise.resolve(new Response('{}', {
status: 200,
headers: { 'Content-Type': 'application/json' }
}));
}
// Handle favicon requests
if (urlString.includes('favicon.ico')) {
return Promise.resolve(new Response('', { status: 204 }));
}
return originalFetch.apply(this, arguments);
};
// Prevent postMessage origin errors
window.addEventListener('message', function(event) {
try {
if (event.origin && event.origin !== window.location.origin) {
event.stopImmediatePropagation();
return false;
}
} catch (e) {
// Silently ignore origin check errors
}
}, true);
// Prevent console errors from missing resources
window.addEventListener('error', function(e) {
if (e.target && e.target.src) {
const src = e.target.src;
if (src.includes('manifest.json') || src.includes('favicon.ico')) {
e.preventDefault();
return false;
}
}
}, true);
// Override console.error to filter out known harmless errors
const originalConsoleError = console.error;
console.error = function(...args) {
const message = args.join(' ');
if (message.includes('manifest.json') ||
message.includes('favicon.ico') ||
message.includes('postMessage') ||
message.includes('target origin')) {
return; // Suppress these specific errors
}
originalConsoleError.apply(console, arguments);
};
}
}
"""
) as demo:
# Global state for cross-tab functionality
sandbox_state = gr.State({})
preview_config_state = gr.State({})
# Global status components that will be defined later
preview_status = None
preview_chat_section = None
config_display = None
with gr.Tabs():
with gr.Tab("Configuration"):
gr.Markdown("# Spaces Configuration")
gr.Markdown("Convert custom assistants from HuggingChat into chat interfaces with HuggingFace Spaces. Configure and download everything needed to deploy a simple HF space using Gradio.")
with gr.Column():
name = gr.Textbox(
label="Space Title",
placeholder="My Course Helper",
value="My Custom Space"
)
description = gr.Textbox(
label="Space Description",
placeholder="A customizable AI chat interface for...",
lines=2,
value=""
)
model = gr.Dropdown(
label="Model",
choices=MODELS,
value=MODELS[0],
info="Choose based on the context and purposes of your space"
)
api_key_var = gr.Textbox(
label="API Key Variable Name",
value="OPENROUTER_API_KEY",
info="Name for the secret in HuggingFace Space settings"
)
access_code = gr.Textbox(
label="Access Code (Optional)",
placeholder="Leave empty for public access, or enter code for student access",
info="If set, students must enter this code to access the chatbot",
type="password"
)
with gr.Accordion("Assistant Configuration", open=True):
gr.Markdown("### Configure your assistant's behavior and capabilities")
gr.Markdown("Define the system prompt and assistant settings. You can use pre-configured templates or custom fields.")
# Main system prompt field - always visible
system_prompt = gr.Textbox(
label="System Prompt",
placeholder="You are a helpful assistant that...",
lines=4,
value="",
info="Define the assistant's role, purpose, and behavior in a single prompt"
)
# Assistant configuration options
enable_research_assistant = gr.Checkbox(
label="Research Template",
value=False,
info="Enable to use pre-configured research assistant settings"
)
examples_text = gr.Textbox(
label="Example Prompts (one per line)",
placeholder="Can you analyze this research paper: https://example.com/paper.pdf\nWhat are the latest findings on climate change adaptation?\nHelp me fact-check claims about renewable energy efficiency",
lines=3,
info="These will appear as clickable examples in the chat interface"
)
with gr.Accordion("Tool Settings", open=True):
enable_dynamic_urls = gr.Checkbox(
label="Enable Dynamic URL Fetching",
value=True, # Enabled by default
info="Allow the assistant to fetch additional URLs mentioned in conversations (enabled by default)",
visible=False # Hidden since it's always enabled
)
enable_web_search = gr.Checkbox(
label="Enable Web Search",
value=False,
info="Allow the assistant to search the web using crawl4ai"
)
web_search_space = gr.Textbox(
label="Web Search Technology",
value="crawl4ai",
info="Uses crawl4ai library for web crawling",
visible=False,
interactive=False
)
enable_vector_rag = gr.Checkbox(
label="Enable Document RAG",
value=False,
info="Upload documents for context-aware responses (PDF, DOCX, TXT, MD)",
visible=HAS_RAG
)
with gr.Column(visible=False) as rag_section:
gr.Markdown("### Document Upload")
file_upload = gr.File(
label="Upload Documents",
file_types=[".pdf", ".docx", ".txt", ".md"],
file_count="multiple"
)
process_btn = gr.Button("Process Documents", variant="secondary")
rag_status = gr.Markdown()
# State to store RAG tool
rag_tool_state = gr.State(None)
with gr.Accordion("URL Grounding (Optional)", open=False):
gr.Markdown("Add URLs to provide context. Content will be fetched and added to the system prompt.")
# Initial URL fields
url1 = gr.Textbox(
label="URL 1",
placeholder="https://example.com/page1",
info="First URL for context grounding"
)
url2 = gr.Textbox(
label="URL 2",
placeholder="https://example.com/page2",
info="Second URL for context grounding"
)
# Additional URL fields (initially hidden)
url3 = gr.Textbox(
label="URL 3",
placeholder="https://example.com/page3",
info="Third URL for context grounding",
visible=False
)
url4 = gr.Textbox(
label="URL 4",
placeholder="https://example.com/page4",
info="Fourth URL for context grounding",
visible=False
)
# URL management buttons
with gr.Row():
add_url_btn = gr.Button("+ Add URLs", size="sm")
remove_url_btn = gr.Button("- Remove URLs", size="sm", visible=False)
url_count = gr.State(2) # Track number of visible URLs
with gr.Accordion("Advanced Settings", open=False):
with gr.Row():
temperature = gr.Slider(
label="Temperature",
minimum=0,
maximum=2,
value=0.7,
step=0.1,
info="Higher = more creative, Lower = more focused"
)
max_tokens = gr.Slider(
label="Max Response Tokens",
minimum=50,
maximum=4096,
value=500,
step=50
)
with gr.Row():
preview_btn = gr.Button("Preview Deployment Package", variant="secondary")
generate_btn = gr.Button("Generate Deployment Package", variant="primary")
status = gr.Markdown(visible=False)
download_file = gr.File(label="Download your zip package", visible=False)
# Connect the research assistant checkbox
enable_research_assistant.change(
toggle_research_assistant,
inputs=[enable_research_assistant],
outputs=[system_prompt, enable_dynamic_urls]
)
# Connect the web search checkbox
enable_web_search.change(
toggle_web_search,
inputs=[enable_web_search],
outputs=[web_search_space]
)
# Connect the URL management buttons
add_url_btn.click(
add_urls,
inputs=[url_count],
outputs=[url3, url4, add_url_btn, remove_url_btn, url_count]
)
remove_url_btn.click(
remove_urls,
inputs=[url_count],
outputs=[url3, url4, add_url_btn, remove_url_btn, url_count]
)
# Connect RAG functionality
enable_vector_rag.change(
toggle_rag_section,
inputs=[enable_vector_rag],
outputs=[rag_section]
)
process_btn.click(
process_documents,
inputs=[file_upload, rag_tool_state],
outputs=[rag_status, rag_tool_state]
)
# Connect the generate button
generate_btn.click(
on_generate,
inputs=[name, description, system_prompt, enable_research_assistant, model, api_key_var, temperature, max_tokens, examples_text, access_code, enable_dynamic_urls, url1, url2, url3, url4, enable_vector_rag, rag_tool_state, enable_web_search],
outputs=[status, download_file, sandbox_state]
)
with gr.Tab("Preview"):
gr.Markdown("# Sandbox Preview")
gr.Markdown("Preview your generated assistant exactly as it will appear in the deployed HuggingFace Space.")
with gr.Column():
# Preview status - assign to global variable
preview_status_comp = gr.Markdown("**Status:** Configure your space in the Configuration tab and click 'Preview Deployment Package' to see your assistant here.", visible=True)
# Simulated chat interface for preview
with gr.Column(visible=False) as preview_chat_section_comp:
preview_chatbot = gr.Chatbot(
value=[],
label="Preview Chat Interface",
height=400,
type="messages" # Use the new messages format
)
preview_msg = gr.Textbox(
label="Test your assistant",
placeholder="Type a message to test your assistant...",
lines=2
)
# URL context fields for preview testing
with gr.Accordion("Test URL Context (Optional)", open=False):
gr.Markdown("Add URLs to test context grounding in the preview")
with gr.Row():
preview_url1 = gr.Textbox(
label="URL 1",
placeholder="https://example.com/page1",
scale=1
)
preview_url2 = gr.Textbox(
label="URL 2",
placeholder="https://example.com/page2",
scale=1
)
with gr.Row():
preview_url3 = gr.Textbox(
label="URL 3",
placeholder="https://example.com/page3",
scale=1,
visible=False
)
preview_url4 = gr.Textbox(
label="URL 4",
placeholder="https://example.com/page4",
scale=1,
visible=False
)
# URL management for preview
with gr.Row():
preview_add_url_btn = gr.Button("+ Add URLs", size="sm")
preview_remove_url_btn = gr.Button("- Remove URLs", size="sm", visible=False)
preview_url_count = gr.State(2)
with gr.Row():
preview_send = gr.Button("Send", variant="primary")
preview_clear = gr.Button("Clear")
export_btn = gr.Button("Export Conversation", variant="secondary")
# Export functionality
export_file = gr.File(label="Download Conversation", visible=False)
# Configuration display - assign to global variable
config_display_comp = gr.Markdown("Configuration will appear here after preview generation.")
# Connect preview chat functionality
preview_send.click(
preview_chat_response,
inputs=[preview_msg, preview_chatbot, preview_config_state, preview_url1, preview_url2, preview_url3, preview_url4],
outputs=[preview_msg, preview_chatbot]
)
preview_msg.submit(
preview_chat_response,
inputs=[preview_msg, preview_chatbot, preview_config_state, preview_url1, preview_url2, preview_url3, preview_url4],
outputs=[preview_msg, preview_chatbot]
)
preview_clear.click(
clear_preview_chat,
outputs=[preview_msg, preview_chatbot]
)
export_btn.click(
export_preview_conversation,
inputs=[preview_chatbot],
outputs=[export_file]
)
# Connect preview URL management buttons
preview_add_url_btn.click(
add_chat_urls,
inputs=[preview_url_count],
outputs=[preview_url3, preview_url4, preview_add_url_btn, preview_remove_url_btn, preview_url_count]
)
preview_remove_url_btn.click(
remove_chat_urls,
inputs=[preview_url_count],
outputs=[preview_url3, preview_url4, preview_add_url_btn, preview_remove_url_btn, preview_url_count]
)
with gr.Tab("Support"):
create_support_docs()
# Connect cross-tab functionality after all components are defined
preview_btn.click(
on_preview_combined,
inputs=[name, description, system_prompt, enable_research_assistant, model, temperature, max_tokens, examples_text, enable_dynamic_urls, enable_vector_rag, enable_web_search],
outputs=[preview_config_state, preview_status_comp, preview_chat_section_comp, config_display_comp]
)
if __name__ == "__main__":
# Configure launch parameters to avoid common development issues
launch_kwargs = {
"server_name": "127.0.0.1", # Use localhost instead of 0.0.0.0
"server_port": 7860,
"share": False, # Disable sharing to avoid origin issues
"debug": False, # Disable debug mode to reduce console errors
"show_error": True, # Show errors in interface
"quiet": False, # Keep logging for debugging
"favicon_path": None, # Disable favicon to avoid 404s
"ssl_verify": False, # Disable SSL verification for local development
"allowed_paths": [], # Empty allowed paths
"blocked_paths": [], # Empty blocked paths
"root_path": None, # No root path
"app_kwargs": {
"docs_url": None, # Disable docs endpoint
"redoc_url": None, # Disable redoc endpoint
}
}
# Override settings for specific environments
if os.environ.get('CODESPACES'):
launch_kwargs.update({
"server_name": "0.0.0.0",
"share": True
})
elif 'devtunnels.ms' in os.environ.get('GRADIO_SERVER_NAME', ''):
launch_kwargs.update({
"server_name": "0.0.0.0",
"share": True
})
print("πŸš€ Starting Chat UI Helper...")
print(f"πŸ“ Server: {launch_kwargs['server_name']}:{launch_kwargs['server_port']}")
print(f"πŸ”— Share: {launch_kwargs['share']}")
demo.launch(**launch_kwargs)