用 Jev 做决策审计追踪：每个答案可记录、可重放、可解释

import logging
import os

import requests

logger = logging.getLogger("decisions")

def decide_with_chat_model(customer_id: str, case: dict) -> str:
    resp = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{
                "role": "user",
                "content": f"按政策 v3 裁决本案，回复决策与一段理由。\nCase: {case}",
            }],
        },
        timeout=30,
    )
    decision_text = resp.json()["choices"][0]["message"]["content"]

    # 审计日志拿到的是一段话。无法设阈值、无法重放、无法解释——
    # 只能重读。这次调用用的是哪个版本的提示词？「大概率批准」
    # 指的是 0.8 的置信度吗？日志说不出来，因为根本没记。
    logger.info("decision for %s: %s", customer_id, decision_text)
    return decision_text

import hashlib
import json

import requests

JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate"
AUTO_EXECUTE_CONFIDENCE = 0.85  # 过闸 且 可逆性 <= 2 -> 自动执行
HUMAN_CONFIDENCE = 0.60         # 低于此，或可逆性 4 -> 人工

QUESTIONS = {
    "decision": {
        "type": "choice",
        "instructions": "按公开政策裁决本案",
        "criteria": {
            "approve": "满足全部政策条件——无需自由裁量",
            "refer": "满足大部分条件但命中已声明的边缘情形",
            "deny": "触碰硬性条件——指明该条件",
        },
    },
    "reversibility": {
        "type": "score",
        "instructions": "如果错了，逆转代价多大？1 = 轻松可逆，4 = 对个人不可逆伤害",
    },
    "needs_human_review": {
        "type": "noul",
        "instructions": "决策生效前是否需要人工确认？",
    },
}

def decide_and_record(subject_ref: str, state: dict) -> dict:
    payload = {"state": state, "questions": QUESTIONS}
    resp = requests.post(JEV_ENDPOINT, json=payload, timeout=5)
    resp.raise_for_status()
    data = resp.json()

    decision = data["decision"]
    reversibility = data["reversibility"]["answer"]
    if data["needs_human_review"]["answer"] or reversibility == 4:
        lane = "human_decision"
    elif decision["confidence"] >= AUTO_EXECUTE_CONFIDENCE and reversibility <= 2:
        lane = "auto_executed"
    else:
        lane = "review_queue"

    # 记录即审计追踪：契约、答案、闸门——一行定长记录，
    # 可在任意未来 schema 版本下重放。
    return {
        "subject_ref": subject_ref,
        "decision": decision["answer"],
        "confidence": decision["confidence"],
        "reversibility": reversibility,
        "lane": lane,
        "logic": {
            "schema_version": "audit-contract@1.3.0",
            "questions_hash": hashlib.sha256(
                json.dumps(payload["questions"], sort_keys=True).encode()
            ).hexdigest()[:16],
            "model_version": "jev-1.13",
        },
        "human_intervention_path": (
            "available_on_request"
            if lane == "auto_executed"
            else "mandatory_before_effect"
        ),
    }

import { z } from "zod";
import OpenAI from "openai";

const Decision = z.object({ decision: z.enum(["approve", "refer", "deny"]) });
const openai = new OpenAI();

export async function decideCase(caseData: unknown) {
  const resp = await openai.chat.completions.create({
    model: "gpt-4o-mini",
    response_format: { type: "json_object" },
    messages: [
      { role: "user", content: `按政策 v3 裁决: ${JSON.stringify(caseData)}` },
    ],
  });
  const { decision } = Decision.parse(
    JSON.parse(resp.choices[0].message.content!),
  );

  // 形状有保证，证据没有。没有置信度、没有 schema 版本、没有
  // 分布——六个月后的审计只能拿到「是什么」，永远拿不到
  // 「为什么」和「多少次是对的」。
  return { decision, auditRecord: { decision } };
}

const JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate";
const AUTO_EXECUTE_CONFIDENCE = 0.85; // 过闸 且 可逆性 <= 2
const HUMAN_CONFIDENCE = 0.6; // 低于此，或可逆性 4 -> 人工

type Lane = "auto_executed" | "review_queue" | "human_decision";

const QUESTIONS = {
  decision: {
    type: "choice",
    instructions: "按公开政策裁决本案",
    criteria: {
      approve: "满足全部政策条件",
      refer: "命中边缘情形——附原因转队列",
      deny: "触碰硬性条件——指明该条件",
    },
  },
  reversibility: {
    type: "score",
    instructions: "如果错了，逆转代价多大？1 = 轻松可逆，4 = 不可逆伤害",
  },
  needs_human_review: {
    type: "noul",
    instructions: "生效前是否需要人工确认？",
  },
} as const;

export async function decideAndRecord(subjectRef: string, state: object) {
  const resp = await fetch(JEV_ENDPOINT, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ state, questions: QUESTIONS }),
  });
  if (!resp.ok) throw new Error("Jev evaluate failed");
  const data = await resp.json();

  const { answer, confidence } = data.decision as {
    answer: string;
    confidence: number;
  };
  const reversibility = data.reversibility.answer as number;
  const lane: Lane =
    data.needs_human_review.answer || reversibility === 4
      ? "human_decision"
      : confidence >= AUTO_EXECUTE_CONFIDENCE && reversibility <= 2
        ? "auto_executed"
        : "review_queue";

  return {
    subject_ref: subjectRef,
    decision: answer,
    confidence,
    reversibility,
    lane,
    logic: { schema_version: "audit-contract@1.3.0", model_version: "jev-1.13" },
    human_intervention_path:
      lane === "auto_executed" ? "available_on_request" : "mandatory_before_effect",
  };
}
// Art. 22 记录 = 这个对象。重放 = 把存储的 state 以新 schema 版本
// 重新提交，然后 diff 车道。