Build a Real-Time Sales Coaching AI Agent with OpenAI Realtime API + LangGraph

Mic/Call Audio
     ↓
OpenAI Realtime API (streaming transcription + voice activity detection)
     ↓
LangGraph State Machine
  ├── transcript_node      → accumulates rolling transcript
  ├── stage_classifier_node → discovery / pitch / objection / pricing / closing
  ├── signal_detector_node  → buying signals, objections, talk-time ratio
  ├── coach_node            → generates a short, actionable tip (or stays silent)
  └── dedupe_node           → suppresses repeat tips within a time window
     ↓
Coaching tip → rep's screen (text overlay or quiet TTS in their earpiece)

import asyncio
import websockets
import json

REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"

async def stream_call_audio(audio_chunks):
    async with websockets.connect(
        REALTIME_URL,
        extra_headers={"Authorization": f"Bearer {OPENAI_API_KEY}"}
    ) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text"],
                "input_audio_transcription": {"model": "whisper-1"}
            }
        }))

        async for chunk in audio_chunks:
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": chunk  # base64-encoded PCM16
            }))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "conversation.item.input_audio_transcription.completed":
                yield event["transcript"]

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class CallState(TypedDict):
    transcript: list[str]
    stage: Literal["discovery", "pitch", "objection", "pricing", "closing"]
    last_tip_at: float
    detected_signals: list[str]
    coaching_tip: str | None

def transcript_node(state: CallState) -> CallState:
    # append latest utterance, trim to last ~2 min for context window discipline
    return state

def stage_classifier_node(state: CallState) -> CallState:
    # cheap heuristic + LLM fallback for ambiguous turns
    state["stage"] = classify_stage(state["transcript"])
    return state

def signal_detector_node(state: CallState) -> CallState:
    state["detected_signals"] = detect_signals(state["transcript"], state["stage"])
    return state

def coach_node(state: CallState) -> CallState:
    if not state["detected_signals"]:
        state["coaching_tip"] = None
        return state
    state["coaching_tip"] = generate_tip(state["detected_signals"], state["stage"])
    return state

def dedupe_node(state: CallState) -> CallState:
    import time
    if state["coaching_tip"] and time.time() - state["last_tip_at"] < 45:
        state["coaching_tip"] = None  # too soon since last tip
    else:
        state["last_tip_at"] = time.time()
    return state

graph = StateGraph(CallState)

graph.add_node("transcript", transcript_node)
graph.add_node("stage", stage_classifier_node)
graph.add_node("signals", signal_detector_node)
graph.add_node("coach", coach_node)
graph.add_node("dedupe", dedupe_node)

graph.set_entry_point("transcript")
graph.add_edge("transcript", "stage")
graph.add_edge("stage", "signals")
graph.add_edge("signals", "coach")
graph.add_edge("coach", "dedupe")
graph.add_edge("dedupe", END)

app = graph.compile()

async for transcript_chunk in stream_call_audio(audio_chunks):
    state["transcript"].append(transcript_chunk)
    state = app.invoke(state)
    if state["coaching_tip"]:
        push_to_rep_screen(state["coaching_tip"])