用 Jev 做 AI 费用分类：银行流水批量入账与置信度门控

import json
import os
import time

import requests

# 基线：先跑关键词规则，规则匹配不了的交给生成式 LLM。
# 两套系统，两种失败模式。
KEYWORD_RULES = {
    "software_saas": ["AWS", "NOTION", "ZOOM.US", "ADOBE"],
    "travel": ["DELTA AIR", "UBER", "LYFT", "MARRIOTT"],
    "meals": ["RESTAURANT", "COFFEE", "SUSHI", "CAFE"],
    "office_supplies": ["OFFICE DEPOT", "STAPLES", "AMZN MKTP"],
}
CATEGORIES = set(KEYWORD_RULES) | {"professional_services", "other"}

def rule_category(line: str) -> str | None:
    upper = line.upper()
    for category, keywords in KEYWORD_RULES.items():
        if any(keyword in upper for keyword in keywords):
            return category
    return None  # "SQ *BLUE BOTTLE"、"MERCH PAYMENT 8842 LLC" 匹配不到任何规则

def generative_category(line: str) -> dict:
    # 每行 1.5-3 秒，每次尝试都付输出 token，而且 json.loads 对幻觉类目
    # 和真实类目一视同仁地放行。
    for attempt in range(3):
        resp = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
            json={
                "model": "gpt-4o-mini",
                "response_format": {"type": "json_object"},
                "messages": [{
                    "role": "user",
                    "content": f'把这行支出归类为 {", ".join(sorted(CATEGORIES))} 之一，'
                               f'以 JSON 返回：{{"category": "..."}}\n流水行：{line}',
                }],
            },
            timeout=30,
        )
        try:
            category = json.loads(
                resp.json()["choices"][0]["message"]["content"]
            )["category"]
            if category in CATEGORIES:  # json_mode 不会帮你做的值校验
                return {"category": category, "confidence": None}
        except (KeyError, json.JSONDecodeError):
            time.sleep(2 ** attempt)  # 每次重试都为完整生成重新付费
    return {"category": "other", "confidence": None}  # 静默失败

def categorize_line(line: str) -> dict:
    category = rule_category(line)
    if category is not None:
        return {"category": category, "confidence": None}  # 规则没有置信度
    return generative_category(line)

import requests

JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate"
AUTO_POST_CONFIDENCE = 0.85   # ≥ 0.85 → 自动入账
REVIEW_MIN_CONFIDENCE = 0.60  # 0.60~0.85 → 复核队列，低于 → 人工

QUESTIONS = {
    "category": {
        "type": "choice",
        "instructions": "对每一行支出，将其归入恰好一个类目",
        "criteria": {
            "software_saas": "SaaS 订阅、云服务、数字工具",
            "travel": "机票、酒店、地面交通",
            "meals": "餐厅、咖啡、客户宴请",
            "office_supplies": "办公用品、设备、仓储采购",
            "professional_services": "法务、会计、咨询、外包",
            "other": "以上都不匹配且无把握",
        },
    },
    "deductibility": {
        "type": "score",
        "instructions": "对每一行评估税务可抵扣程度，1（个人支出）到 4（全额可抵扣）",
    },
    "needs_review": {
        "type": "noul",
        "instructions": "对每一行判断：是否模糊到需要记账人员确认后才能入账？",
    },
}

def categorize_statement(lines: list[dict]) -> list[dict]:
    # 整份账单作为一个 state——三个类型化问题在单次约 70-100ms 的
    # 前向推理中逐行返回答案。
    resp = requests.post(
        JEV_ENDPOINT,
        json={"state": {"lines": lines}, "questions": QUESTIONS},
        timeout=5,
    )
    resp.raise_for_status()
    data = resp.json()

    results = []
    for i, line in enumerate(lines):
        category = data["category"][i]
        flagged = data["needs_review"][i]["answer"]
        if flagged or category["answer"] == "other":
            lane = "human_review"
        elif category["confidence"] >= AUTO_POST_CONFIDENCE:
            lane = "auto_post"
        elif category["confidence"] >= REVIEW_MIN_CONFIDENCE:
            lane = "review"
        else:
            lane = "human_review"
        results.append({
            "line_index": line["line_index"],
            "category": category["answer"],
            "confidence": category["confidence"],
            "deductibility": data["deductibility"][i]["answer"],
            "lane": lane,
        })
    return results

def sweep_statements(statements: list[list[dict]]) -> dict:
    # 仅输入计费 $0.042/M：30 行账单约 1,500 token ≈ $0.00007，
    # 每月一万份账单不到一美元。
    lanes: dict[str, int] = {"auto_post": 0, "review": 0, "human_review": 0}
    for lines in statements:
        for row in categorize_statement(lines):
            lanes[row["lane"]] += 1
    return lanes

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

// 基线：先跑关键词规则，剩下的走结构化输出 LLM。
const KEYWORD_RULES: Record<string, string[]> = {
  software_saas: ["AWS", "NOTION", "ZOOM.US", "ADOBE"],
  travel: ["DELTA AIR", "UBER", "LYFT", "MARRIOTT"],
  meals: ["RESTAURANT", "COFFEE", "SUSHI", "CAFE"],
  office_supplies: ["OFFICE DEPOT", "STAPLES", "AMZN MKTP"],
};

const CATEGORIES = [
  "software_saas",
  "travel",
  "meals",
  "office_supplies",
  "professional_services",
  "other",
] as const;

const Decision = z.object({ category: z.enum(CATEGORIES) });
const client = new OpenAI();

function ruleCategory(line: string): string | null {
  const upper = line.toUpperCase();
  for (const [category, keywords] of Object.entries(KEYWORD_RULES)) {
    if (keywords.some((k) => upper.includes(k))) return category;
  }
  return null; // "MERCH PAYMENT 8842 LLC" 匹配不到任何规则
}

export async function categorizeLine(line: string) {
  const category = ruleCategory(line);
  if (category) return { category, confidence: null }; // 规则没有置信度

  for (let attempt = 0; attempt < 3; attempt++) {
    try {
      const resp = await client.chat.completions.create({
        model: "gpt-4o-mini",
        response_format: { type: "json_object" },
        messages: [
          {
            role: "user",
            content: `把这行支出归类为 ${CATEGORIES.join(", ")} 之一，以 JSON 返回：{"category": "..."}\n流水行：${line}`,
          },
        ],
      });
      // 形状有保证，值是否正确没有保证——
      // 而且没有任何可以驱动自动化的置信度数字。
      return {
        category: Decision.parse(
          JSON.parse(resp.choices[0].message.content!),
        ).category,
        confidence: null,
      };
    } catch {
      await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
    }
  }
  return { category: "other" as const, confidence: null };
}

const JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate";

// 策略写在应用代码里——模型只负责报告事实。
const AUTO_POST_CONFIDENCE = 0.85; // ≥ 0.85 → 自动入账
const REVIEW_MIN_CONFIDENCE = 0.6; // 0.60~0.85 → 复核队列

type StatementLine = {
  line_index: number;
  merchant: string;
  amount: number;
  date: string;
  memo?: string;
};

const QUESTIONS = {
  category: {
    type: "choice",
    instructions: "对每一行支出，将其归入恰好一个类目",
    criteria: {
      software_saas: "SaaS 订阅、云服务、数字工具",
      travel: "机票、酒店、地面交通",
      meals: "餐厅、咖啡、客户宴请",
      office_supplies: "办公用品、设备、仓储采购",
      professional_services: "法务、会计、咨询、外包",
      other: "以上都不匹配且无把握",
    },
  },
  deductibility: {
    type: "score",
    instructions: "对每一行评估税务可抵扣程度，1（个人支出）到 4（全额可抵扣）",
  },
  needs_review: {
    type: "noul",
    instructions: "对每一行判断：是否模糊到需要记账人员确认后才能入账？",
  },
} as const;

export type ExpenseLane = "auto_post" | "review" | "human_review";

export async function categorizeStatement(lines: StatementLine[]) {
  const response = await fetch(JEV_ENDPOINT, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ state: { lines }, questions: QUESTIONS }),
  });
  if (!response.ok) throw new Error("Jev evaluate failed");
  const data = await response.json();

  // 答案逐行返回，与 state.lines 的顺序对齐。
  return lines.map((line, i) => {
    const category = data.category[i] as {
      answer: string;
      confidence: number;
    };
    const flagged = data.needs_review[i].answer as boolean;
    const lane: ExpenseLane =
      flagged || category.answer === "other"
        ? "human_review"
        : category.confidence >= AUTO_POST_CONFIDENCE
          ? "auto_post"
          : category.confidence >= REVIEW_MIN_CONFIDENCE
            ? "review"
            : "human_review";
    return {
      ...line,
      category: category.answer,
      confidence: category.confidence,
      deductibility: data.deductibility[i].answer as number,
      lane,
    };
  });
}

// 金额、税额和按类目汇总留在你的代码里——模型从不做四则运算。
// 30 行账单 ≈ 1,500 输入 token，墙钟约 70-100ms，输出 token $0。