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

Add new tools to DuplexPipeline: calculator, code_interpreter, turn_on_camera, turn_off_camera, increase_volume, and decrease_volume. Implement fallback schema for unknown string tools and assign default client executors for specific tools. Update tests to validate new functionality and ensure correct tool handling in the pipeline.

Browse Source
This commit is contained in:
Xin Wang
2026-02-27 13:59:37 +08:00
parent 8d453e10da
commit d942c85eff
3 changed files with 214 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
engine/tests/test_tool_call_flow.py
View File

@@ -97,6 +97,7 @@ def test_pipeline_uses_default_tools_from_settings(monkeypatch):
"core.duplex_pipeline.settings.tools",
[
"current_time",
"calculator",
{
"name": "weather",
"description": "Get weather by city",
@@ -112,14 +113,36 @@ def test_pipeline_uses_default_tools_from_settings(monkeypatch):
pipeline, _events = _build_pipeline(monkeypatch, [[LLMStreamEvent(type="done")]])
cfg = pipeline.resolved_runtime_config()
assert cfg["tools"]["allowlist"] == ["current_time", "weather"]
assert cfg["tools"]["allowlist"] == ["calculator", "current_time", "weather"]
schemas = pipeline._resolved_tool_schemas()
names = [s.get("function", {}).get("name") for s in schemas if isinstance(s, dict)]
assert "current_time" in names
assert "calculator" in names
assert "weather" in names
def test_pipeline_exposes_unknown_string_tools_with_fallback_schema(monkeypatch):
monkeypatch.setattr("core.duplex_pipeline.settings.tools", ["custom_system_cmd"])
pipeline, _events = _build_pipeline(monkeypatch, [[LLMStreamEvent(type="done")]])
schemas = pipeline._resolved_tool_schemas()
tool_schema = next((s for s in schemas if s.get("function", {}).get("name") == "custom_system_cmd"), None)
assert tool_schema is not None
assert tool_schema.get("function", {}).get("parameters", {}).get("type") == "object"
def test_pipeline_assigns_default_client_executor_for_system_string_tools(monkeypatch):
monkeypatch.setattr("core.duplex_pipeline.settings.tools", ["increase_volume"])
pipeline, _events = _build_pipeline(monkeypatch, [[LLMStreamEvent(type="done")]])
tool_call = {
"type": "function",
"function": {"name": "increase_volume", "arguments": "{}"},
}
assert pipeline._tool_executor(tool_call) == "client"
@pytest.mark.asyncio
async def test_ws_message_parses_tool_call_results():
msg = parse_client_message(