DeepSeek JSON Output：自动分类账单流水

把银行或支付软件导出的流水交给模型分类（餐饮、交通、购物……），用 JSON Output（response_format=json_object）拿到结构化结果。程序逐条校验，不合格就把错误发回模型重试。**金额和日期一律从原文读取，不让模型碰数字。**最后导出成 Excel 能直接打开的 CSV。

先看清成品

• 测试： pytest 2 项通过。
• 本地实测（3 个本地模型 × 6 条流水）： 输出都是合法 JSON，结构校验都通过，没有触发重试。
• 分类准确率：
  • 第一版提示词下，qwen2.5 1.5B 把“机械键盘”分成了通讯，R1 蒸馏 1.5B 把“星巴克 拿铁”分成了购物。
  • 加上分类标准后，qwen2.5 1.5B 全部正确；R1 蒸馏 1.5B 把拿铁分成了交通，qwen3 0.6B 把拿铁分成了购物，仍然有错。
• 没有测过： DeepSeek 官方 API。

没有在 box 上跑的部分： box 上没有 DeepSeek API Key，所以没有调用过 DeepSeek 官方 API。下面的实测结果都来自同一份代码连接本地 Ollama（同样是 OpenAI 兼容接口，模型是 DeepSeek-R1-Distill-Qwen-1.5B 的 Q4_K_M 量化版）。换成官方 API 只需要改环境变量，但效果和速度会和本地小模型差别很大。

合规提醒：API Key 只放在环境变量里，不要写进代码或提交到 Git。不要把身份证号、客户资料、公司机密发给云端 API；这类数据用 B 篇的本地部署处理。如果要把它做成面向公众的服务，需要按《生成式人工智能服务管理暂行办法》《人工智能生成合成内容标识办法》等规定履行相应义务。API 价格以 DeepSeek 官网“模型 & 价格”页为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

41-deepseek-C-src/
├── bill_json.py
├── bills.txt
└── test_bill.py

第 1 步：代码

bill_json.py：

"""把账单流水交给模型分类，用 JSON Output（response_format=json_object）拿到结构化结果，并逐条校验。
校验不通过就把错误原因发回模型重试（最多 2 次）；金额以原始流水为准，模型改了金额会被拦下。
环境变量：DS_BASE_URL（默认 https://api.deepseek.com）、DEEPSEEK_API_KEY、DS_MODEL（默认 deepseek-flash）
用法：python bill_json.py bills.txt [输出.csv]"""
import csv, json, os, re, sys
from openai import OpenAI

BASE = os.environ.get("DS_BASE_URL", "https://api.deepseek.com")
MODEL = os.environ.get("DS_MODEL", "deepseek-flash")
CATS = ["餐饮", "交通", "购物", "通讯", "其他"]
client = OpenAI(base_url=BASE, api_key=os.environ.get("DEEPSEEK_API_KEY") or "local-no-key")

SYSTEM = f"""你是记账助手。把用户给的每一行流水分类，只输出 json，格式：
{{"items": [{{"line": 行号(从1开始), "category": "{'/'.join(CATS)} 之一", "note": "不超过10个字的说明"}}]}}
每一行都要有对应的一项，不要输出金额和日期（程序会从原文读取）。
分类标准（按买的是什么，不按商家）：
- 餐饮：吃的喝的，例如 外卖、咖啡、奶茶、超市买的零食
- 交通：打车、地铁、火车票、机票、加油、停车
- 购物：实物商品，例如 键盘、衣服、日用品（即使是在电商平台买的数码产品，也算购物）
- 通讯：话费、流量、宽带
- 其他：以上都不是"""

def parse_lines(text):
    rows = []
    for ln in text.strip().splitlines():
        m = re.match(r"(\d{4}-\d{2}-\d{2})\s+(.+?)\s+(\d+(?:\.\d{1,2})?)$", ln.strip())
        if not m: raise SystemExit(f"无法解析这一行：{ln}")
        rows.append({"date": m[1], "desc": m[2], "amount": m[3]})
    return rows

def validate(data, n):
    errs = []
    items = data.get("items") if isinstance(data, dict) else None
    if not isinstance(items, list): return ["缺少 items 数组"]
    got = {it.get("line") for it in items if isinstance(it, dict)}
    if got != set(range(1, n + 1)): errs.append(f"行号应为 1..{n}，实际 {sorted(x for x in got if isinstance(x, int))}")
    for it in items:
        if isinstance(it, dict) and it.get("category") not in CATS:
            errs.append(f"第 {it.get('line')} 行分类“{it.get('category')}”不在 {CATS} 里")
    return errs

def classify(rows):
    user = "\n".join(f"{i}. {r['desc']}" for i, r in enumerate(rows, 1))
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": user}]
    extra = {} if "11434" in BASE else {"thinking": {"type": "disabled"}}
    for attempt in range(3):
        r = client.chat.completions.create(model=MODEL, messages=messages, temperature=0,
                                           response_format={"type": "json_object"}, max_tokens=1024, extra_body=extra)
        raw = r.choices[0].message.content or ""
        try: data = json.loads(raw); errs = validate(data, len(rows))
        except json.JSONDecodeError as e: data, errs = None, [f"不是合法 JSON：{e}"]
        if not errs: return data["items"], attempt
        print(f"  第 {attempt+1} 次结果不合格：{'；'.join(errs)}", file=sys.stderr)
        messages += [{"role": "assistant", "content": raw},
                     {"role": "user", "content": "上面的 json 有问题：" + "；".join(errs) + "。请修正后重新输出完整 json。"}]
    raise SystemExit("重试 3 次仍不合格，请人工检查")

if __name__ == "__main__":
    rows = parse_lines(open(sys.argv[1], encoding="utf-8").read())
    items, retries = classify(rows)
    by = {it["line"]: it for it in items}
    out = [{**r, "category": by[i]["category"], "note": by[i].get("note", "")} for i, r in enumerate(rows, 1)]
    for o in out: print(f"{o['date']}  {o['amount']:>8}  {o['category']}  {o['desc']}（{o['note']}）")
    tot = {}
    for o in out: tot[o["category"]] = tot.get(o["category"], 0) + float(o["amount"])
    print("合计：" + "，".join(f"{k} {v:.2f}" for k, v in tot.items()) + f"（重试 {retries} 次）")
    if len(sys.argv) > 2:
        with open(sys.argv[2], "w", newline="", encoding="utf-8-sig") as f:
            w = csv.DictWriter(f, fieldnames=list(out[0])); w.writeheader(); w.writerows(out)
        print(f"已写入 {sys.argv[2]}（utf-8-sig，Excel 可直接打开）")

test_bill.py：

import os; os.environ.setdefault("DS_BASE_URL", "http://127.0.0.1:11434/v1")
from bill_json import parse_lines, validate

def test_parse_amount_from_source():
    r = parse_lines("2026-10-01 美团外卖 黄焖鸡米饭 32.50")
    assert r == [{"date": "2026-10-01", "desc": "美团外卖 黄焖鸡米饭", "amount": "32.50"}]

def test_validate_catches_missing_line_and_bad_category():
    errs = validate({"items": [{"line": 1, "category": "餐饮"}, {"line": 3, "category": "娱乐"}]}, 3)
    assert any("行号" in e for e in errs) and any("娱乐" in e for e in errs)

第 2 步：运行

$ DS_BASE_URL=http://127.0.0.1:11434/v1 DS_MODEL=qwen2.5-local:1.5b python bill_json.py bills.txt 十月.csv
2026-10-01     32.50  餐饮  美团外卖 黄焖鸡米饭（黄焖鸡米饭）
2026-10-02     18.00  交通  滴滴出行 快车（快车）
2026-10-03    299.00  购物  京东 机械键盘（机械键盘）
2026-10-05     50.00  通讯  中国移动 话费充值（话费充值）
2026-10-06     33.00  餐饮  星巴克 拿铁（拿铁）
2026-10-08     73.00  交通  12306 火车票 杭州-上海（火车票 杭州-上海）
合计：餐饮 65.50，交通 91.00，购物 299.00，通讯 50.00（重试 0 次）
已写入 十月.csv（utf-8-sig，Excel 可直接打开）

检查点：合计金额和原始流水逐条相加的结果一致（金额直接取自原文）。分类请人工扫一遍。

调试示例：JSON 是对的，分类是错的

第一版的提示词只写了分类的名称。两个 1.5B 模型输出的都是合法 JSON，结构校验也都通过，但分类有错：

qwen2.5 1.5B：京东 机械键盘 → 通讯
R1 蒸馏 1.5B：星巴克 拿铁  → 购物

结构校验只能保证“格式对”，保证不了“内容对”。我在提示词里加了分类标准：按买的是什么分，不按商家分，并给每类举了例子。之后 qwen2.5 1.5B 全对；R1 蒸馏 1.5B 和 qwen3 0.6B 仍然会把拿铁分错。所以这个工具的结论是：结构问题交给程序校验，语义问题靠更好的模型加上人工抽查，小模型的分类结果不能直接用。

常见问题

模板规则

• 数字和日期从原文读取，模型只负责判断类别。
• 格式用程序校验，语义靠人工抽查。
• 导出 CSV 用 utf-8-sig 编码，Excel 打开不会乱码。

项目资源

让模型只做它擅长的判断，数字和格式交给程序，这样记账结果才可靠。