Intelligent Network for Temporal & Embodied Neuro-symbolic Tasks (INTENT)¶
Overview¶
INTENT is a pattern library built on top of the Spaxiom DSL that transforms raw sensor streams into semantically rich, agent-ready events. While Spaxiom provides the foundational primitives for spatial reasoning, temporal logic, and sensor fusion, INTENT captures recurring behavioral patterns as reusable abstractions.
Think of it this way: - Spaxiom = the axioms of the physical world (sensors, zones, conditions, temporal logic) - INTENT = named, domain-specific behaviors you can plug directly into agents and control systems
INTENT sits between low-level DSL primitives and high-level agent reasoning. Rather than forcing every developer to re-implement "queue detection" or "crowd monitoring" from scratch, INTENT encapsulates these patterns once with proper sensor fusion, temporal filtering, and domain expertise built in.
Why INTENT?¶
INTENT patterns serve three critical purposes:
1. Semantic Compression (100-1000× token reduction)¶
Instead of feeding hundreds of raw sensor readings per second to an LLM, INTENT produces single, meaningful events:
- Raw: [pressure_mat[423] = 0.82, pressure_mat[424] = 0.91, ...] (hundreds of values)
- INTENT: QueueFormed(zone="checkout", length=7, wait_time_est=180s) (one event)
This achieves 64-3286× compression vs. raw telemetry, enabling real-time agentic decision-making without token budget exhaustion.
2. Domain Expertise Encapsulation¶
INTENT patterns embed field-specific knowledge:
- QueueFlow uses queueing theory (Little's Law) to estimate wait times from arrival/service rates
- ADLTracker applies gerontology best practices for fall risk and ADL monitoring
- SafetyMonitor implements ISO safety standards for human-robot collaboration zones
3. Agent-Ready Output¶
INTENT events are designed for LLM consumption:
{
"type": "CrowdingDetected",
"zone": "lobby-west",
"occupancy_pct": 87,
"duration_s": 245,
"recommendation": "Consider opening overflow seating area"
}
Agents can reason about these events without parsing low-level sensor data or performing complex spatial computations.
Installation¶
pip install "spaxiom-dsl[intent]"
Core Pattern Library¶
INTENT ships with 20+ production-tested patterns across multiple domains. Below are four foundational examples:
OccupancyField¶
Tracks spatial occupancy and crowding over floor grids using pressure mats, depth cameras, or thermal sensors.
Capabilities: - Hotspot detection (top-k busiest tiles) - Density heatmaps (% occupied over time) - Flow analysis (entries/exits per zone)
Typical applications: Lobby management, retail heat-mapping, emergency evacuation planning, stadium crowd control
Example:
from spaxiom.intent import OccupancyField
from spaxiom import Zone
lobby = Zone(0, 0, 30, 20) # 30m × 20m
occupancy = OccupancyField(
zone=lobby,
sensors=pressure_mat_grid,
grid_resolution=0.1 # 10cm pixels
)
@on(always())
def monitor_crowding():
pct = occupancy.percent()
if pct > 0.75:
hotspots = occupancy.hotspots(k=3)
agent.notify(f"Crowding detected: {pct:.0%} occupied. "
f"Hotspots: {hotspots}")
QueueFlow¶
Estimates queue characteristics and wait times from entry/exit sensors.
Capabilities: - Queue length estimation - Arrival/service rate calculation (Little's Law) - Wait time prediction - Abandonment detection (people leaving queue)
Typical applications: Checkout optimization, airport security, restaurant drive-thru, hospital triage
Example:
from spaxiom.intent import QueueFlow
queue = QueueFlow(
entry_sensor=line_entry_floor,
service_sensor=checkout_counter_floor,
exit_sensor=exit_floor
)
@on(within(60, always())) # Every 60 seconds
def check_queue():
stats = queue.snapshot()
if stats.length > 10 or stats.wait_time_est > 300:
agent.alert(f"Long queue: {stats.length} people, "
f"~{stats.wait_time_est}s wait")
ADLTracker¶
Monitors Activities of Daily Living (ADL) for elder care and health monitoring.
Capabilities: - Bed occupancy tracking (sleep patterns, bed exit events) - Meal activity detection (kitchen time, eating duration) - Bathing/hygiene monitoring (bathroom usage patterns) - Movement pattern analysis (gait stability, fall risk)
Typical applications: Assisted living facilities, in-home care monitoring, post-op recovery tracking
Example:
from spaxiom.intent import ADLTracker
adl = ADLTracker(
bed_sensor=bedroom_pressure_mat,
kitchen_sensor=kitchen_floor_grid,
bathroom_sensor=bathroom_floor_grid,
mobility_sensors=hallway_depth_cameras
)
@on(adl.bed_exit_after(hours=8))
def morning_routine():
agent.log("Resident up for the day")
@on(adl.no_kitchen_activity(hours=18))
def missed_meals():
agent.alert("No meal activity detected in 18 hours - check on resident")
@on(adl.fall_risk_pattern())
def gait_instability():
agent.notify("Unstable gait pattern detected - increased fall risk")
FmSteward¶
Facility management monitoring across maintenance needs, air quality, and environmental conditions.
Capabilities: - Door cycle counting (maintenance scheduling) - Waste bin level tracking - Air quality monitoring (CO₂, VOC, particulates) - Floor condition detection (spills, wear patterns)
Typical applications: Commercial buildings, hospitals, schools, retail stores
Example:
from spaxiom.intent import FmSteward
fm = FmSteward(
door_sensors=entrance_contacts,
waste_sensors=bin_ultrasonic_levels,
air_quality=iaq_probes,
floor_sensors=floor_grid
)
@on(fm.door_cycles_exceeded(threshold=10000))
def schedule_door_maintenance(door_id):
agent.create_ticket(f"Door {door_id} approaching maintenance cycle limit")
@on(fm.waste_bin_full(threshold=0.85))
def dispatch_cleaning(bin_id):
agent.notify(f"Waste bin {bin_id} at 85% - schedule pickup")
@on(fm.air_quality_poor(co2_ppm=1200))
def ventilation_alert(zone):
agent.adjust_hvac(zone, mode="high_ventilation")
Extended Pattern Catalog¶
Beyond the four foundational patterns, INTENT includes domain-specific patterns for:
Safety & Robotics:
- SafetyMonitor — ISO 13849 compliant human-robot collision avoidance
- CollisionEnvelope — Dynamic safety zones with formal verification
Smart Buildings:
- ConferenceRoomUtilization — Meeting room booking efficiency
- SmartBuildingAgent — Multi-pattern HVAC orchestration
- EnergyOptimizer — Demand-response and decarbonization
Healthcare:
- ContaminationMonitor — Cleanroom particle/pressure tracking
- ORSterilityMonitor — Operating room sterility assurance
- PatientFlowOptimizer — Emergency department throughput
Retail & Hospitality:
- CategoryAggregator — Retail/expo engagement tracking
- QsrFlowOptimizer — Quick-service restaurant throughput
- ShelfMonitor — Out-of-stock detection
Manufacturing & Industrial:
- MachineryHealthMonitor — Vibration/acoustic anomaly detection
- ProductionLineOptimizer — Throughput and bottleneck analysis
Logistics & Supply Chain:
- TrafficFlowMonitor — Logistics corridor throughput
- ColdChainMonitor — Temperature/humidity tracking for pharmaceuticals
Agriculture & Environment:
- IrrigationOptimizer — Crop water stress and soil moisture
- WildfireRiskMonitor — Forest health and fire danger assessment
Public Infrastructure:
- MicroMobilityMonitor — E-scooter/bike safety and traffic flow
- CrowdSafetyMonitor — Stadium/venue crowd dynamics
For detailed examples of each pattern, see the Use Case Atlas (Appendix sections A.1-A.12 in the paper).
Complete Example: Lobby Crowding Agent¶
Here's a full example demonstrating how INTENT integrates with Spaxiom and LLM-based agents:
from spaxiom import Zone, Sensor, on, within, always
from spaxiom.intent import OccupancyField
import openai
# Define the lobby zone
lobby = Zone(0, 0, 30, 20, name="lobby-west")
# Wrap pressure mat sensors in OccupancyField
occupancy = OccupancyField(
zone=lobby,
sensors=pressure_mat_grid,
grid_resolution=0.1 # 10cm resolution
)
# Detect sustained crowding (>75% for 60+ seconds)
crowded = occupancy.percent() > 0.75
sustained_crowding = within(60, crowded)
@on(sustained_crowding)
def handle_crowding():
# Get current occupancy stats
pct = occupancy.percent()
hotspots = occupancy.hotspots(k=3)
# Build context for LLM
context = f"""
ALERT: Lobby crowding detected
- Occupancy: {pct:.0%}
- Duration: 60+ seconds
- Hotspots: {', '.join(f'{h.zone} ({h.density:.0%})' for h in hotspots)}
Available actions:
1. Open overflow seating area (East wing)
2. Dispatch staff to manage flow
3. Activate digital signage for queue guidance
4. Adjust HVAC for increased occupancy
Recommend immediate action.
"""
# Get LLM recommendation
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": context}],
max_tokens=150
)
action = response.choices[0].message.content
print(f"Agent recommendation: {action}")
# Execute recommended action
execute_action(action)
Key benefits demonstrated:
- Token efficiency: Occupancy compressed from ~6000 pressure readings → 1 OccupancyField event
- Semantic clarity: Agent receives "crowding detected" vs. raw sensor arrays
- Domain expertise: Automatic hotspot detection, temporal filtering (60s threshold)
- Agent-ready: LLM receives structured context and can reason about physical space
When to Use INTENT vs. Plain Spaxiom¶
Use INTENT when: - ✅ Your scenario matches a known pattern (queues, occupancy, ADL, facilities) - ✅ You want battle-tested domain expertise (queueing theory, safety standards) - ✅ You're building agent workflows and need token-efficient events - ✅ You need fast time-to-deployment with minimal sensor fusion expertise
Use plain Spaxiom when: - ✅ Your use case is novel or highly specialized - ✅ You need fine-grained control over sensor fusion logic - ✅ You're researching new spatial-temporal patterns - ✅ You want to build your own INTENT-like patterns
Extend INTENT when:
- ✅ You have domain expertise to contribute (e.g., you're an HVAC engineer building ChillerOptimizer)
- ✅ You've built a pattern that could benefit others in your industry
Token Compression Benchmarks¶
INTENT achieves dramatic compression vs. alternatives:
| Scenario | Raw Telemetry | Spaxiom DSL | INTENT | Compression Ratio |
|---|---|---|---|---|
| 1-hour warehouse occupancy | 2.16M tokens | 33,750 tokens | 658 tokens | 3,286× |
| Queue monitoring (retail) | 480K tokens | 12,000 tokens | 234 tokens | 2,051× |
| ADL tracking (elder care) | 720K tokens | 8,100 tokens | 112 tokens | 6,429× |
Comparison with other frameworks: - vs. Kafka raw streams: 1,200-3,000× compression - vs. Apache Flink aggregations: 240-800× compression - vs. OpenTelemetry metrics: 180-600× compression
Design Principles¶
INTENT patterns follow consistent design principles:
1. Type Safety¶
All INTENT events have explicit schemas with units:
@dataclass
class QueueFormedEvent:
zone: str
timestamp: datetime
length: int # people
wait_time_est: Quantity[Time] # seconds
arrival_rate: float # people/min
service_rate: float # people/min
2. Temporal Semantics¶
INTENT patterns handle time windows, buffering, and filtering:
# Only emit if queue persists for 30+ seconds
sustained_queue = within(30, queue.length > 5)
3. Spatial Grounding¶
All events carry spatial context (zones, coordinates):
{
"type": "CrowdingDetected",
"zone": "lobby-west", # Zone identifier
"hotspot_coords": [(12.3, 5.7), (18.1, 9.2)], # Physical coordinates
"affected_area_m2": 45.6
}
4. Privacy-by-Design¶
INTENT patterns minimize PII collection:
- OccupancyField: Counts people, doesn't identify them
- QueueFlow: Measures wait times, doesn't track individuals
- ADLTracker: Detects bed exits, doesn't record video
Aggregate statistics, not individual tracking.
5. Composability¶
INTENT patterns can be combined:
# Multi-pattern smart building
building = SmartBuildingAgent(
occupancy=OccupancyField(...),
air_quality=FmSteward(...),
energy=EnergyOptimizer(...)
)
@on(building.optimize_needed())
def optimize():
# Coordinate HVAC, lighting, and staffing
building.run_optimization_cycle()
Formal Verification Support¶
Critical INTENT patterns (e.g., SafetyMonitor) compile to timed automata for formal verification:
from spaxiom.intent import SafetyMonitor
safety = SafetyMonitor(
red_zone=danger_area,
yellow_zone=caution_area,
robot_sensor=robot_position_tracker
)
# Verify safety property: "If human in red zone, robot velocity must be 0"
# Compiled to UPPAAL timed automaton and model-checked
assert safety.verify_property(
"A[] (human_in_red -> robot.velocity == 0)"
)
Contributing New Patterns¶
Have a domain-specific pattern to contribute? We welcome PRs!
Guidelines:
1. Follow the INTENT pattern structure (see spaxiom/intent/base.py)
2. Include type annotations and unit tests
3. Provide example usage and documentation
4. Demonstrate token compression vs. alternatives
5. Consider privacy implications (minimize PII)
Example pattern structure:
from spaxiom.intent import IntentPattern
from dataclasses import dataclass
@dataclass
class YourPatternEvent:
zone: str
timestamp: datetime
# ... your fields with types and units
class YourPattern(IntentPattern):
def __init__(self, sensors, ...):
super().__init__(name="YourPattern")
# ... setup
def snapshot(self, window_s: float) -> YourPatternEvent:
# Compute event from sensor fusion
...
def percent(self) -> float:
# Domain-specific metrics
...
Resources¶
- Paper: Spaxiom: A Language for Sensor Fusion and Spatiotemporal Reasoning
- GitHub: spaxiom-dsl
- Examples: Use Case Atlas (12 detailed implementations)
- API Docs: Full API Reference
License¶
Apache 2.0