diff --git a/ai-language-learning-phone-tutor-python/app.py b/ai-language-learning-phone-tutor-python/app.py
index 82da591a..cf5b7916 100644
--- a/ai-language-learning-phone-tutor-python/app.py
+++ b/ai-language-learning-phone-tutor-python/app.py
@@ -1,113 +1,502 @@
#!/usr/bin/env python3
-"""AI Language Learning Phone Tutor — call a number, practice a foreign language with AI."""
-import os, json, time, requests, telnyx
+"""AI Language Learning Phone Tutor — call a number, practice a foreign language with AI.
+
+Uses Telnyx Conversation Relay (TeXML + WebSocket). Telnyx handles STT/TTS/call control.
+This app only exchanges text over a WebSocket and forwards to AI Inference.
+
+Flow:
+ 1. Caller dials → Telnyx fetches TeXML from /texml/inbound
+ 2. TeXML tells Telnyx to open a Conversation Relay WebSocket to /ws/conversation-relay
+ 3. Caller speaks → Telnyx transcribes → sends text frame to our WebSocket
+ 4. We forward to AI Inference (Llama-3.3-70B) with a language-tutor system prompt
+ 5. We stream the reply back token-by-token → Telnyx TTS speaks it to the caller
+
+A live dashboard at / shows the call happening in real time (SSE) — handy for demos/video.
+"""
+import json
+import os
+import time
+import threading
+from collections import deque
+from html import escape
+from typing import Any
+
+import requests
from dotenv import load_dotenv
-from flask import Flask, request, jsonify
-import threading, time as _ttl_time
+from flask import Flask, Response, request, jsonify, stream_with_context
+from flask_sock import Sock
+
load_dotenv()
+
app = Flask(__name__)
-client = telnyx.Telnyx(api_key=os.getenv("TELNYX_API_KEY"), public_key=os.getenv("TELNYX_PUBLIC_KEY"))
-TELNYX_PUBLIC_KEY = os.getenv("TELNYX_PUBLIC_KEY", "")
-TELNYX_API_KEY = os.getenv("TELNYX_API_KEY")
+sock = Sock(app)
+
+# --- Configuration -----------------------------------------------------------
+TELNYX_API_KEY = os.getenv("TELNYX_API_KEY", "")
AI_MODEL = os.getenv("AI_MODEL", "meta-llama/Llama-3.3-70B-Instruct")
-INFERENCE_URL = "https://api.telnyx.com/v2/ai/chat/completions"
-active_calls = {}
+INFERENCE_URL = os.getenv("INFERENCE_URL", "https://api.telnyx.com/v2/ai/chat/completions")
+MAX_TOKENS = int(os.getenv("MAX_TOKENS", "200"))
+REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "20"))
+VOICE = os.getenv("VOICE", "Telnyx.Natural.abbie")
+LANGUAGE = os.getenv("LANGUAGE", "en")
+TRANSCRIPTION_PROVIDER = os.getenv("TRANSCRIPTION_PROVIDER", "deepgram")
+WELCOME_GREETING = os.getenv(
+ "WELCOME_GREETING",
+ "Welcome to Language Tutor! Say the name of a language to start: Spanish, French, Japanese, or Mandarin.",
+)
-def _start_ttl_cleanup(*stores, ttl_seconds=3600, interval=300):
- def _cleanup():
- while True:
- _ttl_time.sleep(interval)
- cutoff = _ttl_time.time() - ttl_seconds
- for store in stores:
- expired = [k for k, v in store.items()
- if isinstance(v, dict) and v.get("_ts", _ttl_time.time()) < cutoff]
- for k in expired:
- store.pop(k, None)
- threading.Thread(target=_cleanup, daemon=True).start()
+LANGUAGES = {
+ "spanish": {"name": "Spanish", "code": "es", "flag": "🇪🇸"},
+ "french": {"name": "French", "code": "fr", "flag": "🇫🇷"},
+ "japanese": {"name": "Japanese", "code": "ja", "flag": "🇯🇵"},
+ "mandarin": {"name": "Mandarin", "code": "zh", "flag": "🇨🇳"},
+ "chinese": {"name": "Mandarin", "code": "zh", "flag": "🇨🇳"},
+}
-_start_ttl_cleanup(active_calls)
+TUTOR_PROMPT = (
+ "You are a {lang} language tutor on a phone call. "
+ "Start with a simple greeting in {lang}, then give the English translation. "
+ "Gradually increase difficulty. Correct mistakes gently. Mix {lang} and English. "
+ "Keep each response short for phone conversation — 2-3 sentences max. "
+ "No markdown, no bullet points, no headers. Speak naturally."
+)
-session_history = []
-_start_ttl_cleanup(session_history)
+# Per-call session state
+sessions: dict[str, dict[str, Any]] = {}
-LANGUAGES = {"1": {"name": "Spanish", "code": "es"}, "2": {"name": "French", "code": "fr"}, "3": {"name": "Japanese", "code": "ja"}, "4": {"name": "Mandarin", "code": "zh"}}
+# --- Live event bus ----------------------------------------------------------
+_events: deque[dict[str, Any]] = deque(maxlen=200)
+_subscribers: list[Any] = []
+_bus_lock = threading.Lock()
-def call_inference(messages, max_tokens=200):
- try:
- resp = requests.post(INFERENCE_URL, headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
- json={"model": AI_MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": 0.7}, timeout=15)
- except requests.exceptions.RequestException as e:
- app.logger.error("Inference request failed: %s", e)
- return None
- resp.raise_for_status()
- return resp.json()["choices"][0]["message"]["content"]
-
-@app.route("/webhooks/voice", methods=["POST"])
-def handle_voice():
- # Verify the Telnyx Ed25519 signature before trusting the event.
+
+def emit(kind: str, direction: str, title: str, text: str = "", session: str | None = None, extra: dict | None = None) -> None:
+ evt = {
+ "ts": time.time(),
+ "kind": kind,
+ "dir": direction,
+ "title": title,
+ "text": text,
+ "session": session,
+ "extra": extra or {},
+ }
+ with _bus_lock:
+ _events.append(evt)
+ dead = []
+ for q in _subscribers:
+ try:
+ q.append(evt)
+ except Exception:
+ dead.append(q)
+ for q in dead:
+ try:
+ _subscribers.remove(q)
+ except ValueError:
+ pass
+
+
+def log(label: str, value: Any) -> None:
+ print(f"[{label}] {json.dumps(value, indent=2, sort_keys=True)}", flush=True)
+
+
+# --- AI Inference ------------------------------------------------------------
+
+def call_inference_streamed(messages: list[dict[str, str]], ws) -> str:
+ """Stream the AI reply — sends partial text frames so TTS starts speaking
+ the first words while the LLM is still generating the rest."""
+ headers = {
+ "Authorization": f"Bearer {TELNYX_API_KEY}",
+ "Content-Type": "application/json",
+ }
+ payload = {
+ "model": AI_MODEL,
+ "messages": messages,
+ "max_tokens": MAX_TOKENS,
+ "temperature": 0.7,
+ "stream": True,
+ }
+ full_reply = ""
try:
- client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
- except Exception:
- return jsonify({"error": "invalid signature"}), 401
- payload = request.get_json()
- if not payload:
- return jsonify({"error": "invalid request body"}), 400
- data = payload.get("data", {})
- p = data.get("payload", {})
- event_type = data.get("event_type")
- ccid = p.get("call_control_id")
- call = active_calls.get(ccid)
- if event_type == "call.initiated" and p.get("direction") == "incoming":
- active_calls[ccid] = {"caller": p.get("from"), "state": "language_select", "conversation": []}
- client.calls.actions.answer(ccid)
- return jsonify({"status": "answering"}), 200
- elif event_type == "call.answered" and call:
- client.calls.actions.speak(ccid, payload="Welcome to Language Tutor! Press 1 for Spanish, 2 for French, 3 for Japanese, 4 for Mandarin.", voice="female", language_code="en-US")
- return jsonify({"status": "greeting"}), 200
- elif event_type == "call.speak.ended" and call:
- if call["state"] == "language_select":
- client.calls.actions.gather(ccid, input_type="dtmf speech", timeout_secs=10, min_digits=1, max_digits=1)
- else:
- client.calls.actions.gather(ccid, input_type="speech", end_silence_timeout_secs=3, timeout_secs=20, language_code="en-US")
- return jsonify({"status": "listening"}), 200
- elif event_type == "call.gather.ended" and call:
- digits = p.get("digits", "")
- speech = p.get("speech", {}).get("result", "")
- if call["state"] == "language_select":
- lang_key = digits or speech.strip()[:1]
- lang = LANGUAGES.get(lang_key, LANGUAGES["1"])
- call["language"] = lang
- call["state"] = "tutoring"
- call["conversation"] = [{"role": "system", "content": f"You are a {lang['name']} language tutor. Start with a simple greeting in {lang['name']}, then English translation. Gradually increase difficulty. Correct mistakes gently. Mix {lang['name']} and English. Keep each response short for phone conversation."}]
- intro = call_inference(call["conversation"] + [{"role": "user", "content": "Start the lesson."}])
- if not intro:
- intro = "Sorry, I had trouble generating a response. Let's try again."
- call["conversation"].append({"role": "assistant", "content": intro})
- client.calls.actions.speak(ccid, payload=intro, voice="female", language_code="en-US")
- elif call["state"] == "tutoring" and speech:
- call["conversation"].append({"role": "user", "content": speech})
- response = call_inference(call["conversation"])
- if not response:
- response = "Sorry, I didn't catch that. Could you repeat what you said?"
- call["conversation"].append({"role": "assistant", "content": response})
- client.calls.actions.speak(ccid, payload=response, voice="female", language_code="en-US")
- else:
- client.calls.actions.speak(ccid, payload="Try again! Say something in the language you're learning.", voice="female", language_code="en-US")
- return jsonify({"status": "processing"}), 200
- elif event_type == "call.hangup":
- call = active_calls.pop(ccid, None)
- if call and call.get("conversation"):
- session_history.append({"caller": call["caller"], "language": call.get("language", {}).get("name"), "exchanges": len(call["conversation"]) // 2})
- return jsonify({"status": "ended"}), 200
- return jsonify({"status": "ok"}), 200
-
-@app.route("/sessions", methods=["GET"])
-def list_sessions():
- return jsonify({"sessions": session_history[-50:]}), 200
+ resp = requests.post(INFERENCE_URL, headers=headers, json=payload, timeout=REQUEST_TIMEOUT, stream=True)
+ resp.raise_for_status()
+ for line in resp.iter_lines(decode_unicode=True):
+ if not line or not line.startswith("data: "):
+ continue
+ data = line.removeprefix("data: ").strip()
+ if data == "[DONE]":
+ break
+ try:
+ chunk = json.loads(data)
+ except json.JSONDecodeError:
+ continue
+ delta = chunk.get("choices", [{}])[0].get("delta", {})
+ token = delta.get("content")
+ if token:
+ full_reply += token
+ ws.send(json.dumps({"type": "text", "token": token, "last": False}))
+ ws.send(json.dumps({"type": "text", "token": "", "last": True}))
+ except requests.exceptions.RequestException as exc:
+ app.logger.error("Inference streaming request failed: %s", exc)
+ fallback = "Sorry, I had trouble generating a response. Let's try again."
+ ws.send(json.dumps({"type": "text", "token": fallback, "last": True}))
+ return fallback
+ return full_reply
+
+
+# --- TeXML -------------------------------------------------------------------
+
+def public_base_url() -> str:
+ configured = os.getenv("TELNYX_PUBLIC_BASE_URL", "").strip()
+ if configured:
+ return configured.rstrip("/")
+ return request.url_root.rstrip("/")
+
+
+def conversation_relay_ws_url() -> str:
+ base = public_base_url()
+ if base.startswith("https://"):
+ return "wss://" + base.removeprefix("https://") + "/ws/conversation-relay"
+ if base.startswith("http://"):
+ return "ws://" + base.removeprefix("http://") + "/ws/conversation-relay"
+ return base + "/ws/conversation-relay"
+
+
+def texml_response() -> str:
+ ws_url = escape(conversation_relay_ws_url(), quote=True)
+ action_url = escape(public_base_url() + "/callbacks/conversation-relay", quote=True)
+ greeting = escape(WELCOME_GREETING, quote=True)
+ voice = escape(VOICE, quote=True)
+ language = escape(LANGUAGE, quote=True)
+ provider = escape(TRANSCRIPTION_PROVIDER, quote=True)
+ return f"""
+
+
+
+
+"""
+
+
+# --- HTTP routes -------------------------------------------------------------
+
+@app.route("/texml/inbound", methods=["GET", "POST"])
+def texml_inbound() -> Response:
+ emit("webhook", "web", "TeXML instruction fetch", text="Telnyx requested the call XML (voice_url)")
+ return Response(texml_response(), status=200, mimetype="application/xml")
+
+
+@app.route("/callbacks/conversation-relay", methods=["GET", "POST"])
+def conversation_relay_action() -> Response:
+ body = request.get_json(silent=True) or request.values.to_dict(flat=False)
+ log("conversation_relay.action", body)
+ emit("webhook", "web", "Conversation Relay action callback", text=json.dumps(body)[:200] if isinstance(body, (dict, list)) else str(body)[:200])
+ return Response(status=204)
+
@app.route("/health", methods=["GET"])
-def health():
- return jsonify({"status": "ok", "active": len(active_calls), "sessions": len(session_history)}), 200
+def health() -> Response:
+ return jsonify({"status": "ok", "active_sessions": len(sessions)})
+
+
+@app.route("/api/state", methods=["GET"])
+def api_state() -> Response:
+ with _bus_lock:
+ recent = list(_events)
+ return jsonify({
+ "config": {
+ "model": AI_MODEL,
+ "inference_url": INFERENCE_URL,
+ "voice": VOICE,
+ "welcome_greeting": WELCOME_GREETING,
+ },
+ "sessions": [
+ {
+ "id": sid,
+ "from": s.get("from"),
+ "language": s.get("language", {}).get("name") if isinstance(s.get("language"), dict) else None,
+ "language_flag": s.get("language", {}).get("flag") if isinstance(s.get("language"), dict) else None,
+ "turns": len(s.get("history", [])) // 2,
+ "duration": int(time.time() - s.get("started", time.time())),
+ "history": s.get("history", []),
+ }
+ for sid, s in sessions.items()
+ ],
+ "events": recent,
+ })
+
+
+@app.route("/events")
+def events() -> Response:
+ q: deque[dict[str, Any]] = deque(maxlen=100)
+ with _bus_lock:
+ _subscribers.append(q)
+
+ def stream():
+ try:
+ yield "retry: 3000\n\n"
+ while True:
+ try:
+ evt = q.popleft()
+ except IndexError:
+ yield ": keep-alive\n\n"
+ time.sleep(0.4)
+ continue
+ yield f"data: {json.dumps(evt)}\n\n"
+ finally:
+ with _bus_lock:
+ try:
+ _subscribers.remove(q)
+ except ValueError:
+ pass
+
+ return Response(stream_with_context(stream()), mimetype="text/event-stream",
+ headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
+
+
+# --- Dashboard ---------------------------------------------------------------
+
+DASHBOARD_HTML = """
+
+
+AI Language Tutor — Live Demo
+
+
+
+ AI Language Tutor — Live Demo call & learn
+ Call a Telnyx number, say a language name, and practice with an AI tutor. Telnyx Conversation Relay handles the audio; AI Inference runs the conversation.
+
+ Caller dials→
+ Telnyx · STT/TTS→
+ This app (WebSocket text)→
+ AI Inference
+ →
+ reply→
+ Telnyx · TTS→
+ Caller hears it
+
+
+
+
+ Live frames
connecting…
+
+
+
+ Conversation
no call yet
+ Dial the number to start a call.
Frames will stream here the moment they happen.