16c · 接进真实产品

# schema 只是一份【形状说明书】，不是一份事实核验
import json

FIELD = {"type": "number", "minimum": 0}
print(json.dumps(FIELD))    # ⭐ 它保证 refund_amount 一定是 number，不保证它是【对的】number

# 结构化输出全流程：定义 → 调用 → 校验 → 带错误重试（框架无关，纯标准库）
import json
import re

SCHEMA = {
    "type": "object",
    "required": ["intent", "refund_amount", "reason"],
    "properties": {
        "intent": {"type": "string", "enum": ["refund", "exchange", "other"]},
        # ⭐ 值域直接写进 schema，别只写 {"type": "number"}
        "refund_amount": {"type": "number", "minimum": 0, "maximum": 2000},
        "reason": {"type": "string", "maxLength": 200},
    },
}


def validate(obj, schema, path="$"):
    """极简校验器：返回【人话错误列表】，空列表 = 通过。"""
    errs = []
    kind = schema.get("type")
    if kind == "object":
        if not isinstance(obj, dict):
            return ["%s 应该是对象，实际是 %s" % (path, type(obj).__name__)]
        for k in schema.get("required", []):
            if k not in obj:
                errs.append("%s.%s 缺失（这是必填字段）" % (path, k))
        for k, sub in schema.get("properties", {}).items():
            if k in obj:
                errs += validate(obj[k], sub, "%s.%s" % (path, k))
    elif kind == "number":
        if isinstance(obj, bool) or not isinstance(obj, (int, float)):
            errs.append("%s 应该是数字，实际是 %r" % (path, obj))
        else:
            lo, hi = schema.get("minimum"), schema.get("maximum")
            if lo is not None and obj < lo:
                errs.append("%s=%s 小于下限 %s" % (path, obj, lo))
            if hi is not None and obj > hi:
                errs.append("%s=%s 超过上限 %s" % (path, obj, hi))   # ⭐ 值域校验
    elif kind == "string":
        if not isinstance(obj, str):
            errs.append("%s 应该是字符串，实际是 %r" % (path, obj))
        else:
            allowed = schema.get("enum")
            if allowed and obj not in allowed:
                errs.append("%s=%r 不在允许取值 %s 内" % (path, obj, allowed))
            limit = schema.get("maxLength")
            if limit is not None and len(obj) > limit:
                errs.append("%s 超长（%d > %d）" % (path, len(obj), limit))
    return errs


def extract_json(text):
    """模型爱套代码围栏、爱加开场白，先抠出最外层大括号再解析。"""
    m = re.search(r"\{.*\}", text, re.S)
    if not m:
        raise ValueError("响应里找不到 JSON 对象")
    return json.loads(m.group(0))


def semantic_check(obj, order):
    """⭐ schema 管不了的那一层：和上游【事实】核对。"""
    errs = []
    if obj["intent"] == "refund" and obj["refund_amount"] > order["paid"]:
        errs.append("refund_amount=%s 大于订单实付 %s，不可能"
                    % (obj["refund_amount"], order["paid"]))
    return errs


def call_model(messages):
    """换成任意 SDK 都行，返回一整段文本。"""
    raise NotImplementedError("接你自己的模型调用")


def ask_structured(user_text, order, max_retry=2):
    messages = [{"role": "user", "content": user_text}]
    last = None
    for _ in range(max_retry + 1):
        raw = call_model(messages)
        try:
            obj = extract_json(raw)
            errs = validate(obj, SCHEMA) or semantic_check(obj, order)
        except Exception as e:
            obj, errs = None, ["JSON 解析失败：%s" % e]
        if not errs:
            return obj
        last = errs
        # ⭐ 关键：把「你上次输出了什么 + 错在哪 + 该怎么改」一起回传，不要原样重试
        messages += [
            {"role": "assistant", "content": raw},
            {"role": "user", "content": "上次输出不合法，只输出修正后的 JSON，不要解释：\n"
                                        + "\n".join("- " + e for e in errs)},
        ]
    # ⭐ 必须有出口：重试是有上限的，超了就走降级（转人工 / 返回兜底）
    raise ValueError("重试 %d 次仍不合法：%s" % (max_retry, last))

event: delta
data: {"text": "退"}

event: delta
data: {"text": "款"}

event: done
data: {"chars": 128}

# 取消传播：客户端断开 → 关掉上游模型流 → 别再烧 token（框架无关）
import contextlib
import threading


class Cancellation:
    """一个能被多处观察的取消信号。HTTP 层 / 模型调用层 / 工具执行层共用同一个。"""

    def __init__(self):
        self._flag = threading.Event()

    def cancel(self):
        self._flag.set()

    @property
    def cancelled(self):
        return self._flag.is_set()

    def check(self):
        if self.cancelled:
            raise RuntimeError("客户端已断开")     # ⭐ 用异常让整条调用栈一起退出


def log_abandoned(info):
    """把「被放弃的生成」单独记一笔，才能算出白烧了多少钱。"""
    print("abandoned:", info)


@contextlib.contextmanager
def upstream_stream(open_stream, cancel, **kw):
    s = open_stream(**kw)                          # 换成任意 SDK 的 stream 入口
    try:
        yield s
    finally:
        s.close()                                  # ⭐ finally 里关：取消/异常/正常结束都会走到
        if cancel.cancelled:
            log_abandoned(kw)

# 动作侧护栏：工具调用前的同步硬闸 + 审计留痕（是代码里的 if，不是提示词里的一句话）
import time
import uuid

POLICY = {
    "version": "2026-08-04.1",          # ⭐ 规则带版本号，改了才能和历史审计对上
    "refund": {"max_amount": 2000, "needs_human_above": 500},
}


def _deny(rec, audit, rule, human_msg):
    rec.update(action="deny", rule=rule)          # ⭐ 拦了什么、按哪条拦的，都要落库
    audit.append(rec)
    return {"ok": False, "trace_id": rec["trace_id"],
            "message": "这笔退款我不能直接处理（%s）。已转人工，编号 %s。"
                       % (human_msg, rec["trace_id"])}


def gate_tool_call(name, args, user, audit):
    """在【每一次】工具调用之前同步跑。不能异步，不能采样。"""
    rec = {"trace_id": uuid.uuid4().hex[:8], "layer": "action", "tool": name,
           "policy_version": POLICY["version"], "ts": time.time(), "user": user["id"]}
    if name == "refund":
        amt = args.get("amount")
        if isinstance(amt, bool) or not isinstance(amt, (int, float)) or amt <= 0:
            return _deny(rec, audit, "amount_invalid", "金额格式不对")
        if amt > POLICY["refund"]["max_amount"]:
            return _deny(rec, audit, "over_max", "超出单笔上限")
        if amt > args.get("order_paid", 0):        # ⭐ 和事实核对，不只和常量比
            return _deny(rec, audit, "over_paid", "退款不能大于订单实付")
        if amt > POLICY["refund"]["needs_human_above"]:
            rec["action"] = "escalate"
            audit.append(rec)
            return {"ok": False, "escalate": True, "trace_id": rec["trace_id"]}
    rec["action"] = "allow"
    audit.append(rec)
    return {"ok": True, "trace_id": rec["trace_id"]}