174 lines
4.2 KiB
Python
174 lines
4.2 KiB
Python
"""应用层 Protocol 端口定义。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path # noqa: TC003 — runtime_checkable Protocol 需运行时可见
|
|
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
|
|
|
|
if TYPE_CHECKING:
|
|
import numpy as np
|
|
|
|
from app.harness.pools import Pools
|
|
from app.tree.index import TreeIndex
|
|
from core.types import GeneratedQuestion, PoolConfig
|
|
|
|
|
|
@runtime_checkable
|
|
class EmbeddingProvider(Protocol):
|
|
"""文本嵌入端口。
|
|
|
|
属性:
|
|
dim: 嵌入维度 D。
|
|
"""
|
|
|
|
@property
|
|
def dim(self) -> int: ...
|
|
|
|
def embed(self, texts: str | list[str]) -> np.ndarray:
|
|
"""文本 → 嵌入向量（L2 归一化）。
|
|
|
|
参数:
|
|
texts: 单条文本或文本列表。
|
|
|
|
返回:
|
|
[N, D] ndarray，每行 L2 范数为 1.0。
|
|
"""
|
|
...
|
|
|
|
|
|
@runtime_checkable
|
|
class QuestionGenerator(Protocol):
|
|
"""LLM 驱动的题目生成端口（预留接口）。
|
|
|
|
参数:
|
|
video_id: 视频标识。
|
|
task_type: 题型。
|
|
tree: 视频树索引，提供锚节点上下文。
|
|
exemplars: 风格示例题目列表。
|
|
|
|
返回:
|
|
生成的单条题目。
|
|
"""
|
|
|
|
async def generate(
|
|
self,
|
|
video_id: str,
|
|
task_type: str,
|
|
tree: TreeIndex,
|
|
*,
|
|
exemplars: list[GeneratedQuestion],
|
|
) -> GeneratedQuestion: ...
|
|
|
|
|
|
@runtime_checkable
|
|
class OCRProvider(Protocol):
|
|
"""帧文字转录端口。
|
|
|
|
实现方负责将帧图像发送给 OCR 服务并返回拼接后的文本。
|
|
单帧失败应降级跳过，不得抛出异常中断整体流程。
|
|
|
|
参数:
|
|
frame_paths: 帧文件路径列表。
|
|
|
|
返回:
|
|
"帧1: <行1> | <行2>\\n帧2: ..." 格式文本；无有效结果时空串。
|
|
"""
|
|
|
|
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: ...
|
|
|
|
|
|
@runtime_checkable
|
|
class PoolStrategy(Protocol):
|
|
"""池构建策略端口。
|
|
|
|
应用层端口（非 core 层），因为返回类型 Pools 定义在 app/harness/pools.py。
|
|
两个具体策略（GlobalPoolStrategy / PerCategoryPoolStrategy）实现此接口。
|
|
"""
|
|
|
|
def build(
|
|
self,
|
|
questions: list[GeneratedQuestion],
|
|
correctness: dict[str, bool],
|
|
config: PoolConfig,
|
|
*,
|
|
db_path: Path | None = None,
|
|
) -> Pools: ...
|
|
|
|
def build_incremental(
|
|
self,
|
|
new_task_types: list[str],
|
|
questions: list[GeneratedQuestion],
|
|
correctness: dict[str, bool],
|
|
config: PoolConfig,
|
|
) -> dict[str, dict[str, list[str]]]: ...
|