Temporal Event Detection & Alerting

from mixpeek import Mixpeek

client = Mixpeek(api_key="your-api-key")

# Create alert condition retriever
alert_retriever = client.retrievers.create(
    retriever_name="safety_monitor",
    stages=[
        {
            "stage_id": "feature_search",
            "config": {
                "query_text": "violence OR weapons OR threatening behavior"
            }
        },
        {
            "stage_id": "score_filter",
            "config": {
                "min_score": 0.85
            }
        }
    ],
    webhook_url="https://api.company.com/alerts"
)

# Monitor live stream
stream_results = client.retrievers.execute(
    retriever_id=alert_retriever.id,
    inputs={
        "collection_id": "live_streams",
        "time_window": "last_5_minutes"
    }
)

# Trigger alert if matches found
if len(stream_results['documents']) > 0:
    client.webhooks.trigger(
        webhook_url=alert_retriever.webhook_url,
        payload=stream_results
    )