資料標註與分類

import json
import os
import random
from concurrent.futures import ThreadPoolExecutor

from openai import OpenAI

client = OpenAI(base_url="https://www.tutujin.com/v1", api_key=os.environ["TUTUJIN_API_KEY"], max_retries=3)

LABELS = ["shipping", "quality", "price", "service", "other"]
SYSTEM = (
    "You label customer reviews. Choose exactly one label from: " + ", ".join(LABELS) + ". "
    'Reply with JSON only, for example {"label": "quality", "confidence": 0.9}.'
)


def label(item, model):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["text"]}],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=50,
    )
    raw = resp.choices[0].message.content
    try:
        result = json.loads(raw[raw.find("{"): raw.rfind("}") + 1])
    except ValueError:
        result = {}
    if result.get("label") not in LABELS:
        return {"id": item["id"], "label": None, "confidence": 0.0}
    return {"id": item["id"], "label": result["label"], "confidence": float(result.get("confidence", 0))}


items = [{"id": 1, "text": "Arrived two days late, but the mug itself is lovely."}]

with ThreadPoolExecutor(max_workers=8) as pool:
    first = list(pool.map(lambda it: label(it, "gemini-2.5-flash-lite"), items))

# second opinion for low-confidence rows from another vendor's model
by_id = {it["id"]: it for it in items}
for row in first:
    if row["confidence"] < 0.7:
        second = label(by_id[row["id"]], "qwen3.5-plus")
        row["needs_human"] = second["label"] != row["label"]
        row["label"] = second["label"] if row["label"] is None else row["label"]

# a random 2% always goes to people for quality checks
for row in random.sample(first, max(1, len(first) // 50)):
    row["needs_human"] = True

with open("labels.jsonl", "w", encoding="utf-8") as f:
    for row in first:
        f.write(json.dumps(row, ensure_ascii=False) + "\n")