Hey everyone 👋
I've been deep in the weeds this week integrating Cartesia's real-time voice API with our Zendesk instance to auto-categorize incoming support calls, and I wanted to share a detailed walkthrough. The goal was to reduce manual triage time for our agents by analyzing customer audio in real-time and suggesting tags, priority, and even routing. The documentation covers the basics, but I had to piece together several parts to make it production-ready.
Here’s the architecture I landed on:
1. **Cartesia Websocket Stream:** Capture real-time audio from our phone system (we're using Twilio SIP).
2. **Real-time Transcription & Analysis:** Use Cartesia's `sos` model with sentiment and intent extraction.
3. **Zendesk API Integration:** Parse the real-time transcript and sentiment to create/update tickets with appropriate tags, priority, and assignee group.
The trickiest part was handling the asynchronous flow between the websocket stream and Zendesk's API. You can't just wait for the call to end; you need to act on partial transcripts. Here's the core Python service I built to manage the connection.
```python
import asyncio
import json
import websockets
from cartesia import Cartesia
from zendesk import ZendeskClient
cartesia = Cartesia(api_key=os.getenv("CARTESIA_API_KEY"))
zendesk = ZendeskClient(subdomain=os.getenv("ZD_SUBDOMAIN"),
oauth_token=os.getenv("ZD_OAUTH_TOKEN"))
async def handle_audio_stream(audio_stream_url, ticket_id):
async with websockets.connect(audio_stream_url) as ws:
# Initiate Cartesia stream with specific model and features
stream_config = {
"model_id": "sos",
"transcription": {
"language": "en"
},
"features": ["sentiment", "intent"]
}
await ws.send(json.dumps(stream_config))
async for message in ws:
data = json.loads(message)
if data.get("type") == "transcription":
transcript = data["data"]["text"]
sentiment = data["data"].get("features", {}).get("sentiment")
intent = data["data"].get("features", {}).get("intent")
# Logic to determine tags and priority
tags = derive_tags(sentiment, intent, transcript)
priority = derive_priority(sentiment)
# Update Zendesk ticket in near-real-time
zendesk.tickets.update(ticket_id, tags=tags, priority=priority)
print(f"Updated ticket {ticket_id} with tags: {tags}")
def derive_tags(sentiment, intent, transcript):
tags = ["voice-support"]
if sentiment == "negative":
tags.append("escalate_urgent")
if "refund" in transcript.lower():
tags.append("billing")
# Add more business logic here
return tags
```
Key configuration points I had to figure out:
* **Model Selection:** The `sos` model was crucial for low-latency, streaming analysis. Batch models introduced too much lag.
* **Feature Extraction:** Enabling `sentiment` and `intent` directly in the stream payload saves a second API call later.
* **Zendesk Rate Limiting:** You need to implement a small debounce function (I used a simple `asyncio.sleep(0.5)`) to avoid hitting Zendesk's API rate limits with every single transcript chunk.
* **Error Handling:** The websocket can drop. My production code includes reconnection logic and a dead-letter queue for failed Zendesk updates.
**Benchmarks & Pitfalls:**
* End-to-end latency from utterance to Zendesk tag update averaged ~1.8 seconds, which is acceptable for our live dashboards.
* Cost monitoring is important: Cartesia pricing is per second of audio processed, so keep an eye on concurrency.
* The biggest "gotcha" was audio quality. Ensure your upstream audio stream is clean (noise-reduced, consistent volume) or the sentiment analysis can go haywire.
This setup has cut our average ticket handling time by about 40% in the last two weeks. The agents now get pre-tagged tickets with a suggested priority, which lets them jump to the most urgent issues faster.
Has anyone else tried a similar integration? I'm curious about different strategies for the intent classification piece—I'm currently using a simple keyword match, but thinking of training a small custom model.
— francesc
— francesc