把数据直接丢给 Clef 跑决策任务，最常见的两类失败是：输入侧字段缺失或类型不对，Clef 仍然返回一个看起来正常的评分；输出侧格式与下游约定不一致，解析器直接抛错，或者把结果写进了错误分支。这两种问题都不在决策逻辑本身，而在调用前后的结构对齐。可行的顺序是：先把输入字段定成一份可校验的清单，再把业务约束翻译成明确的输出格式要求，然后用输入校验函数和输出解析函数把两个边界卡住，最后用固定样本跑通一遍完整链路。

适用场景：Clef 的返回要直接进入下游系统或人工审核队列，格式错一次就影响后续动作。操作动作：先落一份输入字段清单（字段名、类型、是否必填、取值范围），再把输出枚举、评分区间、排序规则写成可校验的约束，调用前后各加一层校验函数。验证方式：用正常、缺字段、边界值三组样本跑完整链路，比对实际返回与预期结构。风险边界：Clef 版本或配置变更后约束可能失效，需要重新对齐；约束写得过紧时，原本可用的返回也可能被判成异常。

定义 Clef 输入的必要字段和可选字段

先分清“缺了它决策结果会明显变样”和“有更好但不影响主判断”。前者是必填，后者是可选。这份模板可以直接放到项目里的 schema 文件，字段名按实际接口替换。

{
  "request_id":  {"type": "string",  "required": true,  "note": "调用方唯一标识，用于日志关联"},
  "scene":       {"type": "string",  "required": true,  "enum": ["credit", "risk", "routing"]},
  "user_id":     {"type": "string",  "required": true,  "note": "决策主体"},
  "features":    {"type": "object",  "required": true,  "note": "键名需与配置侧一致"},
  "amount":      {"type": "number",  "required": false, "range": [0, 1000000]},
  "history_cnt": {"type": "integer", "required": false, "range": [0, 10000]},
  "locale":      {"type": "string",  "required": false, "enum": ["zh-CN", "en-US"]}
}

• 取值范围的写法要具体到闭区间或枚举，不要写“合理值”“正数即可”，否则校验函数无法判断。
• 可选字段缺失时的处理要提前定一种：要么调用方显式传 null，要么由配置侧补默认值，两种只能选一种，并写进接口约定。
• features 这类嵌套对象，至少要约定顶层键名；键名拼错通常不会报错，只会让 Clef 在缺失特征上做判断，评分看起来正常但不可解释。

把业务约束翻译成输出格式要求

业务方说的“给个明确结论”“分数别太粗糙”，落到 Clef 的返回上必须换成可判断的句子，并且这份描述要在请求参数或提示词里出现，同时在解析层再校验一次。

• 输出枚举：decision 只接受 allow / deny / review 三个字符串，明确写明不要返回中文、布尔值或整段解释文字。
• 评分范围：score 为 0 到 100 的整数；如果 Clef 倾向输出小数，先约定是取整还是四舍五入，不要让下游各自处理。
• 排序规则：candidates 按 score 降序，score 相同时按 id 升序，这样同一份输入两次调用得到的结果顺序可比。
• 结构层级：结果放在固定键下，例如 {"decision": ..., "score": ..., "reasons": [...]}，并要求返回内容本身就是 JSON，不要包在 Markdown 代码块里、也不要在前后加解释句。

这几条描述可以直接作为请求里的约束段落。写完要检查一遍：每条都必须是“能被一个 if 判断”的，凡是需要人读一遍才能判断的表述，下游解析不了。

写一个输入校验函数，在调用前检查字段完整性

校验放在组装请求之后、真正发出调用之前。下面的骨架是通用写法，字段名和范围按上面那份 schema 替换即可。

import logging
logger = logging.getLogger("clef.input")

REQUIRED_FIELDS = ["request_id", "scene", "user_id", "features"]
RANGES = {"amount": (0, 1000000), "history_cnt": (0, 10000)}

def validate_input(payload):
    errors = []
    if not isinstance(payload, dict):
        return False, ["payload 不是对象"]

    for name in REQUIRED_FIELDS:
        if name not in payload or payload[name] in (None, "", [], {}):
            errors.append("missing required field: %s" % name)

    for name, (low, high) in RANGES.items():
        if name in payload and payload[name] is not None:
            v = payload[name]
            if isinstance(v, bool) or not isinstance(v, (int, float)):
                errors.append("%s 类型不是数值: %s" % (name, type(v).__name__))
            elif not (low <= v <= high):
                errors.append("%s 超出范围 [%s, %s]: %r" % (name, low, high, v))

    if errors:
        logger.warning("input validation failed request_id=%s errors=%s",
                       payload.get("request_id"), errors)
        return False, errors
    return True, []

日志上有两点值得注意：request_id 即使校验失败也要尽量带上，否则出问题时对不上号；不要整条 payload 打进日志，features 里可能有敏感值，只记录字段名、字段类型和越界值。校验不通过就直接返回错误给调用方，不要带上残缺字段去试一次。

写一个输出解析函数，校验 Clef 返回是否符合约束

解析函数要做三件事：把原始返回变成结构化对象、逐条核对输出约束、把不符合约束的情况变成明确异常，而不是用默认值兜底。

import json

ALLOWED_DECISION = {"allow", "deny", "review"}

class DecisionFormatError(ValueError):
    pass

def parse_decision(raw):
    if isinstance(raw, bytes):
        raw = raw.decode("utf-8", errors="strict")
    if not isinstance(raw, str):
        raise DecisionFormatError("返回不是文本: %s" % type(raw).__name__)

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:
        raise DecisionFormatError("JSON 解析失败: %s" % e) from e

    if not isinstance(data, dict):
        raise DecisionFormatError("顶层不是对象")

    decision = data.get("decision")
    if decision not in ALLOWED_DECISION:
        raise DecisionFormatError("decision 不在枚举内: %r" % decision)

    score = data.get("score")
    if isinstance(score, bool) or not isinstance(score, int) or not (0 <= score <= 100):
        raise DecisionFormatError("score 不合法: %r" % score)

    reasons = data.get("reasons", [])
    if not isinstance(reasons, list) or not all(isinstance(r, str) for r in reasons):
        raise DecisionFormatError("reasons 必须是字符串数组")

    return {"decision": decision, "score": score, "reasons": reasons}

异常处理策略放在调用层，不在解析函数里：捕获 DecisionFormatError 后，可以选择重试一次、收紧提示词里的约束再调，或者直接转人工审核。关键是不要吞掉异常返回一个默认决策，否则下游看到的是“格式正常但结论可疑”的结果，比直接报错更难排查。

用一组样本数据跑通输入到输出的完整链路

样本的作用是固定住结构和约束，不追求覆盖真实分布。三组就够起步，把它们写成脚本参数，便于重复执行。

1. case_ok：request_id、scene、user_id、features 齐全，amount=800。预期是输入校验通过并发出调用，返回经过解析后得到 decision 在枚举内、score 为 0 到 100 的整数、reasons 为字符串数组。
2. case_missing：去掉 features。预期是在输入校验阶段就被拦下，日志里出现 missing required field: features，并且不应发出 Clef 请求。
3. case_edge：amount=0，history_cnt 取范围内上限。预期是输入校验通过；若返回的 score 正好落在 0 或 100，仍走同一套解析逻辑，不因为边界值触发异常。

比对方式：把 parse_decision 的输出用 json.dumps(..., sort_keys=True) 序列化后与预期结构做 diff，逐键断言而不是只断言整体字符串，这样能直接看出是哪个字段不符合约束。断言失败时先判断是哪一层拦下的——输入校验还是输出解析，然后调出本次 Clef 的原始返回文本核对，脚本里建议保留原始返回，例如：

python run_clef_pipeline.py `--case` case_ok `--dump-raw`

三组样本都稳定通过之后，再把约束收紧或字段扩展；每改一次 schema 或输出约定，就把这三组重跑一遍，比在生产流量上发现格式不匹配要省事得多。