Every monsoon season in Nepal brings with it a familiar dread: the news of another landslide. These unpredictable events don't just block roads; they sever critical trade arteries, halting the flow of goods, stranding travelers, and causing significant economic disruption. For logistics operators, businesses, and even government agencies, manually sifting through a deluge of real-time news reports to identify specific, actionable impacts is a slow, reactive, and often overwhelming task. What if we could automate this, turning raw, unstructured news into precise, structured event data that allows for proactive decision-making? In this post, I'll walk you through building a resilient system using Generative AI and careful post-processing to do exactly that, transforming the chaos of real-time news into a clear manifest of logistical challenges.
Key Takeaways
- Structured prompting with Pydantic schemas is crucial for reliable, type-safe LLM output, reducing post-processing complexity.
- Robust error handling, including API call retries and parsing fallbacks, is essential for production-grade LLM pipelines dealing with external APIs and potentially malformed responses.
- Few-shot examples in prompts significantly improve LLM accuracy and adherence to specific output formats, even for complex extraction tasks.
- Translating raw extracted events into domain-specific actionable insights requires a secondary processing layer that maps generic locations to known logistical assets.
- The value of Generative AI lies not just in its ability to generate text, but in its capacity to transform unstructured data into structured, decision-ready formats.
The Problem: Unstructured Chaos, Structured Needs
Imagine you're managing a fleet of trucks transporting essential goods across Nepal. A major highway, like the Prithvi Highway connecting Kathmandu to Pokhara, is frequently impacted by landslides. News breaks, but it's often vague: "Landslide in Dhading" or "Road blocked near Mugling." This unstructured information is hard to act on. You need to know: *when* did it happen, *exactly where*, *which specific highway segments* are affected, *how severe* is it, and *what's the source*? Manually tracking this for dozens of potential events across the country is impossible in real-time. My goal was to build a system that could ingest these messy news snippets and spit out a clean, machine-readable list of events, ready for logistical planning.
Data and Sources
For real-time news intelligence, I'll use the NewsAPI.org. It provides a straightforward API to query news articles from various sources. For the Generative AI component, I'll rely on OpenAI's Chat Completions API, specifically `gpt-3.5-turbo` or `gpt-4` for its instruction following capabilities. Our geographical data about Nepali highways will be a simplified, hardcoded dictionary for demonstration, but in a production setting, this would ideally link to a GIS database or a more comprehensive mapping service.
Data accessed on 2024-07-28.
Step 1 — Gathering Real-time Unstructured Intelligence
The first hurdle is reliably fetching relevant news articles about potential disruptions in Nepal. This sub-problem demands a way to programmatically query news sources for keywords indicative of natural disasters and road closures. I opted for NewsAPI.org because of its broad coverage and easy-to-use API. The key is to craft search queries that are specific enough to capture relevant events without being overwhelmed by unrelated news.
import requests
import os
import json
from datetime import datetime, timedelta
from typing import List, Optional
def gather_unstructured_intelligence(api_key: str, query: str = "Nepal landslide OR \"road closure Nepal\" OR \"highway blocked Nepal\"", days_back: int = 1) -> List[dict]:
"""
Fetches relevant news articles from NewsAPI.org.
Solves the sub-problem of reliably getting real-time news about disruptions.
"""
base_url = "https://newsapi.org/v2/everything"
from_date = (datetime.now() - timedelta(days=days_back)).isoformat(timespec='minutes')
params = {
'q': query,
'language': 'en',
'sortBy': 'publishedAt',
'from': from_date,
'apiKey': api_key,
'pageSize': 100 # Max articles per request
}
articles = []
try:
response = requests.get(base_url, params=params, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
data = response.json()
if data['status'] == 'ok':
articles.extend(data['articles'])
# In a real scenario, handle pagination if totalResults > pageSize
# For this demo, we'll assume relevant articles fit within one page.
else:
print(f"NewsAPI error: {data.get('message', 'Unknown error')}")
except requests.exceptions.Timeout:
print("NewsAPI request timed out.")
except requests.exceptions.RequestException as e:
print(f"Error fetching news from NewsAPI: {e}")
return articles
This function uses the `requests` library to query NewsAPI.org. I've included error handling for network issues and API-specific errors, which are common when dealing with external services. Notice the `days_back` parameter, allowing us to specify how far back we want to look, crucial for real-time monitoring where fresh data is paramount.
Step 2 — Defining Structured Output with Pydantic
The core challenge with LLMs is getting them to reliably output data in a consistent, machine-readable format. This is where Pydantic shines. It solves the sub-problem of clearly specifying the desired data structure for LLM output, ensuring type safety and validation. By defining a schema, we give the LLM a clear target and provide our application with a robust way to validate the LLM's response.
from pydantic import BaseModel, Field, HttpUrl
class LandslideEvent(BaseModel):
"""
Represents a structured event of a landslide or road disruption in Nepal.
"""
event_date: str = Field(..., description="Date of the event in YYYY-MM-DD format.")
location_description: str = Field(..., description="A detailed description of the location, e.g., 'near Mugling, Dhading district'.")
affected_highways: List[str] = Field(..., description="List of major highways or roads affected, e.g., ['Prithvi Highway'].")
impact_severity: str = Field(..., description="Severity of impact: 'Minor', 'Moderate', 'Significant', 'Severe'.")
source_url: HttpUrl = Field(..., description="URL of the news article reporting the event.")
relevance_score: float = Field(..., description="A score from 0.0 to 1.0 indicating how relevant this article is to a *landslide/road closure event* in Nepal (1.0 is highly relevant).")
class Config:
json_schema_extra = {
"example": {
"event_date": "2024-07-27",
