필사 모드: LLM structured output 的稳定化 — 消灭解析失败的四层防御

import json, re

FENCE = re.compile(r"^\s*```(?:json)?\s*|\s*```\s*$", re.MULTILINE)

def extract_json(text):
    """1) 剥掉围栏 2) 还不行就只截取第一个对象区间再试"""
    cleaned = FENCE.sub("", text).strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass
    start, end = cleaned.find("{"), cleaned.rfind("}")
    if start != -1 and end > start:
        return json.loads(cleaned[start:end + 1])
    raise ValueError("JSON을 찾을 수 없음")

# 如果自己部署开源模型，约束解码在服务端选项里打开
vllm serve <모델경로> --guided-decoding-backend xgrammar

# 从请求侧传入 schema (例如 OpenAI 兼容端点)
#   "guided_json": { ...JSON Schema... }
#   "guided_choice": ["正面", "负面", "中性"]
#   "guided_regex": "\\d{4}-\\d{2}-\\d{2}"

# 糟糕: 就算没有值也必须给出一个数字
{"price": {"type": "integer"}}

# 良好: 可以表达“没有”
{"price": {"type": ["integer", "null"]},
 "price_found": {"type": "boolean"}}

from pydantic import BaseModel, Field
from typing import Literal, Optional

# 糟糕的 schema
class BadTicket(BaseModel):
    type: str            # 什么都可能进来
    date: str            # 格式五花八门
    info: dict           # 没有结构

# 良好的 schema
class Ticket(BaseModel):
    reasoning: str = Field(description="분류 근거를 두 문장 이내로")
    category: Literal["결제", "배송", "환불", "계정", "기타"]
    urgency: Literal["낮음", "보통", "높음"]
    order_id: Optional[str] = Field(None, description="주문번호. 본문에 없으면 null")
    due_date_iso: Optional[str] = Field(None, description="YYYY-MM-DD 형식")

import json
from pydantic import ValidationError

def generate_structured(client, messages, model_cls, max_attempts=3, max_tokens=1024):
    history = list(messages)
    last_error = None

    for attempt in range(max_attempts):
        resp = client.create(messages=history, max_tokens=max_tokens)
        text = resp.content
        finish = resp.finish_reason

        if finish == "length":
            # 截断靠重新生成解决不了。要么提高上限，要么把任务拆开。
            max_tokens = min(max_tokens * 2, 8192)
            history.append({"role": "user",
                            "content": "출력이 잘렸습니다. 더 짧게, 설명 없이 다시 출력하십시오."})
            continue

        try:
            return model_cls.model_validate(extract_json(text))
        except (ValueError, json.JSONDecodeError, ValidationError) as e:
            last_error = e
            # 把失败的输出和具体错误一起回灌
            history.append({"role": "assistant", "content": text})
            history.append({"role": "user", "content":
                f"위 출력이 스키마 검증에 실패했습니다.\n오류: {e}\n"
                f"오류만 수정한 JSON을 다른 텍스트 없이 출력하십시오."})

    raise RuntimeError(f"{max_attempts}회 시도 후 실패: {last_error}")

def retry_math(p_fail, max_attempts):
    expected_calls = sum(p_fail ** i for i in range(max_attempts))
    residual = p_fail ** max_attempts
    return expected_calls, residual

for p in (0.04, 0.15):
    calls, res = retry_math(p, 3)
    print(f"1회 실패율 {p:.0%} → 평균 호출 {calls:.3f}회 (비용 +{calls - 1:.1%}), 최종 실패율 {res:.4%}")
# 1회 실패율 4% → 평균 호출 1.042회 (비용 +4.2%), 최종 실패율 0.0064%
# 1회 실패율 15% → 평균 호출 1.172회 (비용 +17.2%), 최종 실패율 0.3375%

def estimate_output_tokens(schema_fields, avg_value_tokens, list_len=1):
    """结构开销 + 取值。真实数值还是用分词器测更准确。"""
    per_item = sum(len(name) // 3 + 4 + avg_value_tokens for name in schema_fields)
    return int(per_item * list_len * 1.25)   # 留 25% 余量

fields = ["reasoning", "category", "urgency", "order_id", "due_date_iso"]
print(estimate_output_tokens(fields, avg_value_tokens=12, list_len=1))   # 116
print(estimate_output_tokens(fields, avg_value_tokens=12, list_len=20))  # 2325