207 lines
7.0 KiB
Python
207 lines
7.0 KiB
Python
"""QualityGate - 产出质量管理
|
||
|
||
多维度质量检查：必填字段、字数、JSON Schema、自定义验证器。
|
||
"""
|
||
|
||
import importlib
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from typing import Callable
|
||
|
||
from agentkit.skills.base import Skill
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class QualityCheck:
|
||
"""单条质量检查结果"""
|
||
|
||
name: str
|
||
passed: bool
|
||
message: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class QualityResult:
|
||
"""质量检查汇总结果"""
|
||
|
||
passed: bool
|
||
checks: list[QualityCheck]
|
||
can_retry: bool
|
||
|
||
|
||
class QualityGate:
|
||
"""产出质量管理 — 多维度质量检查"""
|
||
|
||
async def validate(
|
||
self,
|
||
output: dict[str, object],
|
||
skill: Skill,
|
||
skill_context: dict[str, object] | None = None,
|
||
) -> QualityResult:
|
||
"""对产出执行多维度质量检查
|
||
|
||
检查维度：
|
||
1. 必填字段检查
|
||
2. 最低字数检查
|
||
3. JSON Schema 验证（如 skill.config.output_schema 存在）
|
||
4. 自定义验证器（如 skill.config.quality_gate.custom_validator 存在）
|
||
5. 技能匹配验证（如 skill_context 含 intent_keywords）
|
||
"""
|
||
checks: list[QualityCheck] = []
|
||
qg = skill.config.quality_gate
|
||
|
||
# 1. 必填字段检查
|
||
for field in qg.required_fields:
|
||
present = field in output and output[field] is not None
|
||
checks.append(
|
||
QualityCheck(
|
||
name=f"required_field:{field}",
|
||
passed=present,
|
||
message=f"Field '{field}' is missing" if not present else None,
|
||
)
|
||
)
|
||
|
||
# 2. 最低字数检查
|
||
if qg.min_word_count > 0:
|
||
content = output.get("content", "")
|
||
if isinstance(content, str):
|
||
word_count = len(content.split())
|
||
else:
|
||
word_count = len(str(content).split())
|
||
passed = word_count >= qg.min_word_count
|
||
checks.append(
|
||
QualityCheck(
|
||
name="min_word_count",
|
||
passed=passed,
|
||
message=(
|
||
f"Word count {word_count} < minimum {qg.min_word_count}"
|
||
if not passed
|
||
else None
|
||
),
|
||
)
|
||
)
|
||
|
||
# 3. JSON Schema 验证
|
||
if skill.config.output_schema:
|
||
try:
|
||
import jsonschema
|
||
|
||
jsonschema.validate(output, skill.config.output_schema)
|
||
checks.append(QualityCheck(name="schema", passed=True))
|
||
except jsonschema.ValidationError as e:
|
||
checks.append(QualityCheck(name="schema", passed=False, message=str(e)))
|
||
except ImportError:
|
||
# jsonschema 未安装，跳过
|
||
pass
|
||
|
||
# 4. 自定义验证器
|
||
if qg.custom_validator:
|
||
try:
|
||
validator = self._import_validator(qg.custom_validator)
|
||
result = validator(output)
|
||
# 支持异步验证器
|
||
if hasattr(result, "__await__"):
|
||
result = await result
|
||
checks.append(QualityCheck(name="custom", passed=bool(result)))
|
||
except Exception as e:
|
||
# 验证器导入/执行失败，跳过并记录警告
|
||
checks.append(
|
||
QualityCheck(
|
||
name="custom",
|
||
passed=True,
|
||
message=f"Validator skipped: {e}",
|
||
)
|
||
)
|
||
|
||
# 5. 技能匹配验证（轻量级路由一致性检查）
|
||
skill_match_check = self._check_skill_match(output, skill_context)
|
||
if skill_match_check is not None:
|
||
checks.append(skill_match_check)
|
||
|
||
# 警告升级逻辑：当 skill_match 警告存在且其他维度有失败时，升级为失败
|
||
if (
|
||
skill_match_check is not None
|
||
and skill_match_check.message
|
||
and "Warning" in skill_match_check.message
|
||
):
|
||
other_failed = any(not c.passed for c in checks if c.name != "skill_match")
|
||
if other_failed:
|
||
# 升级：将 skill_match 的 passed 也设为 False
|
||
checks = [
|
||
QualityCheck(name=c.name, passed=False, message=c.message)
|
||
if c.name == "skill_match"
|
||
else c
|
||
for c in checks
|
||
]
|
||
|
||
return QualityResult(
|
||
passed=all(c.passed for c in checks),
|
||
checks=checks,
|
||
can_retry=qg.max_retries > 0,
|
||
)
|
||
|
||
# 允许的验证器模块前缀白名单
|
||
_ALLOWED_VALIDATOR_PREFIXES = (
|
||
"agentkit.",
|
||
"app.agent_framework.",
|
||
)
|
||
|
||
@staticmethod
|
||
def _check_skill_match(
|
||
output: dict[str, object],
|
||
skill_context: dict[str, object] | None,
|
||
) -> QualityCheck | None:
|
||
"""第五维度：技能匹配验证
|
||
|
||
当 skill_context 含 intent_keywords 时，检查输出内容是否包含
|
||
至少一个关键词。不匹配时标记为警告（passed=True + message），
|
||
当其他维度也有失败时升级为 passed=False。
|
||
|
||
Returns:
|
||
QualityCheck 或 None（当 skill_context 无效时跳过）
|
||
"""
|
||
if not skill_context:
|
||
return None
|
||
|
||
intent_keywords: list[str] | None = skill_context.get("intent_keywords")
|
||
if not intent_keywords:
|
||
return None
|
||
|
||
# 拼接输出中所有字符串值
|
||
all_text = " ".join(
|
||
str(v) for v in output.values() if isinstance(v, (str, int, float, bool))
|
||
).lower()
|
||
|
||
matched = any(kw.lower() in all_text for kw in intent_keywords)
|
||
if matched:
|
||
return QualityCheck(name="skill_match", passed=True)
|
||
|
||
return QualityCheck(
|
||
name="skill_match",
|
||
passed=True, # 警告级别，不单独拦截
|
||
message="Warning: output may not match routed skill",
|
||
)
|
||
|
||
def _import_validator(self, dotted_path: str) -> Callable:
|
||
"""从点分路径导入自定义验证器函数
|
||
|
||
出于安全考虑，只允许导入白名单前缀下的模块。
|
||
"""
|
||
# 安全校验：只允许白名单前缀的模块
|
||
if not any(dotted_path.startswith(prefix) for prefix in self._ALLOWED_VALIDATOR_PREFIXES):
|
||
raise ImportError(
|
||
f"Validator '{dotted_path}' is not in allowed module prefixes: "
|
||
f"{self._ALLOWED_VALIDATOR_PREFIXES}"
|
||
)
|
||
try:
|
||
module_path, func_name = dotted_path.rsplit(".", 1)
|
||
module = importlib.import_module(module_path)
|
||
handler = getattr(module, func_name)
|
||
if not callable(handler):
|
||
raise ValueError(f"'{dotted_path}' is not callable")
|
||
return handler
|
||
except (ImportError, AttributeError, ValueError) as e:
|
||
raise ImportError(f"Failed to import validator '{dotted_path}': {e}") from e
|