用 Ollama 结构化输出提取中文活动信息：JSON Schema 与时间校验

Ollama 支持结构化输出：在请求里给 format 一个 JSON Schema，模型的回复就会符合这个结构。 这篇做 extract.py：从中文活动通知中提取标题、日期、开始时间、地点、费用、是否报名，写成 CSV；再用规则解析时间，和模型的结果交叉核对。

先看清成品

• 3 条通知在 box 上真实提取，JSON 全部符合 Schema。
• 发现真实问题：3 条的开始时间全错（9点半→09:00，下午两点→13:00，晚上7:00→20:00），但都通过了格式校验。
• 加上规则解析后自动更正并在 CSV 中标出。pytest 4 项通过。

合规提醒：模型在本机运行，数据不出本机；但每个模型有自己的许可证（在 ollama.com/library 的模型页面查看），商用前先确认。本文只用 Ollama 官方安装方式和它自带的本地 API（默认 http://localhost:11434），不使用任何非官方镜像或破解版本。

运行说明：本文所有输出都是在 box 上真实跑出来的（Ollama 0.40.7，8 核 Intel Xeon CPU、15 GB 内存、无 GPU，2026-10-21 20:38–20:56，UTC+8）。box 的网络不允许从官方模型库下载：ollama pull qwen2.5:1.5b 报错 redirect target not allowed: …r2.cloudflarestorage.com resolves to non-public 198.18.0.1。所以我用的是 box 上之前从 GGUF 文件导入的模型，名字带 -local（qwen2.5-local:1.5b、qwen3-local:0.6b、deepseek-r1-distill:1.5b）。你在自己电脑上直接 ollama pull qwen2.5:1.5b，把命令里的模型名换掉即可。速度数据只代表这台 CPU 机器。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

80-ollama-B-src/
├── extract.py
├── notices/
│   ├── n1.txt
│   ├── n2.txt
│   └── n3.txt
└── test_extract.py

第 1 步：脚本

"""用 Ollama 结构化输出（format=JSON Schema）从中文通知里提取活动信息，并在本地再校验一遍。"""
import argparse, csv, json, re, sys
from pathlib import Path
import requests

HOST = "http://localhost:11434"
SCHEMA = {
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "date": {"type": "string", "description": "YYYY-MM-DD"},
        "start_time": {"type": "string", "description": "HH:MM，24 小时制"},
        "location": {"type": "string"},
        "fee_yuan": {"type": "number", "description": "免费填 0"},
        "need_signup": {"type": "boolean"},
    },
    "required": ["title", "date", "start_time", "location", "fee_yuan", "need_signup"],
}
DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$"); TIME = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")

CN = {"零": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "十": 10, "十一": 11, "十二": 12}
TIME_CN = re.compile(r"(上午|早上|中午|下午|晚上)?\s*(\d{1,2}|十[一二]?|[一二两三四五六七八九])\s*[点:：时]\s*(半|\d{1,2})?")

def parse_cn_time(text):
    """规则解析原文中的时间。恰好找到一个时才返回 HH:MM，否则返回 None（交给人工）。"""
    found = set()
    for ap, h, m in TIME_CN.findall(text):
        h = int(h) if h.isdigit() else CN[h]
        if ap in ("下午", "晚上") and h < 12: h += 12
        if ap == "中午" and h < 6: h += 12
        mm = 30 if m == "半" else int(m or 0)
        if h < 24 and mm < 60: found.add(f"{h:02d}:{mm:02d}")
    return found.pop() if len(found) == 1 else None

def validate(d):
    errs = []
    types = {"string": str, "number": (int, float), "boolean": bool}
    for k in SCHEMA["required"]:
        if k not in d: errs.append(f"缺少 {k}"); continue
        t = SCHEMA["properties"][k]["type"]
        if not isinstance(d[k], types[t]) or (t == "number" and isinstance(d[k], bool)):
            errs.append(f"{k} 类型应为 {t}")
    if isinstance(d.get("date"), str) and not DATE.match(d["date"]): errs.append(f"date 格式不对：{d['date']}")
    if isinstance(d.get("start_time"), str) and not TIME.match(d["start_time"]): errs.append(f"start_time 格式不对：{d['start_time']}")
    return errs

def ask(model, text, *, extra="", session=requests):
    body = {"model": model, "stream": False, "format": SCHEMA, "options": {"temperature": 0},
            "messages": [{"role": "system", "content": "从通知中提取字段，只输出 JSON。日期用 YYYY-MM-DD，时间用 24 小时制 HH:MM。" + extra},
                         {"role": "user", "content": text}]}
    r = session.post(f"{HOST}/api/chat", json=body, timeout=300)
    r.raise_for_status()
    return json.loads(r.json()["message"]["content"])

def extract(model, text, *, session=requests):
    d = ask(model, text, session=session)
    errs = validate(d)
    if errs:  # 把错误告诉模型，重试一次
        d = ask(model, text, extra=" 上一次输出的问题：" + "；".join(errs), session=session)
        errs = validate(d)
    rule = parse_cn_time(text)
    if rule and d.get("start_time") != rule:  # 格式对不等于内容对：以规则解析为准并标出
        errs.append(f"start_time 模型给 {d.get('start_time')}，原文解析为 {rule}，已改为 {rule}")
        d["start_time"] = rule
    return d, errs

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("files", nargs="+")
    ap.add_argument("--model", default="qwen2.5:1.5b"); ap.add_argument("-o", default="events.csv")
    a = ap.parse_args(argv)
    rows = []
    for f in a.files:
        d, errs = extract(a.model, Path(f).read_text(encoding="utf-8"))
        d["source"] = Path(f).name; d["problems"] = "；".join(errs)
        rows.append(d); print(json.dumps(d, ensure_ascii=False))
    with open(a.o, "w", newline="", encoding="utf-8-sig") as fh:
        w = csv.DictWriter(fh, fieldnames=["source"] + SCHEMA["required"] + ["problems"], extrasaction="ignore")
        w.writeheader(); w.writerows(rows)
    bad = sum(1 for r in rows if r["problems"])
    print(f"写入 {a.o}：{len(rows)} 条，{bad} 条有自动更正或问题，请人工核对")

if __name__ == "__main__":
    main()

第 2 步：示例通知

# notices/n1.txt
【社区通知】本周六（2026年10月24日）上午9点半，在社区活动中心二楼举办旧物交换市集，免费参加，无需报名，欢迎带上闲置图书和玩具。
# notices/n2.txt
各位家长：少儿编程体验课定于11月1日下午两点开始，地点是滨江图书馆301教室，每人收取材料费30元，名额20个，请提前在群里接龙报名。（年份：2026）
# notices/n3.txt
通知：2026/11/08 晚上7:00 物业将在南门广场召开业主说明会，讨论电梯更换方案。业主凭门禁卡签到，不收费，不用提前报名。

第 3 步：测试

import json
import extract as m

def test_parse_cn_time():
    assert m.parse_cn_time("上午9点半在二楼") == "09:30"
    assert m.parse_cn_time("下午两点开始") == "14:00"
    assert m.parse_cn_time("晚上7:00 召开") == "19:00"
    assert m.parse_cn_time("9点或10点") is None   # 两个时间：交给人工
    assert m.parse_cn_time("没有时间") is None

def test_validate():
    ok = {"title": "a", "date": "2026-10-24", "start_time": "09:30", "location": "b", "fee_yuan": 0, "need_signup": False}
    assert m.validate(ok) == []
    bad = dict(ok, date="2026/10/24", fee_yuan=True)
    errs = m.validate(bad)
    assert any("date" in e for e in errs) and any("fee_yuan" in e for e in errs)

class R:
    def __init__(self, d): self.d = d
    def raise_for_status(self): pass
    def json(self): return {"message": {"content": json.dumps(self.d)}}

def test_schema_sent_and_time_corrected():
    sent = []
    class S:
        def post(self, url, json=None, **k):
            sent.append(json)
            return R({"title": "课", "date": "2026-11-01", "start_time": "13:00", "location": "x", "fee_yuan": 30, "need_signup": True})
    d, errs = m.extract("x", "11月1日下午两点开始", session=S())
    assert sent[0]["format"]["required"] == m.SCHEMA["required"]
    assert d["start_time"] == "14:00" and "原文解析为 14:00" in errs[0]

def test_retry_on_invalid():
    replies = iter([{"title": "a"}, {"title": "a", "date": "2026-10-24", "start_time": "09:30", "location": "b", "fee_yuan": 0, "need_signup": False}])
    sent = []
    class S:
        def post(self, url, json=None, **k):
            sent.append(json); return R(next(replies))
    d, errs = m.extract("x", "无", session=S())
    assert len(sent) == 2 and "上一次输出的问题" in sent[1]["messages"][0]["content"] and errs == []

第 4 步：运行

$ python extract.py notices/*.txt --model qwen2.5-local:1.5b
{"date": "2026-10-24", "fee_yuan": 0, "location": "社区活动中心二楼", "need_signup": false, "start_time": "09:30", "title": "旧物交换市集", "source": "n1.txt", "problems": "start_time 模型给 09:00，原文解析为 09:30，已改为 09:30"}
{"date": "2026-11-01", "fee_yuan": 30, "location": "滨江图书馆301教室", "need_signup": true, "start_time": "14:00", "title": "少儿编程体验课", "source": "n2.txt", "problems": "start_time 模型给 13:00，原文解析为 14:00，已改为 14:00"}
{"date": "2026-11-08", "fee_yuan": 0, "location": "南门广场", "need_signup": false, "start_time": "19:00", "title": "业主说明会", "source": "n3.txt", "problems": "start_time 模型给 20:00，原文解析为 19:00，已改为 19:00"}
写入 events.csv：3 条，3 条有自动更正或问题，请人工核对
$ cat events.csv
source,title,date,start_time,location,fee_yuan,need_signup,problems
n1.txt,旧物交换市集,2026-10-24,09:30,社区活动中心二楼,0,False,start_time 模型给 09:00，原文解析为 09:30，已改为 09:30
n2.txt,少儿编程体验课,2026-11-01,14:00,滨江图书馆301教室,30,True,start_time 模型给 13:00，原文解析为 14:00，已改为 14:00
n3.txt,业主说明会,2026-11-08,19:00,南门广场,0,False,start_time 模型给 20:00，原文解析为 19:00，已改为 19:00
$ python -m pytest -q
4 passed

怎么检查结果？

• CSV 的 problems 列不为空的行，人工看一眼原文。
• 规则解析只在原文恰好有一个时间时才生效；有多个时间（“9点或10点”）时不自动更正。
• 金额、日期也可以按同样思路加规则核对。

调试示例

格式对，内容错。 第一版只做 Schema 校验，运行结果是：

{"title": "旧物交换市集", "start_time": "09:00", …}      原文：上午9点半
{"title": "少儿编程体验课", "start_time": "13:00", …}    原文：下午两点
{"title": "业主说明会", "start_time": "20:00", …}        原文：晚上7:00
写入 events.csv：3 条，0 条需要人工核对

三条全部错了，而且全部通过校验：format 只保证 JSON 结构和类型，不保证内容正确。1.5B 模型对“点半”“下午两点”这类中文时间换算并不可靠。

修复：写一个只认常见写法的规则解析 parse_cn_time()（上午/下午/晚上 + 数字或中文数字 + 点/时/冒号 + 半/分钟）。原文中恰好找到一个时间、且和模型结果不同时，以规则为准并在 problems 列写明“模型给 X，原文解析为 Y”。

常见问题

模板规则

• Schema 管格式，规则或人工管内容。
• 能用规则核对的字段（时间、金额、日期）都核对。
• 自动更正要留下记录。
• temperature 设为 0。

项目资源

让模型填表，让规则查表。