远程连接服务测试

code/voice_chat_server.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import sys
+import uuid
+from copy import deepcopy
+from pathlib import Path
+
+from b1_agent_runtime import run_agent
+from common.io_utils import read_json, write_json
+from common.path_utils import PROJECT_ROOT, bootstrap_project_root, resolve_from_file
+
+bootstrap_project_root()
+
+from skills.text_to_speech import shutdown_tts_worker
+
+
+RESPONSE_PREFIX = "__VOICE_CHAT_RESPONSE__"
+
+
+def _response(payload: dict) -> None:
+    encoded = base64.b64encode(json.dumps(payload, ensure_ascii=False).encode("utf-8")).decode("ascii")
+    print(RESPONSE_PREFIX + encoded, flush=True)
+
+
+def _save_audio(message: dict, session_dir: Path, turn: int) -> Path:
+    if isinstance(message.get("audio_path"), str):
+        path = Path(message["audio_path"]).expanduser().resolve()
+        if not path.is_file():
+            raise ValueError(f"audio file not found: {path}")
+        return path
+    encoded = message.get("audio_b64")
+    if not isinstance(encoded, str) or not encoded:
+        raise ValueError("request requires audio_b64 or audio_path")
+    suffix = str(message.get("suffix", ".wav")).lower()
+    if suffix not in {".wav", ".flac", ".mp3", ".m4a", ".ogg", ".webm"}:
+        raise ValueError(f"unsupported audio suffix: {suffix}")
+    path = session_dir / f"input_{turn:03d}{suffix}"
+    path.write_bytes(base64.b64decode(encoded, validate=True))
+    return path
+
+
+def serve(args: argparse.Namespace) -> int:
+    template_path = Path(args.template).resolve()
+    template = read_json(template_path)
+    session_id = args.session_id or uuid.uuid4().hex[:12]
+    session_dir = (PROJECT_ROOT / "data" / "voice_sessions" / session_id).resolve()
+    output_dir = (PROJECT_ROOT / "outputs" / "voice_sessions" / session_id).resolve()
+    session_dir.mkdir(parents=True, exist_ok=True)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    memory_id = None
+    turn = 1
+    tools_config = args.tools_config or str(PROJECT_ROOT / "configs" / "tools.yaml")
+    memory_config = args.memory_config or str(PROJECT_ROOT / "configs" / "memory.yaml")
+    model_config = args.model_config or str(PROJECT_ROOT / "configs" / "model.yaml")
+    print(f"[VoiceChat] session={session_id}; waiting for audio", file=sys.stderr, flush=True)
+
+    try:
+        for line in sys.stdin:
+            if not line.strip():
+                continue
+            try:
+                message = json.loads(line)
+                if message.get("command") in {"quit", "exit", "shutdown"}:
+                    _response({"status": "bye", "session_id": session_id})
+                    return 0
+                audio_path = _save_audio(message, session_dir, turn)
+                runtime = deepcopy(template)
+                runtime.update({
+                    "execution_mode": "integrated",
+                    "conversation_id": f"{session_id}_turn_{turn:03d}",
+                    "input_type": "audio",
+                    "audio_path": str(audio_path),
+                    "audio_language": message.get("language") or args.language,
+                    "toolset": "multimodal_tools",
+                    "max_turns": max(3, int(runtime.get("max_turns", 6))),
+                    "save_memory": "conversation",
+                    "enable_tts": True,
+                    "tts_language": args.tts_language,
+                    "selected_memory_ids": [memory_id] if memory_id else [],
+                    "use_global_memory": False,
+                    "system_prompt_path": str(resolve_from_file(template["system_prompt_path"], template_path)),
+                })
+                runtime_path = session_dir / f"runtime_{turn:03d}.json"
+                turn_dir = output_dir / f"turn_{turn:03d}"
+                write_json(runtime, runtime_path)
+                result = run_agent(
+                    str(runtime_path), tools_config, memory_config,
+                    model_config, str(turn_dir), args.llm_mode,
+                )
+                saved_path = turn_dir / "saved_memory.json"
+                if saved_path.is_file():
+                    memory_id = read_json(saved_path).get("memory_id") or memory_id
+                audio_output = result.get("final_audio_path")
+                trace = read_json(result["trace_path"])
+                audio_b64 = base64.b64encode(Path(audio_output).read_bytes()).decode("ascii") if audio_output else None
+                _response({
+                    "status": result.get("status", "success"), "session_id": session_id,
+                    "turn": turn, "transcript": (trace.get("input_audio") or {}).get("transcript"),
+                    "answer": result.get("final_answer", ""), "audio_b64": audio_b64,
+                    "audio_suffix": Path(audio_output).suffix if audio_output else None,
+                    "tts": result.get("tts_output"), "output_dir": str(turn_dir),
+                })
+            except Exception as exc:
+                _response({"status": "error", "turn": turn, "error": f"{type(exc).__name__}: {exc}"})
+            turn += 1
+    finally:
+        shutdown_tts_worker()
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Persistent stdin/stdout voice-chat server for an SSH session.")
+    parser.add_argument("--template", required=True, help="Base runtime JSON")
+    parser.add_argument("--tools-config", default=None)
+    parser.add_argument("--memory-config", default=None)
+    parser.add_argument("--model-config", default=None)
+    parser.add_argument("--llm-mode", choices=["mock", "prompt_json"], default=None)
+    parser.add_argument("--language", default="zh")
+    parser.add_argument("--tts-language", choices=["auto", "zh", "en"], default="auto")
+    parser.add_argument("--session-id", default=None)
+    return parser
+
+
+if __name__ == "__main__":
+    raise SystemExit(serve(build_parser().parse_args()))

code/voice_chat_windows_client.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import argparse
+import base64
+import io
+import json
+import shlex
+import subprocess
+import sys
+import threading
+
+
+PREFIX = "__VOICE_CHAT_RESPONSE__"
+
+
+def record_wav(sample_rate: int) -> bytes:
+    try:
+        import sounddevice as sd
+        import soundfile as sf
+    except ImportError as exc:
+        raise RuntimeError("install client audio packages: py -m pip install sounddevice soundfile") from exc
+    chunks = []
+    def callback(indata, frames, time_info, status):
+        if status:
+            print(status, file=sys.stderr)
+        chunks.append(indata.copy())
+    input("按 Enter 开始录音...")
+    with sd.InputStream(samplerate=sample_rate, channels=1, dtype="float32", callback=callback):
+        input("正在录音；再次按 Enter 结束...")
+    if not chunks:
+        raise RuntimeError("没有录到音频")
+    import numpy as np
+    buffer = io.BytesIO()
+    sf.write(buffer, np.concatenate(chunks, axis=0), sample_rate, format="WAV", subtype="PCM_16")
+    return buffer.getvalue()
+
+
+def play_wav(data: bytes) -> None:
+    import sounddevice as sd
+    import soundfile as sf
+    audio, rate = sf.read(io.BytesIO(data), dtype="float32")
+    sd.play(audio, rate)
+    sd.wait()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Windows microphone/speaker bridge to the SSH voice-chat server.")
+    parser.add_argument("--host", required=True, help="SSH target, e.g. user@server")
+    parser.add_argument("--port", type=int, default=22, help="SSH port")
+    parser.add_argument("--remote-root", required=True, help="assignment_B absolute path on server")
+    parser.add_argument("--python", default="python")
+    parser.add_argument("--template", default="agent/data/runtime_input_audio.json")
+    parser.add_argument("--language", default="zh")
+    parser.add_argument("--sample-rate", type=int, default=16000)
+    args = parser.parse_args()
+    remote = (
+        f"cd {shlex.quote(args.remote_root)} && {shlex.quote(args.python)} "
+        f"agent/code/voice_chat_server.py --template {shlex.quote(args.template)} "
+        f"--language {shlex.quote(args.language)}"
+    )
+    process = subprocess.Popen(
+        ["ssh", "-T", "-p", str(args.port), args.host, remote],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1,
+    )
+
+    def show_stderr() -> None:
+        if process.stderr:
+            for line in process.stderr:
+                print(line, file=sys.stderr, end="")
+    threading.Thread(target=show_stderr, daemon=True).start()
+
+    try:
+        while process.poll() is None:
+            wav = record_wav(args.sample_rate)
+            process.stdin.write(json.dumps({"audio_b64": base64.b64encode(wav).decode("ascii"), "suffix": ".wav", "language": args.language}) + "\n")
+            process.stdin.flush()
+            for line in process.stdout:
+                if not line.startswith(PREFIX):
+                    print("[server]", line.rstrip())
+                    continue
+                response = json.loads(base64.b64decode(line[len(PREFIX):]).decode("utf-8"))
+                if response.get("status") == "error":
+                    print("错误:", response.get("error"))
+                else:
+                    print("你:", response.get("transcript") or "(转写见服务端输出)")
+                    print("Agent:", response.get("answer", ""))
+                    if response.get("audio_b64"):
+                        play_wav(base64.b64decode(response["audio_b64"]))
+                break
+            if input("继续对话？[Enter/q] ").strip().lower() in {"q", "quit", "exit"}:
+                break
+    finally:
+        if process.poll() is None:
+            try:
+                process.stdin.write('{"command":"quit"}\n')
+                process.stdin.flush()
+                process.wait(timeout=10)
+            except Exception:
+                process.terminate()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/text_to_speech.py

@@ -6,6 +6,8 @@ import re
 import subprocess
 import sys
 import uuid
+import atexit
+import threading
 from pathlib import Path
 from time import perf_counter
 from typing import Any
@@ -18,6 +20,9 @@ from common.path_utils import PROJECT_ROOT
 TTS_CONFIG_PATH = PROJECT_ROOT / "configs" / "tts.yaml"
 TTS_WORKER_PATH = Path(__file__).resolve().with_name("tts_worker.py")
 SUPPORTED_LANGUAGES = {"auto", "zh", "en"}
+_WORKER_LOCK = threading.Lock()
+_WORKER_PROCESS: subprocess.Popen[str] | None = None
+_WORKER_KEY: tuple[str, str] | None = None
 
 
 def _require(condition: bool, code: str, message: str) -> None:
@@ -35,8 +40,8 @@ def _resolve_setting(
     default: str | None = None,
 ) -> str | None:
     return (
-        _nonempty_string(configured)
-        or _nonempty_string(os.environ.get(environment_name))
+        _nonempty_string(os.environ.get(environment_name))
+        or _nonempty_string(configured)
         or default
     )
 
@@ -114,6 +119,9 @@ def _load_tts_settings() -> dict[str, Any]:
     max_total_chars = runtime.get("max_total_chars", 1200)
     max_chunk_chars = runtime.get("max_chunk_chars", 120)
     silence_sec = runtime.get("inter_chunk_silence_sec", 0.18)
+    persistent_worker = runtime.get("persistent_worker", True)
+    load_jit = runtime.get("load_jit", False)
+    fp16 = runtime.get("fp16", False)
 
     _require(
         isinstance(timeout_sec, int) and not isinstance(timeout_sec, bool) and 30 <= timeout_sec <= 3600,
@@ -135,6 +143,9 @@ def _load_tts_settings() -> dict[str, Any]:
         "TTS-1003",
         "tts.yaml.runtime.inter_chunk_silence_sec must be between 0 and 3",
     )
+    _require(isinstance(persistent_worker, bool), "TTS-1003", "tts.yaml.runtime.persistent_worker must be boolean")
+    _require(isinstance(load_jit, bool), "TTS-1003", "tts.yaml.runtime.load_jit must be boolean")
+    _require(isinstance(fp16, bool), "TTS-1003", "tts.yaml.runtime.fp16 must be boolean")
 
     return {
         "python_executable": _resolve_config_path(python_executable, executable=True),
@@ -146,9 +157,65 @@ def _load_tts_settings() -> dict[str, Any]:
         "max_total_chars": max_total_chars,
         "max_chunk_chars": max_chunk_chars,
         "inter_chunk_silence_sec": float(silence_sec),
+        "persistent_worker": persistent_worker,
+        "load_jit": load_jit,
+        "fp16": fp16,
     }
 
 
+def shutdown_tts_worker() -> None:
+    global _WORKER_PROCESS, _WORKER_KEY
+    with _WORKER_LOCK:
+        process, _WORKER_PROCESS, _WORKER_KEY = _WORKER_PROCESS, None, None
+        if process is not None and process.poll() is None:
+            try:
+                process.stdin.write('{"command":"shutdown","id":"shutdown"}\n')
+                process.stdin.flush()
+                process.wait(timeout=5)
+            except Exception:
+                process.terminate()
+
+
+atexit.register(shutdown_tts_worker)
+
+
+def _get_persistent_worker(settings: dict[str, Any], env: dict[str, str]) -> subprocess.Popen[str]:
+    global _WORKER_PROCESS, _WORKER_KEY
+    key = (settings["python_executable"], settings["model_dir"])
+    if _WORKER_PROCESS is not None and _WORKER_PROCESS.poll() is None and _WORKER_KEY == key:
+        return _WORKER_PROCESS
+    if _WORKER_PROCESS is not None and _WORKER_PROCESS.poll() is None:
+        _WORKER_PROCESS.terminate()
+    process = subprocess.Popen(
+        [settings["python_executable"], str(TTS_WORKER_PATH), "--serve"],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=None,
+        text=True, bufsize=1, env=env,
+    )
+    while True:
+        line = process.stdout.readline()
+        if not line:
+            raise SkillError("TTS-1007", "persistent CosyVoice worker exited during startup")
+        if line.strip() == "__TTS_WORKER_READY__":
+            break
+    _WORKER_PROCESS, _WORKER_KEY = process, key
+    return process
+
+
+def _run_persistent(settings: dict[str, Any], env: dict[str, str], request_path: Path, response_path: Path) -> int:
+    with _WORKER_LOCK:
+        process = _get_persistent_worker(settings, env)
+        request_id = uuid.uuid4().hex
+        process.stdin.write(json.dumps({"id": request_id, "request": str(request_path), "response": str(response_path)}) + "\n")
+        process.stdin.flush()
+        marker = f"__TTS_WORKER_DONE__:{request_id}:"
+        while True:
+            line = process.stdout.readline()
+            if not line:
+                raise SkillError("TTS-1007", "persistent CosyVoice worker exited during synthesis")
+            if line.strip().startswith(marker):
+                return int(line.strip().rsplit(":", 1)[1])
+
+
 def _detect_language(text: str) -> str:
     # 对最终回答使用保守检测：只要出现中文字符，就选中文 zero-shot 音色合成。
     return "zh" if re.search(r"[\u4e00-\u9fff]", text) else "en"
@@ -206,13 +273,13 @@ def _safe_output_path(output_dir: str | None, output_filename: str | None) -> Pa
     return candidate
 
 
-def _worker_error_message(result: dict[str, Any], completed: subprocess.CompletedProcess[str]) -> str:
+def _worker_error_message(result: dict[str, Any], completed: subprocess.CompletedProcess[str] | None) -> str:
     worker_error = result.get("error") if isinstance(result, dict) else None
     if isinstance(worker_error, dict):
         detail = worker_error.get("message")
         if isinstance(detail, str) and detail.strip():
             return detail.strip()
-    stderr = (completed.stderr or "").strip()
+    stderr = (completed.stderr or "").strip() if completed is not None else ""
     return stderr[-2000:] if stderr else "CosyVoice worker failed without an error message"
 
 
@@ -260,6 +327,8 @@ def text_to_speech(
             "prompt_wav": settings["prompt_wav"],
             "prompt_text": settings["prompt_text"],
             "inter_chunk_silence_sec": settings["inter_chunk_silence_sec"],
+            "load_jit": settings["load_jit"],
+            "fp16": settings["fp16"],
         },
     }
 
@@ -276,21 +345,16 @@ def text_to_speech(
     started = perf_counter()
     try:
         try:
-            completed = subprocess.run(
-                [
-                    settings["python_executable"],
-                    str(TTS_WORKER_PATH),
-                    "--request",
-                    str(request_path),
-                    "--response",
-                    str(response_path),
-                ],
-                capture_output=True,
-                text=True,
-                env=env,
-                timeout=settings["timeout_sec"],
-                check=False,
-            )
+            if settings["persistent_worker"]:
+                completed = None
+                returncode = _run_persistent(settings, env, request_path, response_path)
+            else:
+                completed = subprocess.run(
+                    [settings["python_executable"], str(TTS_WORKER_PATH), "--request", str(request_path), "--response", str(response_path)],
+                    capture_output=True, text=True, env=env,
+                    timeout=settings["timeout_sec"], check=False,
+                )
+                returncode = completed.returncode
         except FileNotFoundError as exc:
             raise SkillError(
                 "TTS-1004",
@@ -310,7 +374,7 @@ def text_to_speech(
         response = read_json(response_path)
         _require(isinstance(response, dict), "TTS-1007", "CosyVoice worker response must be an object")
 
-        if completed.returncode != 0 or response.get("status") != "success":
+        if returncode != 0 or response.get("status") != "success":
             raise SkillError("TTS-1006", _worker_error_message(response, completed))
 
         output = response.get("output")

skills/tts_worker.py

@@ -4,6 +4,7 @@ import argparse
 import json
 import sys
 import traceback
+import uuid
 from pathlib import Path
 from time import perf_counter
 from typing import Any
@@ -19,6 +20,27 @@ def _require(value: bool, message: str) -> None:
         raise ValueError(message)
 
 
+_MODEL_CACHE: dict[tuple[str, bool, bool], Any] = {}
+
+
+def _get_model(repo: Path, model_dir: Path, settings: dict[str, Any]) -> tuple[Any, float, bool]:
+    key = (str(model_dir), bool(settings.get("load_jit", False)), bool(settings.get("fp16", False)))
+    if key in _MODEL_CACHE:
+        return _MODEL_CACHE[key], 0.0, True
+    repo_text = str(repo)
+    matcha_text = str(repo / "third_party" / "Matcha-TTS")
+    if repo_text not in sys.path:
+        sys.path.insert(0, repo_text)
+    if matcha_text not in sys.path:
+        sys.path.insert(0, matcha_text)
+    from cosyvoice.cli.cosyvoice import AutoModel
+    started = perf_counter()
+    model = AutoModel(model_dir=str(model_dir), load_jit=key[1], fp16=key[2])
+    elapsed = round((perf_counter() - started) * 1000, 3)
+    _MODEL_CACHE[key] = model
+    return model, elapsed, False
+
+
 def _synthesize(request: dict[str, Any]) -> dict[str, Any]:
     settings = request.get("settings")
     _require(isinstance(settings, dict), "request.settings must be an object")
@@ -42,20 +64,9 @@ def _synthesize(request: dict[str, Any]) -> dict[str, Any]:
     _require(isinstance(prompt_text, str) and prompt_text.strip(), "prompt_text must be a non-empty string")
     _require(isinstance(silence_sec, (int, float)) and 0 <= float(silence_sec) <= 3, "inter_chunk_silence_sec must be between 0 and 3")
 
-    repo_text = str(repo)
-    matcha_text = str(repo / "third_party" / "Matcha-TTS")
-    if repo_text not in sys.path:
-        sys.path.insert(0, repo_text)
-    if matcha_text not in sys.path:
-        sys.path.insert(0, matcha_text)
-
     import torch
     import torchaudio
-    from cosyvoice.cli.cosyvoice import AutoModel
-
-    model_started = perf_counter()
-    cosyvoice = AutoModel(model_dir=str(model_dir))
-    model_load_latency_ms = round((perf_counter() - model_started) * 1000, 3)
+    cosyvoice, model_load_latency_ms, model_cache_hit = _get_model(repo, model_dir, settings)
     sample_rate = int(cosyvoice.sample_rate)
 
     synth_started = perf_counter()
@@ -99,19 +110,13 @@ def _synthesize(request: dict[str, Any]) -> dict[str, Any]:
             "name": "CosyVoice2-0.5B",
             "model_dir": str(model_dir),
             "model_load_latency_ms": model_load_latency_ms,
+            "model_cache_hit": model_cache_hit,
             "synthesis_latency_ms": synthesis_latency_ms,
         },
     }
 
 
-def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description="CosyVoice subprocess worker for the B2 TTS Skill.")
-    parser.add_argument("--request", required=True)
-    parser.add_argument("--response", required=True)
-    args = parser.parse_args(argv)
-
-    request_path = Path(args.request).resolve()
-    response_path = Path(args.response).resolve()
+def _run_request(request_path: Path, response_path: Path) -> int:
     try:
         request = json.loads(request_path.read_text(encoding="utf-8"))
         if not isinstance(request, dict):
@@ -120,20 +125,44 @@ def main(argv: list[str] | None = None) -> int:
         _write_response(response_path, {"status": "success", "output": output, "error": None})
         return 0
     except Exception as exc:
-        _write_response(
-            response_path,
-            {
-                "status": "error",
-                "output": None,
-                "error": {
-                    "type": type(exc).__name__,
-                    "message": str(exc),
-                    "traceback": traceback.format_exc(limit=5),
-                },
-            },
-        )
+        _write_response(response_path, {"status": "error", "output": None, "error": {
+            "type": type(exc).__name__, "message": str(exc), "traceback": traceback.format_exc(limit=5)}})
         return 1
 
 
+def _serve() -> int:
+    print("__TTS_WORKER_READY__", flush=True)
+    for line in sys.stdin:
+        request_id = "unknown"
+        status = 1
+        try:
+            command = json.loads(line)
+            request_id = str(command.get("id") or uuid.uuid4().hex)
+            if command.get("command") == "shutdown":
+                status = 0
+                return 0
+            status = _run_request(Path(command["request"]).resolve(), Path(command["response"]).resolve())
+        finally:
+            print(f"__TTS_WORKER_DONE__:{request_id}:{status}", flush=True)
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="CosyVoice subprocess worker for the B2 TTS Skill.")
+    parser.add_argument("--request")
+    parser.add_argument("--response")
+    parser.add_argument("--serve", action="store_true")
+    args = parser.parse_args(argv)
+
+    if args.serve:
+        return _serve()
+    if not args.request or not args.response:
+        parser.error("--request and --response are required unless --serve is used")
+
+    request_path = Path(args.request).resolve()
+    response_path = Path(args.response).resolve()
+    return _run_request(request_path, response_path)
+
+
 if __name__ == "__main__":
     raise SystemExit(main())