wx44wx/AI-VideoAssistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor assistant configuration management and update documentation

Browse Source 
- Removed legacy agent profile settings from the .env.example and README, streamlining the configuration process.
- Introduced a new local YAML configuration adapter for assistant settings, allowing for easier management of assistant profiles.
- Updated backend integration documentation to clarify the behavior of assistant config sourcing based on backend URL settings.
- Adjusted various service implementations to directly utilize API keys from the new configuration structure.
- Enhanced test coverage for the new local YAML adapter and its integration with backend services.
This commit is contained in:
Xin Wang
2026-03-05 21:24:15 +08:00
parent d0a6419990
commit 935f2fbd1f
17 changed files with 585 additions and 739 deletions
Download Patch File Download Diff File Expand all files Collapse all files

300
engine/tests/test_agent_config.py
View File

@@ -1,293 +1,21 @@
import os
from pathlib import Path
import importlib
import pytest
os.environ.setdefault("LLM_API_KEY", "test-openai-key")
os.environ.setdefault("TTS_API_KEY", "test-tts-key")
os.environ.setdefault("ASR_API_KEY", "test-asr-key")
def test_settings_load_from_environment(monkeypatch):
monkeypatch.setenv("HOST", "127.0.0.1")
monkeypatch.setenv("PORT", "8123")
from app.config import load_settings
import app.config as config_module
importlib.reload(config_module)
settings = config_module.get_settings()
assert settings.host == "127.0.0.1"
assert settings.port == 8123
def _write_yaml(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
def test_assistant_local_config_dir_default_present():
import app.config as config_module
def _full_agent_yaml(llm_model: str = "gpt-4o-mini", llm_key: str = "test-openai-key") -> str:
return f"""
agent:
vad:
type: silero
model_path: data/vad/silero_vad.onnx
threshold: 0.63
min_speech_duration_ms: 100
eou_threshold_ms: 800
llm:
provider: openai_compatible
model: {llm_model}
temperature: 0.2
api_key: {llm_key}
api_url: https://example-llm.invalid/v1
tts:
provider: openai_compatible
api_key: test-tts-key
api_url: https://example-tts.invalid/v1/audio/speech
model: FunAudioLLM/CosyVoice2-0.5B
voice: anna
speed: 1.0
asr:
provider: openai_compatible
api_key: test-asr-key
api_url: https://example-asr.invalid/v1/audio/transcriptions
model: FunAudioLLM/SenseVoiceSmall
interim_interval_ms: 500
min_audio_ms: 300
start_min_speech_ms: 160
pre_speech_ms: 240
final_tail_ms: 120
duplex:
enabled: true
system_prompt: You are a strict test assistant.
barge_in:
min_duration_ms: 200
silence_tolerance_ms: 60
""".strip()
def _dashscope_tts_yaml() -> str:
return _full_agent_yaml().replace(
""" tts:
provider: openai_compatible
api_key: test-tts-key
api_url: https://example-tts.invalid/v1/audio/speech
model: FunAudioLLM/CosyVoice2-0.5B
voice: anna
speed: 1.0
""",
""" tts:
provider: dashscope
api_key: test-dashscope-key
voice: Cherry
speed: 1.0
""",
)
def test_cli_profile_loads_agent_yaml(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
config_dir = tmp_path / "config" / "agents"
_write_yaml(
config_dir / "support.yaml",
_full_agent_yaml(llm_model="gpt-4.1-mini"),
)
settings = load_settings(
argv=["--agent-profile", "support"],
)
assert settings.llm_model == "gpt-4.1-mini"
assert settings.llm_temperature == 0.2
assert settings.vad_threshold == 0.63
assert settings.agent_config_source == "cli_profile"
assert settings.agent_config_path == str((config_dir / "support.yaml").resolve())
def test_cli_path_has_higher_priority_than_env(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
env_file = tmp_path / "config" / "agents" / "env.yaml"
cli_file = tmp_path / "config" / "agents" / "cli.yaml"
_write_yaml(env_file, _full_agent_yaml(llm_model="env-model"))
_write_yaml(cli_file, _full_agent_yaml(llm_model="cli-model"))
monkeypatch.setenv("AGENT_CONFIG_PATH", str(env_file))
settings = load_settings(argv=["--agent-config", str(cli_file)])
assert settings.llm_model == "cli-model"
assert settings.agent_config_source == "cli_path"
assert settings.agent_config_path == str(cli_file.resolve())
def test_default_yaml_is_loaded_without_args_or_env(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
default_file = tmp_path / "config" / "agents" / "default.yaml"
_write_yaml(default_file, _full_agent_yaml(llm_model="from-default"))
monkeypatch.delenv("AGENT_CONFIG_PATH", raising=False)
monkeypatch.delenv("AGENT_PROFILE", raising=False)
settings = load_settings(argv=[])
assert settings.llm_model == "from-default"
assert settings.agent_config_source == "default"
assert settings.agent_config_path == str(default_file.resolve())
def test_missing_required_agent_settings_fail(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "missing-required.yaml"
_write_yaml(
file_path,
"""
agent:
llm:
model: gpt-4o-mini
""".strip(),
)
with pytest.raises(ValueError, match="Missing required agent settings in YAML"):
load_settings(argv=["--agent-config", str(file_path)])
def test_blank_required_provider_key_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "blank-key.yaml"
_write_yaml(file_path, _full_agent_yaml(llm_key=""))
with pytest.raises(ValueError, match="Missing required agent settings in YAML"):
load_settings(argv=["--agent-config", str(file_path)])
def test_missing_tts_api_url_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "missing-tts-url.yaml"
_write_yaml(
file_path,
_full_agent_yaml().replace(
" api_url: https://example-tts.invalid/v1/audio/speech\n",
"",
),
)
with pytest.raises(ValueError, match="Missing required agent settings in YAML"):
load_settings(argv=["--agent-config", str(file_path)])
def test_dashscope_tts_allows_default_url_and_model(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "dashscope-tts.yaml"
_write_yaml(file_path, _dashscope_tts_yaml())
settings = load_settings(argv=["--agent-config", str(file_path)])
assert settings.tts_provider == "dashscope"
assert settings.tts_api_key == "test-dashscope-key"
assert settings.tts_api_url is None
assert settings.tts_model is None
def test_dashscope_tts_requires_api_key(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "dashscope-tts-missing-key.yaml"
_write_yaml(file_path, _dashscope_tts_yaml().replace(" api_key: test-dashscope-key\n", ""))
with pytest.raises(ValueError, match="Missing required agent settings in YAML"):
load_settings(argv=["--agent-config", str(file_path)])
def test_missing_asr_api_url_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "missing-asr-url.yaml"
_write_yaml(
file_path,
_full_agent_yaml().replace(
" api_url: https://example-asr.invalid/v1/audio/transcriptions\n",
"",
),
)
with pytest.raises(ValueError, match="Missing required agent settings in YAML"):
load_settings(argv=["--agent-config", str(file_path)])
def test_agent_yaml_unknown_key_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "bad-agent.yaml"
_write_yaml(file_path, _full_agent_yaml() + "\n unknown_option: true")
with pytest.raises(ValueError, match="Unknown agent config keys"):
load_settings(argv=["--agent-config", str(file_path)])
def test_legacy_siliconflow_section_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "legacy-siliconflow.yaml"
_write_yaml(
file_path,
"""
agent:
siliconflow:
api_key: x
""".strip(),
)
with pytest.raises(ValueError, match="Section 'siliconflow' is no longer supported"):
load_settings(argv=["--agent-config", str(file_path)])
def test_agent_yaml_missing_env_reference_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "bad-ref.yaml"
_write_yaml(
file_path,
_full_agent_yaml(llm_key="${UNSET_LLM_API_KEY}"),
)
with pytest.raises(ValueError, match="Missing environment variable"):
load_settings(argv=["--agent-config", str(file_path)])
def test_agent_yaml_tools_list_is_loaded(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "tools-agent.yaml"
_write_yaml(
file_path,
_full_agent_yaml()
+ """
tools:
- current_time
- name: weather
description: Get weather by city.
parameters:
type: object
properties:
city:
type: string
required: [city]
executor: server
""",
)
settings = load_settings(argv=["--agent-config", str(file_path)])
assert isinstance(settings.tools, list)
assert settings.tools[0] == "current_time"
assert settings.tools[1]["name"] == "weather"
assert settings.tools[1]["executor"] == "server"
def test_agent_yaml_tools_must_be_list(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
file_path = tmp_path / "bad-tools-agent.yaml"
_write_yaml(
file_path,
_full_agent_yaml()
+ """
tools:
weather:
executor: server
""",
)
with pytest.raises(ValueError, match="Agent config key 'tools' must be a list"):
load_settings(argv=["--agent-config", str(file_path)])
settings = config_module.get_settings()
assert isinstance(settings.assistant_local_config_dir, str)
assert settings.assistant_local_config_dir

214
engine/tests/test_backend_adapters.py
View File

@@ -2,24 +2,42 @@ import aiohttp
import pytest
from app.backend_adapters import (
HistoryDisabledBackendAdapter,
HttpBackendAdapter,
NullBackendAdapter,
AssistantConfigSourceAdapter,
LocalYamlAssistantConfigAdapter,
build_backend_adapter,
)
@pytest.mark.asyncio
async def test_build_backend_adapter_without_url_returns_null_adapter():
async def test_without_backend_url_uses_local_yaml_for_assistant_config(tmp_path):
config_dir = tmp_path / "assistants"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "dev_local.yaml").write_text(
"\n".join(
[
"assistant:",
" assistantId: dev_local",
" systemPrompt: local prompt",
" greeting: local greeting",
]
),
encoding="utf-8",
)
adapter = build_backend_adapter(
backend_url=None,
backend_mode="auto",
history_enabled=True,
timeout_sec=3,
assistant_local_config_dir=str(config_dir),
)
assert isinstance(adapter, NullBackendAdapter)
assert isinstance(adapter, AssistantConfigSourceAdapter)
assert await adapter.fetch_assistant_config("assistant_1") is None
payload = await adapter.fetch_assistant_config("dev_local")
assert isinstance(payload, dict)
assert payload.get("__error_code") in (None, "")
assert payload["assistant"]["assistantId"] == "dev_local"
assert payload["assistant"]["systemPrompt"] == "local prompt"
assert (
await adapter.create_call_record(
user_id=1,
@@ -54,7 +72,7 @@ async def test_build_backend_adapter_without_url_returns_null_adapter():
@pytest.mark.asyncio
async def test_http_backend_adapter_create_call_record_posts_expected_payload(monkeypatch):
async def test_http_backend_adapter_create_call_record_posts_expected_payload(monkeypatch, tmp_path):
captured = {}
class _FakeResponse:
@@ -90,15 +108,31 @@ async def test_http_backend_adapter_create_call_record_posts_expected_payload(mo
captured["json"] = json
return _FakeResponse(status=200, payload={"id": "call_123"})
def get(self, url):
_ = url
return _FakeResponse(
status=200,
payload={
"assistant": {
"assistantId": "assistant_9",
"systemPrompt": "backend prompt",
}
},
)
monkeypatch.setattr("app.backend_adapters.aiohttp.ClientSession", _FakeClientSession)
config_dir = tmp_path / "assistants"
config_dir.mkdir(parents=True, exist_ok=True)
adapter = build_backend_adapter(
backend_url="http://localhost:8100",
backend_mode="auto",
history_enabled=True,
timeout_sec=7,
assistant_local_config_dir=str(config_dir),
)
assert isinstance(adapter, HttpBackendAdapter)
assert isinstance(adapter, AssistantConfigSourceAdapter)
call_id = await adapter.create_call_record(
user_id=99,
@@ -119,25 +153,115 @@ async def test_http_backend_adapter_create_call_record_posts_expected_payload(mo
@pytest.mark.asyncio
async def test_backend_mode_disabled_forces_null_even_with_url():
async def test_with_backend_url_uses_backend_for_assistant_config(monkeypatch, tmp_path):
class _FakeResponse:
def __init__(self, status=200, payload=None):
self.status = status
self._payload = payload if payload is not None else {}
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
async def json(self):
return self._payload
def raise_for_status(self):
if self.status >= 400:
raise RuntimeError("http_error")
class _FakeClientSession:
def __init__(self, timeout=None):
self.timeout = timeout
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
def get(self, url):
_ = url
return _FakeResponse(
status=200,
payload={
"assistant": {
"assistantId": "dev_http",
"systemPrompt": "backend prompt",
}
},
)
def post(self, url, json=None):
_ = (url, json)
return _FakeResponse(status=200, payload={"id": "call_1"})
monkeypatch.setattr("app.backend_adapters.aiohttp.ClientSession", _FakeClientSession)
config_dir = tmp_path / "assistants"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "dev_http.yaml").write_text(
"\n".join(
[
"assistant:",
" assistantId: dev_http",
" systemPrompt: local prompt",
]
),
encoding="utf-8",
)
adapter = build_backend_adapter(
backend_url="http://localhost:8100",
backend_mode="auto",
history_enabled=True,
timeout_sec=3,
assistant_local_config_dir=str(config_dir),
)
assert isinstance(adapter, AssistantConfigSourceAdapter)
payload = await adapter.fetch_assistant_config("dev_http")
assert payload["assistant"]["assistantId"] == "dev_http"
assert payload["assistant"]["systemPrompt"] == "backend prompt"
@pytest.mark.asyncio
async def test_backend_mode_disabled_uses_local_assistant_config_even_with_url(monkeypatch, tmp_path):
class _FailIfCalledClientSession:
def __init__(self, timeout=None):
_ = timeout
raise AssertionError("HTTP client should not be created when backend_mode=disabled")
monkeypatch.setattr("app.backend_adapters.aiohttp.ClientSession", _FailIfCalledClientSession)
config_dir = tmp_path / "assistants"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "dev_disabled.yaml").write_text(
"\n".join(
[
"assistant:",
" assistantId: dev_disabled",
" systemPrompt: local disabled prompt",
]
),
encoding="utf-8",
)
adapter = build_backend_adapter(
backend_url="http://localhost:8100",
backend_mode="disabled",
history_enabled=True,
timeout_sec=7,
timeout_sec=3,
assistant_local_config_dir=str(config_dir),
)
assert isinstance(adapter, NullBackendAdapter)
assert isinstance(adapter, AssistantConfigSourceAdapter)
payload = await adapter.fetch_assistant_config("dev_disabled")
assert payload["assistant"]["assistantId"] == "dev_disabled"
assert payload["assistant"]["systemPrompt"] == "local disabled prompt"
@pytest.mark.asyncio
async def test_history_disabled_wraps_backend_adapter():
adapter = build_backend_adapter(
backend_url="http://localhost:8100",
backend_mode="auto",
history_enabled=False,
timeout_sec=7,
)
assert isinstance(adapter, HistoryDisabledBackendAdapter)
assert await adapter.create_call_record(user_id=1, assistant_id="a1", source="debug") is None
assert await adapter.add_transcript(
call_id="c1",
@@ -148,3 +272,53 @@ async def test_history_disabled_wraps_backend_adapter():
end_ms=10,
duration_ms=10,
) is False
@pytest.mark.asyncio
async def test_local_yaml_adapter_rejects_path_traversal_like_assistant_id(tmp_path):
adapter = LocalYamlAssistantConfigAdapter(str(tmp_path))
payload = await adapter.fetch_assistant_config("../etc/passwd")
assert payload == {"__error_code": "assistant.not_found", "assistantId": "../etc/passwd"}
@pytest.mark.asyncio
async def test_local_yaml_translates_agent_schema_to_runtime_services(tmp_path):
config_dir = tmp_path / "assistants"
config_dir.mkdir(parents=True, exist_ok=True)
(config_dir / "default.yaml").write_text(
"\n".join(
[
"agent:",
" llm:",
" provider: openai",
" model: gpt-4o-mini",
" api_key: sk-llm",
" api_url: https://api.example.com/v1",
" tts:",
" provider: openai_compatible",
" model: tts-model",
" api_key: sk-tts",
" api_url: https://tts.example.com/v1/audio/speech",
" voice: anna",
" asr:",
" provider: openai_compatible",
" model: asr-model",
" api_key: sk-asr",
" api_url: https://asr.example.com/v1/audio/transcriptions",
" duplex:",
" system_prompt: You are test assistant",
]
),
encoding="utf-8",
)
adapter = LocalYamlAssistantConfigAdapter(str(config_dir))
payload = await adapter.fetch_assistant_config("default")
assert isinstance(payload, dict)
assistant = payload.get("assistant", {})
services = assistant.get("services", {})
assert services.get("llm", {}).get("apiKey") == "sk-llm"
assert services.get("tts", {}).get("apiKey") == "sk-tts"
assert services.get("asr", {}).get("apiKey") == "sk-asr"
assert assistant.get("systemPrompt") == "You are test assistant"