Innovative AI-Driven Real-Time Alert System with Streamlit
Written on
Chapter 1: Introduction to Real-Time Alerting
Imagine having an AI-driven personal alert assistant that utilizes real-time data to keep you informed. This application can notify you about crucial events like significant stock market fluctuations, changes in shared SharePoint documents, or awaited discounts on Amazon. It is designed to alert you about major updates based on your predefined criteria expressed in natural language.
In this article, we will explore how to develop a comprehensive event-driven weather alert chat application using Python and advanced tools like Streamlit, NATS, and OpenAI. The app gathers real-time weather data, interprets your alert preferences through AI, and delivers timely notifications via the user interface.
This guide, complete with code samples, is invaluable for tech enthusiasts and developers interested in understanding modern real-time alerting systems with Large Language Models (LLMs) and implementing one themselves. You can also access the source code on our GitHub repository to experiment with it firsthand.
Chapter 2: How the Alert Application Functions
Let's delve into the mechanics of how the AI weather alert chat application operates, transforming raw data into actionable insights to keep you ahead of changing weather conditions. At its core, the application features a responsive backend built in Python, leveraging NATS for real-time data processing and message management. By integrating OpenAI's GPT model, the application achieves a conversational AI capable of understanding alert specifications and responding to user inquiries.
Real-Time Weather Data Collection
The process begins with continuous asynchronous weather data collection from various sources. Our application utilizes the api.weatherapi.com service, retrieving real-time weather information every 10 seconds. This data encompasses temperature, humidity, precipitation, and more, covering locations worldwide. Here’s a code snippet that asynchronously fetches the current weather data for Estonia, with the potential for dynamic location input from the user:
async def fetch_weather_data():
api_url = f"http://api.weatherapi.com/v1/current.json?key={weather_api_key}&q=estonia"
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url) as response:
if response.status == 200:
return await response.json()else:
logging.error(f"Error fetching weather data: HTTP {response.status}")
return None
except Exception as e:
logging.error(f"Error fetching weather data: {e}")
return None
The Role of NATS in Data Streaming
In the backend.py file, the integration of NATS for event-driven messaging, continuous weather monitoring, and alerting is demonstrated. The nats.py library allows us to incorporate NATS within the Python code. A connection to the NATS server running in Docker is established as follows:
nats_client = await nats.connect("nats://localhost:4222")
We then define an asynchronous message_handler function to subscribe to and process messages received on the chat subject from the NATS server. If a message begins with "Set Alert:", it updates the user's alert criteria accordingly.
async def message_handler(msg):
nonlocal user_alert_criteria
data = msg.data.decode()
if data.startswith("Set Alert:"):
user_alert_criteria = data[len("Set Alert:"):].strip()
logging.info(f"User alert criteria updated: {user_alert_criteria}")
await nats_client.subscribe("chat", cb=message_handler)
The backend service integrates with both external services like the Weather API and the OpenAI Chat Completion API. If both weather data and user alert criteria are present, the application formulates a prompt for OpenAI's GPT model to determine whether the weather meets the user's specifications. The model analyzes the current weather against the user's criteria, responding with "YES" or "NO" along with a brief weather summary. When the AI confirms that the incoming weather data aligns with the user's conditions, it generates a tailored alert message and publishes it to the chat_response subject on the NATS server, updating the frontend app with the latest information. For example, the notification might read, "Heads up! Rain is expected in Estonia tomorrow. Don't forget your umbrella!"
while True:
current_weather = await fetch_weather_data()
if current_weather and user_alert_criteria:
logging.info(f"Current weather data: {current_weather}")
prompt = f"Use the current weather: {current_weather} information and user alert criteria: {user_alert_criteria}. Identify if the weather meets these criteria and return only YES or NO with a short weather temperature info without explaining why."
response_text = await get_openai_response(prompt)
if response_text and "YES" in response_text:
logging.info("Weather conditions met user criteria.")
ai_response = f"Weather alert! Your specified conditions have been met. {response_text}"
await nats_client.publish("chat_response", payload=ai_response.encode())
else:
logging.info("Weather conditions did not meet user criteria.")else:
logging.info("No current weather data or user alert criteria set.")await asyncio.sleep(10)
Delivering and Receiving Alerts in Real-Time
To illustrate the overall communication flow between the backend and frontend, users input their weather alert preferences through a chat interface built with Streamlit.
alert_criteria = st.text_input("Set your weather alert criteria", key="alert_criteria", disabled=st.session_state['alert_set'])
The Streamlit frontend interacts with the backend service via NATS messaging, publishing these criteria to the NATS server on the chat subject.
def send_message_to_nats_handler(message):
with NATSClient() as client:
client.connect()
client.publish("chat", payload=message.encode())
client.subscribe("chat_response", callback=read_message_from_nats_handler)
client.wait()
if set_alert_btn:
st.session_state['alert_set'] = True
st.success('Alert criteria set')
send_message_to_nats_handler(f"Set Alert: {alert_criteria}")
As previously mentioned, the backend listens to the chat subject, receives criteria, fetches current weather data, and uses AI to assess whether an alert should be issued. If conditions are satisfied, the backend sends an alert message to the chat_response subject. The frontend then receives this message and updates the user interface accordingly.
def read_message_from_nats_handler(msg):
message = msg.payload.decode()
st.session_state['conversation'].append(("AI", message))
st.markdown(f"🔔 AI: {message}", unsafe_allow_html=True)
Try It Out
To explore the real-time weather alert chat application in detail and try it for yourself, please visit our GitHub repository. The repository contains all the necessary code, setup instructions, and additional documentation to help you get started. Once you’ve completed the setup, you can launch the Streamlit frontend alongside the Python backend. Set your weather alert criteria and see how the system processes real-time weather data to keep you informed.
Chapter 3: Building Stream Processing Pipelines
The real-time weather alert chat application exemplifies a powerful use case of NATS for real-time messaging in a distributed system, facilitating effective communication between a user-facing frontend and a data-processing backend. However, several key considerations must be addressed to ensure that the information presented to users is relevant, accurate, and actionable. Currently, the app fetches live raw weather data and directly sends it to OpenAI or the frontend. It may be necessary to transform this data to filter, enrich, aggregate, or normalize it in real-time before it reaches external services, prompting the creation of a stream processing pipeline with multiple stages.
For instance, not all data retrieved from the API will be pertinent to every user, making it essential to filter out irrelevant information at the outset. Additionally, data may arrive in various formats, especially when sourced from multiple APIs for comprehensive alerting, necessitating normalization. The next stage involves enriching the data with additional context or information to enhance its utility. This could include comparing current weather conditions against historical data to identify unusual patterns or adding location-specific insights using another external API, such as providing specific advice for weather conditions in a particular area. Later stages might involve aggregating hourly temperature data to calculate an average daytime temperature or highlight the peak temperature reached throughout the day.
Next Steps
When it comes to data transformation, deployment, operation, and scaling in a production environment, you might consider using dedicated frameworks in Python like GlassFlow to construct sophisticated stream-processing pipelines. GlassFlow offers a fully managed serverless infrastructure for stream processing, eliminating the need for setup or maintenance, while efficiently handling large volumes of data and user requests. It also provides advanced state management capabilities, making it easier to track user alert criteria and other application states. Your application can scale in tandem with its user base without sacrificing performance.
Recommended Content
- Microservices Data Synchronization Using PostgreSQL, Debezium, and NATS
- Training Fraud Detection ML Models with Real-time Data Streams
About the Author
Visit my blog: www.iambobur.com
Chapter 4: Understanding the Competition
In the realm of AI and data-driven applications, it's essential to consider the competitive landscape that Streamlit operates within. Innovations in real-time data processing and AI integration are rapidly evolving, leading to new possibilities and challenges.
In this video, we discuss the competitive landscape for Streamlit, focusing on its unique features and potential challenges it faces against other platforms.
Chapter 5: Exploring UI Innovations
Streamlit continuously evolves, integrating cutting-edge technologies like Gemini and GPT-4 Turbo to enhance its user interface and capabilities.
This video explores Streamlit's UI innovations in conjunction with Gemini, Mixtral 8x7b, and GPT-4 Turbo, showcasing how these advancements can enhance user experiences.