tzt
5078721df4
- model_pool：条目新增 capabilities{vision,tools,context_window}，
归一化缺省全兼容（老条目行为零变化；非法形态回落缺省）；
filter_by_capabilities 硬过滤（context_window=0 视为不限）
- proxy routes：_request_needs 从请求体推断需求（多模态 image_url → vision、
tools 非空 → tools、上下文需求 = 字符/3 + min(max_tokens,4096) 与预扣同口径）；
条目不满足时在池内重定向到首个合格条目
- 与 T-X1/T-X2 组合语义：能力重定向 → 预算降档 → 降级链，逐级独立不互扰
pytest 461 passed（T-X3 后 455 + 6）
95 lines
4.6 KiB
Python
95 lines
4.6 KiB
Python
"""模型池能力位过滤测试（T-X6）：归一化缺省 / 硬过滤 / 请求需求推断。"""
|
|
import pytest
|
|
|
|
from gateway.model_pool import PoolStore, filter_by_capabilities
|
|
from gateway.proxy.routes import _entry_meets, _request_needs
|
|
|
|
pytest.importorskip("fastapi")
|
|
|
|
|
|
def _make_pool(tmp_path) -> PoolStore:
|
|
store = PoolStore(path=tmp_path / "pool.json")
|
|
store.upsert({"id": "text-only", "name": "纯文本", "tier": "budget",
|
|
"backend": "openai", "base_url": "http://a", "model": "text-m",
|
|
"capabilities": {"vision": False, "tools": True,
|
|
"context_window": 8192},
|
|
"enabled": True})
|
|
store.upsert({"id": "vision", "name": "多模态", "tier": "premium",
|
|
"backend": "openai", "base_url": "http://b", "model": "vision-m",
|
|
"capabilities": {"vision": True, "tools": True,
|
|
"context_window": 32768},
|
|
"enabled": True})
|
|
store.upsert({"id": "legacy", "name": "老条目", "tier": "local",
|
|
"backend": "llama_server", "base_url": "http://c",
|
|
"model": "legacy-m", "enabled": True}) # 无 capabilities 字段
|
|
return store
|
|
|
|
|
|
def test_capabilities_default_full_compat(tmp_path):
|
|
"""未声明能力位的老条目：缺省全兼容（vision/tools True、context 0 不限）。"""
|
|
pool = _make_pool(tmp_path)
|
|
e = pool.find_by_model("legacy-m")
|
|
assert e["capabilities"] == {"vision": True, "tools": True, "context_window": 0}
|
|
|
|
|
|
def test_capabilities_invalid_shapes_fall_back(tmp_path):
|
|
pool = _make_pool(tmp_path)
|
|
pool.upsert({"id": "weird", "name": "怪", "tier": "budget", "backend": "openai",
|
|
"base_url": "http://d", "model": "weird-m",
|
|
"capabilities": {"vision": "是", "context_window": "abc"},
|
|
"enabled": True})
|
|
cap = pool.find_by_model("weird-m")["capabilities"]
|
|
assert cap["vision"] is True # 非布尔按缺省
|
|
assert cap["context_window"] == 0 # 非整数回落不限
|
|
|
|
|
|
def test_filter_by_capabilities_hard_rules():
|
|
entries = [{"capabilities": {"vision": False, "tools": True, "context_window": 8192}},
|
|
{"capabilities": {"vision": True, "tools": True, "context_window": 32768}},
|
|
{"capabilities": {"vision": True, "tools": False, "context_window": 0}}]
|
|
out = filter_by_capabilities(entries, need_vision=True)
|
|
assert len(out) == 2
|
|
out = filter_by_capabilities(entries, need_vision=True, need_tools=True)
|
|
assert len(out) == 1
|
|
out = filter_by_capabilities(entries, need_vision=True, need_tools=True,
|
|
min_context_tokens=16384)
|
|
assert len(out) == 1 and out[0]["capabilities"]["context_window"] == 32768
|
|
# context_window=0（不限）不受 min_context 约束
|
|
out = filter_by_capabilities(entries, min_context_tokens=999999)
|
|
assert len(out) == 1 and out[0]["capabilities"]["context_window"] == 0
|
|
|
|
|
|
def test_request_needs_detection():
|
|
body = {"messages": [
|
|
{"role": "user", "content": "普通文本问题"},
|
|
{"role": "user", "content": [{"type": "text", "text": "看这张图"},
|
|
{"type": "image_url",
|
|
"image_url": {"url": "data:image/png;base64,x"}}]},
|
|
], "tools": [{"type": "function", "function": {"name": "f"}}],
|
|
"max_tokens": 3000}
|
|
needs = _request_needs(body)
|
|
assert needs["vision"] is True and needs["tools"] is True
|
|
assert needs["min_context_tokens"] > 3000
|
|
|
|
plain = _request_needs({"messages": [{"role": "user", "content": "hi"}]})
|
|
assert plain["vision"] is False and plain["tools"] is False
|
|
assert plain["min_context_tokens"] > 0
|
|
|
|
|
|
def test_entry_meets_defaults_and_limits():
|
|
legacy = {"capabilities": {"vision": True, "tools": True, "context_window": 0}}
|
|
assert _entry_meets(legacy, {"vision": True, "tools": True,
|
|
"min_context_tokens": 999999})
|
|
limited = {"capabilities": {"vision": True, "tools": True, "context_window": 4096}}
|
|
assert _entry_meets(limited, {"vision": False, "tools": False,
|
|
"min_context_tokens": 4096})
|
|
assert not _entry_meets(limited, {"vision": False, "tools": False,
|
|
"min_context_tokens": 4097})
|
|
|
|
|
|
def test_pool_list_returns_capabilities(tmp_path):
|
|
"""list() 打码输出保留 capabilities 字段（前端可见）。"""
|
|
pool = _make_pool(tmp_path)
|
|
e = next(x for x in pool.list()["entries"] if x["id"] == "vision")
|
|
assert e["capabilities"]["vision"] is True
|