From 2481923e3d0f7e2d4157e42f966a2425f0dbbc42 Mon Sep 17 00:00:00 2001 From: Greg Holmes Date: Thu, 3 Sep 2026 12:09:37 +0100 Subject: [PATCH 1/2] docs(examples): add Voice Agent session recorder --- examples/31-voice-agent-session-recording.py | 146 +++++++++++++++++++ examples/README.md | 1 + 2 files changed, 147 insertions(+) create mode 100644 examples/31-voice-agent-session-recording.py diff --git a/examples/31-voice-agent-session-recording.py b/examples/31-voice-agent-session-recording.py new file mode 100644 index 00000000..08893d21 --- /dev/null +++ b/examples/31-voice-agent-session-recording.py @@ -0,0 +1,146 @@ +""" +Example: Record a Voice Agent session as JSON + +This stores received transcripts, function-call events, and latency reports in +voice-agent-session.json after the session ends. Those fields can contain +sensitive data. Apply your application's consent, redaction, retention, and +storage policies before using this pattern in production. + +The recorder does not capture audio or manufacture fields that the server did +not send, such as audio references or unavailable latency measurements. +""" + +import json +import threading +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv + +from deepgram import DeepgramClient +from deepgram.agent.v1.types import ( + AgentV1InjectUserMessage, + AgentV1Settings, + AgentV1SettingsAgent, + AgentV1SettingsAgentListen, + AgentV1SettingsAgentListenProvider_V1, + AgentV1SettingsAudio, + AgentV1SettingsAudioInput, +) +from deepgram.core.events import EventType +from deepgram.types.speak_settings_v1 import SpeakSettingsV1 +from deepgram.types.speak_settings_v1provider import SpeakSettingsV1Provider_Deepgram +from deepgram.types.think_settings_v1 import ThinkSettingsV1 +from deepgram.types.think_settings_v1provider import ThinkSettingsV1Provider_OpenAi + +load_dotenv() + +OUTPUT_PATH = Path("voice-agent-session.json") +USER_MESSAGE = "What does the first all-female spacewalk symbolize?" + + +def _serialize_event(message: Any) -> dict[str, Any]: + if hasattr(message, "model_dump"): + return message.model_dump(mode="json") + return json.loads(message.json()) + + +@dataclass +class SessionRecorder: + started_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + ended_at: str | None = None + conversation: list[dict[str, Any]] = field(default_factory=list) + function_calls: list[dict[str, Any]] = field(default_factory=list) + latency_reports: list[dict[str, Any]] = field(default_factory=list) + _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + def record(self, message: Any) -> None: + if isinstance(message, bytes): + return + + event = _serialize_event(message) + with self._lock: + if event.get("type") == "ConversationText": + self.conversation.append(event) + elif event.get("type") in {"FunctionCallRequest", "FunctionCallResponse"}: + self.function_calls.append(event) + elif event.get("type") == "LatencyReport": + self.latency_reports.append(event) + + def finish(self) -> None: + with self._lock: + self.ended_at = datetime.now(timezone.utc).isoformat() + + def to_dict(self) -> dict[str, Any]: + with self._lock: + return { + "started_at": self.started_at, + "ended_at": self.ended_at, + "conversation": list(self.conversation), + "function_calls": list(self.function_calls), + "latency_reports": list(self.latency_reports), + } + + +def main() -> None: + recorder = SessionRecorder() + settings_applied_event = threading.Event() + assistant_response_event = threading.Event() + + try: + with DeepgramClient().agent.v1.connect() as agent: + settings = AgentV1Settings( + audio=AgentV1SettingsAudio(input=AgentV1SettingsAudioInput(encoding="linear16", sample_rate=24000)), + agent=AgentV1SettingsAgent( + listen=AgentV1SettingsAgentListen( + provider=AgentV1SettingsAgentListenProvider_V1(type="deepgram", model="nova-3") + ), + think=ThinkSettingsV1( + provider=ThinkSettingsV1Provider_OpenAi(type="open_ai", model="gpt-4o-mini", temperature=0.7), + prompt="You are a helpful AI assistant. Keep your responses brief.", + ), + speak=SpeakSettingsV1( + provider=SpeakSettingsV1Provider_Deepgram(type="deepgram", model="aura-2-asteria-en") + ), + ), + ) + + def on_message(message: Any) -> None: + recorder.record(message) + if isinstance(message, bytes): + return + + if getattr(message, "type", None) == "SettingsApplied": + settings_applied_event.set() + elif getattr(message, "type", None) == "ConversationText": + role = getattr(message, "role", "unknown") + print(f"[{role}] {message.content}") + if role == "assistant": + assistant_response_event.set() + + agent.on(EventType.MESSAGE, on_message) + agent.on(EventType.ERROR, lambda error: print(f"Error: {error}")) + + listener = threading.Thread(target=agent.start_listening, daemon=True) + listener.start() + + agent.send_settings(settings) + if not settings_applied_event.wait(10): + raise TimeoutError("Timed out waiting for agent settings to apply") + + agent.send_inject_user_message(AgentV1InjectUserMessage(content=USER_MESSAGE)) + if not assistant_response_event.wait(30): + raise TimeoutError("Timed out waiting for the agent to respond") + + time.sleep(2) + finally: + recorder.finish() + OUTPUT_PATH.write_text(json.dumps(recorder.to_dict(), indent=2), encoding="utf-8") + print(f"Wrote session record to {OUTPUT_PATH}") + + +if __name__ == "__main__": + main() diff --git a/examples/README.md b/examples/README.md index 943600e8..83fac2e4 100644 --- a/examples/README.md +++ b/examples/README.md @@ -32,6 +32,7 @@ This directory contains comprehensive examples demonstrating how to use the Deep ### 30-39: Voice Agent - **30-voice-agent.py** - Voice Agent configuration and usage +- **31-voice-agent-session-recording.py** - Record selected Voice Agent events as JSON ### 40-49: Text Intelligence (Read) From 518c93524fc2dca19f1b3834356695eac3b84ad2 Mon Sep 17 00:00:00 2001 From: Greg Holmes Date: Thu, 3 Sep 2026 12:27:43 +0100 Subject: [PATCH 2/2] fix(examples): export completed agent sessions safely --- examples/31-voice-agent-session-recording.py | 125 +++++++++++-------- 1 file changed, 72 insertions(+), 53 deletions(-) diff --git a/examples/31-voice-agent-session-recording.py b/examples/31-voice-agent-session-recording.py index 08893d21..296b3143 100644 --- a/examples/31-voice-agent-session-recording.py +++ b/examples/31-voice-agent-session-recording.py @@ -2,15 +2,19 @@ Example: Record a Voice Agent session as JSON This stores received transcripts, function-call events, and latency reports in -voice-agent-session.json after the session ends. Those fields can contain -sensitive data. Apply your application's consent, redaction, retention, and -storage policies before using this pattern in production. +JSON after a successful session. Pass `--output` to choose a persistent +destination; otherwise the example writes a permission-restricted temporary +file. Those fields can contain sensitive data. Apply your application's +consent, redaction, retention, and storage policies before using this pattern +in production. The recorder does not capture audio or manufacture fields that the server did not send, such as audio references or unavailable latency measurements. """ +import argparse import json +import tempfile import threading import time from dataclasses import dataclass, field @@ -38,7 +42,6 @@ load_dotenv() -OUTPUT_PATH = Path("voice-agent-session.json") USER_MESSAGE = "What does the first all-female spacewalk symbolize?" @@ -48,6 +51,19 @@ def _serialize_event(message: Any) -> dict[str, Any]: return json.loads(message.json()) +def _write_session_record(recorder: "SessionRecorder", output_path: Path | None) -> Path: + record = json.dumps(recorder.to_dict(), indent=2) + if output_path is not None: + output_path.write_text(record, encoding="utf-8") + return output_path + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", prefix="voice-agent-session-", suffix=".json", delete=False + ) as output_file: + output_file.write(record) + return Path(output_file.name) + + @dataclass class SessionRecorder: started_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) @@ -86,60 +102,63 @@ def to_dict(self) -> dict[str, Any]: def main() -> None: + parser = argparse.ArgumentParser(description="Record selected Voice Agent events as JSON.") + parser.add_argument("--output", type=Path, help="Persistent path for the completed session record.") + args = parser.parse_args() + recorder = SessionRecorder() settings_applied_event = threading.Event() assistant_response_event = threading.Event() - try: - with DeepgramClient().agent.v1.connect() as agent: - settings = AgentV1Settings( - audio=AgentV1SettingsAudio(input=AgentV1SettingsAudioInput(encoding="linear16", sample_rate=24000)), - agent=AgentV1SettingsAgent( - listen=AgentV1SettingsAgentListen( - provider=AgentV1SettingsAgentListenProvider_V1(type="deepgram", model="nova-3") - ), - think=ThinkSettingsV1( - provider=ThinkSettingsV1Provider_OpenAi(type="open_ai", model="gpt-4o-mini", temperature=0.7), - prompt="You are a helpful AI assistant. Keep your responses brief.", - ), - speak=SpeakSettingsV1( - provider=SpeakSettingsV1Provider_Deepgram(type="deepgram", model="aura-2-asteria-en") - ), + with DeepgramClient().agent.v1.connect() as agent: + settings = AgentV1Settings( + audio=AgentV1SettingsAudio(input=AgentV1SettingsAudioInput(encoding="linear16", sample_rate=24000)), + agent=AgentV1SettingsAgent( + listen=AgentV1SettingsAgentListen( + provider=AgentV1SettingsAgentListenProvider_V1(type="deepgram", model="nova-3") ), - ) - - def on_message(message: Any) -> None: - recorder.record(message) - if isinstance(message, bytes): - return - - if getattr(message, "type", None) == "SettingsApplied": - settings_applied_event.set() - elif getattr(message, "type", None) == "ConversationText": - role = getattr(message, "role", "unknown") - print(f"[{role}] {message.content}") - if role == "assistant": - assistant_response_event.set() - - agent.on(EventType.MESSAGE, on_message) - agent.on(EventType.ERROR, lambda error: print(f"Error: {error}")) - - listener = threading.Thread(target=agent.start_listening, daemon=True) - listener.start() - - agent.send_settings(settings) - if not settings_applied_event.wait(10): - raise TimeoutError("Timed out waiting for agent settings to apply") - - agent.send_inject_user_message(AgentV1InjectUserMessage(content=USER_MESSAGE)) - if not assistant_response_event.wait(30): - raise TimeoutError("Timed out waiting for the agent to respond") - - time.sleep(2) - finally: - recorder.finish() - OUTPUT_PATH.write_text(json.dumps(recorder.to_dict(), indent=2), encoding="utf-8") - print(f"Wrote session record to {OUTPUT_PATH}") + think=ThinkSettingsV1( + provider=ThinkSettingsV1Provider_OpenAi(type="open_ai", model="gpt-4o-mini", temperature=0.7), + prompt="You are a helpful AI assistant. Keep your responses brief.", + ), + speak=SpeakSettingsV1( + provider=SpeakSettingsV1Provider_Deepgram(type="deepgram", model="aura-2-asteria-en") + ), + ), + ) + + def on_message(message: Any) -> None: + recorder.record(message) + if isinstance(message, bytes): + return + + if getattr(message, "type", None) == "SettingsApplied": + settings_applied_event.set() + elif getattr(message, "type", None) == "ConversationText": + role = getattr(message, "role", "unknown") + print(f"[{role}] {message.content}") + if role == "assistant": + assistant_response_event.set() + + agent.on(EventType.MESSAGE, on_message) + agent.on(EventType.ERROR, lambda error: print(f"Error: {error}")) + + listener = threading.Thread(target=agent.start_listening, daemon=True) + listener.start() + + agent.send_settings(settings) + if not settings_applied_event.wait(10): + raise TimeoutError("Timed out waiting for agent settings to apply") + + agent.send_inject_user_message(AgentV1InjectUserMessage(content=USER_MESSAGE)) + if not assistant_response_event.wait(30): + raise TimeoutError("Timed out waiting for the agent to respond") + + time.sleep(2) + + recorder.finish() + output_path = _write_session_record(recorder, args.output) + print(f"Wrote completed session record to {output_path}") if __name__ == "__main__":