From 924160c779f5ab7fdc147228bf8a526dc2e202a6 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 9 Jul 2026 12:20:16 -0400 Subject: [PATCH] =?UTF-8?q?feat(ports):=20=E6=96=B0=E5=A2=9E=20ToolDispatc?= =?UTF-8?q?hFactory/PromptBuilderFactory=20Protocol?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 4 个新 Protocol 类型: - ToolDispatchFn: 工具调度函数签名 - ToolDispatchFactory: per-version 工具调度工厂 - PromptBuilderFn: Prompt 构建函数签名 - PromptBuilderFactory: per-version prompt 构建工厂 含 runtime_checkable isinstance 测试。 Co-Authored-By: Claude Opus 4.6 (1M context) --- app/ports.py | 72 ++++++++++++++++- tests/unit/test_ports_factory.py | 128 +++++++++++++++++++++++++++++++ 2 files changed, 199 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_ports_factory.py diff --git a/app/ports.py b/app/ports.py index 2584c17..4a0b6ce 100644 --- a/app/ports.py +++ b/app/ports.py @@ -3,7 +3,7 @@ from __future__ import annotations from pathlib import Path # noqa: TC003 — runtime_checkable Protocol 需运行时可见 -from typing import TYPE_CHECKING, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable if TYPE_CHECKING: import numpy as np @@ -74,3 +74,73 @@ class OCRProvider(Protocol): """ async def transcribe_frames(self, frame_paths: list[Path]) -> str: ... + + +@runtime_checkable +class ToolDispatchFn(Protocol): + """工具调度函数签名。 + + 参数: + tool_name: 工具名称。 + args: 工具参数字典。 + context: 上下文字典（包含 session_id）。 + + 返回: + 工具执行结果文本。 + """ + + async def __call__( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: ... + + +@runtime_checkable +class ToolDispatchFactory(Protocol): + """per-version 工具调度工厂。 + + 通过注入 skills_dir 生产对应版本的 ToolDispatchFn。 + + 参数: + skills_dir: skill 文件目录（None 则不加载 skill）。 + + 返回: + ToolDispatchFn 实例。 + """ + + def __call__(self, *, skills_dir: Path | None = None) -> ToolDispatchFn: ... + + +@runtime_checkable +class PromptBuilderFn(Protocol): + """Prompt 构建函数签名。 + + 参数: + qa: 待构建 prompt 的题目。 + + 返回: + (system_prompt, user_prompt) 二元组。 + """ + + def __call__(self, qa: GeneratedQuestion) -> tuple[str, str]: ... + + +@runtime_checkable +class PromptBuilderFactory(Protocol): + """per-version prompt 构建工厂。 + + 通过注入 skills_dir 和 prompts_dir 生产对应版本的 PromptBuilderFn。 + + 参数: + skills_dir: skill 文件目录（None 则不加载 skill）。 + prompts_dir: prompt 文件目录（None 则使用默认目录）。 + + 返回: + PromptBuilderFn 实例。 + """ + + def __call__( + self, + *, + skills_dir: Path | None = None, + prompts_dir: Path | None = None, + ) -> PromptBuilderFn: ... diff --git a/tests/unit/test_ports_factory.py b/tests/unit/test_ports_factory.py new file mode 100644 index 0000000..d782b6a --- /dev/null +++ b/tests/unit/test_ports_factory.py @@ -0,0 +1,128 @@ +"""app/ports.py 工厂 Protocol isinstance 检查测试。""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from app.ports import ( + PromptBuilderFactory, + PromptBuilderFn, + ToolDispatchFactory, + ToolDispatchFn, +) + +if TYPE_CHECKING: + from pathlib import Path + + +# --------------------------------------------------------------------------- +# ToolDispatchFn +# --------------------------------------------------------------------------- + + +class TestToolDispatchFnProtocol: + """ToolDispatchFn runtime_checkable isinstance 检查。""" + + def test_conforming_async_callable(self) -> None: + """符合签名的 async callable 通过 isinstance 检查。""" + + async def dispatch(tool_name: str, args: dict, *, context: dict) -> str: + return "ok" + + assert isinstance(dispatch, ToolDispatchFn) + + def test_non_callable_fails(self) -> None: + """非 callable 不通过 isinstance 检查。""" + assert not isinstance("not_a_callable", ToolDispatchFn) + + +# --------------------------------------------------------------------------- +# ToolDispatchFactory +# --------------------------------------------------------------------------- + + +class TestToolDispatchFactoryProtocol: + """ToolDispatchFactory runtime_checkable isinstance 检查。""" + + def test_conforming_factory_class(self) -> None: + """符合签名的工厂类通过 isinstance 检查。""" + + class _Factory: + def __call__(self, *, skills_dir: Path | None = None) -> ToolDispatchFn: + async def _d(tool_name: str, args: dict, *, context: dict) -> str: + return "ok" + + return _d # type: ignore[return-value] + + assert isinstance(_Factory(), ToolDispatchFactory) + + def test_conforming_function(self) -> None: + """符合签名的普通函数通过 isinstance 检查。""" + + def factory(*, skills_dir: Path | None = None) -> ToolDispatchFn: + async def _d(tool_name: str, args: dict, *, context: dict) -> str: + return "ok" + + return _d # type: ignore[return-value] + + assert isinstance(factory, ToolDispatchFactory) + + +# --------------------------------------------------------------------------- +# PromptBuilderFn +# --------------------------------------------------------------------------- + + +class TestPromptBuilderFnProtocol: + """PromptBuilderFn runtime_checkable isinstance 检查。""" + + def test_conforming_callable(self) -> None: + """符合签名的 callable 通过 isinstance 检查。""" + + def builder(qa: object) -> tuple[str, str]: + return ("system", "user") + + assert isinstance(builder, PromptBuilderFn) + + def test_non_callable_fails(self) -> None: + """非 callable 不通过 isinstance 检查。""" + assert not isinstance(42, PromptBuilderFn) + + +# --------------------------------------------------------------------------- +# PromptBuilderFactory +# --------------------------------------------------------------------------- + + +class TestPromptBuilderFactoryProtocol: + """PromptBuilderFactory runtime_checkable isinstance 检查。""" + + def test_conforming_factory_class(self) -> None: + """符合签名的工厂类通过 isinstance 检查。""" + + class _Factory: + def __call__( + self, + *, + skills_dir: Path | None = None, + prompts_dir: Path | None = None, + ) -> PromptBuilderFn: + def _b(qa: object) -> tuple[str, str]: + return ("s", "u") + + return _b # type: ignore[return-value] + + assert isinstance(_Factory(), PromptBuilderFactory) + + def test_conforming_function(self) -> None: + """符合签名的普通函数通过 isinstance 检查。""" + + def factory( + *, skills_dir: Path | None = None, prompts_dir: Path | None = None + ) -> PromptBuilderFn: + def _b(qa: object) -> tuple[str, str]: + return ("s", "u") + + return _b # type: ignore[return-value] + + assert isinstance(factory, PromptBuilderFactory)