大模型返回结果乱码、格式异常？一文搞定所有格式问题

布吉岛的石头

449人浏览 · 2026-05-07 10:30:00

布吉岛的石头 · 2026-05-07 10:30:00 发布

作者：洛水石

阅读时间：约 12 分钟

关键词：大模型 API、乱码、JSON 格式异常、SSE、UTF-8、流式输出

---

[引言：你踩过这些坑吗？](#引言)
[5 大常见格式异常场景](#5-大常见格式异常场景)
[乱码问题根因与解决方案](#乱码问题根因与解决方案)
[JSON 格式异常排查指南](#json-格式异常排查指南)
[SSE 流式输出断流修复](#sse-流式输出断流修复)
[生产环境最佳实践](#生产环境最佳实践)
[总结与参考](#总结与参考)

---

引言：你踩过这些坑吗？

调用 OpenAI、Claude、通义千问、文心一言等大模型 API 时，你是否遇到过以下崩溃瞬间：

**场景1**：明明模型返回了中文，前端却显示一堆 `\uXXXX` 转义符或乱码方块。
**场景2**：SSE 流式输出到一半突然断开，前端只渲染了一半回答。
**场景3**：返回的 JSON 被截断，解析直接抛 `JSONDecodeError`。
**场景4**：Markdown 代码块里嵌套了 JSON，解析器把代码块里的 JSON 当成了外层结构。
**场景5**：代理层转发了响应，编码被改了，中文全变问号。

这些问题看似琐碎，但在生产环境中，一个乱码就能导致用户流失。本文系统性地总结了与大模型 API 交互中所有常见的格式异常，给出根因分析和可落地的代码修复方案。

---

5 大常见格式异常场景

异常类型	典型表现	影响等级
Unicode 转义未解码	`\u4e2d\u6587` 而非中文	⭐⭐
编码不一致导致乱码	中文显示为 � 或 ???	⭐⭐⭐
JSON 被截断/截流	`JSONDecodeError`	⭐⭐⭐⭐
SSE 断流/丢事件	回答只显示一半	⭐⭐⭐⭐
Markdown 嵌套解析错误	代码块内 JSON 被误解析	⭐⭐

Unicode 转义未解码	`\u4e2d\u6587` 而非中文	⭐⭐
编码不一致导致乱码	中文显示为 � 或 ???	⭐⭐⭐
JSON 被截断/截流	`JSONDecodeError`	⭐⭐⭐⭐
SSE 断流/丢事件	回答只显示一半	⭐⭐⭐⭐
Markdown 嵌套解析错误	代码块内 JSON 被误解析	⭐⭐

编码不一致导致乱码	中文显示为 � 或 ???	⭐⭐⭐
JSON 被截断/截流	`JSONDecodeError`	⭐⭐⭐⭐
SSE 断流/丢事件	回答只显示一半	⭐⭐⭐⭐
Markdown 嵌套解析错误	代码块内 JSON 被误解析	⭐⭐

JSON 被截断/截流	`JSONDecodeError`	⭐⭐⭐⭐
SSE 断流/丢事件	回答只显示一半	⭐⭐⭐⭐
Markdown 嵌套解析错误	代码块内 JSON 被误解析	⭐⭐

SSE 断流/丢事件	回答只显示一半	⭐⭐⭐⭐
Markdown 嵌套解析错误	代码块内 JSON 被误解析	⭐⭐

---

乱码问题根因与解决方案

根因 1：API 返回 Unicode 转义符

很多大模型 API（尤其是 OpenAI）默认返回的 JSON 中，中文字符会被转义为 \uXXXX 形式。

错误示范：

import requests

resp = requests.post(url, json=payload)

data = resp.json()

content = data["choices"][0]["message"]["content"]

print(content) # 输出: "\u4f60\u597d" 而非 "你好"

正确做法：

import json

方案 A：请求时要求 API 不转义（OpenAI 支持）

payload = {

"model": "gpt-4",

"messages": [{"role": "user", "content": "你好"}],

"ensure_ascii": False # 部分国产模型支持

}

方案 B：Python 侧解码 Unicode 转义

content = data["choices"][0]["message"]["content"]

content = content.encode('utf-8').decode('unicode_escape')

print(content) # 输出: "你好"

根因 2：响应编码不一致

当请求经过 Nginx、API 网关、CDN 等代理层时，编码可能被修改。

排查步骤：

检查响应头 `Content-Type` 是否包含 `charset=utf-8`
检查代理层是否强制修改编码
使用原始二进制流而非文本流接收响应

错误：让 requests 自动推断编码

resp = requests.post(url, json=payload)

print(resp.encoding) # 可能为 ISO-8859-1

正确：强制使用 UTF-8 解码

resp = requests.post(url, json=payload)

resp.encoding = 'utf-8'

data = resp.json()

根因 3：流式输出编码错误

SSE (Server-Sent Events) 流式输出时，如果前端或后端编码处理不当，会导致逐字乱码。

后端正确做法（Python FastAPI）：

from fastapi import FastAPI

from fastapi.responses import StreamingResponse

import json

app = FastAPI()

@app.post("/chat")

async def chat():

async def generate():

chunks = ["你好", "，", "世界", "！"]

for chunk in chunks:

# 确保每次发送的数据是 UTF-8 编码

data = json.dumps({"content": chunk}, ensure_ascii=False)

yield f"data: {data}\n\n"

return StreamingResponse(

generate(),

media_type="text/event-stream; charset=utf-8"

)

前端正确做法（JavaScript）：

const eventSource = new EventSource('/chat');

// 必须设置正确的解码器

const decoder = new TextDecoder('utf-8');

eventSource.onmessage = (event) => {

const data = JSON.parse(event.data);

console.log(data.content); // 正确显示中文

};

---

JSON 格式异常排查指南

异常 1：JSON 被截断

当模型输出超长内容时，部分 API 会截断响应，导致 JSON 不完整。

典型错误：

{

"choices": [{

"message": {

"content": "这是一段很长的内容..."

}

}]

// 缺少右括号！

修复方案：

import json

def safe_parse_json(text):

"""安全解析可能被截断的 JSON"""

try:

return json.loads(text)

except json.JSONDecodeError as e:

# 尝试修复截断的 JSON

text = text.strip()

# 补全缺失的右括号

open_braces = text.count('{') - text.count('}')

open_brackets = text.count('[') - text.count(']')

text += '}' * open_braces + ']' * open_brackets

try:

return json.loads(text)

except:

# 如果仍然失败，返回原始文本

return {"raw": text, "error": "JSON parse failed"}

使用示例

resp_text = requests.post(url, json=payload).text

result = safe_parse_json(resp_text)

异常 2：Markdown 代码块内嵌 JSON

模型输出中经常包含 Markdown 代码块（json ... ），如果直接用正则提取 JSON，容易误匹配。

错误示范：

import re

错误：贪婪匹配会导致跨代码块匹配

json_pattern = re.compile(r'json\s*(.*?)\s*', re.DOTALL)

正确示范：

import re

def extract_json_from_markdown(text):

"""从 Markdown 中安全提取 JSON"""

# 匹配 json ... 代码块

pattern = re.compile(r'json\s*\n(.*?)\n', re.DOTALL)

matches = pattern.findall(text)

for match in matches:

try:

return json.loads(match.strip())

except:

continue

# 如果没有代码块，尝试直接解析整个文本

try:

return json.loads(text)

except:

return None

异常 3：返回的是 JSONL/NDJSON

部分模型（如 OpenAI 的 batch API）返回 JSON Lines 格式，每行一个 JSON 对象。

def parse_jsonl(text):

"""解析 JSON Lines 格式"""

results = []

for line in text.strip().split('\n'):

line = line.strip()

if line:

results.append(json.loads(line))

return results

---

SSE 流式输出断流修复

断流原因分析

原因	说明	解决方向
代理超时	Nginx/网关默认 60s 超时	调整超时配置
客户端超时	前端 EventSource 默认无超时	添加心跳机制
网络不稳定	移动端/WiFi 切换	实现重连机制
服务器内部错误	模型推理中断	返回 error 事件

代理超时	Nginx/网关默认 60s 超时	调整超时配置
客户端超时	前端 EventSource 默认无超时	添加心跳机制
网络不稳定	移动端/WiFi 切换	实现重连机制
服务器内部错误	模型推理中断	返回 error 事件

客户端超时	前端 EventSource 默认无超时	添加心跳机制
网络不稳定	移动端/WiFi 切换	实现重连机制
服务器内部错误	模型推理中断	返回 error 事件

网络不稳定	移动端/WiFi 切换	实现重连机制
服务器内部错误	模型推理中断	返回 error 事件

完整修复方案

后端（Python FastAPI + 心跳保活）：

import asyncio

from fastapi import FastAPI

from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")

async def chat_stream():

async def event_generator():

try:

# 发送心跳保持连接

for _ in range(30): # 30 秒心跳

await asyncio.sleep(1)

yield ":heartbeat\n\n"

# 实际业务数据

for chunk in model_stream():

data = json.dumps(chunk, ensure_ascii=False)

yield f"data: {data}\n\n"

# 发送结束标记

yield "data: [DONE]\n\n"

except Exception as e:

yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"

return StreamingResponse(

event_generator(),

media_type="text/event-stream",

headers={

"Cache-Control": "no-cache",

"Connection": "keep-alive",

"X-Accel-Buffering": "no" # 禁用 Nginx 缓冲

}

)

前端（带重连的 SSE 客户端）：

class SSEClient {

constructor(url, options = {}) {

this.url = url;

this.options = options;

this.eventSource = null;

this.reconnectAttempts = 0;

this.maxReconnectAttempts = options.maxReconnect || 5;

this.decoder = new TextDecoder('utf-8');

}

connect() {

this.eventSource = new EventSource(this.url);

this.eventSource.onopen = () => {

console.log('SSE connected');

this.reconnectAttempts = 0;

};

this.eventSource.onmessage = (event) => {

if (event.data === '[DONE]') {

this.close();

return;

}

try {

const data = JSON.parse(event.data);

this.options.onMessage?.(data);

} catch (e) {

this.options.onError?.(e);

}

};

this.eventSource.onerror = (error) => {

this.eventSource.close();

if (this.reconnectAttempts < this.maxReconnectAttempts) {

setTimeout(() => {

this.reconnectAttempts++;

this.connect();

}, 1000 * Math.pow(2, this.reconnectAttempts));

} else {

this.options.onError?.(new Error('Max reconnect attempts reached'));

}

};

}

close() {

this.eventSource?.close();

}

Nginx 配置调优

location /chat {

proxy_pass http://backend;

proxy_http_version 1.1;

proxy_set_header Connection "";

proxy_set_header Cache-Control "no-cache";

# 关键：关闭缓冲，确保实时推送

proxy_buffering off;

proxy_cache off;

# 延长超时时间

proxy_read_timeout 300s;

proxy_send_timeout 300s;

# 添加 SSE 支持头

proxy_set_header Accept "text/event-stream";

}

---

生产环境最佳实践

1. 统一编码处理中间件

encoding_middleware.py

import json

from starlette.middleware.base import BaseHTTPMiddleware

class EncodingMiddleware(BaseHTTPMiddleware):

async def dispatch(self, request, call_next):

response = await call_next(request)

# 确保所有 JSON 响应使用 UTF-8 且不转义 Unicode

if response.headers.get("content-type") == "application/json":

body = b""

async for chunk in response.body_iterator:

body += chunk

data = json.loads(body)

body = json.dumps(data, ensure_ascii=False).encode('utf-8')

response.headers["content-length"] = str(len(body))

response.body_iterator = iter([body])

return response

2. 响应格式校验器

from pydantic import BaseModel, validator

from typing import Optional

class LLMResponse(BaseModel):

content: str

finish_reason: Optional[str] = None

@validator('content')

def check_encoding(cls, v):

# 检测是否包含未解码的 Unicode 转义

if r'\u' in v:

v = v.encode('utf-8').decode('unicode_escape')

return v

@validator('content')

def check_garbled(cls, v):

# 检测乱码字符（替换字符 U+FFFD）

if '\ufffd' in v:

raise ValueError("Response contains garbled characters")

return v

3. 全链路日志追踪

import logging

import uuid

logger = logging.getLogger(__name__)

def call_llm_with_trace(payload):

trace_id = str(uuid.uuid4())

logger.info(f"[{trace_id}] LLM request: {json.dumps(payload, ensure_ascii=False)}")

try:

resp = requests.post(API_URL, json=payload, timeout=60)

raw_text = resp.text

# 记录原始响应前 500 字符

logger.info(f"[{trace_id}] Raw response preview: {raw_text[:500]}")

# 安全解析

result = safe_parse_json(raw_text)

logger.info(f"[{trace_id}] Parsed successfully")

return result

except Exception as e:

logger.error(f"[{trace_id}] LLM call failed: {e}")

raise

4. 客户端健壮性处理

// 统一处理各种格式异常

function normalizeLLMResponse(raw) {

// 1. 处理 Unicode 转义

if (typeof raw === 'string' && raw.includes('\\u')) {

raw = raw.replace(/\\u([0-9a-fA-F]{4})/g, (_, hex) =>

String.fromCharCode(parseInt(hex, 16))

);

}

// 2. 提取 Markdown 代码块内容

if (raw.includes('`')) {

const match = raw.match(/(?:json)?\s*\n([\s\S]*?)\n/);

if (match) raw = match[1];

}

// 3. 尝试解析 JSON

try {

return JSON.parse(raw);

} catch {

return { content: raw, format: 'raw' };

}

---

总结与参考

核心要点速查表

问题	根因	解决方案
Unicode 转义	API 默认行为	`ensure_ascii=False` 或客户端解码
乱码方块	编码不一致	强制 UTF-8，检查代理层
JSON 截断	超长输出/网络问题	截断修复 + 超时重试
SSE 断流	代理超时/网络抖动	心跳保活 + 指数退避重连
Markdown 嵌套	正则匹配贪婪	使用非贪婪模式或专用解析器

Unicode 转义	API 默认行为	`ensure_ascii=False` 或客户端解码
乱码方块	编码不一致	强制 UTF-8，检查代理层
JSON 截断	超长输出/网络问题	截断修复 + 超时重试
SSE 断流	代理超时/网络抖动	心跳保活 + 指数退避重连
Markdown 嵌套	正则匹配贪婪	使用非贪婪模式或专用解析器

乱码方块	编码不一致	强制 UTF-8，检查代理层
JSON 截断	超长输出/网络问题	截断修复 + 超时重试
SSE 断流	代理超时/网络抖动	心跳保活 + 指数退避重连
Markdown 嵌套	正则匹配贪婪	使用非贪婪模式或专用解析器

JSON 截断	超长输出/网络问题	截断修复 + 超时重试
SSE 断流	代理超时/网络抖动	心跳保活 + 指数退避重连
Markdown 嵌套	正则匹配贪婪	使用非贪婪模式或专用解析器

SSE 断流	代理超时/网络抖动	心跳保活 + 指数退避重连
Markdown 嵌套	正则匹配贪婪	使用非贪婪模式或专用解析器