can I subscribe `input_audio_buffer.speech_stopped...
# ask-ai
n
can I subscribe
input_audio_buffer.speech_stopped
event of OpenAI Realtime model using session: AgentSession? I am using python sdk
t
Thanks for reaching out, Darshak. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — you’re asking whether you can subscribe to an
input_audio_buffer.speech_stopped
event from the OpenAI Realtime model using
AgentSession
in the LiveKit Python SDK. Short answer: No — you cannot subscribe to an
input_audio_buffer.speech_stopped
event on
AgentSession
.
AgentSession
does not expose an
input_audio_buffer
with that event name. What to use instead: • Listen to the
user_state_changed
event on
AgentSession
. The OpenAI Realtime model performs server-side VAD/turn detection and the session surface exposes those changes via
user_state_changed
. You can treat the transition to the
"listening"
state as equivalent to “speech stopped” / end of the user’s turn. Example (Python SDK):
Copy code
@session.on("user_state_changed")
def on_user_state_changed(ev):
    # ev.new_state will be a string like "listening", "speaking", etc.
    if ev.new_state == "listening":
        print("User has stopped speaking (equivalent to speech_stopped)")
If you’d like, I can: • Show a fuller example with session setup and event wiring. • Help you detect other turn states (e.g., start of speech) or debounce short pauses. • Verify exact state strings provided by your agent model if you paste a sample event payload. Sources: OpenAI Realtime API plugin guide | LiveKit Documentation | Agent session | LiveKit Documentation
n
facing this issue Object of type RealtimeModelBeta is not JSON serializable
Copy code
def to_dict(session_report: SessionReport) -> dict:
        events_dict: list[dict] = []

        for event in session_report.events:
            if event.type == "metrics_collected":
                metrics_event: MetricsCollectedEvent = event
                if metrics_event.metrics.type == "vad_metrics":
                    continue  # vad_metrics are too noisy currently skip it if needed will add it
            events_dict.append(event.model_dump())

        return {
            "job_id": session_report.job_id,
            "room_id": session_report.room_id,
            "room": session_report.room,
            "events": events_dict,
            "audio_recording_path": (
                str(session_report.audio_recording_path.absolute()) if session_report.audio_recording_path else None
            ),
            "audio_recording_started_at": session_report.audio_recording_started_at,
            "options": {
                "allow_interruptions": session_report.options.allow_interruptions,
                "discard_audio_if_uninterruptible": session_report.options.discard_audio_if_uninterruptible,
                "min_interruption_duration": session_report.options.min_interruption_duration,
                "min_interruption_words": session_report.options.min_interruption_words,
                "min_endpointing_delay": session_report.options.min_endpointing_delay,
                "max_endpointing_delay": session_report.options.max_endpointing_delay,
                "max_tool_steps": session_report.options.max_tool_steps,
                "user_away_timeout": session_report.options.user_away_timeout,
                "min_consecutive_speech_delay": session_report.options.min_consecutive_speech_delay,
                "preemptive_generation": session_report.options.preemptive_generation,
            },
            "chat_history": session_report.chat_history.to_dict(exclude_timestamp=False),
            "timestamp": session_report.timestamp,
        }

    webhook_url = os.getenv("OBSERVABILITY_URL", None)
    if not webhook_url:
        return

    try:
        report: SessionReport = ctx.make_session_report()
        report_dict = to_dict(report)

        buffer = io.BytesIO(
            json.dumps(report_dict, indent=2).encode("utf-8")
        )

        filename = f"session_report_{ctx.room.name}.json"

        data = aiohttp.FormData()
        data.add_field(
            name="file",
            value=buffer,
            filename=filename,
            content_type="application/json",
        )
        data.add_field(
            name="caseId",
            value=case_id,
        )

        async with aiohttp.ClientSession() as session:
            t0 = time.time()
            <http://logging.info|logging.info>(f"Sending session report: {case_id}")
            async with <http://session.post|session.post>(webhook_url, data=data) as resp:
t
Thanks — you’re hitting the error because a
RealtimeModelBeta
instance is present somewhere in the data you’re passing to
json.dumps
. Python’s JSON encoder can’t serialize complex objects like live model instances (they contain methods, sockets, clients, etc.), so you need to remove or replace them with plain serializable data before dumping. What I’m responding to: your traceback "Object of type RealtimeModelBeta is not JSON serializable" when building and sending
report_dict
as JSON. Quick options (pick one): • Remove the model instance from the report entirely and store only serializable metadata (recommended). • Pre-walk and sanitize the dict to replace non-serializable objects with safe representations. • Provide a custom JSON encoder that converts
RealtimeModelBeta
to a serializable dict. Short examples you can apply 1) Replace the model object with metadata when building the report • At the code point where the report is created, ensure you do not include the live model instance. Replace it with a small dict containing only strings/ints you need (e.g., model name, voice, provider):
Copy code
# Example: ensure session_report.llm (or wherever model lives) is not a live object
llm_info = {
    "provider": "openai_realtime",
    "model": getattr(session.llm, "model", None) or str(type(session.llm)),
    "voice": getattr(session.llm, "voice", None),
}
# store llm_info in SessionReport instead of the full object
2) Sanitize the dict before json.dumps • Walk the dict recursively and replace any RealtimeModelBeta instances:
Copy code
def sanitize_for_json(obj):
    if isinstance(obj, dict):
        return {k: sanitize_for_json(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [sanitize_for_json(x) for x in obj]
    # import the class or check by name if not directly available
    if type(obj).__name__ == "RealtimeModelBeta":
        return {
            "type": "RealtimeModelBeta",
            "model": getattr(obj, "model", None),
            "voice": getattr(obj, "voice", None),
        }
    try:
        json.dumps(obj)
        return obj
    except (TypeError, ValueError):
        return repr(obj)
Then use:
Copy code
buffer = io.BytesIO(json.dumps(sanitize_for_json(report_dict), indent=2).encode("utf-8"))
3) Custom JSONEncoder • If you prefer an encoder, implement one that knows how to handle RealtimeModelBeta:
Copy code
import json

class SafeJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if type(obj).__name__ == "RealtimeModelBeta":
            return {
                "type": "RealtimeModelBeta",
                "model": getattr(obj, "model", None),
                "voice": getattr(obj, "voice", None),
            }
        return super().default(obj)

json.dumps(report_dict, indent=2, cls=SafeJSONEncoder)
How to choose: • If you control SessionReport shape: remove model instances at creation time and store only metadata — simplest and safest. • If you can’t change SessionReport: sanitize the dict or use the encoder before sending. Next steps I can help with • If you paste the
SessionReport
class fields or a sample of
report_dict
(or the traceback showing which field contains RealtimeModelBeta), I’ll point exactly where to change it. • I can provide a ready-to-run sanitization function tailored to your structure. • I can show a full example replacing the model in your current flow (where you call
ctx.make_session_report()
). Which would you like me to do next? Sources: OpenAI Realtime API plugin guide | LiveKit Documentation | Gemini Live API plugin | LiveKit Documentation | Using the Testing Framework with Realtime Models | LiveKit