""" Viz LLM - Gradio App A RAG-powered assistant for data visualization guidance, powered by Jina-CLIP-v2 embeddings and research from the field of information graphics. Now with Datawrapper integration for chart generation! """ import os import io import asyncio import pandas as pd import gradio as gr from dotenv import load_dotenv from src.rag_pipeline import create_pipeline from src.datawrapper_client import create_and_publish_chart, get_iframe_html from datetime import datetime, timedelta from collections import defaultdict from src.vanna import VannaComponent # Load environment variables load_dotenv() # Rate limiting: Track requests per user (IP-based) # Format: {ip: [(timestamp1, timestamp2, ...)]} rate_limit_tracker = defaultdict(list) DAILY_LIMIT = 20 # Initialize the RAG pipeline print("Initializing Graphics Design Pipeline...") try: pipeline = create_pipeline( retrieval_k=5, model=os.getenv("LLM_MODEL", "meta-llama/Llama-3.1-8B-Instruct"), temperature=float(os.getenv("LLM_TEMPERATURE", "0.2")) ) print("✓ Pipeline initialized successfully") except Exception as e: print(f"✗ Error initializing pipeline: {e}") raise # Initialize Vanna print("Initializing Vanna...") try: vanna = VannaComponent( hf_model="Qwen/Qwen3-VL-30B-A3B-Instruct", hf_token=os.getenv("HF_TOKEN_VANNA"), hf_provider="novita", connection_string=os.getenv("SUPABASE_CONNECTION") ) print("✓ Vanna initialized successfully") except Exception as e: print(f"✗ Error initializing Vanna: {e}") raise def check_rate_limit(request: gr.Request) -> tuple[bool, int]: """Check if user has exceeded rate limit""" if request is None: return True, DAILY_LIMIT # Allow if no request object user_id = request.client.host now = datetime.now() cutoff = now - timedelta(days=1) # Remove old requests (older than 24 hours) rate_limit_tracker[user_id] = [ ts for ts in rate_limit_tracker[user_id] if ts > cutoff ] remaining = DAILY_LIMIT - len(rate_limit_tracker[user_id]) if remaining <= 0: return False, 0 # Add current request rate_limit_tracker[user_id].append(now) return True, remaining - 1 def recommend_stream(message: str, history: list, request: gr.Request): """ Streaming version of design recommendation function Args: message: User's design query history: Chat history request: Gradio request object for rate limiting Yields: Response chunks """ # Check rate limit allowed, remaining = check_rate_limit(request) if not allowed: yield "⚠️ **Rate limit exceeded.** You've reached the maximum of 20 queries per day. Please try again in 24 hours." return try: response_stream = pipeline.generate_recommendations(message, stream=True) full_response = "" for chunk in response_stream: full_response += chunk yield full_response # Add rate limit info at the end if remaining <= 5: yield full_response + f"\n\n---\n*You have {remaining} queries remaining today.*" except Exception as e: yield f"Error generating response: {str(e)}\n\nPlease check your environment variables (HF_TOKEN, SUPABASE_URL, SUPABASE_KEY) and try again." def generate_chart_from_csv(csv_file, user_prompt): """ Generate a Datawrapper chart from uploaded CSV and user prompt. Args: csv_file: Uploaded CSV file user_prompt: User's description of the chart Returns: HTML string with iframe or error message """ if not csv_file: return "
Please upload a CSV file to generate a chart.
" if not user_prompt or user_prompt.strip() == "": return "
Please describe what chart you want to create.
" try: # Show loading message loading_html = """

🎨 Creating your chart...

Analyzing your data and selecting the best visualization...

""" # Read CSV file df = pd.read_csv(csv_file) # Create and publish chart (async function, need to run in event loop) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) result = loop.run_until_complete( create_and_publish_chart(df, user_prompt, pipeline) ) loop.close() if result.get("success"): # Get the iframe HTML iframe_html = get_iframe_html(result.get('public_url'), height=500) # Create HTML with iframe, reasoning, and edit button chart_html = f"""
{iframe_html}
Why this chart?

{result['reasoning']}

✏️ Open in Datawrapper
""" return chart_html else: error_msg = result.get("error", "Unknown error") return f"""

❌ Chart Generation Failed

{error_msg}

Please check your CSV format and try again.

""" except Exception as e: return f"""

❌ Error

{str(e)}

Please ensure your CSV is properly formatted and try again.

""" def csv_to_cards_html(csv_text: str) -> str: """ Transforme le CSV brut retourné par Vanna en cartes HTML. """ try: df = pd.read_csv(io.StringIO(csv_text.strip())) if df.empty: return "
Aucune donnée trouvée.
" cards_html = "" for _, row in df.iterrows(): title = row.get("title", "Sans titre") source_url = row.get("source_url", "#") author = row.get("author", "Inconnu") published_date = row.get("published_date", "") if not published_date == "nan": published_date = "" image_url = row.get("image_url", "") if not image_url == "nan": image_url = "https://fpoimg.com/800x600?text=Image+not+found" cards_html += f"""
{title}

{title}

{author}

{published_date}

🔗 Voir la source
""" html = f"""
{cards_html}
""" return html except Exception as e: return f"
Erreur lors du parsing du CSV : {e}
" async def search_inspiration_from_database(user_prompt): """ Search inspiration posts from user prompt in database. Args: user_prompt: User's description of the inspiration query Returns: HTML string displaying cards or an error message """ if not user_prompt or user_prompt.strip() == "": return """
Please describe what kind of inspiration you want to search for.
""" try: response = await vanna.ask(user_prompt) print("response :", repr(response)) clean_response = response.strip() if clean_response.startswith("⚠️") or "Aucun CSV détecté" in clean_response: return f"""

❌ No valid data found

The AI couldn't generate any data for this request. Try being more specific — for example: "Show me spotlights from 2020 about design".

""" csv_text = ( clean_response .strip("```") .replace("csv", "") .replace("CSV", "") ) if "," not in csv_text: return f"""

❌ No valid CSV detected

The model didn't return any structured data. Try rephrasing your query to be more precise.

""" cards_html = csv_to_cards_html(csv_text) return cards_html except Exception as e: return f"""

❌ Error

{str(e)}

Please try again.

""" # Minimal CSS to fix UI artifacts and style the mode selector custom_css = """ /* Hide retry/undo buttons that appear as artifacts */ .chatbot button[aria-label="Retry"], .chatbot button[aria-label="Undo"] { display: none !important; } /* Remove overflow-y scroll from textarea */ textarea[data-testid="textbox"] { overflow-y: hidden !important; } /* Mode selector buttons */ .mode-button { font-size: 1.1em; padding: 12px 24px; margin: 5px; } """ # Create Gradio interface with dual-mode layout with gr.Blocks( title="Viz LLM", css=custom_css ) as demo: gr.Markdown(""" # 📊 Viz LLM Get design recommendations or generate charts with AI-powered data visualization assistance. """) # Mode selector buttons with gr.Row(): ideation_btn = gr.Button("💡 Ideation Mode", variant="primary", elem_classes="mode-button") chart_gen_btn = gr.Button("📊 Chart Generation Mode", variant="secondary", elem_classes="mode-button") inspiration_btn = gr.Button("✨ Inspiration Mode", variant="secondary", elem_classes="mode-button") # Ideation Mode: Chat interface (shown by default, wrapped in Column) with gr.Column(visible=True) as ideation_container: ideation_interface = gr.ChatInterface( fn=recommend_stream, type="messages", examples=[ "What's the best chart type for showing trends over time?", "How do I create an effective infographic for complex data?", "What are best practices for data visualization accessibility?", "How should I design a dashboard for storytelling?", "What visualization works best for comparing categories?" ], cache_examples=False, api_name="recommend" ) # Chart Generation Mode: Chart controls and output (hidden by default) with gr.Column(visible=False) as chart_gen_container: csv_upload = gr.File( label="📁 Upload CSV File", file_types=[".csv"], type="filepath" ) chart_prompt_input = gr.Textbox( label="Describe your chart", placeholder="E.g., 'Show sales trends over time' or 'Compare revenue by category'", lines=2 ) generate_chart_btn = gr.Button("Generate Chart", variant="primary", size="lg") chart_output = gr.HTML( value="
Upload a CSV file and describe your visualization above, then click Generate Chart.
", label="Generated Chart" ) # Inspiration Mode: with gr.Column(visible=False) as inspiration_container: with gr.Row(): inspiration_prompt_input = gr.Textbox( placeholder="Ask for an inspiration...", show_label=False, scale=4, container=False ) inspiration_search_btn = gr.Button("🔍 Search", variant="primary", scale=1) inspiration_cards_html = gr.HTML("") # Mode switching functions def switch_to_ideation(): return [ gr.update(variant="primary"), # ideation_btn gr.update(variant="secondary"), # chart_gen_btn gr.update(variant="secondary"), # inspiration_btn gr.update(visible=True), # ideation_container gr.update(visible=False), # chart_gen_container gr.update(visible=False), # inspiration_container ] def switch_to_chart_gen(): return [ gr.update(variant="secondary"), # ideation_btn gr.update(variant="primary"), # chart_gen_btn gr.update(variant="secondary"), # inspiration_btn gr.update(visible=False), # ideation_container gr.update(visible=True), # chart_gen_container gr.update(visible=False), # inspiration_container ] def switch_to_inspiration(): return [ gr.update(variant="secondary"), # ideation_btn gr.update(variant="secondary"), # chart_gen_btn gr.update(variant="primary"), # inspiration_btn gr.update(visible=False), # ideation_container gr.update(visible=False), # chart_gen_container gr.update(visible=True), # inspiration_container ] # Wire up mode switching ideation_btn.click( fn=switch_to_ideation, inputs=[], outputs=[ideation_btn, chart_gen_btn, inspiration_btn, ideation_container, chart_gen_container, inspiration_container] ) chart_gen_btn.click( fn=switch_to_chart_gen, inputs=[], outputs=[ideation_btn, chart_gen_btn, inspiration_btn, ideation_container, chart_gen_container, inspiration_container] ) inspiration_btn.click( fn=switch_to_inspiration, inputs=[], outputs=[ideation_btn, chart_gen_btn, inspiration_btn, ideation_container, chart_gen_container, inspiration_container] ) # Generate chart when button is clicked generate_chart_btn.click( fn=generate_chart_from_csv, inputs=[csv_upload, chart_prompt_input], outputs=[chart_output] ) # Search inspiration when button is clicked inspiration_search_btn.click( fn=search_inspiration_from_database, inputs=[inspiration_prompt_input], outputs=[inspiration_cards_html] ) # Knowledge base section (below both interfaces) gr.Markdown(""" ### About Viz LLM **Ideation Mode:** Get design recommendations based on research papers, design principles, and examples from the field of information graphics and data visualization. **Chart Generation Mode:** Upload your CSV data and describe your visualization goal. The AI will analyze your data, select the optimal chart type, and generate a publication-ready chart using Datawrapper. **Inspiration Mode:** Coming soon. **Credits:** Special thanks to the researchers whose work informed this model: Robert Kosara, Edward Segel, Jeffrey Heer, Matthew Conlen, John Maeda, Kennedy Elliott, Scott McCloud, and many others. --- **Usage Limits:** This service is limited to 20 queries per day per user to manage costs. Responses are optimized for English.
Embeddings: Jina-CLIP-v2 | Charts: Datawrapper API
""") # Launch configuration if __name__ == "__main__": # Check for required environment variables required_vars = ["SUPABASE_URL", "SUPABASE_KEY", "HF_TOKEN", "DATAWRAPPER_ACCESS_TOKEN"] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: print(f"⚠️ Warning: Missing environment variables: {', '.join(missing_vars)}") print("Please set these in your .env file or as environment variables") if "DATAWRAPPER_ACCESS_TOKEN" in missing_vars: print("Note: DATAWRAPPER_ACCESS_TOKEN is required for chart generation mode") # Launch the app demo.launch( server_name="0.0.0.0", server_port=7860, share=False, show_api=True )