feat(python): voice-agents v1 — escalation-handoff

recipes/python/voice-agents/v1/escalation-handoff/README.md

@@ -0,0 +1,60 @@
+# Escalation Handoff (Voice Agents v1)
+
+Detect when a voice agent conversation needs human intervention and perform a warm handoff with a structured context packet.
+
+## What it does
+
+Configures a Deepgram Voice Agent with an `escalate_to_human` function that the LLM calls when it detects the conversation should be transferred to a human agent. Escalation triggers include: the user explicitly asking for a human, repeated inability to resolve an issue, or negative caller sentiment. When triggered, the agent outputs a structured JSON context packet containing the reason for escalation, a conversation summary, and sentiment — suitable for routing to a CRM or ticketing system. The agent then delivers a handoff message to the caller.
+
+## Key parameters
+
+| Parameter | Value | Description |
+|-----------|-------|-------------|
+| `think.functions` | `[escalate_to_human]` | Function definition the LLM invokes to escalate |
+| `FunctionCallRequest.input` | `dict` | Contains `reason`, `summary`, and `sentiment` from the LLM |
+| `send_function_call_response()` | method | Acknowledge the function call so the LLM continues |
+| `send_inject_agent_message()` | method | Deliver a handoff message to the caller |
+
+## Escalation triggers
+
+The LLM is prompted to call `escalate_to_human` when any of these occur:
+
+1. **Explicit request** — the caller says "let me speak to a human" or similar
+2. **Repeated failure** — the agent cannot resolve the issue after multiple attempts
+3. **Negative sentiment** — the caller expresses frustration or dissatisfaction
+
+## Example output
+
+```
+Agent configured with escalation handoff
+Connection opened
+Event: SettingsApplied
+ESCALATION CONTEXT PACKET:
+{
+  "escalation": true,
+  "call_id": "fc-abc123",
+  "reason": "User explicitly requested a human agent",
+  "summary": "Caller asked about billing discrepancy, AI could not access account details",
+  "sentiment": "negative"
+}
+Event: ConversationText
+Connection closed
+```
+
+## Prerequisites
+
+- Python 3.10+
+- Set `DEEPGRAM_API_KEY` environment variable
+- Install: `pip install -r recipes/python/requirements.txt`
+
+## Run
+
+```bash
+python example.py
+```
+
+## Test
+
+```bash
+pytest example_test.py -v
+```

recipes/python/voice-agents/v1/escalation-handoff/example.py

@@ -0,0 +1,66 @@
+"""Escalation Handoff — Voice Agent detects when a caller needs a human
+and outputs a structured context packet via function calling."""
+
+import json
+
+from deepgram import DeepgramClient
+from deepgram.agent.v1.types import (
+    AgentV1FunctionCallRequest, AgentV1InjectAgentMessage,
+    AgentV1SendFunctionCallResponse, 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
+
+ESCALATE_FN = {
+    "name": "escalate_to_human",
+    "description": "Call when the user asks for a human, is repeatedly frustrated, or the issue is unresolvable.",
+    "parameters": {"type": "object", "properties": {
+        "reason": {"type": "string", "description": "Why escalation is needed"},
+        "summary": {"type": "string", "description": "Conversation summary"},
+        "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
+    }, "required": ["reason", "summary", "sentiment"]},
+}
+
+def main():
+    client = DeepgramClient()
+    with client.agent.v1.connect() as agent:
+        agent.send_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"),
+                    prompt="You are a support agent. Help the caller. If you cannot resolve their issue, they ask for a human, or they seem frustrated, call escalate_to_human.",
+                    functions=[ESCALATE_FN],
+                ),
+                speak=SpeakSettingsV1(provider=SpeakSettingsV1Provider_Deepgram(type="deepgram", model="aura-2-thalia-en")),
+            ),
+        ))
+        print("Agent configured with escalation handoff")
+
+        def on_message(msg) -> None:
+            if isinstance(msg, AgentV1FunctionCallRequest):
+                args = json.loads(msg.input) if isinstance(msg.input, str) else msg.input
+                packet = {"escalation": True, "call_id": msg.id, **args}
+                print(f"ESCALATION CONTEXT PACKET:\n{json.dumps(packet, indent=2)}")
+                agent.send_function_call_response(AgentV1SendFunctionCallResponse(
+                    type="FunctionCallResponse", id=msg.id, name=msg.name, content='{"status":"transferring"}'))
+                agent.send_inject_agent_message(AgentV1InjectAgentMessage(
+                    message="I'm transferring you to a human agent now. They'll have full context. Please hold."))
+            elif isinstance(msg, bytes):
+                print(f"Audio: {len(msg)} bytes")
+            else:
+                print(f"Event: {getattr(msg, 'type', type(msg).__name__)}")
+
+        agent.on(EventType.OPEN, lambda _: print("Connection opened"))
+        agent.on(EventType.MESSAGE, on_message)
+        agent.on(EventType.CLOSE, lambda _: print("Connection closed"))
+        agent.start_listening()
+
+if __name__ == "__main__":
+    main()

recipes/python/voice-agents/v1/escalation-handoff/example_test.py

@@ -0,0 +1,16 @@
+import subprocess
+from pathlib import Path
+
+def test_example_runs():
+    """Runs the escalation-handoff example and verifies it produces output."""
+    example = Path(__file__).parent / "example.py"
+    result = subprocess.run(
+        ["python", str(example)],
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    assert result.returncode == 0, (
+        f"Example failed\nSTDOUT: {result.stdout}\nSTDERR: {result.stderr}"
+    )
+    assert result.stdout.strip(), "Example produced no output"