Fabio
c36510c21e
Some checks failed
Linux Build / Linux cpu (push) Has been cancelled
Linux Build / Linux cpu-clang (push) Has been cancelled
Linux Build / Linux vulkan (push) Has been cancelled
mac-build / macOS CPU (push) Has been cancelled
Nix Build / Nix (cpu) (push) Has been cancelled
Nix Build / Nix (vulkan) (push) Has been cancelled
windows-build / Windows CPU (push) Has been cancelled
988 lines
40 KiB
Python
988 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""Run a true PersonaPlex speech/live multi-turn conversation.
|
|
|
|
Example turns JSON:
|
|
|
|
[
|
|
{
|
|
"label": "turn_01_part_availability",
|
|
"audio": "questions/turn_01.wav",
|
|
"user_question": "My dishwasher stopped draining. Can you tell me if the replacement pump is in stock?",
|
|
"assistant_instruction": "Answer as a helpful appliance repair dispatcher. Be concise and specific."
|
|
},
|
|
{
|
|
"label": "turn_02_delay",
|
|
"audio": "questions/turn_02.wav",
|
|
"user_question": "If the part is not in stock, how long would the alternative take?",
|
|
"assistant_instruction": "Continue the same support call and explain the delay clearly."
|
|
},
|
|
{
|
|
"label": "turn_03_cost",
|
|
"audio": "questions/turn_03.wav",
|
|
"user_question": "What should I expect for labor cost, and can I still book a morning appointment?",
|
|
"assistant_instruction": "Continue the same support call and answer both cost and scheduling."
|
|
}
|
|
]
|
|
|
|
The request body is raw PCM over HTTP/1.1 chunked transfer encoding, and the
|
|
SSE response reader runs concurrently with upload. This validates the live path;
|
|
it is not equivalent to uploading a complete WAV first.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import csv
|
|
import json
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import uuid
|
|
import wave
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Iterator
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
LOG_ROOT = REPO_ROOT / "logs" / "streaming_test"
|
|
DEFAULT_SERVER_BIN = REPO_ROOT / "build" / "debug" / "bin" / "audiocpp_server"
|
|
|
|
|
|
def timestamp_slug() -> str:
|
|
return datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def write_json(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
def path_is_under(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(root.resolve())
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def wav_info(path: Path) -> dict[str, Any]:
|
|
with wave.open(str(path), "rb") as wav:
|
|
channels = wav.getnchannels()
|
|
sample_width = wav.getsampwidth()
|
|
sample_rate = wav.getframerate()
|
|
frames = wav.getnframes()
|
|
return {
|
|
"path": str(path),
|
|
"bytes": path.stat().st_size,
|
|
"channels": channels,
|
|
"sample_width_bytes": sample_width,
|
|
"sample_rate": sample_rate,
|
|
"frames": frames,
|
|
"duration_s": frames / float(sample_rate),
|
|
"audio_bytes_per_second": channels * sample_width * sample_rate,
|
|
}
|
|
|
|
|
|
def load_pcm16_wav(path: Path, max_seconds: float | None) -> tuple[bytes, dict[str, Any]]:
|
|
info = wav_info(path)
|
|
if info["sample_width_bytes"] != 2:
|
|
raise RuntimeError(f"{path}: expected 16-bit PCM WAV, got sample width {info['sample_width_bytes']}")
|
|
if info["channels"] <= 0 or info["sample_rate"] <= 0:
|
|
raise RuntimeError(f"{path}: invalid WAV format")
|
|
with wave.open(str(path), "rb") as wav:
|
|
frames = wav.getnframes()
|
|
if max_seconds is not None:
|
|
frames = min(frames, int(max_seconds * wav.getframerate()))
|
|
pcm = wav.readframes(frames)
|
|
info["streamed_frames"] = frames
|
|
info["streamed_duration_s"] = frames / float(info["sample_rate"])
|
|
info["streamed_pcm_bytes"] = len(pcm)
|
|
return pcm, info
|
|
|
|
|
|
def write_pcm16_wav(path: Path, pcm: bytes, sample_rate: int, channels: int = 1) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with wave.open(str(path), "wb") as wav:
|
|
wav.setnchannels(channels)
|
|
wav.setsampwidth(2)
|
|
wav.setframerate(sample_rate)
|
|
wav.writeframes(pcm)
|
|
|
|
|
|
def analyze_pcm16(pcm: bytes, sample_rate: int) -> dict[str, Any]:
|
|
if not pcm:
|
|
return {
|
|
"duration_s": 0.0,
|
|
"max_peak": 0.0,
|
|
"first_peak_gt_0_01_s": None,
|
|
"last_peak_gt_0_01_s": None,
|
|
"silent_spans_peak_lt_0_01": [],
|
|
}
|
|
sample_count = len(pcm) // 2
|
|
samples = struct.unpack("<" + "h" * sample_count, pcm[:sample_count * 2])
|
|
win = max(1, sample_rate // 10)
|
|
first: float | None = None
|
|
last: float | None = None
|
|
silent_start: float | None = None
|
|
silent_spans: list[tuple[float, float]] = []
|
|
max_peak = 0.0
|
|
for offset in range(0, sample_count, win):
|
|
chunk = samples[offset:offset + win]
|
|
peak = max(abs(value) for value in chunk) / 32768.0 if chunk else 0.0
|
|
max_peak = max(max_peak, peak)
|
|
t = offset / float(sample_rate)
|
|
if peak > 0.01:
|
|
if first is None:
|
|
first = t
|
|
last = t
|
|
if peak < 0.01:
|
|
if silent_start is None:
|
|
silent_start = t
|
|
elif silent_start is not None:
|
|
if t - silent_start >= 0.5:
|
|
silent_spans.append((round(silent_start, 2), round(t, 2)))
|
|
silent_start = None
|
|
duration_s = sample_count / float(sample_rate)
|
|
if silent_start is not None and duration_s - silent_start >= 0.5:
|
|
silent_spans.append((round(silent_start, 2), round(duration_s, 2)))
|
|
return {
|
|
"duration_s": duration_s,
|
|
"max_peak": max_peak,
|
|
"first_peak_gt_0_01_s": first,
|
|
"last_peak_gt_0_01_s": last,
|
|
"silent_spans_peak_lt_0_01": silent_spans,
|
|
}
|
|
|
|
|
|
def read_http_headers(reader: Any) -> tuple[str, dict[str, str]]:
|
|
status = reader.readline().decode("iso-8859-1", "replace").strip()
|
|
headers: dict[str, str] = {}
|
|
while True:
|
|
line = reader.readline().decode("iso-8859-1", "replace")
|
|
if line in ("\r\n", "\n", ""):
|
|
break
|
|
key, _, value = line.partition(":")
|
|
headers[key.strip().lower()] = value.strip()
|
|
return status, headers
|
|
|
|
|
|
def parse_sse_data(buffer: str) -> tuple[list[str], str]:
|
|
events: list[str] = []
|
|
while "\n\n" in buffer:
|
|
block, buffer = buffer.split("\n\n", 1)
|
|
data_lines: list[str] = []
|
|
for line in block.splitlines():
|
|
if line.startswith("data:"):
|
|
data_lines.append(line[5:].strip())
|
|
if data_lines:
|
|
events.append("\n".join(data_lines))
|
|
return events, buffer
|
|
|
|
|
|
def iter_chunked_response(reader: Any) -> Iterator[bytes]:
|
|
while True:
|
|
size_line = reader.readline().decode("ascii", "replace").strip()
|
|
if not size_line:
|
|
continue
|
|
size = int(size_line.split(";", 1)[0], 16)
|
|
if size == 0:
|
|
reader.readline()
|
|
return
|
|
chunk = reader.read(size)
|
|
reader.read(2)
|
|
yield chunk
|
|
|
|
|
|
def wait_for_health(host: str, port: int, timeout_s: float) -> None:
|
|
deadline = time.monotonic() + timeout_s
|
|
last_error: Exception | None = None
|
|
request = (
|
|
f"GET /health HTTP/1.1\r\n"
|
|
f"Host: {host}:{port}\r\n"
|
|
"Connection: close\r\n"
|
|
"\r\n"
|
|
).encode("utf-8")
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
with socket.create_connection((host, port), timeout=1.0) as sock:
|
|
sock.sendall(request)
|
|
response = sock.recv(256)
|
|
if b"200 OK" in response:
|
|
return
|
|
except Exception as exc:
|
|
last_error = exc
|
|
time.sleep(0.5)
|
|
raise RuntimeError(f"server health timeout for {host}:{port}: {last_error}")
|
|
|
|
|
|
def normalize_turns(turns_path: Path, repo_root: Path) -> list[dict[str, Any]]:
|
|
raw = load_json(turns_path)
|
|
if not isinstance(raw, list):
|
|
raise RuntimeError("--turns-json must contain a JSON array")
|
|
if not raw:
|
|
raise RuntimeError("--turns-json must contain at least one turn")
|
|
turns: list[dict[str, Any]] = []
|
|
for index, item in enumerate(raw, 1):
|
|
if not isinstance(item, dict):
|
|
raise RuntimeError(f"turn {index}: expected object")
|
|
label = str(item.get("label") or f"turn_{index:02d}")
|
|
audio_raw = item.get("audio")
|
|
question = str(item.get("user_question") or "").strip()
|
|
instruction = str(item.get("assistant_instruction") or "").strip()
|
|
if not audio_raw:
|
|
raise RuntimeError(f"{label}: missing audio")
|
|
if not question:
|
|
raise RuntimeError(f"{label}: missing user_question")
|
|
if not instruction:
|
|
raise RuntimeError(f"{label}: missing assistant_instruction")
|
|
audio_path = Path(str(audio_raw))
|
|
if not audio_path.is_absolute():
|
|
audio_path = (turns_path.parent / audio_path).resolve()
|
|
if not audio_path.exists():
|
|
audio_path = (repo_root / str(audio_raw)).resolve()
|
|
if not audio_path.exists():
|
|
raise RuntimeError(f"{label}: audio path does not exist: {audio_path}")
|
|
turns.append({
|
|
"label": label,
|
|
"audio": audio_path,
|
|
"user_question": question,
|
|
"assistant_instruction": instruction,
|
|
"expected_answer_points": item.get("expected_answer_points", []),
|
|
})
|
|
return turns
|
|
|
|
|
|
def stream_turn(
|
|
host: str,
|
|
port: int,
|
|
model: str,
|
|
turn: dict[str, Any],
|
|
output_dir: Path,
|
|
audio_chunk_ms: float,
|
|
pace_factor: float,
|
|
max_input_seconds: float | None,
|
|
request_timeout_s: float,
|
|
extra_options: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
pcm_input, input_info = load_pcm16_wav(turn["audio"], max_input_seconds)
|
|
sample_rate = int(input_info["sample_rate"])
|
|
channels = int(input_info["channels"])
|
|
bytes_per_second = max(1, int(input_info["audio_bytes_per_second"]))
|
|
chunk_bytes = max(channels * 2, int(bytes_per_second * audio_chunk_ms / 1000.0))
|
|
frame_bytes = channels * 2
|
|
chunk_bytes = max(frame_bytes, (chunk_bytes // frame_bytes) * frame_bytes)
|
|
|
|
query = {
|
|
"model": model,
|
|
"input": turn["assistant_instruction"],
|
|
"sample_rate": str(sample_rate),
|
|
"channels": str(channels),
|
|
"sample_format": "s16le",
|
|
"stream_format": "sse",
|
|
"response_format": "pcm",
|
|
}
|
|
query.update(extra_options)
|
|
target = "/v1/audio/speech/live?" + urllib.parse.urlencode(query)
|
|
|
|
turn_dir = output_dir / turn["label"]
|
|
turn_dir.mkdir(parents=True, exist_ok=True)
|
|
upload_log_path = turn_dir / "upload_chunks.csv"
|
|
events_path = turn_dir / "events.jsonl"
|
|
response_headers_path = turn_dir / "response_headers.json"
|
|
output_wav = turn_dir / "assistant_response.wav"
|
|
request_manifest_path = turn_dir / "request_manifest.json"
|
|
|
|
request_headers = (
|
|
f"POST {target} HTTP/1.1\r\n"
|
|
f"Host: {host}:{port}\r\n"
|
|
"Content-Type: application/octet-stream\r\n"
|
|
"Accept: text/event-stream\r\n"
|
|
"Transfer-Encoding: chunked\r\n"
|
|
"Connection: close\r\n"
|
|
"Expect:\r\n"
|
|
"\r\n"
|
|
).encode("utf-8")
|
|
|
|
manifest = {
|
|
"label": turn["label"],
|
|
"endpoint": "/v1/audio/speech/live",
|
|
"model": model,
|
|
"user_question": turn["user_question"],
|
|
"assistant_instruction": turn["assistant_instruction"],
|
|
"expected_answer_points": turn["expected_answer_points"],
|
|
"input_audio": input_info,
|
|
"audio_chunk_ms": audio_chunk_ms,
|
|
"audio_chunk_bytes": chunk_bytes,
|
|
"pace_factor": pace_factor,
|
|
"client_input_streaming": "raw PCM sent with HTTP/1.1 Transfer-Encoding: chunked",
|
|
"client_response_reading": "SSE response reader runs concurrently with upload",
|
|
}
|
|
write_json(request_manifest_path, manifest)
|
|
|
|
start = time.perf_counter()
|
|
upload_complete_ms: float | None = None
|
|
first_event_ms: float | None = None
|
|
first_delta_ms: float | None = None
|
|
done_ms: float | None = None
|
|
ttft_ms: float | None = None
|
|
first_audio_before_input_end: bool | None = None
|
|
request_start_to_first_audio_ms: float | None = None
|
|
input_end_ms: float | None = None
|
|
overlap_ms: float | None = None
|
|
delta_events = 0
|
|
output_pcm = bytearray()
|
|
response_headers: dict[str, Any] = {}
|
|
response_events: list[dict[str, Any]] = []
|
|
errors: list[str] = []
|
|
reader_error: list[Exception] = []
|
|
|
|
def read_response(reader: Any) -> None:
|
|
nonlocal first_event_ms, first_delta_ms, done_ms, ttft_ms, first_audio_before_input_end
|
|
nonlocal request_start_to_first_audio_ms, input_end_ms, overlap_ms, delta_events
|
|
try:
|
|
status, headers = read_http_headers(reader)
|
|
response_headers.update({"status": status, "headers": headers})
|
|
write_json(response_headers_path, response_headers)
|
|
if not status.startswith("HTTP/1.1 200"):
|
|
body = reader.read().decode("utf-8", "replace")
|
|
raise RuntimeError(f"{turn['label']}: {status}: {body}")
|
|
|
|
sse_buffer = ""
|
|
with events_path.open("w", encoding="utf-8") as events_handle:
|
|
response_chunks = (
|
|
iter_chunked_response(reader)
|
|
if "chunked" in headers.get("transfer-encoding", "").lower()
|
|
else [reader.read()]
|
|
)
|
|
for raw_chunk in response_chunks:
|
|
if not raw_chunk:
|
|
continue
|
|
sse_buffer += raw_chunk.decode("utf-8", "replace").replace("\r\n", "\n")
|
|
events, sse_buffer = parse_sse_data(sse_buffer)
|
|
for data in events:
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
if first_event_ms is None:
|
|
first_event_ms = elapsed_ms
|
|
if data == "[DONE]":
|
|
done_ms = elapsed_ms
|
|
events_handle.write(json.dumps({"elapsed_ms": elapsed_ms, "data": data}) + "\n")
|
|
events_handle.flush()
|
|
continue
|
|
event = json.loads(data)
|
|
event_type = event.get("type")
|
|
if event_type == "speech.audio.delta":
|
|
delta_events += 1
|
|
if first_delta_ms is None:
|
|
first_delta_ms = elapsed_ms
|
|
output_pcm.extend(base64.b64decode(event["audio"]))
|
|
elif event_type == "speech.audio.done":
|
|
done_ms = elapsed_ms
|
|
timing = event.get("timing", {})
|
|
ttft_ms = timing.get("ttft_ms")
|
|
first_audio_before_input_end = timing.get("first_audio_before_input_end")
|
|
request_start_to_first_audio_ms = timing.get("request_start_to_first_audio_ms")
|
|
input_end_ms = timing.get("input_end_ms")
|
|
overlap_ms = timing.get("overlap_ms")
|
|
if request_start_to_first_audio_ms is None:
|
|
request_start_to_first_audio_ms = ttft_ms
|
|
elif event_type == "error":
|
|
errors.append(event.get("error", {}).get("message", json.dumps(event)))
|
|
response_events.append({"elapsed_ms": elapsed_ms, "event": event})
|
|
events_handle.write(json.dumps(response_events[-1], ensure_ascii=False) + "\n")
|
|
events_handle.flush()
|
|
except Exception as exc:
|
|
reader_error.append(exc)
|
|
|
|
with socket.create_connection((host, port), timeout=request_timeout_s) as sock:
|
|
sock.settimeout(request_timeout_s)
|
|
reader = sock.makefile("rb")
|
|
reader_thread = threading.Thread(target=read_response, args=(reader,), daemon=True)
|
|
reader_thread.start()
|
|
with upload_log_path.open("w", newline="", encoding="utf-8") as upload_handle:
|
|
writer = csv.writer(upload_handle)
|
|
writer.writerow(["index", "elapsed_ms", "bytes", "audio_offset_bytes", "audio_time_end_s"])
|
|
sock.sendall(request_headers)
|
|
offset = 0
|
|
index = 0
|
|
while offset < len(pcm_input):
|
|
end = min(len(pcm_input), offset + chunk_bytes)
|
|
chunk = pcm_input[offset:end]
|
|
sock.sendall(f"{len(chunk):X}\r\n".encode("ascii"))
|
|
sock.sendall(chunk)
|
|
sock.sendall(b"\r\n")
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
writer.writerow([
|
|
index,
|
|
round(elapsed_ms, 3),
|
|
len(chunk),
|
|
end,
|
|
round(end / float(bytes_per_second), 6),
|
|
])
|
|
upload_handle.flush()
|
|
offset = end
|
|
index += 1
|
|
if pace_factor > 0.0 and offset < len(pcm_input):
|
|
time.sleep((audio_chunk_ms / 1000.0) * pace_factor)
|
|
sock.sendall(b"0\r\n\r\n")
|
|
upload_complete_ms = (time.perf_counter() - start) * 1000.0
|
|
|
|
try:
|
|
sock.shutdown(socket.SHUT_WR)
|
|
except OSError:
|
|
pass
|
|
reader_thread.join(timeout=request_timeout_s)
|
|
if reader_thread.is_alive():
|
|
raise RuntimeError(f"{turn['label']}: timed out waiting for live speech response")
|
|
if reader_error:
|
|
raise reader_error[0]
|
|
|
|
if errors:
|
|
raise RuntimeError(f"{turn['label']}: " + "; ".join(errors))
|
|
if request_start_to_first_audio_ms is None:
|
|
raise RuntimeError(f"{turn['label']}: response did not include first-audio timing")
|
|
if ttft_ms is None and first_audio_before_input_end is not True:
|
|
raise RuntimeError(f"{turn['label']}: response did not include live TTFT timing")
|
|
if not output_pcm:
|
|
raise RuntimeError(f"{turn['label']}: response produced no audio")
|
|
|
|
write_pcm16_wav(output_wav, bytes(output_pcm), sample_rate=sample_rate, channels=1)
|
|
audio_stats = analyze_pcm16(bytes(output_pcm), sample_rate)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
result = {
|
|
"label": turn["label"],
|
|
"user_question": turn["user_question"],
|
|
"assistant_instruction": turn["assistant_instruction"],
|
|
"expected_answer_points": turn["expected_answer_points"],
|
|
"elapsed_ms": elapsed_ms,
|
|
"upload_complete_ms": upload_complete_ms,
|
|
"first_event_ms": first_event_ms,
|
|
"first_delta_ms": first_delta_ms,
|
|
"done_ms": done_ms,
|
|
"ttft_ms": ttft_ms,
|
|
"first_audio_before_input_end": first_audio_before_input_end,
|
|
"request_start_to_first_audio_ms": request_start_to_first_audio_ms,
|
|
"input_end_ms": input_end_ms,
|
|
"overlap_ms": overlap_ms,
|
|
"first_event_before_upload_complete": (
|
|
first_event_ms < upload_complete_ms
|
|
if first_event_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"first_delta_before_upload_complete": (
|
|
first_delta_ms < upload_complete_ms
|
|
if first_delta_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"client_observed_ttft_after_upload_ms": (
|
|
first_delta_ms - upload_complete_ms
|
|
if first_delta_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"delta_events": delta_events,
|
|
"input_audio": input_info,
|
|
"output_audio": audio_stats,
|
|
"artifacts": {
|
|
"request_manifest": str(request_manifest_path),
|
|
"upload_chunks": str(upload_log_path),
|
|
"events": str(events_path),
|
|
"response_headers": str(response_headers_path),
|
|
"assistant_response_wav": str(output_wav),
|
|
},
|
|
}
|
|
write_json(turn_dir / "summary.json", result)
|
|
return result
|
|
|
|
|
|
def stream_continuous_conversation(
|
|
host: str,
|
|
port: int,
|
|
model: str,
|
|
turns: list[dict[str, Any]],
|
|
output_dir: Path,
|
|
system_prompt: str,
|
|
audio_chunk_ms: float,
|
|
pace_factor: float,
|
|
turn_gap_seconds: float,
|
|
final_gap_seconds: float,
|
|
max_input_seconds: float | None,
|
|
request_timeout_s: float,
|
|
extra_options: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
if turn_gap_seconds < 0.0:
|
|
raise RuntimeError("--turn-gap-seconds must be non-negative")
|
|
if final_gap_seconds < 0.0:
|
|
raise RuntimeError("--final-gap-seconds must be non-negative")
|
|
|
|
input_parts: list[bytes] = []
|
|
input_infos: list[dict[str, Any]] = []
|
|
sample_rate: int | None = None
|
|
channels: int | None = None
|
|
sample_width = 2
|
|
timeline: list[dict[str, Any]] = []
|
|
cursor_s = 0.0
|
|
for index, turn in enumerate(turns):
|
|
pcm, info = load_pcm16_wav(turn["audio"], max_input_seconds)
|
|
current_sample_rate = int(info["sample_rate"])
|
|
current_channels = int(info["channels"])
|
|
if sample_rate is None:
|
|
sample_rate = current_sample_rate
|
|
channels = current_channels
|
|
if current_sample_rate != sample_rate or current_channels != channels:
|
|
raise RuntimeError("continuous PersonaPlex test requires all turn WAVs to share sample rate and channels")
|
|
input_parts.append(pcm)
|
|
input_infos.append(info)
|
|
duration_s = len(pcm) / float(max(1, current_sample_rate * current_channels * sample_width))
|
|
timeline.append({
|
|
"label": turn["label"],
|
|
"user_question": turn["user_question"],
|
|
"start_s": round(cursor_s, 6),
|
|
"end_s": round(cursor_s + duration_s, 6),
|
|
"audio": info,
|
|
"expected_answer_points": turn["expected_answer_points"],
|
|
})
|
|
cursor_s += duration_s
|
|
if index + 1 < len(turns) and turn_gap_seconds > 0.0:
|
|
silence_frames = int(round(turn_gap_seconds * current_sample_rate))
|
|
silence = b"\x00" * silence_frames * current_channels * sample_width
|
|
input_parts.append(silence)
|
|
timeline.append({
|
|
"label": f"gap_after_{turn['label']}",
|
|
"start_s": round(cursor_s, 6),
|
|
"end_s": round(cursor_s + silence_frames / float(current_sample_rate), 6),
|
|
"silence_s": silence_frames / float(current_sample_rate),
|
|
})
|
|
cursor_s += silence_frames / float(current_sample_rate)
|
|
|
|
if final_gap_seconds > 0.0:
|
|
silence_frames = int(round(final_gap_seconds * sample_rate))
|
|
silence = b"\x00" * silence_frames * channels * sample_width
|
|
input_parts.append(silence)
|
|
timeline.append({
|
|
"label": "final_response_gap",
|
|
"start_s": round(cursor_s, 6),
|
|
"end_s": round(cursor_s + silence_frames / float(sample_rate), 6),
|
|
"silence_s": silence_frames / float(sample_rate),
|
|
})
|
|
cursor_s += silence_frames / float(sample_rate)
|
|
|
|
if sample_rate is None or channels is None:
|
|
raise RuntimeError("continuous PersonaPlex test requires at least one turn")
|
|
|
|
pcm_input = b"".join(input_parts)
|
|
bytes_per_second = max(1, sample_rate * channels * sample_width)
|
|
chunk_bytes = max(channels * sample_width, int(bytes_per_second * audio_chunk_ms / 1000.0))
|
|
frame_bytes = channels * sample_width
|
|
chunk_bytes = max(frame_bytes, (chunk_bytes // frame_bytes) * frame_bytes)
|
|
|
|
query = {
|
|
"model": model,
|
|
"input": system_prompt,
|
|
"sample_rate": str(sample_rate),
|
|
"channels": str(channels),
|
|
"sample_format": "s16le",
|
|
"stream_format": "sse",
|
|
"response_format": "pcm",
|
|
}
|
|
query.update(extra_options)
|
|
target = "/v1/audio/speech/live?" + urllib.parse.urlencode(query)
|
|
|
|
conversation_dir = output_dir / "continuous"
|
|
conversation_dir.mkdir(parents=True, exist_ok=True)
|
|
upload_log_path = conversation_dir / "upload_chunks.csv"
|
|
events_path = conversation_dir / "events.jsonl"
|
|
response_headers_path = conversation_dir / "response_headers.json"
|
|
output_wav = conversation_dir / "assistant_response.wav"
|
|
request_manifest_path = conversation_dir / "request_manifest.json"
|
|
input_wav = conversation_dir / "user_input_stream.wav"
|
|
write_pcm16_wav(input_wav, pcm_input, sample_rate=sample_rate, channels=channels)
|
|
|
|
request_headers = (
|
|
f"POST {target} HTTP/1.1\r\n"
|
|
f"Host: {host}:{port}\r\n"
|
|
"Content-Type: application/octet-stream\r\n"
|
|
"Accept: text/event-stream\r\n"
|
|
"Transfer-Encoding: chunked\r\n"
|
|
"Connection: close\r\n"
|
|
"Expect:\r\n"
|
|
"\r\n"
|
|
).encode("utf-8")
|
|
|
|
manifest = {
|
|
"endpoint": "/v1/audio/speech/live",
|
|
"model": model,
|
|
"system_prompt": system_prompt,
|
|
"input_wav": str(input_wav),
|
|
"input_duration_s": len(pcm_input) / float(bytes_per_second),
|
|
"timeline": timeline,
|
|
"audio_chunk_ms": audio_chunk_ms,
|
|
"audio_chunk_bytes": chunk_bytes,
|
|
"pace_factor": pace_factor,
|
|
"turn_gap_seconds": turn_gap_seconds,
|
|
"final_gap_seconds": final_gap_seconds,
|
|
"client_input_streaming": "all user turns sent over one HTTP/1.1 chunked live request",
|
|
"client_response_reading": "one SSE response reader runs concurrently with upload",
|
|
}
|
|
write_json(request_manifest_path, manifest)
|
|
|
|
start = time.perf_counter()
|
|
upload_complete_ms: float | None = None
|
|
first_event_ms: float | None = None
|
|
first_delta_ms: float | None = None
|
|
done_ms: float | None = None
|
|
ttft_ms: float | None = None
|
|
first_audio_before_input_end: bool | None = None
|
|
request_start_to_first_audio_ms: float | None = None
|
|
input_end_ms: float | None = None
|
|
overlap_ms: float | None = None
|
|
delta_events = 0
|
|
output_pcm = bytearray()
|
|
response_headers: dict[str, Any] = {}
|
|
response_events: list[dict[str, Any]] = []
|
|
errors: list[str] = []
|
|
reader_error: list[Exception] = []
|
|
|
|
def read_response(reader: Any) -> None:
|
|
nonlocal first_event_ms, first_delta_ms, done_ms, ttft_ms, first_audio_before_input_end
|
|
nonlocal request_start_to_first_audio_ms, input_end_ms, overlap_ms, delta_events
|
|
try:
|
|
status, headers = read_http_headers(reader)
|
|
response_headers.update({"status": status, "headers": headers})
|
|
write_json(response_headers_path, response_headers)
|
|
if not status.startswith("HTTP/1.1 200"):
|
|
body = reader.read().decode("utf-8", "replace")
|
|
raise RuntimeError(f"continuous: {status}: {body}")
|
|
|
|
sse_buffer = ""
|
|
with events_path.open("w", encoding="utf-8") as events_handle:
|
|
response_chunks = (
|
|
iter_chunked_response(reader)
|
|
if "chunked" in headers.get("transfer-encoding", "").lower()
|
|
else [reader.read()]
|
|
)
|
|
for raw_chunk in response_chunks:
|
|
if not raw_chunk:
|
|
continue
|
|
sse_buffer += raw_chunk.decode("utf-8", "replace").replace("\r\n", "\n")
|
|
events, sse_buffer = parse_sse_data(sse_buffer)
|
|
for data in events:
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
if first_event_ms is None:
|
|
first_event_ms = elapsed_ms
|
|
if data == "[DONE]":
|
|
done_ms = elapsed_ms
|
|
events_handle.write(json.dumps({"elapsed_ms": elapsed_ms, "data": data}) + "\n")
|
|
events_handle.flush()
|
|
continue
|
|
event = json.loads(data)
|
|
event_type = event.get("type")
|
|
if event_type == "speech.audio.delta":
|
|
delta_events += 1
|
|
if first_delta_ms is None:
|
|
first_delta_ms = elapsed_ms
|
|
output_pcm.extend(base64.b64decode(event["audio"]))
|
|
elif event_type == "speech.audio.done":
|
|
done_ms = elapsed_ms
|
|
timing = event.get("timing", {})
|
|
ttft_ms = timing.get("ttft_ms")
|
|
first_audio_before_input_end = timing.get("first_audio_before_input_end")
|
|
request_start_to_first_audio_ms = timing.get("request_start_to_first_audio_ms")
|
|
input_end_ms = timing.get("input_end_ms")
|
|
overlap_ms = timing.get("overlap_ms")
|
|
if request_start_to_first_audio_ms is None:
|
|
request_start_to_first_audio_ms = ttft_ms
|
|
elif event_type == "error":
|
|
errors.append(event.get("error", {}).get("message", json.dumps(event)))
|
|
response_events.append({"elapsed_ms": elapsed_ms, "event": event})
|
|
events_handle.write(json.dumps(response_events[-1], ensure_ascii=False) + "\n")
|
|
events_handle.flush()
|
|
except Exception as exc:
|
|
reader_error.append(exc)
|
|
|
|
with socket.create_connection((host, port), timeout=request_timeout_s) as sock:
|
|
sock.settimeout(request_timeout_s)
|
|
reader = sock.makefile("rb")
|
|
reader_thread = threading.Thread(target=read_response, args=(reader,), daemon=True)
|
|
reader_thread.start()
|
|
with upload_log_path.open("w", newline="", encoding="utf-8") as upload_handle:
|
|
writer = csv.writer(upload_handle)
|
|
writer.writerow(["index", "elapsed_ms", "bytes", "audio_offset_bytes", "audio_time_end_s"])
|
|
sock.sendall(request_headers)
|
|
offset = 0
|
|
index = 0
|
|
while offset < len(pcm_input):
|
|
end = min(len(pcm_input), offset + chunk_bytes)
|
|
chunk = pcm_input[offset:end]
|
|
sock.sendall(f"{len(chunk):X}\r\n".encode("ascii"))
|
|
sock.sendall(chunk)
|
|
sock.sendall(b"\r\n")
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
writer.writerow([
|
|
index,
|
|
round(elapsed_ms, 3),
|
|
len(chunk),
|
|
end,
|
|
round(end / float(bytes_per_second), 6),
|
|
])
|
|
upload_handle.flush()
|
|
offset = end
|
|
index += 1
|
|
if pace_factor > 0.0 and offset < len(pcm_input):
|
|
time.sleep((audio_chunk_ms / 1000.0) * pace_factor)
|
|
sock.sendall(b"0\r\n\r\n")
|
|
upload_complete_ms = (time.perf_counter() - start) * 1000.0
|
|
|
|
try:
|
|
sock.shutdown(socket.SHUT_WR)
|
|
except OSError:
|
|
pass
|
|
reader_thread.join(timeout=request_timeout_s)
|
|
if reader_thread.is_alive():
|
|
raise RuntimeError("continuous: timed out waiting for live speech response")
|
|
if reader_error:
|
|
raise reader_error[0]
|
|
|
|
if errors:
|
|
raise RuntimeError("continuous: " + "; ".join(errors))
|
|
if request_start_to_first_audio_ms is None:
|
|
raise RuntimeError("continuous: response did not include first-audio timing")
|
|
if ttft_ms is None and first_audio_before_input_end is not True:
|
|
raise RuntimeError("continuous: response did not include live TTFT timing")
|
|
if not output_pcm:
|
|
raise RuntimeError("continuous: response produced no audio")
|
|
|
|
write_pcm16_wav(output_wav, bytes(output_pcm), sample_rate=sample_rate, channels=1)
|
|
audio_stats = analyze_pcm16(bytes(output_pcm), sample_rate)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000.0
|
|
result = {
|
|
"label": "continuous",
|
|
"elapsed_ms": elapsed_ms,
|
|
"upload_complete_ms": upload_complete_ms,
|
|
"first_event_ms": first_event_ms,
|
|
"first_delta_ms": first_delta_ms,
|
|
"done_ms": done_ms,
|
|
"ttft_ms": ttft_ms,
|
|
"first_audio_before_input_end": first_audio_before_input_end,
|
|
"request_start_to_first_audio_ms": request_start_to_first_audio_ms,
|
|
"input_end_ms": input_end_ms,
|
|
"overlap_ms": overlap_ms,
|
|
"first_event_before_upload_complete": (
|
|
first_event_ms < upload_complete_ms
|
|
if first_event_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"first_delta_before_upload_complete": (
|
|
first_delta_ms < upload_complete_ms
|
|
if first_delta_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"client_observed_ttft_after_upload_ms": (
|
|
first_delta_ms - upload_complete_ms
|
|
if first_delta_ms is not None and upload_complete_ms is not None
|
|
else None
|
|
),
|
|
"delta_events": delta_events,
|
|
"input_audio": {
|
|
"sample_rate": sample_rate,
|
|
"channels": channels,
|
|
"streamed_duration_s": len(pcm_input) / float(bytes_per_second),
|
|
"streamed_pcm_bytes": len(pcm_input),
|
|
"turns": input_infos,
|
|
},
|
|
"output_audio": audio_stats,
|
|
"timeline": timeline,
|
|
"artifacts": {
|
|
"request_manifest": str(request_manifest_path),
|
|
"input_wav": str(input_wav),
|
|
"upload_chunks": str(upload_log_path),
|
|
"events": str(events_path),
|
|
"response_headers": str(response_headers_path),
|
|
"assistant_response_wav": str(output_wav),
|
|
},
|
|
}
|
|
write_json(conversation_dir / "summary.json", result)
|
|
return result
|
|
|
|
|
|
def run_client(args: argparse.Namespace) -> dict[str, Any]:
|
|
output_dir = args.output_dir
|
|
if not path_is_under(output_dir, LOG_ROOT):
|
|
raise RuntimeError(f"--output-dir must be under {LOG_ROOT}")
|
|
if args.audio_chunk_ms <= 0.0:
|
|
raise RuntimeError("--audio-chunk-ms must be positive")
|
|
if args.pace_factor < 0.0:
|
|
raise RuntimeError("--pace-factor must be non-negative")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
server_config = load_json(args.server_config)
|
|
host = args.host or server_config.get("host", "127.0.0.1")
|
|
port = int(args.port or server_config["port"])
|
|
models = server_config.get("models") or []
|
|
model = args.model or (models[0]["id"] if models else "")
|
|
if not model:
|
|
raise RuntimeError("--model is required when server config has no models")
|
|
write_json(output_dir / "server_config_snapshot.json", server_config)
|
|
|
|
option_pairs: dict[str, str] = {}
|
|
for raw in args.request_option:
|
|
key, sep, value = raw.partition("=")
|
|
if not sep or not key:
|
|
raise RuntimeError(f"--request-option must be key=value, got: {raw}")
|
|
option_pairs[key] = value
|
|
|
|
turns = normalize_turns(args.turns_json, args.repo_root)
|
|
server: subprocess.Popen[str] | None = None
|
|
server_log_handle = None
|
|
server_command: list[str] | None = None
|
|
try:
|
|
if args.start_server:
|
|
server_command = [
|
|
str(args.server_bin),
|
|
"--config",
|
|
str(args.server_config),
|
|
"--log-file",
|
|
str(output_dir / "framework.log"),
|
|
]
|
|
server_log_handle = (output_dir / "server_stdout.log").open("w", encoding="utf-8")
|
|
server = subprocess.Popen(
|
|
server_command,
|
|
cwd=args.repo_root,
|
|
stdout=server_log_handle,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
wait_for_health(str(host), port, args.health_timeout_s)
|
|
|
|
effective_system_prompt = args.system_prompt
|
|
if not effective_system_prompt:
|
|
effective_system_prompt = turns[0]["assistant_instruction"]
|
|
results = [stream_continuous_conversation(
|
|
str(host),
|
|
port,
|
|
model,
|
|
turns,
|
|
output_dir,
|
|
effective_system_prompt,
|
|
args.audio_chunk_ms,
|
|
args.pace_factor,
|
|
args.turn_gap_seconds,
|
|
args.final_gap_seconds,
|
|
args.max_input_seconds,
|
|
args.request_timeout_s,
|
|
option_pairs,
|
|
)]
|
|
|
|
summary = {
|
|
"measurement_kind": "personaplex_true_speech_live_continuous_multiturn",
|
|
"repo": str(args.repo_root),
|
|
"server_command": server_command,
|
|
"server_config": str(args.server_config),
|
|
"turns_json": str(args.turns_json),
|
|
"model": model,
|
|
"host": host,
|
|
"port": port,
|
|
"system_prompt": effective_system_prompt,
|
|
"audio_chunk_ms": args.audio_chunk_ms,
|
|
"pace_factor": args.pace_factor,
|
|
"turn_gap_seconds": args.turn_gap_seconds,
|
|
"final_gap_seconds": args.final_gap_seconds,
|
|
"max_input_seconds": args.max_input_seconds,
|
|
"request_options": option_pairs,
|
|
"turns": results,
|
|
"artifacts": {
|
|
"summary": str(output_dir / "summary.json"),
|
|
"server_config_snapshot": str(output_dir / "server_config_snapshot.json"),
|
|
"framework_log": str(output_dir / "framework.log") if args.start_server else None,
|
|
"server_stdout": str(output_dir / "server_stdout.log") if args.start_server else None,
|
|
},
|
|
}
|
|
write_json(output_dir / "summary.json", summary)
|
|
return summary
|
|
finally:
|
|
if server is not None:
|
|
server.terminate()
|
|
try:
|
|
server.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
server.kill()
|
|
server.wait(timeout=10)
|
|
if server_log_handle is not None:
|
|
server_log_handle.close()
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
default_output = LOG_ROOT / f"personaplex_live_multiturn_{timestamp_slug()}"
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repo-root", type=Path, default=REPO_ROOT)
|
|
parser.add_argument("--server-bin", type=Path, default=DEFAULT_SERVER_BIN)
|
|
parser.add_argument("--server-config", type=Path, required=True)
|
|
parser.add_argument("--turns-json", type=Path, required=True)
|
|
parser.add_argument("--start-server", action=argparse.BooleanOptionalAction, default=True)
|
|
parser.add_argument("--host", default="")
|
|
parser.add_argument("--port", type=int, default=0)
|
|
parser.add_argument("--model", default="")
|
|
parser.add_argument("--audio-chunk-ms", type=float, default=200.0)
|
|
parser.add_argument("--turn-gap-seconds", type=float, default=1.2)
|
|
parser.add_argument("--final-gap-seconds", type=float, default=6.0)
|
|
parser.add_argument(
|
|
"--system-prompt",
|
|
default="",
|
|
)
|
|
parser.add_argument(
|
|
"--pace-factor",
|
|
type=float,
|
|
default=1.0,
|
|
help="sleep audio_chunk_ms * pace_factor between chunks; 1.0 approximates real time",
|
|
)
|
|
parser.add_argument("--max-input-seconds", type=float)
|
|
parser.add_argument("--request-option", action="append", default=[])
|
|
parser.add_argument("--output-dir", type=Path, default=default_output)
|
|
parser.add_argument("--health-timeout-s", type=float, default=180.0)
|
|
parser.add_argument("--request-timeout-s", type=float, default=1800.0)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
summary = run_client(parse_args())
|
|
except Exception as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps({
|
|
"output_dir": summary["artifacts"]["summary"].rsplit("/", 1)[0],
|
|
"model": summary["model"],
|
|
"turns": [
|
|
{
|
|
"label": turn["label"],
|
|
"input_duration_s": turn["input_audio"]["streamed_duration_s"],
|
|
"output_duration_s": turn["output_audio"]["duration_s"],
|
|
"ttft_ms": turn["ttft_ms"],
|
|
"first_audio_before_input_end": turn["first_audio_before_input_end"],
|
|
"request_start_to_first_audio_ms": turn["request_start_to_first_audio_ms"],
|
|
"overlap_ms": turn["overlap_ms"],
|
|
"first_delta_ms": turn["first_delta_ms"],
|
|
"upload_complete_ms": turn["upload_complete_ms"],
|
|
"first_delta_before_upload_complete": turn["first_delta_before_upload_complete"],
|
|
"delta_events": turn["delta_events"],
|
|
"silent_spans": turn["output_audio"]["silent_spans_peak_lt_0_01"],
|
|
"wav": turn["artifacts"]["assistant_response_wav"],
|
|
"input_wav": turn["artifacts"].get("input_wav"),
|
|
}
|
|
for turn in summary["turns"]
|
|
],
|
|
}, indent=2, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|