VisualBot / app.py
ganeshkumar383's picture
Update app.py
d578a12 verified
"""
VISUAL CONVERSATIONAL INTELLIGENCE ENGINE
==========================================
A pluggable, image-grounded multi-turn conversational system.
Architecture:
- Session-based image memory (stored once, queried multiple times)
- Vision-Language Model (BLIP) for image-question answering
- REST-style core logic (pure functions)
- Gradio UI for demonstration
Academic Purpose:
Demonstrates AI system design for visual question answering with
conversational context, suitable for research evaluation.
"""
import gradio as gr
from PIL import Image
from transformers import BlipProcessor, BlipForQuestionAnswering
import torch
from typing import Optional, Tuple, List
import uuid
import re
# ============================================================================
# SESSION MEMORY MANAGEMENT
# ============================================================================
class SessionMemory:
"""
Manages session state for image-grounded conversations.
Each session stores:
- uploaded_image: PIL Image object
- conversation_history: List of (question, answer) tuples
- session_id: Unique identifier for the session
"""
def __init__(self):
self.sessions = {}
def create_session(self) -> str:
"""Create a new session and return its ID."""
session_id = str(uuid.uuid4())
self.sessions[session_id] = {
'uploaded_image': None,
'conversation_history': []
}
return session_id
def store_image(self, session_id: str, image: Image.Image) -> None:
"""Store an image in session memory."""
if session_id in self.sessions:
self.sessions[session_id]['uploaded_image'] = image
def get_image(self, session_id: str) -> Optional[Image.Image]:
"""Retrieve the stored image from session."""
if session_id in self.sessions:
return self.sessions[session_id]['uploaded_image']
return None
def add_to_history(self, session_id: str, question: str, answer: str) -> None:
"""Add a Q&A pair to conversation history."""
if session_id in self.sessions:
self.sessions[session_id]['conversation_history'].append((question, answer))
def get_history(self, session_id: str) -> List[Tuple[str, str]]:
"""Retrieve conversation history."""
if session_id in self.sessions:
return self.sessions[session_id]['conversation_history']
return []
def reset_session(self, session_id: str) -> None:
"""Clear all session data (image + conversation history)."""
if session_id in self.sessions:
self.sessions[session_id] = {
'uploaded_image': None,
'conversation_history': []
}
# ============================================================================
# VISION-LANGUAGE MODEL INITIALIZATION
# ============================================================================
class VisualQAEngine:
"""
Core inference engine using BLIP (Bootstrapping Language-Image Pre-training).
BLIP is a vision-language model that can answer questions about images.
We use the pretrained model without any fine-tuning.
"""
def __init__(self, model_name: str = "Salesforce/blip-vqa-base"):
"""
Initialize the BLIP model and processor.
Args:
model_name: HuggingFace model identifier
"""
print(f"Loading model: {model_name}")
self.processor = BlipProcessor.from_pretrained(model_name)
self.model = BlipForQuestionAnswering.from_pretrained(model_name)
# Use GPU if available, otherwise CPU (for HuggingFace Spaces compatibility)
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self.model.to(self.device)
print(f"Model loaded on device: {self.device}")
def answer_question(self, image: Image.Image, question: str) -> str:
"""
Generate an answer to a question about the image.
This is a PURE FUNCTION suitable for REST APIs:
- Takes image + question as input
- Returns answer as output
- No side effects
Args:
image: PIL Image object
question: User's question about the image
Returns:
Generated answer grounded in the image
"""
# Preprocess image and question
inputs = self.processor(image, question, return_tensors="pt").to(self.device)
# Generate answer using the vision-language model
with torch.no_grad():
outputs = self.model.generate(**inputs, max_length=50)
# Decode the generated answer
answer = self.processor.decode(outputs[0], skip_special_tokens=True)
return answer
# ============================================================================
# APPLICATION LOGIC (REST-STYLE PURE FUNCTIONS)
# ============================================================================
def validate_question(question: str, image: Optional[Image.Image]) -> Tuple[bool, str]:
"""
Validate that conditions are met for answering a question.
Validation rules:
1. Image must be uploaded
2. Question must not be empty
Args:
question: User's input question
image: Stored image (or None)
Returns:
(is_valid, error_message)
"""
if image is None:
return False, "⚠️ Please upload an image first before asking questions."
if not question or question.strip() == "":
return False, "⚠️ Please enter a question."
return True, ""
def process_question(
vqa_engine: VisualQAEngine,
session_memory: SessionMemory,
session_id: str,
question: str
) -> Tuple[str, List[Tuple[str, str]]]:
"""
Process a user question and generate an image-grounded answer.
This function orchestrates the core conversational flow:
1. Validate inputs
2. Retrieve image from session
3. Generate answer using vision-language model
4. Update conversation history
5. Return answer + updated history
Args:
vqa_engine: Visual QA inference engine
session_memory: Session storage
session_id: Current session identifier
question: User's question
Returns:
(answer, updated_conversation_history)
"""
# Retrieve stored image
image = session_memory.get_image(session_id)
# Validate inputs
is_valid, error_msg = validate_question(question, image)
if not is_valid:
return error_msg, session_memory.get_history(session_id)
# Generate image-grounded answer
answer = vqa_engine.answer_question(image, question)
# Update conversation history
session_memory.add_to_history(session_id, question, answer)
# Return answer and updated history
return answer, session_memory.get_history(session_id)
def handle_image_upload(
session_memory: SessionMemory,
session_id: str,
image: Image.Image
) -> str:
"""
Handle image upload and store in session memory.
Args:
session_memory: Session storage
session_id: Current session identifier
image: Uploaded PIL Image
Returns:
Confirmation message
"""
if image is None:
return "⚠️ No image uploaded."
# Store image in session
session_memory.store_image(session_id, image)
return "✅ Image uploaded successfully! You can now ask questions about this image."
def reset_conversation(
session_memory: SessionMemory,
session_id: str
) -> Tuple[str, List, None]:
"""
Reset the conversation (clear image and history).
Args:
session_memory: Session storage
session_id: Current session identifier
Returns:
(status_message, empty_history, None_for_image)
"""
session_memory.reset_session(session_id)
return "🔄 Conversation reset. Please upload a new image.", [], None
# ============================================================================
# GRADIO UI INTERFACE
# ============================================================================
def format_history_for_chatbot(history: List[Tuple[str, str]]) -> List[dict]:
"""
Convert internal (question, answer) tuples into
Gradio v4 Chatbot message format.
"""
messages = []
for q, a in history:
messages.append({"role": "user", "content": q})
messages.append({"role": "assistant", "content": a})
return messages
def generate_visual_topic_suggestions(
vqa_engine: VisualQAEngine,
image: Image.Image
) -> List[str]:
"""
Generate guided visual topic suggestions using the SAME BLIP VQA model.
IMPORTANT:
- This is GUIDANCE ONLY, not object detection
- Uses a small, fixed set of internal prompts
- Extracts 1-4 single-word nouns only
- Does NOT claim to list all objects
Args:
vqa_engine: Visual QA inference engine
image: Uploaded PIL Image
Returns:
List of 1-4 single-word topic suggestions
"""
if image is None:
return []
# Fixed set of internal prompts for guidance
internal_prompts = [
"What is the main object in the image?",
"Is there a furniture item?",
"Is there an electronic device?",
"Is there a plant?"
]
suggestions = []
for prompt in internal_prompts:
try:
answer = vqa_engine.answer_question(image, prompt)
# Extract single-word nouns only
words = re.findall(r'\b[a-zA-Z]+\b', answer.lower())
# Filter out common stop words and keep only meaningful nouns
stop_words = {'yes', 'no', 'the', 'a', 'an', 'is', 'are', 'there', 'not'}
meaningful_words = [w for w in words if w not in stop_words and len(w) > 2]
if meaningful_words:
suggestions.append(meaningful_words[0])
except:
continue
# Return unique suggestions, max 4
unique_suggestions = list(dict.fromkeys(suggestions))[:4]
return unique_suggestions
def clear_chat_only(
session_memory: SessionMemory,
session_id: str
) -> Tuple[str, List]:
"""
Clear conversation history only (keep image).
Args:
session_memory: Session storage
session_id: Current session identifier
Returns:
(status_message, empty_history)
"""
if session_id in session_memory.sessions:
session_memory.sessions[session_id]['conversation_history'] = []
return "💬 Chat cleared. Image retained.", []
def remove_image_only(
session_memory: SessionMemory,
session_id: str
) -> Tuple[str, None]:
"""
Remove image only (keep conversation history).
Args:
session_memory: Session storage
session_id: Current session identifier
Returns:
(status_message, None_for_image)
"""
if session_id in session_memory.sessions:
session_memory.sessions[session_id]['uploaded_image'] = None
return "🖼️ Image removed. Chat history retained.", None
def get_session_metadata(
session_memory: SessionMemory,
session_id: str
) -> str:
"""
Get session metadata for Advanced Mode display.
Args:
session_memory: Session storage
session_id: Current session identifier
Returns:
Formatted metadata string
"""
if session_id not in session_memory.sessions:
return "Session ID: Unknown\nImage Loaded: No\nConversation Turns: 0"
session = session_memory.sessions[session_id]
image_loaded = "Yes" if session['uploaded_image'] is not None else "No"
turn_count = len(session['conversation_history'])
return f"""**Session ID:** `{session_id[:8]}...`
**Image Loaded:** {image_loaded}
**Conversation Turns:** {turn_count}"""
def create_gradio_interface(vqa_engine: VisualQAEngine, session_memory: SessionMemory) -> gr.Blocks:
"""
Create the Gradio UI for the Visual Conversational Intelligence Engine.
UI Components:
- Mode selector (Basic / Advanced)
- Image upload with guided topic suggestions
- Question input with type selector (Advanced Mode)
- Chat history display
- Advanced controls and metadata (Advanced Mode only)
"""
# Custom CSS for visual polish and theming
custom_css = """
.mode-selector {font-size: 16px; font-weight: bold;}
.topic-chip {margin: 4px; padding: 8px 16px; border-radius: 16px; background: #e3f2fd; cursor: pointer;}
.capability-box {background: #f5f5f5; padding: 16px; border-radius: 8px; margin: 8px 0;}
.metadata-box {background: #fafafa; padding: 12px; border-radius: 6px; font-family: monospace;}
"""
with gr.Blocks(title="Visual Conversational Intelligence Engine", css=custom_css) as demo:
# Session state (hidden)
session_id = gr.State(value=session_memory.create_session())
# Mode state (Basic = default)
mode_state = gr.State(value="Basic")
# Header
gr.Markdown("""
# 🔍 Visual Conversational Intelligence Engine
**An image-grounded multi-turn conversational system for academic demonstration**
""")
# MODE SELECTOR (TOP OF UI)
with gr.Row():
mode_selector = gr.Radio(
choices=["Basic Mode", "Advanced Mode"],
value="Basic Mode",
label="Interface Mode",
info="Basic Mode: Student-friendly interface | Advanced Mode: Research/admin view with additional controls",
elem_classes="mode-selector"
)
# BASIC MODE INSTRUCTIONS (shown only in Basic Mode)
basic_instructions = gr.Markdown("""
### 🎓 How to use (Student View):
1. **Upload an image** 📤
2. **Ask questions** about the image 💬
3. **Continue the conversation** - ask follow-up questions without re-uploading
4. **Reset** to start over with a new image 🔄
**Note:** All answers are strictly grounded in the uploaded image.
""", visible=True)
# MAIN LAYOUT (TWO COLUMNS)
with gr.Row():
# LEFT COLUMN: IMAGE UPLOAD SECTION
with gr.Column(scale=1):
with gr.Group():
gr.Markdown("### 📤 Upload Image")
image_input = gr.Image(
type="pil",
label="Drag and drop or click to upload",
height=400
)
upload_status = gr.Textbox(
label="Status",
interactive=False,
lines=1
)
upload_btn = gr.Button("📥 Upload Image", variant="primary", size="lg")
# GUIDED VISUAL TOPIC SUGGESTIONS (shown after upload)
gr.Markdown("#### 💡 Suggested Visual Topics (Guidance Only)")
gr.Markdown("*Click a topic to prefill your question. These are suggestions, not exhaustive object lists.*")
with gr.Row():
topic_btn_1 = gr.Button("", visible=False, size="sm")
topic_btn_2 = gr.Button("", visible=False, size="sm")
topic_btn_3 = gr.Button("", visible=False, size="sm")
topic_btn_4 = gr.Button("", visible=False, size="sm")
# RIGHT COLUMN: CHAT / CONVERSATION SECTION
with gr.Column(scale=1):
with gr.Group():
gr.Markdown("### 💬 Ask Questions")
chatbot = gr.Chatbot(
label="Conversation History",
height=400
)
question_input = gr.Textbox(
label="Your Question",
placeholder="Ask a question about the uploaded image...",
lines=2
)
with gr.Row():
submit_btn = gr.Button("🚀 Ask Question", variant="primary", size="lg")
reset_btn_basic = gr.Button("🔄 Reset All", variant="secondary", size="lg")
# ADVANCED MODE PANEL (shown only in Advanced Mode)
with gr.Group(visible=False) as advanced_panel:
gr.Markdown("## 🔬 Advanced Controls & Metadata")
with gr.Row():
# QUESTION TYPE SELECTOR (GUIDANCE ONLY)
with gr.Column(scale=1):
gr.Markdown("### Question Type Selector (Guidance)")
question_type = gr.Dropdown(
choices=[
"Object Presence",
"Object Attribute (Color / Shape)",
"Scene Understanding",
"Yes / No Verification"
],
label="Select Question Type",
info="This is for guidance only. It does not change AI logic.",
value="Object Presence"
)
# SESSION METADATA PANEL (READ-ONLY)
with gr.Column(scale=1):
gr.Markdown("### Session Metadata")
session_metadata = gr.Markdown(
"**Session ID:** Not initialized\n**Image Loaded:** No\n**Conversation Turns:** 0"
)
refresh_metadata_btn = gr.Button("🔄 Refresh Metadata", size="sm")
# CAPABILITY / SCOPE INDICATOR (STATIC INFO BOX)
with gr.Row():
gr.Markdown("""
### ⚙️ System Capabilities & Limitations
**What this system CAN do:**
- ✅ Image-grounded Question Answering
- ✅ Single-image Conversational Memory
- ✅ Multi-turn dialogue about the same image
**What this system CANNOT do:**
- ❌ Exhaustive Object Listing (not object detection)
- ❌ Multi-image Reasoning
- ❌ Precise Counting (VQA models have known limitations)
- ❌ Open-domain knowledge questions unrelated to the image
*This is an academic prototype demonstrating AI system design, not a production object detection system.*
""")
# ADVANCED RESET CONTROLS
with gr.Row():
gr.Markdown("### Reset Controls")
with gr.Row():
clear_chat_btn = gr.Button("💬 Clear Chat Only", variant="secondary")
remove_image_btn = gr.Button("🖼️ Remove Image Only", variant="secondary")
full_reset_btn = gr.Button("🔄 Full Reset (Image + Chat)", variant="stop")
# Footer
gr.Markdown("""
---
**Academic Prototype** | Demonstrates AI system design for visual question answering
**Tech Stack:** Python • HuggingFace BLIP • Gradio • Session-based Memory
""")
# ====================================================================
# EVENT HANDLERS
# ====================================================================
def toggle_mode(mode_choice):
"""
Toggle between Basic and Advanced Mode.
Mode toggle does NOT reset session or image.
"""
is_advanced = (mode_choice == "Advanced Mode")
return {
advanced_panel: gr.update(visible=is_advanced),
basic_instructions: gr.update(visible=not is_advanced),
reset_btn_basic: gr.update(visible=not is_advanced)
}
def upload_image_handler(image, session_id):
"""
Handle image upload event.
Stores image and generates guided topic suggestions.
"""
status = handle_image_upload(session_memory, session_id, image)
# Generate guided topic suggestions
suggestions = generate_visual_topic_suggestions(vqa_engine, image)
# Update topic buttons
updates = []
for i in range(4):
if i < len(suggestions):
updates.append(gr.update(value=suggestions[i], visible=True))
else:
updates.append(gr.update(value="", visible=False))
return [status] + updates
def topic_click_handler(topic_text):
"""
Handle topic chip click.
Prefills question input with suggested topic.
User can edit before submitting.
"""
return f"What is the {topic_text} in the image?"
def ask_question_handler(question, session_id):
"""
Handle question submission.
Uses existing process_question logic (unchanged).
"""
answer, history = process_question(
vqa_engine, session_memory, session_id, question
)
formatted_history = format_history_for_chatbot(history)
return formatted_history, ""
def question_type_change_handler(question_type):
"""
Handle question type selector change.
Optionally prefills question input with example.
This is GUIDANCE ONLY.
"""
examples = {
"Object Presence": "Is there a [object] in the image?",
"Object Attribute (Color / Shape)": "What color is the [object]?",
"Scene Understanding": "What is happening in the image?",
"Yes / No Verification": "Is the [object] [attribute]?"
}
return examples.get(question_type, "")
def refresh_metadata_handler(session_id):
"""
Refresh session metadata display.
"""
return get_session_metadata(session_memory, session_id)
def clear_chat_handler(session_id):
"""
Clear chat only (Advanced Mode).
"""
status, history = clear_chat_only(session_memory, session_id)
return status, []
def remove_image_handler(session_id):
"""
Remove image only (Advanced Mode).
"""
status, image = remove_image_only(session_memory, session_id)
return status, image
def full_reset_handler(session_id):
"""
Full reset (Advanced Mode).
"""
status, history, image = reset_conversation(session_memory, session_id)
return status, [], image, "", "", "", ""
def basic_reset_handler(session_id):
"""
Basic mode reset.
"""
status, history, image = reset_conversation(session_memory, session_id)
return status, [], image
# ====================================================================
# WIRE UP EVENTS
# ====================================================================
# Mode toggle
mode_selector.change(
fn=toggle_mode,
inputs=[mode_selector],
outputs=[advanced_panel, basic_instructions, reset_btn_basic]
)
# Image upload
upload_btn.click(
fn=upload_image_handler,
inputs=[image_input, session_id],
outputs=[upload_status, topic_btn_1, topic_btn_2, topic_btn_3, topic_btn_4]
)
# Topic chip clicks
topic_btn_1.click(
fn=topic_click_handler,
inputs=[topic_btn_1],
outputs=[question_input]
)
topic_btn_2.click(
fn=topic_click_handler,
inputs=[topic_btn_2],
outputs=[question_input]
)
topic_btn_3.click(
fn=topic_click_handler,
inputs=[topic_btn_3],
outputs=[question_input]
)
topic_btn_4.click(
fn=topic_click_handler,
inputs=[topic_btn_4],
outputs=[question_input]
)
# Question submission
submit_btn.click(
fn=ask_question_handler,
inputs=[question_input, session_id],
outputs=[chatbot, question_input]
)
question_input.submit(
fn=ask_question_handler,
inputs=[question_input, session_id],
outputs=[chatbot, question_input]
)
# Question type selector (Advanced Mode)
question_type.change(
fn=question_type_change_handler,
inputs=[question_type],
outputs=[question_input]
)
# Metadata refresh (Advanced Mode)
refresh_metadata_btn.click(
fn=refresh_metadata_handler,
inputs=[session_id],
outputs=[session_metadata]
)
# Advanced reset controls
clear_chat_btn.click(
fn=clear_chat_handler,
inputs=[session_id],
outputs=[upload_status, chatbot]
)
remove_image_btn.click(
fn=remove_image_handler,
inputs=[session_id],
outputs=[upload_status, image_input]
)
full_reset_btn.click(
fn=full_reset_handler,
inputs=[session_id],
outputs=[upload_status, chatbot, image_input, topic_btn_1, topic_btn_2, topic_btn_3, topic_btn_4]
)
# Basic mode reset
reset_btn_basic.click(
fn=basic_reset_handler,
inputs=[session_id],
outputs=[upload_status, chatbot, image_input]
)
return demo
# ============================================================================
# MAIN APPLICATION ENTRY POINT
# ============================================================================
def main():
"""
Initialize and launch the Visual Conversational Intelligence Engine.
"""
print("=" * 60)
print("VISUAL CONVERSATIONAL INTELLIGENCE ENGINE")
print("=" * 60)
# Initialize core components
print("\n[1/3] Initializing Vision-Language Model...")
vqa_engine = VisualQAEngine(model_name="Salesforce/blip-vqa-base")
print("\n[2/3] Setting up session memory...")
session_memory = SessionMemory()
print("\n[3/3] Creating Gradio interface...")
demo = create_gradio_interface(vqa_engine, session_memory)
print("\n" + "=" * 60)
print("🚀 Launching application...")
print("=" * 60)
# Launch the application
demo.launch(
share=True, # Set to True for public sharing
server_name="0.0.0.0", # Allow external access
server_port=7860 # Standard Gradio port
)
if __name__ == "__main__":
main()