feat(ports): 新增 ToolDispatchFactory/PromptBuilderFactory Protocol

4 个新 Protocol 类型:
- ToolDispatchFn: 工具调度函数签名
- ToolDispatchFactory: per-version 工具调度工厂
- PromptBuilderFn: Prompt 构建函数签名
- PromptBuilderFactory: per-version prompt 构建工厂

含 runtime_checkable isinstance 测试。

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-09 12:20:16 -04:00
parent e7be42570d
commit 924160c779
2 changed files with 199 additions and 1 deletions
+71 -1
View File
@@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
from pathlib import Path # noqa: TC003 — runtime_checkable Protocol 需运行时可见 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: if TYPE_CHECKING:
import numpy as np import numpy as np
@@ -74,3 +74,73 @@ class OCRProvider(Protocol):
""" """
async def transcribe_frames(self, frame_paths: list[Path]) -> str: ... 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: ...
+128
View File
@@ -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)