Python + Gemini JSON Schema 实现可校验的工单路由

客服消息里同时出现“扣款”“登录不上”“发票”，规则越堆越难维护。通俗答案是：让大语言模型（LLM）理解自然语言，但绝不能直接把它的一段自由文本送进工单系统。把输出限制为 JSON，再用本地规则复核，才能把“会聊天的模型”变成可靠的分类协作者。你会学到 JSON Schema、超时、失败回退和人工复核这四个可迁移到任何自动化场景的基本功。

为什么关键词方案迟早失控

关键词只能匹配字面。用户说“银行卡被扣了两次”，不含“退款”也许仍应进入支付队列；“想问电子发票”又不该误判为支付故障。模型擅长处理这种语义差异，但模型输出本质上是不确定的：可能多解释一句，可能给未知队列，也可能在网络波动时根本没有结果。因此职责要拆开：模型给候选意图；Schema 约束形状；程序检查允许值、置信度和敏感升级条件；不通过就转人工，而不是猜。

Gemini 的 generateContent 接口支持通过 responseMimeType 与 responseJsonSchema 请求 JSON 输出。Schema 不是事实校验器：它只能规定字段、类型和枚举，不能保证“支付”这个判断真的正确。生产系统还应保留抽样标注集，观察各队列的误分率。

环境准备

需要 Python 3.10+，本例只使用标准库，无须安装第三方依赖。到 Google AI Studio 创建密钥后设置环境变量；不要把密钥写进源码或提交到 Git。

export GEMINI_API_KEY='你的密钥'
python3 route_ticket.py '银行卡重复扣款，客服一直没回复'

完整代码

保存为 route_ticket.py。示例对 HTTP 设置 20 秒超时；即使接口返回合法 JSON，也会拒绝低置信度与升级关键词。模型名使用官方生成内容页面示例中的 gemini-3.8-flash；上线前请在目标账号核对可用模型与配额。

import json
import os
import sys
import urllib.error
import urllib.request

ALLOWED = {"payment", "account", "invoice", "human_review"}
SCHEMA = {
    "type": "object",
    "properties": {
        "queue": {"type": "string", "enum": sorted(ALLOWED)},
        "confidence": {"type": "number"},
        "reason": {"type": "string"},
    },
    "required": ["queue", "confidence", "reason"],
    "additionalProperties": False,
}

def call_model(message: str) -> dict:
    key = os.environ.get("GEMINI_API_KEY")
    if not key:
        raise RuntimeError("缺少 GEMINI_API_KEY 环境变量")
    body = {
        "contents": [{"parts": [{"text": (
            "将消息路由到 payment/account/invoice/human_review。"
            "不确定时选 human_review。消息：" + message)}]}],
        "generationConfig": {
            "responseMimeType": "application/json",
            "responseJsonSchema": SCHEMA,
            "temperature": 0,
        },
    }
    url = ("https://generativelanguage.googleapis.com/v1beta/models/"
           "gemini-3.8-flash:generateContent?key=" + key)
    request = urllib.request.Request(
        url, data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(request, timeout=20) as response:
        payload = json.load(response)
    text = payload["candidates"][0]["content"]["parts"][0]["text"]
    return json.loads(text)

def safe_route(message: str) -> dict:
    try:
        result = call_model(message)
        risky = any(word in message for word in ("盗刷", "泄露", "起诉"))
        if (result.get("queue") not in ALLOWED or
                not isinstance(result.get("confidence"), (int, float)) or
                result["confidence"] < 0.80 or risky):
            return {"queue": "human_review", "reason": "本地安全门禁"}
        return result
    except (KeyError, ValueError, urllib.error.URLError, TimeoutError) as exc:
        return {"queue": "human_review", "reason": f"调用或解析失败: {exc}"}

if __name__ == "__main__":
    print(json.dumps(safe_route(" ".join(sys.argv[1:])), ensure_ascii=False))

逐段看懂这段程序

SCHEMA 把队列锁为四个枚举值，并要求三个字段全部出现；temperature: 0 让相同输入更稳定，但不是正确性的保证。urlopen(..., timeout=20) 防止网络连接无限等待。最关键的是 safe_route：它将 API 失败、字段缺失、低置信度和敏感投诉统一降级到人工队列。这样即使供应商限流或模型响应异常，也不会让订单卡在半自动状态。

成功时可能输出 {"queue":"payment","confidence":0.94,"reason":"重复扣款"}；没有密钥时则直接指出环境变量缺失。本次任务已在本机用 python3 -m py_compile 做过语法检查，但未实际调用线上 API，不能把示例输出当作模型实测结果。