为什么模型返回的有效 JSON 仍会触发解析器语法错误

# 普通的 print 输出（会隐藏问题字符）
print(f"Received: {raw_response}")

# 使用 repr() 进行原始字符检查（暴露隐藏字符）
print(repr(raw_response[:80]))

# 包含 BOM 的原始响应示例
raw_response = '\ufeff{\n  "customer": "Acme Corp",\n  "issue": "billing dispute"\n}'

```json
\{
  "customer": "Acme Corp",
  "issue": "billing dispute"
\}
```

如果你的管道仅使用简单的 `.strip()` 去除首尾的空白字符，那些反引号（`` ` ``）和 `json` 标识符依然会残留在字符串的头部。解析器在读取到反引号时，同样会抛出语法解析错误。

---

## 幕后黑手 3：模型自带的前导语与后记

LLM 本质上是一个对话模型，它们经常难以克制“说话”的本能。即使你设置了严格的系统提示词，模型有时仍会在 JSON 前后加上解释性文字：

```text
好的，这是为您提取的客户信息 JSON 数据：
\{
  "customer": "Acme Corp",
  "issue": "billing dispute"
\}
希望对您有所帮助！

import json
import re
from typing import Any, Dict

def extract_json(raw: str) -> Dict[str, Any]:
    """
    从原始 LLM 文本响应中清理并提取有效的 JSON 对象。
    支持处理 UTF-8 BOM、Markdown 代码块围栏以及前后的对话性文本。
    """
    if not raw:
        raise ValueError("接收到的 JSON 解析输入为空字符串。")

    # 1. 清理 UTF-8 字节顺序标记 (BOM)
    if raw.startswith("\ufeff"):
        raw = raw.lstrip("\ufeff")

    # 规范化换行符并去除首尾空白字符
    raw = raw.strip()

    # 2. 使用正则表达式提取 Markdown 代码块中的内容
    # 支持忽略大小写的 ```json 或 纯 ``` 围栏
    markdown_pattern = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE)
    match = markdown_pattern.search(raw)

    if match:
        raw = match.group(1).strip()
    else:
        # 3. 回退策略：定位最外层的花括号 \{\}
        # 适用于模型在 JSON 前后添加了对话性前导语或总结的情况
        start = raw.find("\{")
        end = raw.rfind("\}")
        if start != -1 and end > start:
            raw = raw[start:end + 1]
        else:
            # 如果预期返回的是数组，则寻找方括号 []
            start_array = raw.find("[")
            end_array = raw.rfind("]")
            if start_array != -1 and end_array > start_array:
                raw = raw[start_array:end_array + 1]

    # 4. 执行反序列化
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        # 抛出异常时附带 repr 打印，方便在日志中排查不可见字符
        raise ValueError(f"JSON 解析失败。原始片段: \{repr(raw[:100])\}。错误详情: \{e\}") from e

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

class CustomerIssue(BaseModel):
    customer: str = Field(..., min_length=1)
    issue: str = Field(..., min_length=1)
    priority: int = Field(..., ge=1, le=5)

def process_pipeline(raw_llm_output: str) -> Optional[CustomerIssue]:
    try:
        # 提取并转换为字典
        parsed_dict = extract_json(raw_llm_output)
        # 使用 Pydantic 校验数据结构
        validated_data = CustomerIssue(**parsed_dict)
        return validated_data
    except (ValueError, ValidationError) as err:
        print(f"管道处理错误: \{err\}")
        # 在此处实现重试逻辑或降级方案
        return None

def run_test_suite():
    fixtures = [
        (
            "标准干净的 JSON",
            '\{"customer": "Acme Corp", "issue": "billing", "priority": 2\}'
        ),
        (
            "带有 UTF-8 BOM 前缀",
            '\ufeff\{"customer": "Acme Corp", "issue": "billing", "priority": 2\}'
        ),
        (
            "Markdown 代码块包裹",
            '```json\n{"customer": "Acme Corp", "issue": "billing", "priority": 2}\n```'
        ),
        (
            "BOM 与 Markdown 代码块混合",
            '\ufeff```json\n{"customer": "Acme Corp", "issue": "billing", "priority": 2}\n```'
        ),
        (
            "带有前导语和结束语的复杂文本",
            '这是为您提取的数据：\n\{\n  "customer": "Acme Corp",\n  "issue": "billing",\n  "priority": 2\n\}\n如有疑问请随时联系我。'
        )
    ]

    print("开始运行解析器测试...")
    for name, raw_input in fixtures:
        try:
            result = extract_json(raw_input)
            assert result["customer"] == "Acme Corp"
            assert result["priority"] == 2
            print(f"[成功] 测试用例: \{name\}")
        except Exception as e:
            print(f"[失败] 测试用例: \{name\}，错误原因: \{e\}")

if __name__ == "__main__":
    run_test_suite()

import requests
import json

# 配置指向 n1n.ai 的统一客户端
API_URL = "https://api.n1n.ai/v1/chat/completions"
API_KEY = "your_n1n_api_key"

headers = \{
    "Authorization": f"Bearer \{API_KEY\}",
    "Content-Type": "application/json"
\}

# 定义目标 JSON Schema
json_schema = \{
    "name": "CustomerIssueSchema",
    "strict": True,
    "schema": \{
        "type": "object",
        "properties": \{
            "customer": \{"type": "string"\},
            "issue": \{"type": "string"\},
            "priority": \{"type": "integer", "minimum": 1, "maximum": 5\}
        \},
        "required": ["customer", "issue", "priority"],
        "additionalProperties": False
    \}
\}

payload = \{
    "model": "gpt-4o-mini",
    "messages": [
        \{
            "role": "system",
            "content": "你是一个数据提取助手。请提取客户问题详情。"
        \},
        \{
            "role": "user",
            "content": "Acme Corp 提交了一个账单争议申诉。这是一个 2 级的高优先级事件。"
        \}
    ],
    "response_format": \{
        "type": "json_schema",
        "json_schema": json_schema
    \}
\}

response = requests.post(API_URL, json=payload, headers=headers)
result = response.json()

# 此时，返回的内容已被强制约束为符合 Schema 的有效 JSON 字符串
raw_content = result["choices"][0]["message"]["content"]
parsed_data = json.loads(raw_content)
print(parsed_data)