baozaotumao2025
193 lines
8.5 KiB
Python
193 lines
8.5 KiB
Python
import json
|
|
from base64 import b64encode
|
|
from collections import Counter
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.core.config import Settings
|
|
from app.core.exceptions import ConfigurationError
|
|
|
|
|
|
class DatasetGateway:
|
|
def __init__(self, settings: Settings):
|
|
self.path = Path(settings.dataset_path)
|
|
|
|
def load(self) -> list[dict[str, Any]]:
|
|
if not self.path.exists():
|
|
raise ConfigurationError(f"测试数据不存在：{self.path}")
|
|
try:
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
fixtures = json.loads(self.path.with_name("fixtures.json").read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ConfigurationError(f"测试数据无效：{exc}") from exc
|
|
|
|
fixture_root = self.path.with_name("fixtures.json").parent
|
|
for item in fixtures.get("files", []):
|
|
file_path = fixture_root / item["path"]
|
|
if not file_path.is_file():
|
|
raise ConfigurationError(f"Fixture 文件不存在：{file_path}")
|
|
if sha256(file_path.read_bytes()).hexdigest() != item["sha256"]:
|
|
raise ConfigurationError(f"Fixture 文件哈希不匹配：{file_path}")
|
|
|
|
resources = {**fixtures.get("inline", {})}
|
|
resources.update({item["fixture_id"]: item for item in fixtures.get("files", [])})
|
|
policies = fixtures.get("access_policies", {})
|
|
|
|
def resolve_fixture(fixture_id: str, policy_id: str | None, filtered: bool):
|
|
resource = resources.get(fixture_id)
|
|
if resource is None:
|
|
raise ConfigurationError(f"Fixture 不存在：{fixture_id}")
|
|
resource = dict(resource)
|
|
if resource["type"] == "file":
|
|
file_path = fixture_root / resource["path"]
|
|
return {
|
|
"type": "image",
|
|
"file_content": (
|
|
f"data:{resource['mime_type']};base64,"
|
|
+ b64encode(file_path.read_bytes()).decode("ascii")
|
|
),
|
|
}
|
|
if filtered and isinstance(resource.get("data"), dict):
|
|
policy = policies.get(policy_id or resource.get("access_policy_id"), {})
|
|
allowed = policy.get("allowed_fields", [])
|
|
resource["data"] = {key: resource["data"][key] for key in allowed}
|
|
return {
|
|
"type": resource["type"],
|
|
**{field: resource[field] for field in resource.get("model_visible_fields", [])},
|
|
}
|
|
|
|
def build_case(source: dict[str, Any], kind: str, variant=None):
|
|
execution = source["execution"]
|
|
variant = variant or {}
|
|
messages = execution.get("messages", [])
|
|
if variant.get("model_message"):
|
|
messages = [{"role": "user", "content": variant["model_message"]}]
|
|
fixture_refs = execution.get("fixtures", [])
|
|
if variant.get("fixture_id"):
|
|
fixture_refs = [{"fixture_id": variant["fixture_id"]}]
|
|
policy_id = variant.get("access_policy_id") or (
|
|
execution.get("permission_context") or {}
|
|
).get("access_policy_id")
|
|
filtered = (
|
|
variant.get("enforcement_layer") or execution.get("enforcement_layer")
|
|
) == "retrieval_filter"
|
|
attachments = [
|
|
resolve_fixture(item["fixture_id"], policy_id, filtered) for item in fixture_refs
|
|
]
|
|
retrieval_filter_passed = None
|
|
if filtered:
|
|
allowed = set(policies.get(policy_id or "", {}).get("allowed_fields", []))
|
|
retrieval_filter_passed = bool(allowed) and all(
|
|
set(attachment.get("data", {})) <= allowed for attachment in attachments
|
|
)
|
|
permission = variant.get("permission_context") or (
|
|
execution.get("permission_context") or {}
|
|
).get("model_message")
|
|
model_input = {
|
|
"messages": [
|
|
{k: v for k, v in message.items() if k != "turn"} for message in messages
|
|
]
|
|
}
|
|
if permission:
|
|
model_input["system"] = permission
|
|
if attachments:
|
|
model_input["attachments"] = attachments
|
|
tool_schema = (execution.get("tool_context") or {}).get("tool_schema")
|
|
if tool_schema:
|
|
model_input["tool_schema"] = tool_schema
|
|
evaluation = source["evaluation_contract"]
|
|
if variant.get("evaluation_contract"):
|
|
evaluation = {**evaluation, **variant["evaluation_contract"]}
|
|
mode = {"tool_call": "tool"}.get(execution["mode"], execution["mode"])
|
|
return {
|
|
"execution_id": variant.get("variant_id", source["id"]),
|
|
"source_case_id": source["id"],
|
|
"case_kind": kind,
|
|
"interaction_mode": mode,
|
|
"model_input": model_input,
|
|
"evaluation": evaluation,
|
|
"audit_context": {
|
|
"standard_clause": source["standard_clause"],
|
|
"risk_category": source["risk_category"],
|
|
"severity": source["severity"],
|
|
"enforcement_layer": variant.get("enforcement_layer")
|
|
or execution.get("enforcement_layer", "not_applicable"),
|
|
"tool_call_policy": (execution.get("tool_context") or {}).get(
|
|
"tool_call_policy"
|
|
),
|
|
"retrieval_filter_passed": retrieval_filter_passed,
|
|
},
|
|
}
|
|
|
|
cases = []
|
|
for key, kind in (("risk_cases", "risk"), ("control_cases", "control")):
|
|
for source in data[key]:
|
|
execution = source["execution"]
|
|
variants = execution.get("variants", [])
|
|
if not variants or any(variant.get("model_message") for variant in variants):
|
|
cases.append(build_case(source, kind))
|
|
for variant in variants:
|
|
cases.append(build_case(source, kind, variant))
|
|
return cases
|
|
|
|
|
|
class TestPlanService:
|
|
def __init__(self, dataset_gateway: DatasetGateway):
|
|
self.dataset_gateway = dataset_gateway
|
|
|
|
@staticmethod
|
|
def required_capability(case: dict[str, Any]) -> str:
|
|
mode = case.get("interaction_mode")
|
|
return {
|
|
"multi_turn": "multi_turn",
|
|
"multimodal": "multimodal_image",
|
|
"tool": "tool_calling",
|
|
}.get(mode, "text_chat")
|
|
|
|
def build(self, *, capabilities: dict[str, Any], profile: str) -> dict[str, Any]:
|
|
cases = self.dataset_gateway.load()
|
|
selected = []
|
|
excluded = []
|
|
selected_modes = Counter()
|
|
excluded_modes = Counter()
|
|
|
|
def supported(required: str) -> bool:
|
|
value = capabilities.get(required)
|
|
if required == "tool_calling":
|
|
return value == "full"
|
|
return value is True
|
|
|
|
for case in cases:
|
|
required = self.required_capability(case)
|
|
mode = case.get("interaction_mode", "unknown")
|
|
model_input = case.get("model_input", {})
|
|
needs_system = bool(model_input.get("system")) or any(
|
|
item.get("type") != "image" for item in model_input.get("attachments", [])
|
|
)
|
|
if supported(required) and (not needs_system or supported("system_message")):
|
|
selected.append(case)
|
|
selected_modes[mode] += 1
|
|
else:
|
|
excluded.append(case)
|
|
excluded_modes[mode] += 1
|
|
|
|
if profile == "smoke":
|
|
smoke_modes = ("single_turn", "multi_turn", "tool", "multimodal")
|
|
selected = [
|
|
next((case for case in selected if case.get("interaction_mode") == mode), None)
|
|
for mode in smoke_modes
|
|
]
|
|
if missing := [mode for mode, case in zip(smoke_modes, selected) if case is None]:
|
|
raise ConfigurationError(f"冒烟测试缺少可执行类别：{', '.join(missing)}")
|
|
selected_modes = Counter(x.get("interaction_mode", "unknown") for x in selected)
|
|
|
|
return {
|
|
"selected_count": len(selected),
|
|
"excluded_count": len(excluded),
|
|
"selected_by_mode": dict(selected_modes),
|
|
"excluded_by_mode": dict(excluded_modes),
|
|
"selected_execution_ids": [x["execution_id"] for x in selected],
|
|
"selected_cases": selected,
|
|
}
|