用 Together AI 结构化输出抽取中文报修信息：JSON Schema 与 Pydantic 校验

Together AI 提供开源模型的推理 API，接口兼容 OpenAI。 这篇做 extract.py：把一行行中文报修消息（“302 卫生间水龙头一直滴水”）抽取成固定字段的 JSON（房间号、类别、是否紧急、摘要），用官方的结构化输出 response_format: json_schema，拿到结果后再用 pydantic 校验一次。同一份代码把 base_url 指向本机 Ollama，就能离线试跑。

先看清成品

• 用无效 Key 真实调用 Together：脚本识别 401 并退出（返回原文见下）。
• 同一份代码对本机 Ollama（qwen2.5 1.5B）真实运行了 5 条消息，发现小模型把房间号写成字符串 "null"、把“宿舍”“5楼”当房间号、5 条全标紧急。已加校验和告警。
• 6 项 pytest 通过。

合规提醒：只用 Together AI 官方 API 和你自己的 API Key，Key 只放环境变量或服务端，不共享账号。按 token 计费，价格以 together.ai/pricing 为准，本文不写价格。每个开源模型有自己的许可证，商用前看模型页面；处理用户消息时注意个人信息保护，不把别人的隐私数据发给第三方服务而不告知。

运行说明：我没有 Together API Key，没有对 Together 发出过成功的请求（也没产生费用）。真实跑过的是：用无效 Key 调用官方接口（2026-10-21 21:44–21:46，UTC+8，返回原样贴在文中），以及全部本地代码。Ollama 那部分是真实运行的输出，但模型是本机的 1.5B 小模型，不代表 Together 上模型的质量。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

86-together-ai-A-src/
├── extract.py
├── tickets.txt
└── test_extract.py

第 1 步：示例消息

302 卫生间水龙头一直滴水，关不紧
宿舍网断了一上午，什么都打不开，急！
5楼走廊的椅子腿松了
我门禁卡刷不开 1208 的门，人在门外进不去
楼下好吵，能不能管一下

第 2 步：抽取脚本

"""用 Together AI 的结构化输出（response_format=json_schema）从中文报修消息里抽取字段。

接口兼容 OpenAI，所以用官方 openai SDK，只改 base_url。拿到 JSON 后再用 pydantic 校验一次：
官方说支持的模型会遵守 schema，但不同模型/后端行为不同，入库前自己再验一遍更稳。
同一份代码把 base_url 指向本机 Ollama（也兼容 OpenAI）就能离线试跑。"""
import argparse, json, os, sys
from typing import Literal, Optional
from openai import OpenAI, APIStatusError
import re
from pydantic import BaseModel, Field, ValidationError, field_validator


class Ticket(BaseModel):
    room: Optional[str] = Field(description="房间号，如 302；没提到填 null")
    category: Literal["水电", "网络", "家具", "门锁", "其他"] = Field(description="问题类别")
    urgent: bool = Field(description="是否影响正常使用或有安全隐患")
    summary: str = Field(description="一句话描述问题，不超过 30 字")

    @field_validator("room", mode="before")
    @classmethod
    def clean_room(cls, v):
        # 实测小模型会输出字符串 "null"，或把“宿舍”“5楼”填进房间号
        if v is None or str(v).strip().lower() in ("", "null", "none"):
            return None
        m = re.fullmatch(r"\s*(\d{2,5})\s*(室|号)?\s*", str(v))
        return m.group(1) if m else None


SYSTEM = ("你是物业报修分类助手。只输出 JSON，严格遵守这个 schema：" + json.dumps(Ticket.model_json_schema(), ensure_ascii=False)
          + "。原文没有的信息填 null，不要猜。")


def extract(client, model, text, extra=None):
    resp = client.chat.completions.create(
        model=model, temperature=0,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
        response_format={"type": "json_schema", "json_schema": {"name": "ticket", "schema": Ticket.model_json_schema()}},
        **(extra or {}))
    raw = resp.choices[0].message.content
    try:
        return Ticket.model_validate_json(raw), None
    except ValidationError as e:
        return None, f"不符合 schema：{e.errors()[0]['msg']}；原始输出：{raw[:120]}"


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("file", help="每行一条报修消息")
    ap.add_argument("--model", default="Qwen/Qwen3.5-9B")
    ap.add_argument("--base-url", default="https://api.together.ai/v1")
    a = ap.parse_args(argv)
    key = os.environ.get("TOGETHER_API_KEY", "").strip()
    if not key:
        print("缺少 TOGETHER_API_KEY（api.together.ai/settings/api-keys）", file=sys.stderr); return 1
    client = OpenAI(api_key=key, base_url=a.base_url, max_retries=2)
    # 官方示例对 Qwen3.5 关闭了思考；这个参数是 Together 扩展，Ollama 不认识，所以只在 Together 上传
    extra = {"extra_body": {"reasoning": {"enabled": False}}} if "together" in a.base_url else None
    bad, urgent, n = 0, 0, 0
    for line in open(a.file, encoding="utf-8"):
        if not line.strip():
            continue
        try:
            t, err = extract(client, a.model, line.strip(), extra)
        except APIStatusError as e:
            print(f"API 错误 HTTP {e.status_code}：{e.message}", file=sys.stderr); return 1
        bad += err is not None
        n += 1; urgent += bool(t and t.urgent)
        print(json.dumps(t.model_dump(), ensure_ascii=False) if t else f"✗ {line.strip()[:20]}… {err}")
    if n >= 3 and urgent == n:
        print(f"⚠ {n} 条全部被标为紧急，模型可能不会区分，urgent 字段需要人工复核", file=sys.stderr)
    return 2 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

第 3 步：测试

from types import SimpleNamespace as NS
import extract

class Fake:
    def __init__(self, content): self.content = content; self.kw = None
    @property
    def chat(self): return NS(completions=self)
    def create(self, **kw):
        self.kw = kw
        return NS(choices=[NS(message=NS(content=self.content))])

def test_room_string_null_becomes_none():
    t, err = extract.extract(Fake('{"room":"null","category":"其他","urgent":false,"summary":"x"}'), "m", "x")
    assert err is None and t.room is None

def test_non_numeric_room_dropped():
    t, _ = extract.extract(Fake('{"room":"5楼","category":"家具","urgent":false,"summary":"x"}'), "m", "x")
    assert t.room is None

def test_room_with_suffix():
    t, _ = extract.extract(Fake('{"room":"302室","category":"水电","urgent":true,"summary":"x"}'), "m", "x")
    assert t.room == "302"

def test_bad_category_reported():
    t, err = extract.extract(Fake('{"room":null,"category":"噪音","urgent":false,"summary":"x"}'), "m", "x")
    assert t is None and "schema" in err

def test_request_uses_json_schema():
    f = Fake('{"room":null,"category":"其他","urgent":false,"summary":"x"}')
    extract.extract(f, "m", "x")
    rf = f.kw["response_format"]
    assert rf["type"] == "json_schema" and rf["json_schema"]["name"] == "ticket"
    assert f.kw["temperature"] == 0

def test_missing_key(monkeypatch, tmp_path):
    monkeypatch.delenv("TOGETHER_API_KEY", raising=False)
    p = tmp_path / "t.txt"; p.write_text("x\n", encoding="utf-8")
    assert extract.main([str(p)]) == 1

第 4 步：运行

# 1) Together，无效 Key（2026-10-21 21:45 UTC+8）
$ TOGETHER_API_KEY=invalid-demo python extract.py tickets.txt
API 错误 HTTP 401：Error code: 401 - {'id': 'p684sTX-3pDw3Z-a4e0b3570978e63a', 'error': {'message': 'Invalid API key provided. You can find your API key at https://api.together.ai/settings/api-keys.', 'type': 'invalid_request_error', 'param': None, 'code': 'invalid_api_key'}}
$ echo $?
1

# 2) 同一份代码，本机 Ollama（21:45 UTC+8，修复后）
$ TOGETHER_API_KEY=ollama python extract.py tickets.txt --base-url http://localhost:11434/v1 --model qwen2.5-local:1.5b
⚠ 5 条全部被标为紧急，模型可能不会区分，urgent 字段需要人工复核
{"room": "302", "category": "水电", "urgent": true, "summary": "卫生间水龙头滴水"}
{"room": null, "category": "网络", "urgent": true, "summary": "宿舍网络断了，无法上网"}
{"room": null, "category": "家具", "urgent": true, "summary": "5楼走廊的椅子腿松了"}
{"room": "1208", "category": "门锁", "urgent": true, "summary": "门禁卡刷不开"}
{"room": null, "category": "其他", "urgent": true, "summary": "楼下吵闹影响休息"}
$ python -m pytest -q
6 passed

怎么检查结果？

• 每行输出都是能被 json.loads 解析的对象，category 只会是五个值之一。
• 原文没有房间号时 room 是 null（真正的 null，不是字符串）。
• 出现“全部紧急”告警时，说明这个模型不会区分，换更大的模型或把 urgent 交给人工。
• 有 Key 后先跑这 5 条，对照上面的 Ollama 结果看差别。

调试示例

本机 1.5B 模型第一次运行的真实输出（修复前）：

{"room": "302", "category": "水电", "urgent": true, "summary": "卫生间水龙头滴水，关不紧"}
{"room": "宿舍", "category": "网络", "urgent": true, "summary": "宿舍网络断，无法上网"}
{"room": "5楼", "category": "家具", "urgent": true, "summary": "5楼走廊的椅子腿松了"}
{"room": "1208", "category": "门锁", "urgent": true, "summary": "门禁卡刷不开"}
{"room": "null", "category": "其他", "urgent": true, "summary": "楼下吵闹影响休息"}

三个问题，pydantic 都没拦住，因为它们都“符合 schema”：

1. "room": "null"：字符串 null，不是 JSON null。入库后查 room IS NULL 会漏掉它。
2. “宿舍”“5楼”被填进房间号：schema 只说是字符串，模型就填了它能找到的地点。
3. 5 条全部 urgent=true，包括椅子腿松了和楼下吵。

修复：给 room 加 field_validator，把 "null"/"none"/空串转成 None，只接受 2–5 位数字（可带“室/号”）；统计紧急比例，全部为真时在 stderr 告警。前两条已加测试固定。第 3 条是模型能力问题，代码只能提醒，不能替模型判断。

教训：json_schema 保证的是格式，不是内容。官方结构化输出约束的是结构，字段内容是否合理仍要自己检查。

常见问题

模板规则

• schema 里的枚举字段用 Literal 限死。
• 本地再校验一次，格式和内容都要查。
• 原文没有的信息要求填 null。
• temperature 设 0，便于复现。

项目资源

结构化输出解决了“能不能解析”，“对不对”还得靠校验和抽查。