README / app.py
skenoo's picture
Upload 4 files
9feb1f1 verified
Raw History Blame Contribute Delete
13.1 kB
#!/usr/bin/env python
"""
SKENAI-RENO Event Processing Demo
---------------------------------
This Gradio app demonstrates how the SKENAI-RENO event processing pipeline
enriches events with AI insights using Hugging Face models.
"""
import gradio as gr
import requests
import json
import os
import time
import numpy as np
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta
# Configure API access (in production, use Secrets)
API_KEY = os.environ.get("HF_API_KEY", "")
HF_ENDPOINT = "https://api-inference.huggingface.co/models/"
# Sample models to use for different event types
MODEL_MAPPING = {
"user_feedback": "distilbert-base-uncased-finetuned-sst-2-english",
"transaction": "bert-base-uncased",
"system_event": "gpt2",
"achievement": "roberta-base"
}
def process_event(event_type, event_content, user_id="user123", game_id="game456"):
"""Process an event through the SKENAI-RENO pipeline with Hugging Face model."""
# Select model based on event type
model_id = MODEL_MAPPING.get(event_type, "distilbert-base-uncased-finetuned-sst-2-english")
# Prepare payload for Hugging Face model
payload = {
"inputs": event_content,
}
# For demo purposes, simulate API call if no key is provided
if not API_KEY:
# Simulate processing delay
time.sleep(1)
# Generate simulated AI insights based on event type
if event_type == "user_feedback":
ai_insights = [
{"label": "POSITIVE", "score": 0.92},
{"label": "NEGATIVE", "score": 0.08}
]
elif event_type == "transaction":
ai_insights = {
"transaction_type": "purchase",
"value_assessment": "high",
"confidence": 0.87
}
elif event_type == "achievement":
ai_insights = {
"achievement_level": "significant",
"percentile": 85,
"engagement_impact": "positive"
}
else:
ai_insights = {
"classification": "routine",
"priority": "low",
"action_required": False
}
else:
# Call Hugging Face Inference API with actual key
try:
headers = {"Authorization": f"Bearer {API_KEY}"}
response = requests.post(
f"{HF_ENDPOINT}{model_id}",
headers=headers,
json=payload,
timeout=10
)
ai_insights = response.json()
except Exception as e:
ai_insights = {"error": str(e)}
# Current timestamp for the demo
current_time = datetime.now().isoformat()
# Format response as it would appear in the SKENAI-RENO pipeline
enriched_event = {
"event_id": f"evt-{int(time.time())}-{hash(event_content) % 10000}",
"type": event_type,
"content": event_content,
"user_id": user_id,
"game_id": game_id,
"timestamp": current_time,
"ai_enrichment": {
"model_id": model_id,
"insights": ai_insights,
"processed_timestamp": current_time
},
"processing": {
"lambda_execution_time_ms": 120,
"sqs_wait_time_ms": 15,
"dynamodb_write_time_ms": 8
}
}
# Add relevant additional fields based on event type
if event_type == "transaction":
enriched_event["transaction_details"] = {
"amount": 4.99,
"currency": "USD",
"item_id": "coins_500",
"payment_method": "app_store"
}
elif event_type == "achievement":
enriched_event["achievement_details"] = {
"achievement_id": "level_completion",
"level": 10,
"score": 95,
"completion_time_sec": 320
}
return json.dumps(enriched_event, indent=2)
def generate_metrics_visualization():
"""Generate a sample visualization of event processing metrics."""
# Create sample data for the past 7 days
dates = [(datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d') for i in range(7)]
dates.reverse()
# Sample event counts by type
event_types = ["user_feedback", "transaction", "system_event", "achievement"]
data = []
for date in dates:
for event_type in event_types:
# Generate some random data with trends
if event_type == "transaction":
# Transactions increasing over time
count = np.random.randint(800, 1200) + (dates.index(date) * 50)
elif event_type == "user_feedback":
# Feedback relatively stable
count = np.random.randint(400, 600)
elif event_type == "achievement":
# Achievements with weekly pattern (higher on weekends)
day_of_week = datetime.strptime(date, '%Y-%m-%d').weekday()
weekend_boost = 200 if day_of_week >= 5 else 0
count = np.random.randint(300, 500) + weekend_boost
else:
# System events decreasing (good sign, fewer errors)
count = np.random.randint(100, 300) - (dates.index(date) * 10)
count = max(count, 50) # Ensure not negative
data.append({
"Date": date,
"Event Type": event_type,
"Count": count
})
# Create DataFrame
df = pd.DataFrame(data)
# Create plot
fig = px.line(
df,
x="Date",
y="Count",
color="Event Type",
title="SKENAI-RENO: Event Processing Volume (Last 7 Days)",
markers=True
)
fig.update_layout(
xaxis_title="Date",
yaxis_title="Number of Events",
legend_title="Event Type",
template="plotly_white"
)
return fig
def create_processing_pipeline_visualization():
"""Create a visual representation of the SKENAI-RENO event processing pipeline."""
# Sample data for pipeline stages and their performance metrics
stages = [
"Redpanda",
"Lambda Connector",
"SQS Queue",
"AI Processor Lambda",
"DynamoDB"
]
avg_latency = [5, 25, 15, 120, 8]
error_rate = [0.01, 0.05, 0.001, 0.2, 0.02]
# Create DataFrame
df = pd.DataFrame({
"Stage": stages,
"Average Latency (ms)": avg_latency,
"Error Rate (%)": [e * 100 for e in error_rate]
})
# Create bar chart for latency
fig_latency = px.bar(
df,
x="Stage",
y="Average Latency (ms)",
title="Pipeline Stage Latency",
color="Average Latency (ms)",
color_continuous_scale="Viridis"
)
# Create bar chart for error rates
fig_errors = px.bar(
df,
x="Stage",
y="Error Rate (%)",
title="Pipeline Stage Error Rates",
color="Error Rate (%)",
color_continuous_scale="Reds"
)
return fig_latency, fig_errors
# Define the Gradio interface
with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo")) as demo:
gr.Markdown("# SKENAI-RENO Event Processing Demo")
gr.Markdown("""
Experience the power of AI-enriched event processing with SKENAI-RENO's serverless pipeline.
This demo simulates how events from gaming and enterprise applications are processed
and enriched with AI insights in real-time.
""")
with gr.Tabs():
with gr.TabItem("Event Processing Simulator"):
with gr.Row():
with gr.Column(scale=2):
gr.Markdown("### Input Event Details")
with gr.Row():
event_type = gr.Dropdown(
choices=["user_feedback", "transaction", "system_event", "achievement"],
value="user_feedback",
label="Event Type"
)
with gr.Row():
user_id = gr.Textbox(value="user123", label="User ID")
game_id = gr.Textbox(value="game456", label="Game/App ID")
with gr.Row():
event_content = gr.Textbox(
lines=5,
placeholder="Enter event content here...",
label="Event Content"
)
with gr.Row():
submit_btn = gr.Button("Process Event", variant="primary")
with gr.Column(scale=3):
gr.Markdown("### Enriched Event Output")
output = gr.JSON()
with gr.Row():
gr.Markdown("### Example Events")
with gr.Row():
example1 = gr.Button("User Feedback Example")
example2 = gr.Button("Transaction Example")
example3 = gr.Button("Achievement Example")
# Set up example event handlers
example1.click(
lambda: ["user_feedback", "user123", "game456", "I love this game! The new level design is amazing."],
outputs=[event_type, user_id, game_id, event_content]
)
example2.click(
lambda: ["transaction", "user789", "game456", "User purchased 500 coins for $4.99"],
outputs=[event_type, user_id, game_id, event_content]
)
example3.click(
lambda: ["achievement", "user456", "game789", "User completed level 10 with score 95/100"],
outputs=[event_type, user_id, game_id, event_content]
)
with gr.TabItem("System Analytics"):
gr.Markdown("### SKENAI-RENO Event Processing Analytics")
gr.Markdown("""
This dashboard shows sample analytics of the SKENAI-RENO event processing pipeline.
In a production environment, these metrics would be generated from real-time CloudWatch data.
""")
with gr.Row():
metrics_plot = gr.Plot()
with gr.Row():
with gr.Column():
latency_plot = gr.Plot()
with gr.Column():
error_plot = gr.Plot()
refresh_btn = gr.Button("Refresh Analytics")
with gr.TabItem("About SKENAI-RENO"):
gr.Markdown("""
# About SKENAI-RENO
SKENAI-RENO is a fully serverless event processing pipeline designed for high-reliability,
AI-enhanced insights, and seamless integration with existing systems.
## Key Features
* **Serverless Architecture**: AWS Lambda, SQS, DynamoDB
* **AI Enrichment**: Hugging Face models integrated directly into the event pipeline
* **High Reliability**: Idempotent processing, dead-letter queues, comprehensive monitoring
* **Enterprise Ready**: Secure, scalable, and compliant
## Technical Architecture
SKENAI-RENO uses a multi-stage serverless architecture:
1. **Event Sources**: Capture events from various sources (gaming, web, mobile, IoT)
2. **Redpanda**: High-performance event streaming platform
3. **Lambda Connector**: Processes events from Redpanda and sends to SQS
4. **SQS Queue**: Decouples components and provides buffering
5. **AI Processor Lambda**: Enriches events with AI insights using Hugging Face models
6. **DynamoDB**: Stores processed events for analysis and retrieval
## Integration Options
* REST API for direct event submission
* SDK libraries for major platforms
* Webhook integrations
* Custom connectors for enterprise systems
## Learn More
For enterprise solutions and partnership opportunities, visit [skenai-reno.com](https://skenai-reno.com)
""")
# Set up event handlers
submit_btn.click(
process_event,
inputs=[event_type, event_content, user_id, game_id],
outputs=output
)
# Initialize and refresh the analytics
refresh_btn.click(
lambda: [generate_metrics_visualization(), *create_processing_pipeline_visualization()],
outputs=[metrics_plot, latency_plot, error_plot]
)
# Initialize the analytics on page load
demo.load(
lambda: [generate_metrics_visualization(), *create_processing_pipeline_visualization()],
outputs=[metrics_plot, latency_plot, error_plot]
)
# Launch the app
if __name__ == "__main__":
demo.launch()