ChatGPT复制格式实战指南：从原理到避坑

1次阅读

共计 3211 个字符，预计需要花费 9 分钟才能阅读完成。

最近在对接 ChatGPT API 时，发现它的响应格式处理起来比想象中复杂得多。主要遇到这几个问题：

流式响应像拼图 ：API 返回的数据是分块的，需要自己拼接完整内容，稍不注意就会漏数据
Markdown 嵌套太自由 ：表格、代码块这些结构可能出现在任意位置，正则表达式经常失效
格式突变防不胜防 ：同一个对话中，前一条返回标准 JSON，下一条可能突然变成纯文本

我见过最典型的翻车现场：

直接按固定偏移量截取内容，结果用户输入带 emoji 时位置全错
用字符串匹配提取 Markdown 表格，遇到单元格内换行直接崩盘
没处理流式响应终止符，导致最后 10% 的内容永远丢失

# 危险示范！实际千万别这么写
response = """{"choices": [{"text":"| 姓名 | 年龄 |\n|------|------|\n| 张三 | 25  |"}]}"""
table_start = response.find('|')  # 太依赖固定格式了

优点：零依赖，代码简单
缺点：脆弱到令人发指，格式稍有变化就崩溃

import re

# 提取 Markdown 表格的改良版
markdown_table = re.compile(r'\|(.+?)\|.*?\n\|[-:|]+\n((?:\|.*?\|\n?)+)')

优点：能处理简单结构化数据
缺点：复杂嵌套结构仍然无解，维护成本高

from openai.util import convert_to_dict
import markdown_it

# 官方推荐方式
response_dict = convert_to_dict(api_response) 
md = markdown_it.MarkdownIt()
html = md.render(response_dict['choices'][0]['text'])

优点：稳定性最佳，支持复杂结构
缺点：需要学习额外 API

import json

def parse_basic_response(raw_response):
    try:
        data = json.loads(raw_response)
        if 'choices' not in data:
            raise ValueError("非标准 API 响应结构")

        # 安全访问嵌套字段
        first_choice = data.get('choices', [{}])[0]
        return first_choice.get('message', {}).get('content', "")
    except json.JSONDecodeError:
        # 兼容非 JSON 响应
        return raw_response

async def handle_stream_response(stream):
    full_content = []
    async for chunk in stream:
        if chunk['object'] == 'chat.completion.chunk':
            delta = chunk['choices'][0].get('delta', {})
            if 'content' in delta:
                full_content.append(delta['content'])

    # 合并时注意换行符处理
    return ''.join(full_content).replace('\n\n','\n')

from bs4 import BeautifulSoup
import html

def extract_safe_markdown_tables(md_text):
    # 先转换 Markdown 为 HTML
    html_content = markdown_to_html(md_text)

    # 使用 BeautifulSoup 防护 XSS
    soup = BeautifulSoup(html_content, 'html.parser')
    for tag in soup.find_all():
        if tag.name not in {'table', 'tr', 'td', 'th'}:
            tag.decompose()

    # 二次转义防注入
    safe_html = html.escape(str(soup))
    return safe_html

API 偶尔会漏发 Content-Type 头，建议双重验证：

content_type = response.headers.get('Content-Type', '')
if 'application/json' not in content_type:
    # 即使没有 header 也尝试 JSON 解析
    try:
        return json.loads(raw_text)
    except:
        return {'raw': raw_text}  # 保留原始数据

遇到长文生成时，推荐使用生成器逐步处理：

def chunked_processor(stream, chunk_size=1024):
    buffer = ""
    for chunk in stream:
        buffer += chunk
        if len(buffer) >= chunk_size:
            processed = process_part(buffer)
            yield processed
            buffer = ""
    if buffer:  # 处理剩余部分
        yield process_part(buffer)

def check_format_consistency(messages):
    formats = set()
    for msg in messages:
        if '```' in msg['content']:
            formats.add('code')
        elif '|' in msg['content'] and '-' in msg['content']:
            formats.add('table')

    if len(formats) > 1:
        logger.warning(f'检测到混合格式: {formats}')

用 10KB 响应体测试三种方案（单位：ms）：

方法	平均耗时	峰值内存
直接字符串处理	1.2	2MB
正则表达式	4.8	3MB
官方库 + 自定义处理	6.4	5MB

def sanitize_prompt(prompt):
    # 阻止试图改变格式的注入
    forbidden_patterns = [
        r'```.*?```',  # 禁止代码块
        r'\{\{.*?\}\}',  # 禁止模板语法
    ]
    for pattern in forbidden_patterns:
        prompt = re.sub(pattern, '', prompt, flags=re.DOTALL)
    return prompt

def filter_sensitive_info(text):
    patterns = {'credit_card': r'\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14})\b',
        'phone': r'\b1[3-9]\d{9}\b'
    }
    for _, pattern in patterns.items():
        text = re.sub(pattern, '[REDACTED]', text)
    return text