149 lines
5.6 KiB
Python
149 lines
5.6 KiB
Python
"""模型能力档案: 不同模型的参数差异都收敛到 yaml,加新模型不用改代码。"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field, fields
|
||
from pathlib import Path
|
||
from typing import List, Optional
|
||
|
||
import yaml
|
||
|
||
from .llm_params import THINKING_TRANSPORTS
|
||
|
||
REASONING_REPLAY_POLICIES = {
|
||
"none",
|
||
"tool_turn",
|
||
"conversation",
|
||
"provider_managed",
|
||
}
|
||
|
||
|
||
def model_profile_of(caps: object) -> str:
|
||
"""返回能力对象的 canonical profile；兼容测试和旧调用方的鸭子类型。"""
|
||
profile = getattr(caps, "profile", "")
|
||
if profile:
|
||
return str(profile)
|
||
return f"{getattr(caps, 'family', '')}.{getattr(caps, 'variant', '')}".strip(".")
|
||
|
||
|
||
@dataclass
|
||
class ModelCapabilities:
|
||
model_id: str = ""
|
||
family: str = ""
|
||
variant: str = ""
|
||
display_name: str = "" # UI 展示用,如 "DeepSeek V4 Flash";空时前端 fallback 拼 family.variant
|
||
|
||
# 上下文
|
||
max_context: int = 128_000
|
||
reliable_context: int = 64_000
|
||
max_output: int = 4096
|
||
|
||
# Tool calling
|
||
parallel_tools: bool = False
|
||
tool_calling_quality: str = "good"
|
||
|
||
# 思考开关
|
||
thinking_enabled: bool = False
|
||
# none=不猜 provider 默认值；extra_body=显式发送 thinking.type，effort 同体透传。
|
||
thinking_transport: str = "none"
|
||
reasoning_effort_levels: List[str] = field(default_factory=list)
|
||
default_reasoning_effort: str = ""
|
||
# 是否清除历史 thinking。None=不发送该 provider 可选字段；False=保留历史推理。
|
||
thinking_clear: Optional[bool] = None
|
||
# 历史 reasoning 发回模型的策略。持久化原始响应与 provider-bound 输入分离：
|
||
# none=剥离；tool_turn=仅当前工具链；conversation=同模型会话；
|
||
# provider_managed=同模型不透明签名/block。
|
||
reasoning_replay: str = "none"
|
||
|
||
# 原生输入模态。附件仍以文件引用落库，发模型前才物化成 provider content blocks。
|
||
input_modalities: List[str] = field(default_factory=lambda: ["text"])
|
||
|
||
# 代码 / 沙盒
|
||
code_quality: str = "good"
|
||
enable_run_python: bool = False
|
||
|
||
# 工程参数
|
||
max_iterations: int = 120 # 单轮自主步数 backstop;空转防护见 loop 无进展熔断,不靠这个砍正经长任务
|
||
optimal_temperature: float = 0.3
|
||
|
||
# provider 特性
|
||
prompt_caching: bool = False
|
||
extended_thinking: bool = False
|
||
|
||
# 计费兜底(CNY / million tokens)。provider / LiteLLM cost map 缺失时使用。
|
||
input_cny_per_mtoken: float = 0.0
|
||
output_cny_per_mtoken: float = 0.0
|
||
# 前缀缓存命中价(DeepSeek 等自动缓存 prompt 前缀,命中部分按此价,通常 ~0.1x input)。
|
||
# 0 = 不区分,缓存命中按 input 全价记(安全兜底,不会少记)。
|
||
cache_hit_cny_per_mtoken: float = 0.0
|
||
|
||
# API 接入
|
||
api_base: str = ""
|
||
api_key_env: str = ""
|
||
|
||
@classmethod
|
||
def load(cls, name: str, models_dir: Path) -> "ModelCapabilities":
|
||
"""name: '<family>.<variant>',如 'deepseek_v4.flash'。"""
|
||
if "." in name:
|
||
family, variant = name.split(".", 1)
|
||
else:
|
||
family, variant = name, "default"
|
||
|
||
path = Path(models_dir) / f"{family}.yaml"
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"模型档案不存在: {path}")
|
||
|
||
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
variants = data.get("variants", {})
|
||
if variant not in variants:
|
||
raise ValueError(
|
||
f"档案 {path} 没有 variant={variant};可选: {list(variants)}"
|
||
)
|
||
|
||
var = dict(variants[variant])
|
||
# 已下架 profile 可保留为一版隐藏别名，确保存量 task / scheduled job 能续跑；
|
||
# 真正能力和记账身份统一归目标 variant，不维持旧模型双轨。
|
||
seen_aliases = {variant}
|
||
while var.get("alias_of"):
|
||
target = str(var["alias_of"]).strip()
|
||
if not target or target in seen_aliases or target not in variants:
|
||
raise ValueError(f"档案 {path} 的 variant={variant} alias_of 无效")
|
||
seen_aliases.add(target)
|
||
variant = target
|
||
var = dict(variants[target])
|
||
valid_keys = {f.name for f in fields(cls)}
|
||
kwargs = {k: v for k, v in var.items() if k in valid_keys}
|
||
kwargs["family"] = data.get("family", family)
|
||
kwargs["variant"] = variant
|
||
caps = cls(**kwargs)
|
||
if caps.thinking_transport not in THINKING_TRANSPORTS:
|
||
raise ValueError(
|
||
f"档案 {path} 的 thinking_transport={caps.thinking_transport!r} 无效;"
|
||
f"可选: {sorted(THINKING_TRANSPORTS)}"
|
||
)
|
||
if caps.reasoning_replay not in REASONING_REPLAY_POLICIES:
|
||
raise ValueError(
|
||
f"档案 {path} 的 reasoning_replay={caps.reasoning_replay!r} 无效;"
|
||
f"可选: {sorted(REASONING_REPLAY_POLICIES)}"
|
||
)
|
||
if caps.thinking_enabled and caps.thinking_transport == "none":
|
||
raise ValueError(
|
||
f"档案 {path} 开启 thinking 时必须声明可验证的 thinking_transport"
|
||
)
|
||
if (
|
||
caps.default_reasoning_effort
|
||
and caps.default_reasoning_effort not in caps.reasoning_effort_levels
|
||
):
|
||
raise ValueError(
|
||
f"档案 {path} 的 default_reasoning_effort="
|
||
f"{caps.default_reasoning_effort!r} 不在 reasoning_effort_levels 中"
|
||
)
|
||
return caps
|
||
|
||
@property
|
||
def profile(self) -> str:
|
||
return f"{self.family}.{self.variant}"
|
||
|
||
@property
|
||
def native_image_input(self) -> bool:
|
||
return "image" in self.input_modalities
|