sentinel

Real-time OODA Loop Agent using Gemini Live API.

This module implements the SentinelAgent that maintains a bidirectional WebSocket session for real-time multimodal interaction (voice + vision).

The agent follows the OODA loop pattern: - Observe: Receive sensor data via WebSocket - Orient: Process context and understand situation - Decide: Generate actions via function calling - Act: Execute local functions and return results

Example

>>> from unsprawl.sentinel import SentinelAgent
>>>
>>> agent = SentinelAgent()
>>> await agent.run_session("Monitor traffic in sector A-7")

Attributes

logger

Classes

SentinelAgent

Real-time multimodal OODA loop agent over WebSockets.

StateTracker

Flink-like State Machine for Event Pattern Detection.

Functions

_default_optimize_traffic(sector_id)

Placeholder traffic optimization function.

Module Contents

logger
async _default_optimize_traffic(sector_id)

Placeholder traffic optimization function.

In production, this would call the Warp-based DifferentiableTraffic optimizer.

class SentinelAgent(model='gemini-live-2.5-flash-preview', api_key=None, response_modalities=None)

Real-time multimodal OODA loop agent over WebSockets.

Maintains a persistent connection to Gemini Live API for bidirectional streaming of audio, video, and text.

Parameters:

  • model (str) – The Gemini Live model to use.

  • api_key (str | None) – Google API key. If None, reads from GOOGLE_API_KEY environment variable.

  • response_modalities (list[str]) – Output modalities. Options: [“TEXT”], [“AUDIO”], [“TEXT”, “AUDIO”].

Notes

  • Uses client.aio.live.connect() for WebSocket streaming

  • Supports tool calls for local function execution

  • OODA = Observe-Orient-Decide-Act (military decision loop)

model = 'gemini-live-2.5-flash-preview'
response_modalities = ['TEXT']
client
_functions: dict[str, Callable[Ellipsis, Any]]
register_function(name, func)

Register a function that can be called by the model.

Parameters:

  • name (str) – Function name as it will appear to the model.

  • func (Callable) – The function to execute when called.

_build_tools()

Build the tools configuration for the Live session.

async _handle_tool_call(session, tool_call)

Handle a tool call from the model and send the response back.

Parameters:

  • session (AsyncSession) – The active Live session.

  • tool_call (LiveServerToolCall) – The tool call request from the model.

async run_session(initial_prompt, on_message=None, max_turns=10)

Run a Live API session with the given initial prompt.

Parameters:

  • initial_prompt (str) – The initial message to send to establish context.

  • on_message (Callable[[str], None] | None) – Optional callback for each received text message.

  • max_turns (int) – Maximum number of conversation turns before closing.

Returns:

List of text responses received during the session.

Return type:

list[str]

async send_observation(session, observation)

Send an observation to an active session.

Parameters:

  • session (AsyncSession) – The active Live session.

  • observation (str) – The observation text to send.

async watch()

Listen to the Urban Telemetry Stream via Redpanda.

async intervene()

Trigger Gemini Intervention on Alert.

class StateTracker(window_seconds=10)

Flink-like State Machine for Event Pattern Detection.

window_seconds = 10
buffer
update(velocity)
check_condition()

Check if avg velocity < 5.0 for the window duration.