"""Bounded AI conversation primitives used by the explicit Mock profile. The module is deliberately provider-neutral. A real provider is only allowed when a separately configured adapter is injected; the default factory never turns a missing provider into a Mock call silently. """ from __future__ import annotations import hashlib import io import json import math import re import threading import time import uuid import wave from collections.abc import Callable, Iterator, Mapping from dataclasses import dataclass from pathlib import Path from typing import Any, Protocol MAX_PROMPT_BYTES = 32 * 1024 MAX_VARIABLE_BYTES = 4096 _PLACEHOLDER = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") _SECRET_KEYS = { "api_key", "apikey", "access_token", "secret", "password", "authorization", "private_key", "credential", "token", } class AIConfigError(ValueError): """A local, deterministic configuration or contract failure.""" def __init__(self, code: str, detail: str) -> None: super().__init__(detail) self.code = code self.detail = detail class AIProviderError(RuntimeError): """A provider failure that must not be hidden by a fallback.""" def __init__(self, code: str, detail: str, retryable: bool = False) -> None: super().__init__(detail) self.code = code self.detail = detail self.retryable = retryable @dataclass(frozen=True) class StreamChunk: sequence: int text: str = "" audio: bytes = b"" done: bool = False late: bool = False provider_model: str | None = None class LLMStream(Protocol): def stream( self, messages: list[dict[str, str]], config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: return iter(()) class TTSStream(Protocol): def stream( self, text: str, config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: return iter(()) class ASRStream(Protocol): def transcribe( self, audio: bytes, config: Mapping[str, Any], turn_id: str ) -> Iterator[dict[str, Any]]: return iter(()) def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def config_digest(config: Mapping[str, Any]) -> str: return hashlib.sha256(canonical(config).encode("utf-8")).hexdigest() def _reject_sensitive_keys(value: Any, path: str = "config") -> None: if isinstance(value, Mapping): for key, child in value.items(): name = str(key).lower() if ( name in _SECRET_KEYS or name in {"url", "endpoint", "base_url"} or name.endswith("_url") ): raise AIConfigError( "AI_CONFIG_SECRET_OR_URL", f"{path}.{key} cannot contain a credential or provider URL", ) _reject_sensitive_keys(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): _reject_sensitive_keys(child, f"{path}[{index}]") def load_prompt(path: str | Path, max_bytes: int = MAX_PROMPT_BYTES) -> str: prompt_path = Path(path) try: if not prompt_path.is_file() or prompt_path.is_symlink(): raise AIConfigError( "PROMPT_FILE_INVALID", "prompt file must be a regular file" ) raw = prompt_path.read_bytes() except AIConfigError: raise except OSError as exc: raise AIConfigError( "PROMPT_FILE_UNREADABLE", "prompt file cannot be read" ) from exc if not 1 <= len(raw) <= max_bytes: raise AIConfigError( "PROMPT_TOO_LARGE", "prompt UTF-8 bytes exceed the configured limit" ) try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: raise AIConfigError("PROMPT_NOT_UTF8", "prompt file must be UTF-8") from exc if "\x00" in text: raise AIConfigError("PROMPT_INVALID", "prompt must not contain NUL bytes") return text def render_prompt( template: str, variables: Mapping[str, Any] | None = None, allowed_variables: list[str] | tuple[str, ...] = (), ) -> str: if not isinstance(template, str) or not template: raise AIConfigError("PROMPT_INVALID", "prompt must be a non-empty string") allowed = set(allowed_variables) if len(allowed) != len(allowed_variables) or any( not isinstance(name, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) for name in allowed_variables ): raise AIConfigError( "PROMPT_VARIABLES_INVALID", "allowed prompt variables are invalid" ) variables = variables or {} unknown_values = set(variables) - allowed if unknown_values: raise AIConfigError( "PROMPT_VARIABLE_NOT_ALLOWED", f"prompt variables are not allowed: {sorted(unknown_values)}", ) def replace(match: re.Match[str]) -> str: name = match.group(1) if name not in allowed: raise AIConfigError( "PROMPT_VARIABLE_NOT_ALLOWED", f"prompt variable is not allowed: {name}" ) if name not in variables: raise AIConfigError( "PROMPT_VARIABLE_MISSING", f"prompt variable is missing: {name}" ) value = variables[name] if not isinstance(value, (str, int, float, bool)) or value is None: raise AIConfigError( "PROMPT_VARIABLE_INVALID", f"prompt variable has an unsupported type: {name}", ) rendered = str(value) if len(rendered.encode("utf-8")) > MAX_VARIABLE_BYTES: raise AIConfigError( "PROMPT_VARIABLE_TOO_LARGE", f"prompt variable is too large: {name}" ) return rendered rendered = _PLACEHOLDER.sub(replace, template) if len(rendered.encode("utf-8")) > MAX_PROMPT_BYTES: raise AIConfigError( "PROMPT_TOO_LARGE", "rendered prompt exceeds the configured limit" ) return rendered def _string(value: Any, field: str, maximum: int = 128) -> str: if ( not isinstance(value, str) or not value or len(value) > maximum or any(char.isspace() for char in value) or "://" in value ): raise AIConfigError("AI_CONFIG_INVALID", f"{field} must be a non-empty token") return value def _ensure_fields(value: Mapping[str, Any], allowed: set[str], field: str) -> None: extra = set(value) - allowed if extra: raise AIConfigError( "AI_CONFIG_FIELDS_INVALID", f"unsupported {field} fields: {sorted(extra)}" ) def normalize_agent_config( source: Mapping[str, Any], version_id: str | None = None ) -> dict[str, Any]: """Normalize legacy profile entries into one immutable AI snapshot shape.""" if not isinstance(source, Mapping): raise AIConfigError( "AI_CONFIG_INVALID", "agent configuration must be an object" ) try: config = json.loads(canonical(dict(source))) except (TypeError, ValueError) as exc: raise AIConfigError( "AI_CONFIG_INVALID", "agent configuration is not JSON-compatible" ) from exc if version_id is not None: config["agent_version_id"] = version_id if not config.get("agent_version_id"): raise AIConfigError( "AI_CONFIG_VERSION_REQUIRED", "agent_version_id is required" ) config.setdefault("immutable", True) llm = config.get("llm") if isinstance(llm, str): config["llm"] = { "provider_ref": llm, "model": "mock-chat-v1" if llm == "mock" else f"{llm}-chat", "temperature": 0.2, "max_tokens": 256, "timeout_ms": 5000, } tts = config.get("tts") if isinstance(tts, str): config["tts"] = { "provider_ref": tts, "model": "mock-tts-v1" if tts == "mock" else f"{tts}-tts", "voice": "mock-neutral" if tts == "mock" else "configured", "speed": 1.0, "format": { "encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1, }, "timeout_ms": 5000, } asr = config.get("asr") if isinstance(asr, str): config["asr"] = { "provider_ref": asr, "language": "zh-CN", "input": { "encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1, "sample_width_bytes": 2, }, "interim": True, "timeout_ms": 5000, } prompt = config.get("prompt") if isinstance(prompt, str): config["prompt"] = {"text": prompt, "allowed_variables": []} elif prompt is None: config["prompt"] = { "text": "You are a concise telephone assistant. Answer the caller's last statement.", "allowed_variables": [], } conversation = config.get("conversation", {}) if not isinstance(conversation, Mapping): raise AIConfigError("AI_CONFIG_INVALID", "conversation must be an object") conversation = dict(conversation) config["conversation"] = conversation defaults = { "opening": "", "allow_interrupt": True, "silence_timeout_ms": 3000, "max_duration_ms": 120000, "max_turns": 20, "sentence_max_chars": 80, "max_pending_audio_chunks": 32, } for key, value in defaults.items(): conversation.setdefault(key, value) return validate_agent_config(config) def validate_agent_config(config: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(config, Mapping): raise AIConfigError( "AI_CONFIG_INVALID", "agent configuration must be an object" ) _reject_sensitive_keys(config) try: result = json.loads(canonical(dict(config))) except (TypeError, ValueError) as exc: raise AIConfigError( "AI_CONFIG_INVALID", "agent configuration is not JSON-compatible" ) from exc allowed_top = { "agent_version_id", "immutable", "llm", "prompt", "tts", "asr", "conversation", "metadata", } extra = set(result) - allowed_top if extra: raise AIConfigError( "AI_CONFIG_FIELDS_INVALID", f"unsupported AI config fields: {sorted(extra)}" ) _string(result.get("agent_version_id"), "agent_version_id") if not result.get("immutable"): raise AIConfigError( "AI_CONFIG_NOT_IMMUTABLE", "published AI versions must be immutable" ) llm = result.get("llm") if not isinstance(llm, Mapping): raise AIConfigError("AI_CONFIG_INVALID", "llm configuration is required") _ensure_fields( llm, { "provider_ref", "credential_ref", "model", "temperature", "max_tokens", "timeout_ms", }, "llm", ) _string(llm.get("provider_ref"), "llm.provider_ref") model = _string(llm.get("model"), "llm.model") if llm["provider_ref"] == "mock" and not model.startswith("mock-"): raise AIConfigError( "LLM_MODEL_UNSUPPORTED", "mock LLM models must use the mock- namespace" ) for field, low, high in (("temperature", 0.0, 2.0),): if field in llm and ( type(llm[field]) not in {int, float} or not low <= llm[field] <= high ): raise AIConfigError( "LLM_PARAMETER_INVALID", f"{field} is outside provider limits" ) for field in ("max_tokens", "timeout_ms"): if field in llm and (type(llm[field]) is not int or llm[field] < 1): raise AIConfigError( "LLM_PARAMETER_INVALID", f"{field} must be a positive integer" ) prompt = result.get("prompt") if not isinstance(prompt, Mapping) or not isinstance(prompt.get("text"), str): raise AIConfigError("PROMPT_INVALID", "prompt.text is required") _ensure_fields(prompt, {"text", "allowed_variables", "max_bytes"}, "prompt") prompt_bytes = len(prompt["text"].encode("utf-8")) max_prompt = prompt.get("max_bytes", MAX_PROMPT_BYTES) if ( type(max_prompt) is not int or not 1 <= max_prompt <= MAX_PROMPT_BYTES or not 1 <= prompt_bytes <= max_prompt ): raise AIConfigError("PROMPT_TOO_LARGE", "prompt exceeds its UTF-8 byte limit") variables = prompt.get("allowed_variables", []) if not isinstance(variables, list) or len(variables) > 32: raise AIConfigError( "PROMPT_VARIABLES_INVALID", "allowed_variables must be a list of at most 32 names", ) render_prompt(prompt["text"], {}, variables) if not _PLACEHOLDER.search( prompt["text"] ) else None for match in _PLACEHOLDER.finditer(prompt["text"]): if match.group(1) not in variables: raise AIConfigError( "PROMPT_VARIABLE_NOT_ALLOWED", f"prompt variable is not allowed: {match.group(1)}", ) tts = result.get("tts") if not isinstance(tts, Mapping): raise AIConfigError("AI_CONFIG_INVALID", "tts configuration is required") _ensure_fields( tts, { "provider_ref", "credential_ref", "model", "voice", "speed", "format", "timeout_ms", }, "tts", ) _string(tts.get("provider_ref"), "tts.provider_ref") tts_model = _string(tts.get("model"), "tts.model") voice = _string(tts.get("voice"), "tts.voice") if tts["provider_ref"] == "mock" and ( not tts_model.startswith("mock-") or not voice.startswith("mock-") ): raise AIConfigError( "TTS_VOICE_INCOMPATIBLE", "mock TTS model and voice must use the mock- namespace", ) speed = tts.get("speed", 1.0) if type(speed) not in {int, float} or not 0.25 <= speed <= 3.0: raise AIConfigError( "TTS_PARAMETER_INVALID", "tts.speed is outside provider limits" ) audio_format = tts.get("format") if not isinstance(audio_format, Mapping): raise AIConfigError("TTS_FORMAT_INVALID", "tts.format is required") _ensure_fields( audio_format, {"encoding", "sample_rate_hz", "channels"}, "tts.format" ) if audio_format.get("encoding") not in {"pcm_s16le", "pcma"}: raise AIConfigError("TTS_FORMAT_INVALID", "tts.format.encoding is unsupported") if ( type(audio_format.get("sample_rate_hz")) is not int or not 8000 <= audio_format["sample_rate_hz"] <= 48000 ): raise AIConfigError( "TTS_FORMAT_INVALID", "tts sample rate is outside the supported range" ) if audio_format.get("channels") != 1: raise AIConfigError("TTS_FORMAT_INVALID", "telephone TTS must be mono") asr = result.get("asr") if not isinstance(asr, Mapping): raise AIConfigError("AI_CONFIG_INVALID", "asr configuration is required") _ensure_fields( asr, { "provider_ref", "credential_ref", "model", "language", "input", "interim", "timeout_ms", }, "asr", ) _string(asr.get("provider_ref"), "asr.provider_ref") if "model" in asr: _string(asr.get("model"), "asr.model") _string(asr.get("language"), "asr.language", 32) input_format = asr.get("input") if ( not isinstance(input_format, Mapping) or input_format.get("encoding") != "pcm_s16le" ): raise AIConfigError("ASR_FORMAT_INVALID", "ASR input must be PCM16LE") _ensure_fields( input_format, {"encoding", "sample_rate_hz", "channels", "sample_width_bytes"}, "asr.input", ) if input_format.get("channels") != 1 or input_format.get("sample_width_bytes") != 2: raise AIConfigError("ASR_FORMAT_INVALID", "ASR input must be mono 16-bit audio") if ( type(input_format.get("sample_rate_hz")) is not int or not 8000 <= input_format["sample_rate_hz"] <= 48000 ): raise AIConfigError( "ASR_FORMAT_INVALID", "ASR sample rate is outside the supported range" ) conversation = result.get("conversation") if not isinstance(conversation, Mapping): raise AIConfigError( "CONVERSATION_CONFIG_INVALID", "conversation configuration is required" ) _ensure_fields( conversation, { "opening", "allow_interrupt", "silence_timeout_ms", "max_duration_ms", "max_turns", "sentence_max_chars", "max_pending_audio_chunks", }, "conversation", ) opening = conversation.get("opening", "") if not isinstance(opening, str) or len(opening.encode("utf-8")) > MAX_PROMPT_BYTES: raise AIConfigError("CONVERSATION_CONFIG_INVALID", "opening is invalid") for field in ( "silence_timeout_ms", "max_duration_ms", "max_turns", "sentence_max_chars", "max_pending_audio_chunks", ): if type(conversation.get(field)) is not int or conversation[field] < 1: raise AIConfigError( "CONVERSATION_CONFIG_INVALID", f"conversation.{field} must be positive" ) if conversation["max_duration_ms"] > 3600000 or conversation["max_turns"] > 1000: raise AIConfigError( "CONVERSATION_CONFIG_INVALID", "conversation limits are too large" ) if type(conversation.get("allow_interrupt")) is not bool: raise AIConfigError( "CONVERSATION_CONFIG_INVALID", "allow_interrupt must be boolean" ) return result def _chunks(text: str, width: int = 8) -> list[str]: return [text[index : index + width] for index in range(0, len(text), width)] or [""] def _pcm_tone(text: str, sample_rate: int = 16000, speed: float = 1.0) -> bytes: try: duration_ms = max(50, min(8000, int(max(1, len(text)) * 42 / max(0.25, speed)))) frames = max(1, int(sample_rate * duration_ms / 1000)) seed = int(hashlib.sha256(text.encode("utf-8")).hexdigest()[:8], 16) frequency = 320 + seed % 360 output = bytearray() for index in range(frames): value = int(9000 * math.sin(2 * math.pi * frequency * index / sample_rate)) output.extend(value.to_bytes(2, "little", signed=True)) return bytes(output) except (OverflowError, TypeError, ValueError, ZeroDivisionError) as exc: raise AIConfigError( "AUDIO_GENERATION_FAILED", "mock audio could not be generated" ) from exc class MockLLM: def __init__( self, chunk_delay_s: float = 0.0, late_chunks_after_cancel: int = 2 ) -> None: self.chunk_delay_s = max(0.0, chunk_delay_s) self.late_chunks_after_cancel = max(0, late_chunks_after_cancel) def stream( self, messages: list[dict[str, str]], config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: last = messages[-1]["content"] if messages else "" reply = f"已收到：{last}。请继续。" late_left = self.late_chunks_after_cancel for sequence, part in enumerate(_chunks(reply)): late = cancelled.is_set() if late and late_left <= 0: return if late: late_left -= 1 if self.chunk_delay_s: time.sleep(self.chunk_delay_s) yield StreamChunk(sequence=sequence, text=part, late=late) yield StreamChunk(sequence=len(reply), done=True, late=cancelled.is_set()) class MockTTS: def __init__( self, chunk_size: int = 640, chunk_delay_s: float = 0.0, late_chunks_after_cancel: int = 2, ) -> None: self.chunk_size = max(64, chunk_size) self.chunk_delay_s = max(0.0, chunk_delay_s) self.late_chunks_after_cancel = max(0, late_chunks_after_cancel) def stream( self, text: str, config: Mapping[str, Any], turn_id: str, cancelled: threading.Event, ) -> Iterator[StreamChunk]: try: tts = config["tts"] sample_rate = int(tts["format"]["sample_rate_hz"]) raw = _pcm_tone(text, sample_rate, float(tts.get("speed", 1.0))) except (KeyError, TypeError, ValueError, OverflowError) as exc: raise AIProviderError( "TTS_CONFIG_INVALID", "TTS configuration cannot produce audio" ) from exc late_left = self.late_chunks_after_cancel for sequence, offset in enumerate(range(0, len(raw), self.chunk_size)): late = cancelled.is_set() if late and late_left <= 0: return if late: late_left -= 1 if self.chunk_delay_s: time.sleep(self.chunk_delay_s) yield StreamChunk( sequence=sequence, audio=raw[offset : offset + self.chunk_size], late=late, ) yield StreamChunk(sequence=len(raw), done=True, late=cancelled.is_set()) class MockASR: def transcribe( self, audio: bytes, config: Mapping[str, Any], turn_id: str ) -> Iterator[dict[str, Any]]: validate_wav(audio) fingerprint = hashlib.sha256(audio).hexdigest()[:12] text = f"mock audio {fingerprint}" yield {"turn_id": turn_id, "sequence": 0, "text": text, "is_final": False} yield {"turn_id": turn_id, "sequence": 1, "text": text, "is_final": True} def validate_wav(data: bytes) -> dict[str, int]: if not isinstance(data, bytes) or len(data) < 44: raise AIConfigError( "AUDIO_INVALID", "audio must be a non-empty WAV byte string" ) try: with wave.open(io.BytesIO(data), "rb") as wav: metadata = { "channels": wav.getnchannels(), "sample_width_bytes": wav.getsampwidth(), "sample_rate_hz": wav.getframerate(), "frames": wav.getnframes(), } except (EOFError, wave.Error) as exc: raise AIConfigError( "AUDIO_INVALID", "audio is not a readable WAV file" ) from exc if metadata["channels"] != 1 or metadata["sample_width_bytes"] != 2: raise AIConfigError("AUDIO_FORMAT_INVALID", "audio must be mono PCM16") if not 8000 <= metadata["sample_rate_hz"] <= 48000 or metadata["frames"] < 1: raise AIConfigError( "AUDIO_FORMAT_INVALID", "audio sample rate or duration is invalid" ) return metadata def pcm_to_wav(raw: bytes, sample_rate_hz: int = 16000) -> bytes: if not isinstance(raw, bytes) or not raw or len(raw) % 2: raise AIConfigError("AUDIO_INVALID", "PCM16 bytes are invalid") output = io.BytesIO() with wave.open(output, "wb") as wav: wav.setnchannels(1) wav.setsampwidth(2) wav.setframerate(sample_rate_hz) wav.writeframes(raw) return output.getvalue() def resample_pcm16(raw: bytes, input_rate_hz: int, output_rate_hz: int) -> bytes: """Use nearest-frame selection for bounded telephone-rate test audio.""" if len(raw) % 2 or input_rate_hz < 1 or output_rate_hz < 1: raise AIConfigError("AUDIO_INVALID", "PCM16 resampling arguments are invalid") if input_rate_hz == output_rate_hz: return raw samples = [ int.from_bytes(raw[offset : offset + 2], "little", signed=True) for offset in range(0, len(raw), 2) ] output_frames = max(1, round(len(samples) * output_rate_hz / input_rate_hz)) selected = bytearray() for index in range(output_frames): source = min(len(samples) - 1, round(index * input_rate_hz / output_rate_hz)) selected.extend(samples[source].to_bytes(2, "little", signed=True)) return bytes(selected) def pcm16_to_alaw( raw: bytes, input_rate_hz: int = 16000, output_rate_hz: int = 8000 ) -> bytes: """Resample PCM16 and encode the PCMA/8000 phone payload.""" raw = resample_pcm16(raw, input_rate_hz, output_rate_hz) output = bytearray() for offset in range(0, len(raw), 2): sample = int.from_bytes(raw[offset : offset + 2], "little", signed=True) sign = 0x80 if sample < 0 else 0 magnitude = min(32635, abs(sample)) if magnitude < 256: exponent = 0 mantissa = magnitude >> 4 else: exponent = 1 value = magnitude >> 8 while value > 1 and exponent < 7: value >>= 1 exponent += 1 mantissa = (magnitude >> (exponent + 3)) & 0x0F output.append((sign | (exponent << 4) | mantissa) ^ 0xD5) return bytes(output) def _segments(text: str, maximum: int) -> tuple[list[str], str]: parts: list[str] = [] current = "" for char in text: current += char if char in "。！？!?；;\n" or len(current) >= maximum: parts.append(current) current = "" return parts, current class ConversationEngine: """One bounded, cancellable conversation engine for text/audio tests and Cells.""" def __init__( self, config: Mapping[str, Any], llm: LLMStream | None = None, tts: TTSStream | None = None, asr: ASRStream | None = None, journal: Any | None = None, player_confirmed: bool = True, monotonic: Callable[[], float] = time.monotonic, ) -> None: self.config = validate_agent_config(config) adapters = {"llm": llm, "tts": tts, "asr": asr} for name, adapter in adapters.items(): if adapter is None and self.config[name]["provider_ref"] != "mock": raise AIConfigError( "REAL_PROVIDER_ADAPTER_REQUIRED", f"{name} provider requires an injected provider-specific adapter", ) self.llm = llm if llm is not None else MockLLM() self.tts = tts if tts is not None else MockTTS() self.asr = asr if asr is not None else MockASR() refs = {self.config[name]["provider_ref"] for name in adapters} self.mode = ( "mock" if refs == {"mock"} else ("real" if len(refs) == 1 else "mixed") ) self.journal = journal self.player_confirmed = player_confirmed self.monotonic = monotonic self.history: list[dict[str, str]] = [] self._lock = threading.Lock() self._active: tuple[str, threading.Event] | None = None def _event(self, event: str, **facts: Any) -> None: if self.journal is not None: self.journal.append("ai", event, mode=self.mode, **facts) def interrupt(self) -> str | None: with self._lock: active = self._active if active is not None: active[1].set() return active[0] return None def speak(self, text: str, turn_id: str | None = None) -> dict[str, Any]: if not isinstance(text, str) or not text: raise AIConfigError("TEXT_INVALID", "text to speak is required") turn_id = turn_id or f"turn_{uuid.uuid4().hex}" cancelled = threading.Event() return self._synthesize(text, turn_id, cancelled) def _synthesize( self, text: str, turn_id: str, cancelled: threading.Event ) -> dict[str, Any]: audio = bytearray() sent_chunks = 0 discarded = 0 first_audio_ms: float | None = None provider_model: str | None = None try: limit = int(self.config["conversation"]["max_pending_audio_chunks"]) except (KeyError, TypeError, ValueError, OverflowError) as exc: raise AIProviderError( "AI_CONFIG_INVALID", "audio queue limit is invalid" ) from exc pending = 0 for chunk in self.tts.stream(text, self.config, turn_id, cancelled): if chunk.provider_model is not None: provider_model = chunk.provider_model if chunk.late or cancelled.is_set(): discarded += 1 continue pending += 1 if pending > limit: raise AIProviderError( "AI_BACKPRESSURE", "pending audio queue exceeded its bound" ) if chunk.audio: if first_audio_ms is None: first_audio_ms = self.monotonic() * 1000 self._event("tts.first_audio", turn_id=turn_id) audio.extend(chunk.audio) sent_chunks += 1 pending -= 1 state = ( "cancelled" if cancelled.is_set() else ("playback_confirmed" if self.player_confirmed else "sent") ) self._event( "tts.completed", turn_id=turn_id, audio_bytes=len(audio), sent_chunks=sent_chunks, discarded_late_chunks=discarded, playback_state=state, ) return { "audio": bytes(audio), "sent_chunks": sent_chunks, "discarded_late_chunks": discarded, "playback_state": state, "first_audio_ms": first_audio_ms, "provider_model": provider_model, } def run_text( self, text: str, variables: Mapping[str, Any] | None = None ) -> dict[str, Any]: if ( not isinstance(text, str) or not text or len(text.encode("utf-8")) > 16 * 1024 ): raise AIConfigError("TEXT_INVALID", "input text is empty or too large") turn_id = f"turn_{uuid.uuid4().hex}" cancelled = threading.Event() with self._lock: self._active = (turn_id, cancelled) started_ms = self.monotonic() * 1000 reply = "" audio = bytearray() segments: list[dict[str, Any]] = [] discarded = 0 first_token_ms: float | None = None first_audio_ms: float | None = None llm_provider_model: str | None = None tts_provider_model: str | None = None try: prompt = render_prompt( self.config["prompt"]["text"], variables, self.config["prompt"]["allowed_variables"], ) messages = [ {"role": "system", "content": prompt}, *self.history, {"role": "user", "content": text}, ] self._event( "turn.started", turn_id=turn_id, input_sha256=hashlib.sha256(text.encode()).hexdigest(), ) self._event( "asr.final", turn_id=turn_id, text_sha256=hashlib.sha256(text.encode()).hexdigest(), ) pending = "" for chunk in self.llm.stream(messages, self.config, turn_id, cancelled): if chunk.provider_model is not None: llm_provider_model = chunk.provider_model if chunk.late or cancelled.is_set(): discarded += 1 continue if chunk.text: if first_token_ms is None: first_token_ms = self.monotonic() * 1000 self._event("llm.first_token", turn_id=turn_id) reply += chunk.text pending += chunk.text completed, pending = _segments( pending, int(self.config["conversation"]["sentence_max_chars"]) ) for segment in completed: synthesized = self._synthesize(segment, turn_id, cancelled) audio.extend(synthesized["audio"]) tts_provider_model = ( synthesized.get("provider_model") or tts_provider_model ) discarded += int(synthesized["discarded_late_chunks"]) if ( first_audio_ms is None and synthesized["first_audio_ms"] is not None ): first_audio_ms = synthesized["first_audio_ms"] - started_ms segments.append( { "text": segment, "audio_bytes": len(synthesized["audio"]), "playback_state": synthesized["playback_state"], } ) if cancelled.is_set(): self._event( "turn.cancelled", turn_id=turn_id, discarded_late_chunks=discarded ) return { "status": "cancelled", "turn_id": turn_id, "text": reply, "audio": bytes(audio), "segments": segments, "discarded_late_chunks": discarded, "provider_returned_model": llm_provider_model, "tts_provider_returned_model": tts_provider_model, "tts_model_evidence": { "configured_model": self.config["tts"]["model"], "provider_echoed_model": tts_provider_model is not None, }, } if pending: synthesized = self._synthesize(pending, turn_id, cancelled) audio.extend(synthesized["audio"]) tts_provider_model = ( synthesized.get("provider_model") or tts_provider_model ) discarded += int(synthesized["discarded_late_chunks"]) if first_audio_ms is None and synthesized["first_audio_ms"] is not None: first_audio_ms = synthesized["first_audio_ms"] - started_ms segments.append( { "text": pending, "audio_bytes": len(synthesized["audio"]), "playback_state": synthesized["playback_state"], } ) self.history.extend( ( {"role": "user", "content": text}, {"role": "assistant", "content": reply}, ) ) finished_ms = self.monotonic() * 1000 self._event( "turn.completed", turn_id=turn_id, response_sha256=hashlib.sha256(reply.encode()).hexdigest(), audio_bytes=len(audio), discarded_late_chunks=discarded, ) return { "status": "completed", "turn_id": turn_id, "text": reply, "audio": bytes(audio), "segments": segments, "discarded_late_chunks": discarded, "llm_first_token_ms": None if first_token_ms is None else first_token_ms - started_ms, "tts_first_audio_ms": first_audio_ms, "provider_returned_model": llm_provider_model, "tts_provider_returned_model": tts_provider_model, "tts_model_evidence": { "configured_model": self.config["tts"]["model"], "provider_echoed_model": tts_provider_model is not None, }, "duration_ms": max(0.0, finished_ms - started_ms), } except AIProviderError as exc: self._event("turn.failed", turn_id=turn_id, error_code=exc.code) return {"status": "failed", "turn_id": turn_id, "reason_code": exc.code} finally: with self._lock: if self._active is not None and self._active[0] == turn_id: self._active = None def run_audio( self, audio: bytes, variables: Mapping[str, Any] | None = None ) -> dict[str, Any]: turn_id = f"turn_{uuid.uuid4().hex}" segments = list(self.asr.transcribe(audio, self.config, turn_id)) final = next( (item for item in reversed(segments) if item.get("is_final")), None ) if ( not isinstance(final, Mapping) or not isinstance(final.get("text"), str) or not final["text"] ): raise AIProviderError("ASR_NO_FINAL", "ASR did not produce a final segment") result = self.run_text(final["text"], variables) result["asr_segments"] = segments return result def build_mock_config( version_id: str, prompt: str, model: str = "mock-chat-v1", tts_model: str = "mock-tts-v1", voice: str = "mock-neutral", ) -> dict[str, Any]: return normalize_agent_config( { "agent_version_id": version_id, "immutable": True, "llm": { "provider_ref": "mock", "model": model, "temperature": 0.2, "max_tokens": 256, "timeout_ms": 5000, }, "prompt": {"text": prompt, "allowed_variables": []}, "tts": { "provider_ref": "mock", "model": tts_model, "voice": voice, "speed": 1.0, "format": { "encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1, }, "timeout_ms": 5000, }, "asr": { "provider_ref": "mock", "language": "zh-CN", "input": { "encoding": "pcm_s16le", "sample_rate_hz": 16000, "channels": 1, "sample_width_bytes": 2, }, "interim": True, "timeout_ms": 5000, }, "conversation": { "opening": "", "allow_interrupt": True, "silence_timeout_ms": 3000, "max_duration_ms": 120000, "max_turns": 20, "sentence_max_chars": 80, "max_pending_audio_chunks": 32, }, } ) __all__ = [ "AIConfigError", "AIProviderError", "ConversationEngine", "MockASR", "MockLLM", "MockTTS", "StreamChunk", "build_mock_config", "config_digest", "load_prompt", "normalize_agent_config", "pcm16_to_alaw", "pcm_to_wav", "render_prompt", "resample_pcm16", "validate_agent_config", "validate_wav", ]