6. 让系统知道什么时候不该自动决定

来源：Jev

预计阅读时间：45～60 分钟
本篇目标：为 Decision Inbox 增加人工复核通道，让低置信度、模糊判断和高风险动作无法悄悄进入自动处理路径。

上一篇的策略有意留下了一个漏洞：0.501 和 0.499 会落入两个不同分支，尽管它们都接近“无法判断”。一个可靠的判断系统不能只有“是”和“否”，还要有“目前证据不足，先别自动处理”。

TypeSafe 官方把置信度的用途分成三条路径：高置信度可以自动行动，中等置信度应谨慎处理，低置信度不应行动；具体边界取决于误判代价。官方还强调，校准是对一组预测的统计性质，并不保证某一次高置信度回答必然正确。详见 Confidence 文档 和 System One 文档。

这一篇不会寻找一个“放之四海而皆准”的神奇阈值，而是给项目建立可解释的拒绝机制。

一、先分清 probability 和 confidence

在当前输出中，这几个数字的含义不同：

字段	表达什么	本项目怎样使用
category_probabilities	各分类选项的概率分布	查看第一名是否明显胜出
category_confidence	分类分布集中程度的摘要	低于门槛时人工复核
urgency_score	紧急等级的概率加权位置	决定优先级
urgency_confidence	紧急等级分布集中程度的摘要	低于门槛时人工复核
action_requested_probability	“明确要求行动”为真的概率	高区间视为是，低区间视为否，中间不决定

Choice 和 Score 都返回概率分布及由分布计算出的 confidence。分布越集中，置信度通常越高。Noul 不带单独的 confidence，它返回的就是命题为真的概率。

因此，不能把三者统一写成 if value > 0.7：

• urgency_score 的范围是 0～2，不是置信度；
• category_confidence 说明选择是否明确，不说明选择了哪个类别；
• Noul 的 0.1 和 0.9 都很明确，只是方向相反，0.5 附近才最不明确。

二、为不同结果定义不同的“暂不决定”条件

课程先采用一组保守的演示边界：

category_confidence < 0.65       → 人工复核
urgency_confidence < 0.65        → 人工复核
0.35 < action probability < 0.65 → 人工复核
category == other                → 人工复核

这些边界不是官方保证，也不应直接复制到生产环境。它们只是清楚展示三类拒绝条件：

1. 模型在多个类别或等级之间犹豫；
2. Noul 的真假方向不够明确；
3. 现有类别没有覆盖输入，即使模型对 other 很有把握，也应让人看看是否需要新增类别。

把 other 送人工复核是业务政策，不是模型能力限制。如果你的应用允许“其他”消息直接进入普通收件箱，可以修改这条规则。

三、风险高低决定自动化边界

置信度高不等于动作一定可以自动执行。还要考虑判断错了会发生什么。

可以用下面的简单分级：

动作	出错后是否容易恢复	建议
排序、加本地标签、生成草稿	容易	高置信度时可自动完成
移动文件、修改日程、对外发送回复	有一定影响	至少让用户确认
付款、删除数据、批准权限	影响大或难恢复	不因一次模型判断而自动执行

Decision Inbox 目前只生成本地建议，属于低风险。即便如此，我们仍然增加人工复核队列，为以后接入真正动作保留安全边界。

要注意，“先确认再执行”和“自动执行”不是一回事。官方 Confidence 文档中的转账示例，即使在高置信度分支也先进入确认再执行。阈值是风险控制的一部分，不是越过确认的通行证；你仍要按自己的动作风险设计权限与确认机制。

四、替换策略模块

用下面的完整内容替换 decision_policy.py。相比上一篇，它增加了 collect_review_reasons()，并把人工复核放在所有自动路由之前。

import math

URGENCY_THRESHOLD = 1.5

CATEGORY_CONFIDENCE_MIN = 0.65
URGENCY_CONFIDENCE_MIN = 0.65
ACTION_NO_MAX = 0.35
ACTION_YES_MIN = 0.65


def collect_review_reasons(judgment: dict) -> list[str]:
    reasons = []
    if not isinstance(judgment, dict):
        return ["invalid_judgment"]
    if judgment.get("category") not in {"billing", "technical", "scheduling", "feedback", "other"}:
        reasons.append("invalid_category")
    for field, upper in (("category_confidence", 1), ("urgency_confidence", 1),
                         ("action_requested_probability", 1), ("urgency_score", 2)):
        value = judgment.get(field)
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or not 0 <= value <= upper:
            reasons.append("invalid_" + field)
    if reasons:
        return reasons

    if judgment["category_confidence"] < CATEGORY_CONFIDENCE_MIN:
        reasons.append("low_category_confidence")

    if judgment["urgency_confidence"] < URGENCY_CONFIDENCE_MIN:
        reasons.append("low_urgency_confidence")

    action_probability = judgment["action_requested_probability"]
    if ACTION_NO_MAX < action_probability < ACTION_YES_MIN:
        reasons.append("ambiguous_action_request")

    if judgment["category"] == "other":
        reasons.append("uncovered_category")

    return reasons


def build_recommendation(record: dict) -> dict:
    judgment = record.get("judgment")
    review_reasons = collect_review_reasons(judgment)

    if review_reasons:
        return {
            **record,
            "recommendation": {
                "topic": judgment.get("category", "unknown") if isinstance(judgment, dict) else "unknown",
                "queue": "manual_review",
                "priority": "needs_review",
                "suggested_action": "inspect_before_routing",
                "automation_allowed": False,
                "reasons": review_reasons,
            },
        }

    is_urgent = judgment["urgency_score"] >= URGENCY_THRESHOLD
    action_requested = (
        judgment["action_requested_probability"] >= ACTION_YES_MIN
    )

    if is_urgent and action_requested:
        queue = "today"
        priority = "high"
        suggested_action = "prepare_reply_draft"
    elif action_requested:
        queue = "follow_up"
        priority = "normal"
        suggested_action = "prepare_reply_draft"
    else:
        queue = "read_only"
        priority = "low"
        suggested_action = "read_or_archive"

    return {
        **record,
        "recommendation": {
            "topic": judgment["category"],
            "queue": queue,
            "priority": priority,
            "suggested_action": suggested_action,
            "automation_allowed": True,
            "reasons": [
                f"category={judgment['category']}",
                f"urgency_score={judgment['urgency_score']:.3f}",
                (
                    "action_requested_probability="
                    f"{judgment['action_requested_probability']:.3f}"
                ),
            ],
        },
    }

这里有几个值得留意的设计决定。

人工复核优先于业务路由

函数先收集 review_reasons。只要有一项成立，就不会继续根据紧急程度和行动请求自动分流。这样可避免出现“已经知道模型不确定，却仍然顺手给出自动建议”的矛盾状态。

Noul 使用两个边界，而不是一个阈值

现在的含义是：

0.00～0.35  可以按“没有明确行动请求”处理
0.35～0.65  暂不决定，送人工复核
0.65～1.00  可以按“有明确行动请求”处理

代码使用严格不等号，因此恰好等于 0.35 归入“否”，恰好等于 0.65 归入“是”。这只是明确的工程约定，后面会用测试锁定。

automation_allowed 只描述当前低风险步骤

这里的 True 表示可以自动完成“本地排序或准备草稿”这类可恢复步骤，不表示可以自动把草稿发出去。字段名如果脱离动作范围就容易被误用，所以真实系统最好同时保存具体允许的动作，例如 allowed_actions: ["add_local_label"]。

五、让路由脚本认识第四个队列

打开 route_inbox.py，找到 queue_order 和 counts，分别替换为：

    queue_order = {
        "manual_review": 0,
        "today": 1,
        "follow_up": 2,
        "read_only": 3,
    }

    counts = {
        "manual_review": 0,
        "today": 0,
        "follow_up": 0,
        "read_only": 0,
    }

人工复核排在最前面，是为了让人先处理系统无法稳定决定的消息，而不是让它们沉到列表底部。

同时修改 load_decisions 的逐条检查：只要求每项是 dict，缺少／非法 judgment 交给策略进入 manual_review。排序不能直接读取可能缺失的 urgency_score。在 route_inbox.py 中 import math，并新增：

def urgency_for_sort(item):
    judgment = item.get("judgment")
    value = judgment.get("urgency_score") if isinstance(judgment, dict) else None
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
        return 0.0
    return value if 0 <= value <= 2 else 0.0

把排序键第二项改为 -urgency_for_sort(item)，保留队列优先级为第一项。非法数值不参与自动决策，排序也不能使人工复核流程崩溃。缺失 judgment 的记录仍需保留 id 与原消息；若顶层不是对象，读取阶段直接拒绝整个产物并报告位置。

修改后先做语法检查：

macOS 或 Linux：

python -m py_compile decision_policy.py route_inbox.py

Windows PowerShell：

.\.venv\Scripts\python.exe -m py_compile decision_policy.py route_inbox.py

命令没有输出，表示两个文件通过了语法检查。

六、更新测试，覆盖不确定性边界

用下面内容替换 test_decision_policy.py：

import unittest

from decision_policy import build_recommendation


def make_record(
    *,
    category: str = "technical",
    category_confidence: float = 0.9,
    urgency: float = 1.0,
    urgency_confidence: float = 0.9,
    action_probability: float = 0.9,
) -> dict:
    return {
        "id": "test-message",
        "source": "test",
        "text": "test content",
        "judgment": {
            "category": category,
            "category_probabilities": {category: 1.0},
            "category_confidence": category_confidence,
            "urgency_score": urgency,
            "urgency_confidence": urgency_confidence,
            "action_requested_probability": action_probability,
        },
    }


class DecisionPolicyTests(unittest.TestCase):
    def test_clear_urgent_action_goes_to_today(self) -> None:
        result = build_recommendation(
            make_record(urgency=1.8, action_probability=0.9)
        )
        recommendation = result["recommendation"]
        self.assertEqual(recommendation["queue"], "today")
        self.assertTrue(recommendation["automation_allowed"])

    def test_clear_non_action_goes_to_read_only(self) -> None:
        result = build_recommendation(
            make_record(urgency=1.8, action_probability=0.2)
        )
        self.assertEqual(result["recommendation"]["queue"], "read_only")

    def test_low_category_confidence_requires_review(self) -> None:
        result = build_recommendation(
            make_record(category_confidence=0.4)
        )
        recommendation = result["recommendation"]
        self.assertEqual(recommendation["queue"], "manual_review")
        self.assertIn("low_category_confidence", recommendation["reasons"])
        self.assertFalse(recommendation["automation_allowed"])

    def test_low_urgency_confidence_requires_review(self) -> None:
        result = build_recommendation(
            make_record(urgency_confidence=0.4)
        )
        self.assertEqual(result["recommendation"]["queue"], "manual_review")

    def test_ambiguous_action_requires_review(self) -> None:
        result = build_recommendation(
            make_record(action_probability=0.5)
        )
        self.assertIn(
            "ambiguous_action_request",
            result["recommendation"]["reasons"],
        )

    def test_other_category_requires_review(self) -> None:
        result = build_recommendation(make_record(category="other"))
        self.assertIn(
            "uncovered_category",
            result["recommendation"]["reasons"],
        )

    def test_action_boundaries_are_explicit(self) -> None:
        low = build_recommendation(make_record(action_probability=0.35))
        high = build_recommendation(make_record(action_probability=0.65))

        self.assertEqual(low["recommendation"]["queue"], "read_only")
        self.assertEqual(high["recommendation"]["queue"], "follow_up")


if __name__ == "__main__":
    unittest.main()

运行：

macOS 或 Linux：

python -m unittest -v test_decision_policy.py

Windows PowerShell：

.\.venv\Scripts\python.exe -m unittest -v test_decision_policy.py

预期七个测试全部通过。它们证明四件事：

• 明确判断仍然能进入正常队列；
• Choice 和 Score 低置信度会阻断自动路由；
• Noul 中间区间会阻断自动路由；
• 两个边界值的归属是明确且可重复的。

七、重新生成收件箱

这一步仍然不需要重新调用 Jev。直接对上一篇的 decisions.json 重新应用策略：

macOS 或 Linux：

python route_inbox.py

Windows PowerShell：

.\.venv\Scripts\python.exe route_inbox.py

如果四条示例中有低置信度或模糊结果，统计里会出现 manual_review。如果没有，也不代表代码失效，只说明这批实际结果没有触发门槛。单元测试已经用可控输入验证过复核路径。

再运行一个独立检查：

macOS 或 Linux：

python -c "import json; d=json.load(open('output/routed_inbox.json', encoding='utf-8')); allowed={'manual_review','today','follow_up','read_only'}; assert all(x['recommendation']['queue'] in allowed for x in d); assert all(not x['recommendation']['automation_allowed'] for x in d if x['recommendation']['queue']=='manual_review'); print('review policy verified')"

Windows PowerShell：

.\.venv\Scripts\python.exe -c "import json; d=json.load(open('output/routed_inbox.json', encoding='utf-8')); allowed={'manual_review','today','follow_up','read_only'}; assert all(x['recommendation']['queue'] in allowed for x in d); assert all(not x['recommendation']['automation_allowed'] for x in d if x['recommendation']['queue']=='manual_review'); print('review policy verified')"

看到 review policy verified，说明人工复核队列中的所有记录都明确禁止自动处理。

八、不要用“调低阈值”消灭人工复核

第一次加入复核队列后，很多人会发现需要人工看的条目比预想中多，于是不断降低门槛，直到列表看起来很干净。这相当于为了让仪表盘好看而关闭报警器。

更合理的处理顺序是：

1. 查看复核样例，确认问题来自哪里；
2. 如果输入缺少上下文，补充 state；
3. 如果类别重叠，重写 Choice 描述；
4. 如果大量消息落入 other，考虑新增有意义的类别；
5. 用标注数据比较修改前后的准确率与复核率；
6. 最后才决定是否调整阈值。

复核率本身也是系统成本。门槛过严会让自动化失去意义，门槛过松会放大误判。两者之间的取舍必须通过真实数据和错误代价决定，而不是凭几条演示消息拍脑袋。

九、常见误区

1. 高置信度就是一定正确

不是。置信度用于表达模型对判断的确定程度；校准是在一组预测上衡量概率与实际结果是否匹配，不能保证单次答案。高风险操作仍要确认、权限控制和可恢复设计。

2. 低置信度时再问同一个问题，直到变高

重复调用可能得到不同数值，但不能自动补全输入中不存在的信息。应该先检查问题定义、状态信息和类别边界。不要把“多试几次挑一个最自信的”当成可靠策略。

3. 把多个置信度简单平均

分类不确定、紧急度明确，与分类明确、紧急度不确定，可能需要完全不同的处理。平均后会掩盖具体风险。当前策略使用“任一关键判断不可靠就复核”，更保守也更容易解释。

4. 记录了 reasons 就等于获得模型解释

我们的 reasons 只记录触发了哪些确定性规则。它不是模型的思维过程，也不能证明语义判断正确。不要把程序生成的规则标签包装成“AI 给出的完整理由”。

十、完成本篇的验收

确认你已经做到：

• 能区分 Choice/Score 的 confidence、Score 的 score 和 Noul 的概率；
• decision_policy.py 为低置信度、Noul 模糊区间和 other 类别提供人工复核路径；
• 七个策略测试全部通过，包括两个边界值测试；
• manual_review 队列中的记录均为 automation_allowed: false；
• 明白阈值应随动作风险变化，而不是所有判断共用一个数；
• 项目仍然不执行对外发送、删除、付款等高风险动作。

现在 Decision Inbox 已经有了完整的判断与拒绝闭环，但演示阈值仍然只是我们的初始假设。下一篇会建立一份小型中文评估集，用可重复的指标回答：分类是否准确、复核率是否可接受，以及调整问题描述后到底有没有变好。

异常输入验收

在策略测试中分别把 category_confidence 改为 NaN、True、-0.1、1.1，删除 urgency_score，或者删除整个 judgment，均应进入 manual_review，不能抛出未处理 KeyError 或继续自动路由。Score 的本课量表是 0～2，概率／confidence 是 0～1，不要混用范围。JSON 读取同样拒绝 NaN／Infinity。