1. 为什么数据格式是LLM应用里最容易被低估的一环
做LLM应用开发的人，十有八九把注意力放在模型选型、提示词调优、RAG召回率这些“显性”环节上，而数据格式往往被当成一个“随便拼拼字符串”的小事。我见过太多项目，模型能力没问题、检索也没问题，最后卡在数据格式上——要么是JSON解析失败导致整条链路崩掉，要么是特殊token被转义吃掉，要么是流式输出拼接时多了一个逗号让前端渲染炸掉。这些问题排查起来极其耗时，因为它们不报“模型错误”，报的是“解析错误”，而解析错误的信息量往往少得可怜。
LLM场景下的数据格式，和传统后端开发里的数据格式有本质区别。传统开发里，数据格式是给人看的、给程序解析的，结构稳定、边界清晰。但LLM场景里，数据格式同时要服务三个对象：模型本身（它需要特定的对话结构才能正确理解角色和指令）、调用方程序（需要结构化数据做后续处理）、最终用户（需要可读的输出）。这三者的需求经常冲突，而数据格式就是调和冲突的那层“接口协议”。
这篇文章面向的是已经上手调用过LLM API、但在多轮对话、工具调用、流式输出、批量推理这些场景里被数据格式折腾过的开发者。我会把LLM生态里最常见的几类数据格式拆开讲清楚：它们长什么样、为什么长这样、什么场景该用哪个、以及我在实际项目里踩过的那些坑。不管你是用OpenAI风格的消息数组，还是Anthropic的content block，还是自己定义的一套内部格式，底层的设计逻辑是相通的。
先给一个整体认知：LLM的数据格式大致分四层——对话消息层（messages）、内容块层（content blocks）、工具调用层（tool/function calling）、流式传输层（streaming chunks）。很多人只关注第一层，后面三层出了问题就抓瞎。下面逐层拆。
2. 对话消息格式：messages数组的三种主流形态
2.1 role-content二元组：最基础也最容易踩坑的格式
最经典的对话格式就是messages数组，每个元素是一个对象，包含role和content两个字段。这是OpenAI Chat Completions API带火的格式，现在几乎成了行业默认标准。一个典型的多轮对话长这样：
[ {"role": "system", "content": "你是一个严谨的技术助手。"}, {"role": "user", "content": "解释一下什么是向量数据库"}, {"role": "assistant", "content": "向量数据库是..."}, {"role": "user", "content": "那它和传统数据库的区别呢？"} ]看起来简单，但坑非常多。第一个坑是role的取值。不同厂商支持的role不完全一样。OpenAI支持system、user、assistant、tool、function（旧版），而有些模型只支持user和assistant，你传system它直接报错或者静默忽略。我遇到过一个国产模型，system消息必须放在数组第一位，放中间会被当成普通user消息处理，导致角色设定完全失效。
第二个坑是content的类型。早期content只能是字符串，现在很多API支持content是数组（用于多模态）。如果你写代码时假设content永远是字符串，遇到多模态输入就会崩。稳妥的做法是在解析前先判断类型：
def normalize_content(content): if isinstance(content, str): return [{"type": "text", "text": content}] if isinstance(content, list): return content raise ValueError(f"不支持的content类型: {type(content)}")第三个坑是空content。有些场景下assistant消息的content是空的（比如纯工具调用），但content字段不能省略，得传空字符串或者null。不同API对null的处理不一样，有的接受null，有的要求空字符串，有的直接拒绝。我的经验是统一传空字符串，兼容性最好。
2.2 多模态content数组：text、image、audio的混排规则
当对话涉及图片、音频时，content就从字符串变成了数组，每个元素是一个内容块，用type字段区分。OpenAI的格式大致是：
{ "role": "user", "content": [ {"type": "text", "text": "这张图里有什么？"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}} ] }这里有几个实操细节值得说。图片的传入方式有两种：URL和base64。URL方式简单，但要求图片可公网访问，内网环境用不了；base64方式通用，但体积会膨胀约33%，大图容易超token限制。我一般对小于500KB的图用base64，大图先压缩再传。
内容块的顺序会影响模型理解。把text放在image前面还是后面，模型的注意力分布是不一样的。实测下来，对于“描述这张图”这类任务，text在前效果更稳；对于“根据这段文字找图中对应区域”这类任务，image在前更好。这不是玄学，是因为自回归模型对靠前的内容有更强的条件约束。
音频和视频的支持还在演进中，格式不统一。有的用input_audio，有的用audio_url，字段名和编码要求都不一样。如果你的应用要跨厂商，建议在内部定义一套统一的内容块格式，调用不同API时再做适配转换。这个适配层看起来麻烦，但比在每个业务代码里写if-else要干净得多。
2.3 消息数组的截断与拼接：token预算下的取舍策略
多轮对话不可能无限增长，必须截断。截断策略直接决定了数据格式怎么组织。常见的有三种：
- 滑动窗口：保留最近N轮，简单粗暴，但会丢失早期的重要上下文。
- 系统消息+最近N轮：system消息永远保留，后面滑动。这是最常用的策略。
- 摘要压缩：把早期对话用模型总结成一段话，替换掉原始消息。省token但引入额外调用成本和信息损失。
我在实际项目里用的是混合策略：system消息固定保留，最近K轮完整保留，更早的对话按“用户问题+助手回答首句”的方式压缩成简版。这样既保留了话题脉络，又控制了token。实现上就是在构造messages数组前做一次预处理：
def build_messages(history, system_prompt, max_recent=6, max_total_tokens=3000): messages = [{"role": "system", "content": system_prompt}] recent = history[-max_recent:] older = history[:-max_recent] # 早期对话压缩 for turn in older: messages.append({"role": "user", "content": turn["user"][:50]}) messages.append({"role": "assistant", "content": turn["assistant"][:30] + "..."}) messages.extend(recent) return messages注意截断时不能把user和assistant拆散，否则模型会困惑。有些API要求messages必须以user开头、assistant结尾（或者交替出现），截断时要保证这个约束。我见过因为截断导致连续两条user消息，模型直接开始“自问自答”的案例。
3. 内容块格式：从纯文本到结构化输出的桥梁
3.1 content block的设计哲学：为什么不用纯字符串
纯字符串的content有一个根本问题：无法表达结构。当模型需要返回代码、表格、引用、思考过程时，全塞在一个字符串里，调用方就得靠正则去解析，脆弱且易错。content block就是为了解决这个问题——把不同类型的内容用不同的块来表达，每块有自己的类型和字段。
Anthropic的content block格式是比较有代表性的设计：
{ "role": "assistant", "content": [ {"type": "text", "text": "让我计算一下。"}, {"type": "tool_use", "id": "toolu_01", "name": "calculator", "input": {"expr": "2+3"}} ] }这种设计的好处是可扩展。新增一种内容类型（比如思考块、引用块），只要加一个新的type，不影响已有解析逻辑。坏处是复杂度上升，调用方必须处理多种块类型，还要处理块之间的顺序和依赖关系。
我的建议是：如果你的应用只做纯文本问答，用字符串content就够了，别过度设计。一旦涉及工具调用、多模态、结构化输出中的任意一个，就果断上content block。因为到那时候你会发现，字符串content根本不够用，后期改造的成本远大于一开始就用对格式。
3.2 结构化输出：JSON mode、schema约束与解析容错
让模型输出JSON是极常见的需求，但“让模型输出JSON”和“让模型输出合法JSON”是两回事。早期大家靠提示词里写“请输出JSON格式”，然后祈祷模型听话。现在主流API都提供了更强的约束机制：
| 机制 | 原理 | 可靠性 | 适用场景 |
|---|---|---|---|
| 提示词约束 | 在prompt里描述格式 | 低 | 简单场景、原型验证 |
| JSON mode | 解码时限制只能输出JSON token | 中高 | 通用JSON输出 |
| Schema约束 | 按JSON Schema约束解码 | 高 | 严格结构化数据 |
| 后处理修复 | 解析失败时用规则/模型修复 | 中 | 兜底方案 |
JSON mode的原理是在解码阶段屏蔽掉不可能构成合法JSON的token，但它只保证“语法合法”，不保证“字段符合预期”。Schema约束更进一步，能保证字段名、类型、枚举值都符合定义。如果你的API支持Schema约束，强烈建议用，能省掉大量后处理代码。
但即使有Schema约束，解析容错也不能省。我遇到过模型在JSON字符串里输出未转义换行符的情况，也遇到过数字被输出成字符串的情况。稳妥的解析流程是：
import json def safe_parse_json(text): # 第一层：直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 第二层：提取代码块中的JSON import re match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass # 第三层：找第一个{到最后一个}之间的内容 start = text.find('{') end = text.rfind('}') if start != -1 and end != -1: try: return json.loads(text[start:end+1]) except json.JSONDecodeError: pass raise ValueError(f"无法解析JSON: {text[:200]}")这个三层解析策略在我多个项目里救过场。第三层尤其有用，因为模型经常在JSON前后加“好的，这是结果：”之类的废话。
3.3 思考块与引用块：新兴内容类型的处理方式
最近一年，带“思考过程”的模型越来越多，对应的content block类型也出现了。思考块（thinking block）的特点是：模型先输出一段内部推理，再输出最终答案。对调用方来说，思考块通常不需要展示给用户，但需要保留在对话历史里，因为下一轮模型可能需要参考。
处理思考块有两个注意点。第一，不要把它当普通文本拼进prompt，否则会污染上下文。第二，多轮对话时思考块要不要回传，不同API要求不一样。有的要求回传（保持推理连续性），有的要求丢弃（节省token）。我一般按API文档来，文档没写的就做实验：回传和不回传各跑一遍，看输出质量差异。
引用块（citation block）常见于RAG场景，模型在回答里标注“这句话来自文档X的第Y段”。格式上通常是文本块加一个引用元数据数组。处理引用块的关键是引用ID和实际文档的映射，这个映射要在应用层维护好，别指望模型记住。我见过模型编造引用ID的情况，所以展示引用前一定要校验ID是否存在。
4. 工具调用格式：function calling的数据契约
4.1 工具定义格式：JSON Schema的写法与陷阱
工具调用的第一步是把工具定义传给模型。主流格式是JSON Schema：
{ "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]} }, "required": ["city"] } } }这里有几个容易踩的坑。description写得好不好，直接决定模型会不会正确调用。我见过description写“查询天气”的，模型经常把城市名和日期搞混；改成“查询指定城市在今天的天气，参数city必须是城市中文名”之后，准确率明显上升。description要具体到“参数该填什么、不该填什么”。
enum约束很有用，但别滥用。如果枚举值太多（超过10个），模型选择困难，反而不如用string加description说明。required字段要谨慎，把非必填字段标成required会导致模型强行编造值。
还有一个隐蔽的坑：工具名不能有特殊字符。有的API只允许字母数字下划线，你写个get-weather就报错。统一用下划线命名最稳。
4.2 工具调用与结果回传：id匹配的完整链路
模型决定调用工具后，返回的assistant消息里会带tool_calls（或tool_use）块，每个调用有一个id。调用方执行完工具后，要把结果以tool角色（或tool_result块）回传，并且id必须严格匹配。完整链路：
// 第一轮：模型请求调用 {"role": "assistant", "content": null, "tool_calls": [ {"id": "call_abc", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\":\"北京\"}"}} ]} // 调用方执行后回传 {"role": "tool", "tool_call_id": "call_abc", "content": "{\"temp\": 25, \"condition\": \"晴\"}"} // 第二轮：模型基于结果生成回答 {"role": "assistant", "content": "北京今天25度，晴天。"}id匹配出错的后果很严重：轻则模型忽略结果重新调用，重则API直接报错。我建议在代码里用一个字典维护id -> 执行结果的映射，回传前校验id存在。另外，arguments是字符串不是对象，需要二次解析。很多新手在这里栽跟头，直接把arguments当dict用，结果报类型错误。
并行工具调用是另一个复杂点。模型可能一次返回多个tool_calls，调用方要全部执行完再一起回传。回传顺序不影响匹配（靠id），但建议按原顺序回传，可读性好。如果某个工具执行失败，回传的内容里要包含错误信息，让模型知道失败了，而不是传空字符串——传空字符串模型会以为成功但结果为空，可能产生幻觉。
4.3 工具调用失败的降级处理
工具调用不可能100%成功。网络超时、参数错误、下游服务挂了，都会失败。降级策略分三层：
第一层，参数校验失败。模型传的参数不符合schema，比如该传数字传了字符串。这时候不要直接报错给模型，而是尝试类型转换（字符串数字转数字），转换失败再回传错误信息让模型重试。
第二层，执行超时。给每个工具调用设超时（比如5秒），超时后回传“执行超时”而不是无限等待。模型收到超时信息后，可能会换个方式调用或者直接告诉用户失败。
第三层，连续失败熔断。如果同一个工具连续失败3次，就停止让模型重试，直接返回兜底回答。否则模型可能陷入“调用-失败-再调用”的死循环，烧token还出不来。
def execute_tool_with_fallback(tool_call, max_retries=3): for attempt in range(max_retries): try: result = execute_tool(tool_call) return {"tool_call_id": tool_call["id"], "content": json.dumps(result)} except TimeoutError: if attempt == max_retries - 1: return {"tool_call_id": tool_call["id"], "content": "工具执行超时"} except Exception as e: if attempt == max_retries - 1: return {"tool_call_id": tool_call["id"], "content": f"工具执行失败: {str(e)}"}5. 流式传输格式：SSE chunk的拼接与解析
5.1 SSE协议下chunk的边界问题
流式输出用的是Server-Sent Events（SSE），每个chunk是一行data: {...}。问题在于，一个chunk不一定是完整JSON。网络传输会随意切分，你可能收到半个JSON对象。所以不能收到一个chunk就json.loads，必须先按SSE协议拼接。
SSE的规则是：以\n\n分隔事件，每个事件内可以有多个data:行。正确的解析流程是维护一个缓冲区，按\n\n切分，对每个完整事件再解析data行。我见过直接按\n切分的代码，遇到data内容里有换行就崩。
class SSEParser: def __init__(self): self.buffer = "" def feed(self, chunk): self.buffer += chunk events = [] while "\n\n" in self.buffer: event_str, self.buffer = self.buffer.split("\n\n", 1) data_lines = [l[5:].strip() for l in event_str.split("\n") if l.startswith("data:")] if data_lines: events.append("\n".join(data_lines)) return events注意data:后面可能有一个空格，也可能没有，l[5:].strip()能兼容两种情况。还有[DONE]标记，表示流结束，要单独处理，不能当JSON解析。
5.2 增量内容拼接：text delta与tool call delta
流式返回的内容是增量的。文本增量好办，直接拼接字符串。但工具调用的增量很麻烦，因为arguments是分片传的：
{"delta": {"tool_calls": [{"index": 0, "function": {"arguments": "{\"ci"}}]}} {"delta": {"tool_calls": [{"index": 0, "function": {"arguments": "ty\":\"北"}}]}} {"delta": {"tool_calls": [{"index": 0, "function": {"arguments": "京\"}"}}]}}必须按index聚合，把arguments片段拼起来，等流结束后再解析。如果多个工具并行调用，index会不同，要分别聚合。我建议用一个字典{index: {"name": "", "arguments": ""}}来累积，流结束后统一处理。
还有一个坑：第一个chunk通常包含role和空的content，别把它当有效内容。最后一个chunk通常包含finish_reason，也没有内容。解析时要判断字段是否存在，别硬取。
5.3 流式场景下的错误恢复
流式传输中途断网怎么办？如果已经输出了一部分内容，重试会导致重复输出。我的做法是：记录已输出的内容长度，重试时带上已输出内容作为上下文，让模型接着写。但这样成本高，且模型可能接不上。
更实用的做法是：前端做去重。重试后把新流的内容和已显示的内容做最长公共前缀匹配，只追加新增部分。这个逻辑在前端做比在后端做更合适，因为前端知道用户已经看到了什么。
如果断流发生在工具调用参数传输中途，那基本没法恢复，只能整个重来。所以工具调用场景下，我一般建议关闭流式，等完整结果返回再处理，牺牲一点响应速度换稳定性。
6. 格式选型与跨厂商适配的实战经验
6.1 内部统一格式与外部适配层的设计
跨厂商调用时，最忌讳的是业务代码里直接拼各家的格式。正确做法是定义一套内部统一格式，在调用层做适配。内部格式我一般这样定义：
@dataclass class Message: role: str # system/user/assistant/tool content: List[ContentBlock] tool_calls: Optional[List[ToolCall]] = None tool_call_id: Optional[str] = None @dataclass class ContentBlock: type: str # text/image/tool_use/tool_result data: dict业务代码只操作Message，调用不同API时用adapter转换。这样换厂商时只改adapter，业务代码不动。适配层要处理的差异包括：role命名（有的用human/ai）、content结构（字符串vs数组）、工具调用字段名（tool_callsvstool_use）、流式chunk结构。这些差异看着琐碎，但集中在一个文件里处理，比散落各处强太多。
6.2 常见格式错误的排查清单
最后给一份排查清单，遇到格式问题时按顺序检查：
| 现象 | 可能原因 | 排查方法 |
|---|---|---|
| 400错误，提示messages格式 | role取值不支持 | 打印messages，对照API文档 |
| 模型忽略system消息 | system位置不对 | 把system放数组第一位 |
| JSON解析失败 | 模型输出带markdown包裹 | 用三层解析策略 |
| 工具调用id不匹配 | 回传时id写错 | 校验id字典 |
| 流式输出乱码 | chunk边界切分错误 | 用SSEParser按\n\n切 |
| 多模态图片不识别 | base64格式或前缀错误 | 检查data:image/png;base64,前缀 |
| token超限 | 历史消息未截断 | 加滑动窗口或摘要压缩 |
这份清单是我从多次线上事故里总结的，每次遇到新问题就加一行。建议你也维护一份，格式问题排查有了清单，效率能提升好几倍。
6.3 我踩过的三个真实坑
第一个坑：JSON mode下模型输出空对象。有次做信息抽取，开了JSON mode，结果模型对某些输入返回{}。排查发现是schema里所有字段都是optional，模型觉得没信息可填就返回空。改成至少一个required字段后正常。教训是：schema设计要引导模型输出，别给太多“可以不填”的自由。
第二个坑：流式拼接时把finish_reason当内容。有个chunk是{"choices":[{"delta":{},"finish_reason":"stop"}]}，delta是空对象，我的代码没判断就直接取delta.content，取到None拼进结果，导致输出末尾多个“None”。修复很简单，加个判断，但这个bug在测试环境没复现，因为测试时网络快，chunk合并方式不同。教训是：流式解析要对每个字段做存在性判断，别假设结构固定。
第三个坑：工具调用参数里的中文被转义。模型返回的arguments里中文变成了\u5317\u4eac，我直接当字符串传给下游，下游解析失败。修复是在解析arguments后做一次unicode解码。这个坑的隐蔽性在于，英文参数完全正常，只有中文出问题，测试用例没覆盖中文就漏了。
数据格式这件事，说到底是“契约”问题。模型、调用方、用户三方之间的契约越清晰，系统越稳。花时间把格式定义好、把解析写健壮、把错误处理做全，比事后救火划算得多。我现在的习惯是：任何新的LLM集成，先花半天把数据格式和解析层写好，再写业务逻辑。这半天省下的调试时间，通常是好几倍。
