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))

# 取消传播：客户端断开 → 关掉上游模型流 → 别再烧 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)

import time
import uuid

POLICY = {"version": "2026-09-06.1", "max_cents": 200000, "approval_above_cents": 50000}
# 本地假订单；生产中在可信数据库中按租户、订单和用户查询。
ORDERS = {("tenant-a", "o1"): {"owner": "alice", "paid_cents": 100000, "refunded_cents": 10000}}

def gate_tool_call(name, args, user, audit):
    rec = {"trace_id": uuid.uuid4().hex, "policy_version": POLICY["version"],
           "ts": time.time(), "user": user.get("id"), "tool": name}
    def finish(action, rule):
        audit.append({**rec, "action": action, "rule": rule})
        return {"ok": action == "allow", "action": action, "rule": rule, "trace_id": rec["trace_id"]}
    if name != "refund":
        return finish("deny", "unknown_tool")
    if not isinstance(args, dict) or set(args) != {"order_id", "amount_cents"}:
        return finish("deny", "invalid_arguments")
    cents = args["amount_cents"]
    if type(cents) is not int or cents <= 0:  # 同时拒绝 bool、float、NaN
        return finish("deny", "invalid_amount")
    if not isinstance(args["order_id"], str):
        return finish("deny", "invalid_order_id")
    order = ORDERS.get((user.get("tenant"), args["order_id"]))
    if order is None or order["owner"] != user.get("id"):
        return finish("deny", "not_authorized")
    if cents > min(POLICY["max_cents"], order["paid_cents"] - order["refunded_cents"]):
        return finish("deny", "insufficient_refundable_balance")
    if cents > POLICY["approval_above_cents"]:
        return finish("requires_approval", "human_approval_required")
    return finish("allow", "precheck_passed")

if __name__ == "__main__":
    audit = []
    session_user = {"id": "alice", "tenant": "tenant-a"}
    print(gate_tool_call("refund", {"order_id": "o1", "amount_cents": 40000}, session_user, audit))
    print(gate_tool_call("unregistered", {}, session_user, audit))