mkbk/Owner
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[项目边界]：完成全双工Agent入口和配置冻结，包含run-agent-live入口、配置校验和迁移文档

Browse Source
This commit is contained in:
mkbk
2026-06-18 21:40:23 +08:00
parent 696ed2c30e
commit 9fd8eaf7eb
8 changed files with 448 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+24
View File
@@ -2,6 +2,7 @@ OWNER_LLM_BASE_URL=https://token-plan-cn.xiaomimimo.com/v1
OWNER_LLM_API_KEY=
OWNER_LLM_MODEL=mimo-v2.5
OWNER_LLM_API_STYLE=chat_completions
OWNER_ASSISTANT_MODE=turn_based_voice_pet
OWNER_REALTIME_TRANSCRIPT_ENABLED=1
OWNER_REALTIME_TRANSCRIPT_IDLE_TIMEOUT_MS=1500
OWNER_AUDIO_INPUT_DEVICE=
@@ -54,6 +55,29 @@ OWNER_END_CHIME_ENABLED=1
OWNER_END_CHIME_FILE=assets/sounds/codex-notification.wav
OWNER_END_CHIME_FREQUENCY_HZ=880
OWNER_END_CHIME_DURATION_MS=140
OWNER_AUDIO_APM_PROVIDER=webrtc
OWNER_AUDIO_AEC_ENABLED=1
OWNER_AUDIO_NS_ENABLED=1
OWNER_AUDIO_AGC_ENABLED=1
OWNER_AUDIO_APM_REQUIRED=1
OWNER_AUDIO_FRAME_MS=20
OWNER_AUDIO_RING_BUFFER_MS=3000
OWNER_INTERRUPT_ENABLED=1
OWNER_INTERRUPT_TARGET_LATENCY_MS=200
OWNER_STREAMING_STT_PROVIDER=faster_whisper
OWNER_STREAMING_STT_PRODUCT_CANDIDATE=sensevoice
OWNER_STREAMING_TTS_PROVIDER=cosyvoice
OWNER_MEMORY_ENABLED=0
OWNER_MEMORY_PROVIDER=faiss_sqlite
OWNER_MEMORY_TOP_K=5
OWNER_MEMORY_AUTO_SAVE_SENSITIVE=0
OWNER_TOOL_ROUTER_ENABLED=0
OWNER_TOOL_MAX_CALLS_PER_TURN=5
OWNER_TOOL_TIMEOUT_MS=30000
OWNER_OPENINTERPRETER_ENABLED=0
OWNER_OPENINTERPRETER_COMMAND=openinterpreter
OWNER_BROWSER_PLAYWRIGHT_ENABLED=0
OWNER_COMPUTER_CONTROL_ENABLED=0
OWNER_WAKE_WORD=小杰小杰
OWNER_SAMPLE_RATE=16000
OWNER_CHANNELS=1
README.md
+46
View File
@@ -8,6 +8,7 @@
- `owner_voice_pet run-live`：真实常驻语音循环。
- `owner_voice_pet run-live --once`：只跑一轮，便于验收。
- `owner_voice_pet run-agent-live --check-config`：全双工 Agent 新入口的配置检查。当前用于迁移期验收，真实运行仍走 `run-live`。
- `.env` 直接读取配置，不要求导出 shell 环境变量。
- 唤醒词检测使用本地 `sherpa-onnx` KWS 模型，不走云端 ASR。
- `OWNER_SPEECH_PROVIDER=local`：默认除 LLM 外全用本地语音链路；`cloud` 仅作为显式兼容选项。
@@ -36,6 +37,7 @@ OWNER_LLM_BASE_URL=https://token-plan-cn.xiaomimimo.com/v1
OWNER_LLM_API_KEY=
OWNER_LLM_MODEL=mimo-v2.5
OWNER_LLM_API_STYLE=chat_completions
OWNER_ASSISTANT_MODE=turn_based_voice_pet
OWNER_REALTIME_TRANSCRIPT_ENABLED=1
OWNER_REALTIME_TRANSCRIPT_IDLE_TIMEOUT_MS=1500
OWNER_WAKE_PROVIDER=local_kws
@@ -81,6 +83,29 @@ OWNER_END_CHIME_ENABLED=1
OWNER_END_CHIME_FILE=assets/sounds/codex-notification.wav
OWNER_END_CHIME_FREQUENCY_HZ=880
OWNER_END_CHIME_DURATION_MS=140
OWNER_AUDIO_APM_PROVIDER=webrtc
OWNER_AUDIO_AEC_ENABLED=1
OWNER_AUDIO_NS_ENABLED=1
OWNER_AUDIO_AGC_ENABLED=1
OWNER_AUDIO_APM_REQUIRED=1
OWNER_AUDIO_FRAME_MS=20
OWNER_AUDIO_RING_BUFFER_MS=3000
OWNER_INTERRUPT_ENABLED=1
OWNER_INTERRUPT_TARGET_LATENCY_MS=200
OWNER_STREAMING_STT_PROVIDER=faster_whisper
OWNER_STREAMING_STT_PRODUCT_CANDIDATE=sensevoice
OWNER_STREAMING_TTS_PROVIDER=cosyvoice
OWNER_MEMORY_ENABLED=0
OWNER_MEMORY_PROVIDER=faiss_sqlite
OWNER_MEMORY_TOP_K=5
OWNER_MEMORY_AUTO_SAVE_SENSITIVE=0
OWNER_TOOL_ROUTER_ENABLED=0
OWNER_TOOL_MAX_CALLS_PER_TURN=5
OWNER_TOOL_TIMEOUT_MS=30000
OWNER_OPENINTERPRETER_ENABLED=0
OWNER_OPENINTERPRETER_COMMAND=openinterpreter
OWNER_BROWSER_PLAYWRIGHT_ENABLED=0
OWNER_COMPUTER_CONTROL_ENABLED=0
```
`OWNER_WAKE_PROVIDER=local_kws` 表示唤醒词“小杰小杰”由本地模型检测。唤醒命中后会先本地播报 `OWNER_WAKE_ACK_TEXT=我在`，再开始听取问题。`OWNER_SPEECH_PROVIDER=local` 表示正式问题 STT、实时字幕和 TTS 都走本地模型或 macOS 本地能力；云端只接收 final 文本和本次会话历史用于 LLM 回复。
@@ -95,6 +120,27 @@ OWNER_END_CHIME_DURATION_MS=140
`OWNER_END_CHIME_ENABLED=1` 表示对话自然结束或追问超时恢复待机前会播放一声项目内置提示音，默认文件是 `assets/sounds/codex-notification.wav`。提示音不走 TTS，也不会写入上下文；如果 `OWNER_END_CHIME_FILE` 指向的文件缺失，会回退到本地合成短音，`OWNER_END_CHIME_FREQUENCY_HZ` 和 `OWNER_END_CHIME_DURATION_MS` 只影响这个回退音。设置 `OWNER_END_CHIME_ENABLED=0` 可以关闭。
## 全双工 Agent 迁移入口
`add-full-duplex-agent-voice-assistant` 的实现会分阶段推进。为避免破坏当前已经可用的真人语音循环，新架构使用独立入口：
```bash
.venv/bin/python -m owner_voice_pet run-agent-live --check-config
```
这个命令会把有效运行模式固定为 `full_duplex_agent` 并输出全双工配置摘要，但当前不会启动未接线的全双工运行时。真实语音交互仍继续使用 `run-live`。后续实现会在这个入口下逐步接入 WebRTC APM、持续监听、Streaming STT/TTS、长期记忆和 Tool Router。
全双工 Agent 相关配置默认只做规划和安全关闭：`OWNER_AUDIO_APM_PROVIDER=webrtc` 代表目标音频底座，`OWNER_STREAMING_STT_PROVIDER=faster_whisper` 和 `OWNER_STREAMING_TTS_PROVIDER=cosyvoice` 是后续 provider 目标；`OWNER_MEMORY_ENABLED=0`、`OWNER_TOOL_ROUTER_ENABLED=0`、`OWNER_OPENINTERPRETER_ENABLED=0`、`OWNER_BROWSER_PLAYWRIGHT_ENABLED=0`、`OWNER_COMPUTER_CONTROL_ENABLED=0` 默认关闭，避免尚未完成安全边界前执行长期记忆或工具任务。
可选依赖按能力分组，不会被默认安装强制拉入：
```bash
.venv/bin/python -m pip install -e '.[full-duplex]'
.venv/bin/python -m pip install -e '.[streaming-stt]'
.venv/bin/python -m pip install -e '.[memory]'
.venv/bin/python -m pip install -e '.[browser]'
```
## 本地模型
首次运行前必须准备本地语音模型；同一脚本会下载 wake、VAD、2025 中文 CTC STT 和 GTCRN denoiser 模型：
openspec/changes/add-full-duplex-agent-voice-assistant/tasks.md
+5 -5
View File
@@ -1,10 +1,10 @@
## 1. OpenSpec 与项目边界
- [ ] 1.1 冻结 full-duplex Agent 实施入口命名；前置条件：人工确认 `run-agent-live` 或 `run-live --mode full-duplex`；优先级：P0；验收标准：README 和配置说明使用同一入口名；测试要点：CLI help 中能看到一致入口。
- [ ] 1.2 增加全双工模式配置读取规划；前置条件：入口命名已确认；优先级：P0；验收标准：配置包含 `OWNER_ASSISTANT_MODE`、APM、VAD、STT、TTS、memory、tool router 开关；测试要点：无效 provider 触发配置错误。
- [ ] 1.3 规划依赖分组；前置条件：provider 候选已确认；优先级：P0；验收标准：依赖按 audio、stt、tts、memory、tools 分组且默认不强制安装重依赖；测试要点：最小安装仍可跑现有 turn-based 测试。
- [ ] 1.4 定义迁移兼容策略；前置条件：现有 `run-live` 行为梳理完成；优先级：P0；验收标准：turn-based 路径在全双工开发期保持可用；测试要点：旧 simulate-live 和 run-live 回归不破坏。
- [ ] 1.5 建立模块提交门禁；前置条件：本变更 OpenSpec 已合并；优先级：P0；验收标准：每个主要任务组完成后验证并中文 commit；测试要点：`git status --short` 不包含未提交中间状态。
- [x] 1.1 冻结 full-duplex Agent 实施入口命名；前置条件：人工确认 `run-agent-live` 或 `run-live --mode full-duplex`；优先级：P0；验收标准：README 和配置说明使用同一入口名；测试要点：CLI help 中能看到一致入口。
- [x] 1.2 增加全双工模式配置读取规划；前置条件：入口命名已确认；优先级：P0；验收标准：配置包含 `OWNER_ASSISTANT_MODE`、APM、VAD、STT、TTS、memory、tool router 开关；测试要点：无效 provider 触发配置错误。
- [x] 1.3 规划依赖分组；前置条件：provider 候选已确认；优先级：P0；验收标准：依赖按 audio、stt、tts、memory、tools 分组且默认不强制安装重依赖；测试要点：最小安装仍可跑现有 turn-based 测试。
- [x] 1.4 定义迁移兼容策略；前置条件：现有 `run-live` 行为梳理完成；优先级：P0；验收标准：turn-based 路径在全双工开发期保持可用；测试要点：旧 simulate-live 和 run-live 回归不破坏。
- [x] 1.5 建立模块提交门禁；前置条件：本变更 OpenSpec 已合并；优先级：P0；验收标准：每个主要任务组完成后验证并中文 commit；测试要点：`git status --short` 不包含未提交中间状态。
## 2. WebRTC APM 与音频环形缓冲
pyproject.toml
+5
View File
@@ -15,6 +15,11 @@ dependencies = []
[project.optional-dependencies]
audio = ["numpy>=1.26", "sounddevice>=0.4.7"]
speech = ["numpy>=1.26", "sounddevice>=0.4.7", "sherpa-onnx>=1.13.3"]
full-duplex = ["numpy>=1.26", "sounddevice>=0.4.7"]
streaming-stt = ["faster-whisper>=1.0"]
memory = ["faiss-cpu>=1.8", "numpy>=1.26"]
browser = ["playwright>=1.44"]
agent-tools = ["playwright>=1.44"]
ui = ["PySide6>=6.7"]
test = []
src/owner_voice_pet/cli.py
+70
View File
@@ -38,6 +38,11 @@ def main(argv: list[str] | None = None) -> int:
subparsers.add_parser("device-check", help="Validate local microphone and speaker availability")
live = subparsers.add_parser("run-live", help="Run real repeated live voice conversation")
live.add_argument("--once", action="store_true", help="Run one completed live turn and exit")
agent_live = subparsers.add_parser(
"run-agent-live",
help="Run planned full-duplex Agent voice assistant entry point",
)
agent_live.add_argument("--check-config", action="store_true", help="Validate and print full-duplex Agent config")
simulate = subparsers.add_parser("simulate-live", help="Run live pipeline with simulated microphone frames")
simulate.add_argument("--turns", type=int, default=2, help="Number of simulated turns. Default: 2")
simulate.add_argument("--fixture", default=None, help="Replay simulated microphone frames from JSONL")
@@ -58,6 +63,7 @@ def main(argv: list[str] | None = None) -> int:
print(
json.dumps(
{
"assistant_mode": config.assistant_mode,
"wake_word": config.wake_word,
"sample_rate": config.sample_rate,
"channels": config.channels,
@@ -113,6 +119,29 @@ def main(argv: list[str] | None = None) -> int:
"end_chime_file": str(config.end_chime_file),
"end_chime_frequency_hz": config.end_chime_frequency_hz,
"end_chime_duration_ms": config.end_chime_duration_ms,
"audio_apm_provider": config.audio_apm_provider,
"audio_aec_enabled": config.audio_aec_enabled,
"audio_ns_enabled": config.audio_ns_enabled,
"audio_agc_enabled": config.audio_agc_enabled,
"audio_apm_required": config.audio_apm_required,
"audio_frame_ms": config.audio_frame_ms,
"audio_ring_buffer_ms": config.audio_ring_buffer_ms,
"interrupt_enabled": config.interrupt_enabled,
"interrupt_target_latency_ms": config.interrupt_target_latency_ms,
"streaming_stt_provider": config.streaming_stt_provider,
"streaming_stt_product_candidate": config.streaming_stt_product_candidate,
"streaming_tts_provider": config.streaming_tts_provider,
"memory_enabled": config.memory_enabled,
"memory_provider": config.memory_provider,
"memory_top_k": config.memory_top_k,
"memory_auto_save_sensitive": config.memory_auto_save_sensitive,
"tool_router_enabled": config.tool_router_enabled,
"tool_max_calls_per_turn": config.tool_max_calls_per_turn,
"tool_timeout_ms": config.tool_timeout_ms,
"openinterpreter_enabled": config.openinterpreter_enabled,
"openinterpreter_command": config.openinterpreter_command,
"browser_playwright_enabled": config.browser_playwright_enabled,
"computer_control_enabled": config.computer_control_enabled,
},
ensure_ascii=False,
sort_keys=True,
@@ -171,6 +200,47 @@ def main(argv: list[str] | None = None) -> int:
return 1
return 0 if summary.completed_turns > 0 or summary.interrupted else 1
if args.command == "run-agent-live":
config = replace(AppConfig.from_dotenv(args.env_file), assistant_mode="full_duplex_agent")
errors = config.validate_basic()
if errors:
print(
json.dumps(
{
"success": False,
"command": "run-agent-live",
"errors": [str(error) for error in errors],
},
ensure_ascii=False,
sort_keys=True,
)
)
return 1
data = {
"success": bool(args.check_config),
"command": "run-agent-live",
"assistant_mode": config.assistant_mode,
"audio_apm_provider": config.audio_apm_provider,
"streaming_stt_provider": config.streaming_stt_provider,
"streaming_tts_provider": config.streaming_tts_provider,
"memory_provider": config.memory_provider,
"memory_enabled": config.memory_enabled,
"tool_router_enabled": config.tool_router_enabled,
"openinterpreter_enabled": config.openinterpreter_enabled,
"browser_playwright_enabled": config.browser_playwright_enabled,
"computer_control_enabled": config.computer_control_enabled,
"turn_based_entry": "run-live",
"full_duplex_runtime_ready": False,
}
if not args.check_config:
data["code"] = "FULL_DUPLEX_RUNTIME_NOT_IMPLEMENTED"
data["message"] = (
"run-agent-live is the reserved full-duplex Agent entry point; "
"runtime wiring will be implemented by later OpenSpec tasks."
)
print(json.dumps(data, ensure_ascii=False, sort_keys=True))
return 0 if args.check_config else 1
if args.command == "simulate-live":
try:
data = run_simulated_live(
src/owner_voice_pet/config.py
+137
View File
@@ -8,6 +8,7 @@ from .models import ErrorCode, ProviderError
@dataclass(frozen=True, slots=True)
class AppConfig:
assistant_mode: str = "turn_based_voice_pet"
wake_word: str = "小杰小杰"
sample_rate: int = 16000
channels: int = 1
@@ -68,6 +69,29 @@ class AppConfig:
end_chime_file: Path = Path("assets/sounds/codex-notification.wav")
end_chime_frequency_hz: int = 880
end_chime_duration_ms: int = 140
audio_apm_provider: str = "webrtc"
audio_aec_enabled: bool = True
audio_ns_enabled: bool = True
audio_agc_enabled: bool = True
audio_apm_required: bool = True
audio_frame_ms: int = 20
audio_ring_buffer_ms: int = 3000
interrupt_enabled: bool = True
interrupt_target_latency_ms: int = 200
streaming_stt_provider: str = "faster_whisper"
streaming_stt_product_candidate: str = "sensevoice"
streaming_tts_provider: str = "cosyvoice"
memory_enabled: bool = False
memory_provider: str = "faiss_sqlite"
memory_top_k: int = 5
memory_auto_save_sensitive: bool = False
tool_router_enabled: bool = False
tool_max_calls_per_turn: int = 5
tool_timeout_ms: int = 30000
openinterpreter_enabled: bool = False
openinterpreter_command: str = "openinterpreter"
browser_playwright_enabled: bool = False
computer_control_enabled: bool = False
@classmethod
def from_dotenv(cls, path: str | Path = ".env", prefix: str = "OWNER_") -> "AppConfig":
@@ -77,7 +101,13 @@ class AppConfig:
value = values.get(f"{prefix}{name}")
return default if value is None or value == "" else value
def get_bool(name: str, default: str = "0") -> bool:
return (get(name, default) or default).lower() not in {"0", "false", "no"}
return cls(
assistant_mode=(
get("ASSISTANT_MODE", "turn_based_voice_pet") or "turn_based_voice_pet"
).lower(),
wake_word=get("WAKE_WORD", "小杰小杰") or "小杰小杰",
sample_rate=int(get("SAMPLE_RATE", "16000") or "16000"),
channels=int(get("CHANNELS", "1") or "1"),
@@ -156,6 +186,37 @@ class AppConfig:
),
end_chime_frequency_hz=int(get("END_CHIME_FREQUENCY_HZ", "880") or "880"),
end_chime_duration_ms=int(get("END_CHIME_DURATION_MS", "140") or "140"),
audio_apm_provider=(get("AUDIO_APM_PROVIDER", "webrtc") or "webrtc").lower(),
audio_aec_enabled=get_bool("AUDIO_AEC_ENABLED", "1"),
audio_ns_enabled=get_bool("AUDIO_NS_ENABLED", "1"),
audio_agc_enabled=get_bool("AUDIO_AGC_ENABLED", "1"),
audio_apm_required=get_bool("AUDIO_APM_REQUIRED", "1"),
audio_frame_ms=int(get("AUDIO_FRAME_MS", "20") or "20"),
audio_ring_buffer_ms=int(get("AUDIO_RING_BUFFER_MS", "3000") or "3000"),
interrupt_enabled=get_bool("INTERRUPT_ENABLED", "1"),
interrupt_target_latency_ms=int(get("INTERRUPT_TARGET_LATENCY_MS", "200") or "200"),
streaming_stt_provider=(
get("STREAMING_STT_PROVIDER", "faster_whisper") or "faster_whisper"
).lower(),
streaming_stt_product_candidate=(
get("STREAMING_STT_PRODUCT_CANDIDATE", "sensevoice") or "sensevoice"
).lower(),
streaming_tts_provider=(
get("STREAMING_TTS_PROVIDER", "cosyvoice") or "cosyvoice"
).lower(),
memory_enabled=get_bool("MEMORY_ENABLED", "0"),
memory_provider=(get("MEMORY_PROVIDER", "faiss_sqlite") or "faiss_sqlite").lower(),
memory_top_k=int(get("MEMORY_TOP_K", "5") or "5"),
memory_auto_save_sensitive=get_bool("MEMORY_AUTO_SAVE_SENSITIVE", "0"),
tool_router_enabled=get_bool("TOOL_ROUTER_ENABLED", "0"),
tool_max_calls_per_turn=int(get("TOOL_MAX_CALLS_PER_TURN", "5") or "5"),
tool_timeout_ms=int(get("TOOL_TIMEOUT_MS", "30000") or "30000"),
openinterpreter_enabled=get_bool("OPENINTERPRETER_ENABLED", "0"),
openinterpreter_command=(
get("OPENINTERPRETER_COMMAND", "openinterpreter") or "openinterpreter"
),
browser_playwright_enabled=get_bool("BROWSER_PLAYWRIGHT_ENABLED", "0"),
computer_control_enabled=get_bool("COMPUTER_CONTROL_ENABLED", "0"),
)
@classmethod
@@ -174,6 +235,16 @@ class AppConfig:
def validate_basic(self) -> list[ProviderError]:
errors: list[ProviderError] = []
if self.assistant_mode not in {"turn_based_voice_pet", "full_duplex_agent"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_ASSISTANT_MODE must be turn_based_voice_pet or full_duplex_agent",
False,
"config",
"startup",
)
)
if self.sample_rate <= 0:
errors.append(
ProviderError(
@@ -451,6 +522,12 @@ class AppConfig:
for name, value in {
"OWNER_END_CHIME_FREQUENCY_HZ": self.end_chime_frequency_hz,
"OWNER_END_CHIME_DURATION_MS": self.end_chime_duration_ms,
"OWNER_AUDIO_FRAME_MS": self.audio_frame_ms,
"OWNER_AUDIO_RING_BUFFER_MS": self.audio_ring_buffer_ms,
"OWNER_INTERRUPT_TARGET_LATENCY_MS": self.interrupt_target_latency_ms,
"OWNER_MEMORY_TOP_K": self.memory_top_k,
"OWNER_TOOL_MAX_CALLS_PER_TURN": self.tool_max_calls_per_turn,
"OWNER_TOOL_TIMEOUT_MS": self.tool_timeout_ms,
}.items():
if value <= 0:
errors.append(
@@ -462,6 +539,66 @@ class AppConfig:
"startup",
)
)
if self.audio_apm_provider not in {"webrtc", "fake", "disabled"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_AUDIO_APM_PROVIDER must be webrtc, fake, or disabled",
False,
"config",
"startup",
)
)
if self.streaming_stt_provider not in {"faster_whisper", "sensevoice", "sherpa_onnx"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_STREAMING_STT_PROVIDER must be faster_whisper, sensevoice, or sherpa_onnx",
False,
"config",
"startup",
)
)
if self.streaming_stt_product_candidate not in {"sensevoice", "faster_whisper", "sherpa_onnx"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_STREAMING_STT_PRODUCT_CANDIDATE must be sensevoice, faster_whisper, or sherpa_onnx",
False,
"config",
"startup",
)
)
if self.streaming_tts_provider not in {"cosyvoice", "macos_say"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_STREAMING_TTS_PROVIDER must be cosyvoice or macos_say",
False,
"config",
"startup",
)
)
if self.memory_provider not in {"faiss_sqlite"}:
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_MEMORY_PROVIDER must be faiss_sqlite",
False,
"config",
"startup",
)
)
if self.openinterpreter_enabled and not self.openinterpreter_command.strip():
errors.append(
ProviderError(
ErrorCode.CONFIG_MISSING_VALUE,
"OWNER_OPENINTERPRETER_COMMAND must be non-empty when Open Interpreter is enabled",
False,
"config",
"startup",
)
)
return errors
def api_url(self, path: str) -> str:
tests/test_cli_acceptance.py
+41
View File
@@ -42,6 +42,7 @@ class CliAcceptanceTests(unittest.TestCase):
self.assertEqual(code, 0)
self.assertTrue(data["llm_api_key_present"])
self.assertEqual(data["llm_model"], "file-model")
self.assertEqual(data["assistant_mode"], "turn_based_voice_pet")
self.assertEqual(data["pipeline_mode"], "live_turn_based")
self.assertEqual(data["endpoint_mode"], "primary_speaker")
self.assertEqual(data["realtime_transcript_idle_timeout_ms"], 1500)
@@ -49,8 +50,48 @@ class CliAcceptanceTests(unittest.TestCase):
self.assertTrue(data["end_chime_enabled"])
self.assertEqual(data["end_chime_file"], "assets/sounds/codex-notification.wav")
self.assertEqual(data["end_chime_duration_ms"], 140)
self.assertEqual(data["audio_apm_provider"], "webrtc")
self.assertTrue(data["audio_aec_enabled"])
self.assertTrue(data["audio_ns_enabled"])
self.assertTrue(data["audio_agc_enabled"])
self.assertEqual(data["audio_frame_ms"], 20)
self.assertEqual(data["audio_ring_buffer_ms"], 3000)
self.assertTrue(data["interrupt_enabled"])
self.assertEqual(data["interrupt_target_latency_ms"], 200)
self.assertEqual(data["streaming_stt_provider"], "faster_whisper")
self.assertEqual(data["streaming_stt_product_candidate"], "sensevoice")
self.assertEqual(data["streaming_tts_provider"], "cosyvoice")
self.assertFalse(data["memory_enabled"])
self.assertEqual(data["memory_provider"], "faiss_sqlite")
self.assertFalse(data["tool_router_enabled"])
self.assertFalse(data["openinterpreter_enabled"])
self.assertFalse(data["browser_playwright_enabled"])
self.assertFalse(data["computer_control_enabled"])
self.assertNotIn("secret-value", str(data))
def test_help_lists_full_duplex_agent_entrypoint(self) -> None:
buffer = io.StringIO()
with redirect_stdout(buffer), self.assertRaises(SystemExit) as raised:
main(["--help"])
self.assertEqual(raised.exception.code, 0)
self.assertIn("run-agent-live", buffer.getvalue())
def test_run_agent_live_check_config_reports_reserved_entrypoint(self) -> None:
code, data = self.call("run-agent-live", "--check-config")
self.assertEqual(code, 0)
self.assertTrue(data["success"])
self.assertEqual(data["command"], "run-agent-live")
self.assertEqual(data["assistant_mode"], "full_duplex_agent")
self.assertFalse(data["full_duplex_runtime_ready"])
self.assertEqual(data["turn_based_entry"], "run-live")
def test_run_agent_live_without_runtime_returns_explicit_not_implemented(self) -> None:
code, data = self.call("run-agent-live")
self.assertEqual(code, 1)
self.assertFalse(data["success"])
self.assertEqual(data["code"], "FULL_DUPLEX_RUNTIME_NOT_IMPLEMENTED")
def test_security_check_command_has_no_leaks(self) -> None:
code, data = self.call("security-check")
self.assertEqual(code, 0)
tests/test_models_config.py
+120
View File
@@ -74,6 +74,7 @@ class ModelsConfigTests(unittest.TestCase):
)
)
config = AppConfig.from_dotenv(path)
self.assertEqual(config.assistant_mode, "turn_based_voice_pet")
self.assertEqual(config.llm_base_url, "https://token-plan-cn.xiaomimimo.com/v1")
self.assertEqual(config.llm_api_key, "secret-value")
self.assertEqual(config.llm_model, "test-model")
@@ -122,14 +123,133 @@ class ModelsConfigTests(unittest.TestCase):
self.assertEqual(str(config.end_chime_file), "assets/sounds/codex-notification.wav")
self.assertEqual(config.end_chime_frequency_hz, 880)
self.assertEqual(config.end_chime_duration_ms, 140)
self.assertEqual(config.audio_apm_provider, "webrtc")
self.assertTrue(config.audio_aec_enabled)
self.assertTrue(config.audio_ns_enabled)
self.assertTrue(config.audio_agc_enabled)
self.assertTrue(config.audio_apm_required)
self.assertEqual(config.audio_frame_ms, 20)
self.assertEqual(config.audio_ring_buffer_ms, 3000)
self.assertTrue(config.interrupt_enabled)
self.assertEqual(config.interrupt_target_latency_ms, 200)
self.assertEqual(config.streaming_stt_provider, "faster_whisper")
self.assertEqual(config.streaming_stt_product_candidate, "sensevoice")
self.assertEqual(config.streaming_tts_provider, "cosyvoice")
self.assertFalse(config.memory_enabled)
self.assertEqual(config.memory_provider, "faiss_sqlite")
self.assertEqual(config.memory_top_k, 5)
self.assertFalse(config.memory_auto_save_sensitive)
self.assertFalse(config.tool_router_enabled)
self.assertEqual(config.tool_max_calls_per_turn, 5)
self.assertEqual(config.tool_timeout_ms, 30000)
self.assertFalse(config.openinterpreter_enabled)
self.assertEqual(config.openinterpreter_command, "openinterpreter")
self.assertFalse(config.browser_playwright_enabled)
self.assertFalse(config.computer_control_enabled)
self.assertTrue(config.llm_stream)
self.assertEqual(config.validate_basic(), [])
def test_full_duplex_agent_config_values_are_read_from_dotenv(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = f"{tmp}/.env"
with open(path, "w", encoding="utf-8") as handle:
handle.write(
"\n".join(
[
"OWNER_ASSISTANT_MODE=full_duplex_agent",
"OWNER_AUDIO_APM_PROVIDER=fake",
"OWNER_AUDIO_AEC_ENABLED=0",
"OWNER_AUDIO_NS_ENABLED=0",
"OWNER_AUDIO_AGC_ENABLED=0",
"OWNER_AUDIO_APM_REQUIRED=0",
"OWNER_AUDIO_FRAME_MS=10",
"OWNER_AUDIO_RING_BUFFER_MS=1200",
"OWNER_INTERRUPT_ENABLED=0",
"OWNER_INTERRUPT_TARGET_LATENCY_MS=180",
"OWNER_STREAMING_STT_PROVIDER=sherpa_onnx",
"OWNER_STREAMING_STT_PRODUCT_CANDIDATE=faster_whisper",
"OWNER_STREAMING_TTS_PROVIDER=macos_say",
"OWNER_MEMORY_ENABLED=1",
"OWNER_MEMORY_PROVIDER=faiss_sqlite",
"OWNER_MEMORY_TOP_K=3",
"OWNER_MEMORY_AUTO_SAVE_SENSITIVE=1",
"OWNER_TOOL_ROUTER_ENABLED=1",
"OWNER_TOOL_MAX_CALLS_PER_TURN=2",
"OWNER_TOOL_TIMEOUT_MS=1000",
"OWNER_OPENINTERPRETER_ENABLED=1",
"OWNER_OPENINTERPRETER_COMMAND=/tmp/openinterpreter",
"OWNER_BROWSER_PLAYWRIGHT_ENABLED=1",
"OWNER_COMPUTER_CONTROL_ENABLED=1",
]
)
)
config = AppConfig.from_dotenv(path)
self.assertEqual(config.assistant_mode, "full_duplex_agent")
self.assertEqual(config.audio_apm_provider, "fake")
self.assertFalse(config.audio_aec_enabled)
self.assertFalse(config.audio_ns_enabled)
self.assertFalse(config.audio_agc_enabled)
self.assertFalse(config.audio_apm_required)
self.assertEqual(config.audio_frame_ms, 10)
self.assertEqual(config.audio_ring_buffer_ms, 1200)
self.assertFalse(config.interrupt_enabled)
self.assertEqual(config.interrupt_target_latency_ms, 180)
self.assertEqual(config.streaming_stt_provider, "sherpa_onnx")
self.assertEqual(config.streaming_stt_product_candidate, "faster_whisper")
self.assertEqual(config.streaming_tts_provider, "macos_say")
self.assertTrue(config.memory_enabled)
self.assertEqual(config.memory_provider, "faiss_sqlite")
self.assertEqual(config.memory_top_k, 3)
self.assertTrue(config.memory_auto_save_sensitive)
self.assertTrue(config.tool_router_enabled)
self.assertEqual(config.tool_max_calls_per_turn, 2)
self.assertEqual(config.tool_timeout_ms, 1000)
self.assertTrue(config.openinterpreter_enabled)
self.assertEqual(config.openinterpreter_command, "/tmp/openinterpreter")
self.assertTrue(config.browser_playwright_enabled)
self.assertTrue(config.computer_control_enabled)
self.assertEqual(config.validate_basic(), [])
def test_speech_provider_must_be_cloud_or_local(self) -> None:
config = AppConfig(speech_provider="invalid")
errors = config.validate_basic()
self.assertTrue(any("OWNER_SPEECH_PROVIDER" in error.message for error in errors))
def test_full_duplex_agent_config_is_validated(self) -> None:
config = AppConfig(
assistant_mode="invalid",
audio_apm_provider="invalid",
audio_frame_ms=0,
audio_ring_buffer_ms=0,
interrupt_target_latency_ms=0,
streaming_stt_provider="invalid",
streaming_stt_product_candidate="invalid",
streaming_tts_provider="invalid",
memory_provider="invalid",
memory_top_k=0,
tool_max_calls_per_turn=0,
tool_timeout_ms=0,
openinterpreter_enabled=True,
openinterpreter_command="",
)
errors = config.validate_basic()
self.assertTrue(any("OWNER_ASSISTANT_MODE" in error.message for error in errors))
self.assertTrue(any("OWNER_AUDIO_APM_PROVIDER" in error.message for error in errors))
self.assertTrue(any("OWNER_AUDIO_FRAME_MS" in error.message for error in errors))
self.assertTrue(any("OWNER_AUDIO_RING_BUFFER_MS" in error.message for error in errors))
self.assertTrue(any("OWNER_INTERRUPT_TARGET_LATENCY_MS" in error.message for error in errors))
self.assertTrue(any("OWNER_STREAMING_STT_PROVIDER" in error.message for error in errors))
self.assertTrue(any("OWNER_STREAMING_STT_PRODUCT_CANDIDATE" in error.message for error in errors))
self.assertTrue(any("OWNER_STREAMING_TTS_PROVIDER" in error.message for error in errors))
self.assertTrue(any("OWNER_MEMORY_PROVIDER" in error.message for error in errors))
self.assertTrue(any("OWNER_MEMORY_TOP_K" in error.message for error in errors))
self.assertTrue(any("OWNER_TOOL_MAX_CALLS_PER_TURN" in error.message for error in errors))
self.assertTrue(any("OWNER_TOOL_TIMEOUT_MS" in error.message for error in errors))
self.assertTrue(any("OWNER_OPENINTERPRETER_COMMAND" in error.message for error in errors))
def test_realtime_transcript_idle_timeout_must_be_non_negative(self) -> None:
config = AppConfig(realtime_transcript_idle_timeout_ms=-1)
errors = config.validate_basic()