1. 为什么数据格式是LLM应用的第一道坎做LLM应用开发这几年我踩过最多的坑不是模型选型也不是提示词设计而是数据格式。听起来很基础对吧但实际情况是我见过太多项目在原型阶段跑得挺欢一到接入真实数据就各种报错——JSON解析失败、字段对不上、编码乱码、嵌套层级太深导致token爆炸。这些问题归根结底都是数据格式没处理好。LLM常见数据格式这个话题说白了就是搞清楚大语言模型在训练、推理、微调、应用集成这几个环节里数据到底长什么样、该怎么组织。它解决的核心问题是让你在构建LLM应用时知道用什么格式喂数据、用什么格式存数据、用什么格式传数据避免在格式转换上浪费大量时间。不管你是刚接触LLM的开发者还是已经在做RAG、Agent、微调的老手这篇文章里的内容都能直接拿去用。我自己的经验是LLM项目里大概有30%到40%的调试时间花在数据格式问题上。这个比例高得离谱但只要你把常见格式和它们的使用场景摸清楚了这部分时间至少能砍掉一半。下面我就按实际项目中的使用频率从最核心的几种格式开始拆解。2. LLM全链路中的核心数据格式拆解2.1 JSON与JSONL微调数据的主力格式JSON大家都不陌生但在LLM场景里它有一个特殊变体叫JSONLJSON Lines这个才是微调数据的主力格式。JSONL的特点是每行一个独立的JSON对象行与行之间用换行符分隔。为什么不用标准JSON数组因为JSONL支持流式读取你不需要把整个文件加载到内存里就能逐行处理。对于动辄几个GB的微调数据集来说这个特性太重要了。我拿一个实际的微调数据例子来说明。假设你要做指令微调每条数据通常包含三个核心字段instruction指令、input输入上下文、output期望输出。用JSONL格式写出来是这样的{instruction: 将以下文本翻译成英文, input: 今天天气很好, output: The weather is nice today} {instruction: 总结这段话的核心观点, input: 长文本内容..., output: 核心观点是...}每一行都是一个完整的训练样本互不干扰。这种格式的好处是你在预处理阶段可以用脚本逐行清洗、过滤、转换不用担心内存溢出。而且很多训练框架比如Hugging Face的datasets库原生支持JSONL的流式加载你直接传文件路径就行。但JSONL也有坑。最常见的问题是字段名不统一。比如有的数据集用instruction有的用prompt有的用query。你在合并多个数据源的时候必须先做字段映射否则训练脚本会直接报KeyError。我的做法是写一个统一的预处理脚本把所有数据源的字段名标准化成同一套命名规范然后再合并。这个脚本大概长这样import json def normalize_sample(sample, field_map): normalized {} for target_field, source_field in field_map.items(): normalized[target_field] sample.get(source_field, ) return normalized field_map { instruction: prompt, input: context, output: response } with open(raw_data.jsonl, r) as fin, open(normalized.jsonl, w) as fout: for line in fin: sample json.loads(line.strip()) normalized normalize_sample(sample, field_map) fout.write(json.dumps(normalized, ensure_asciiFalse) \n)注意ensure_asciiFalse这个参数不加的话中文会被转义成\uXXXX形式虽然不影响解析但文件体积会膨胀好几倍而且人眼没法直接检查数据。实操心得JSONL文件在写入时一定要确保每行末尾有换行符最后一行也不例外。有些工具在读取时如果最后一行没有换行符会直接忽略掉那条数据而且不报错。这个坑我踩过排查了半天才发现是文件末尾少了换行。2.2 ChatML与对话格式多轮对话的标准化方案做对话类应用的时候数据格式就变成了多轮对话的结构。目前业界比较通用的方案是ChatMLChat Markup Language风格它的核心思想是用特殊token来标记不同角色的发言边界。一个典型的多轮对话样本长这样{ messages: [ {role: system, content: 你是一个专业的翻译助手}, {role: user, content: 把你好翻译成英文}, {role: assistant, content: Hello}, {role: user, content: 那谢谢呢}, {role: assistant, content: Thank you} ] }这个格式的关键在于role字段通常有system、user、assistant三种角色。system消息用来设定模型的行为准则user是用户输入assistant是模型回复。在训练时通常只对assistant角色的内容计算lossuser和system部分会被mask掉。这个细节很重要如果你不小心把user的内容也纳入loss计算模型会学会自己跟自己对话输出质量会明显下降。不同模型对对话格式的要求有细微差别。有的模型要求system消息必须放在第一条有的模型支持在对话中间插入system消息。有的模型用|im_start|和|im_end|作为特殊token有的用[INST]和[/INST]。你在准备数据之前一定要先查清楚目标模型的对话模板chat template。很多训练框架提供了tokenizer.apply_chat_template()方法你直接把messages列表传进去它会自动帮你转换成模型需要的格式。这个函数一定要用不要自己手动拼接字符串否则很容易漏掉特殊token或者空格。我见过一个典型的错误案例某开发者手动拼接对话历史用\n分隔不同角色的发言结果模型训练出来之后回复里经常莫名其妙地出现user:这样的前缀。原因就是训练数据里没有明确的分隔标记模型分不清哪些内容该生成、哪些不该生成。用标准的ChatML格式加上正确的chat template这个问题就不会出现。2.3 嵌入向量与向量数据库格式做RAG应用的时候数据格式就变成了向量。文本经过嵌入模型处理后会变成一个浮点数数组通常是768维、1024维或1536维。这个数组本身没什么好说的但它在存储和检索时的格式选择很关键。向量数据库通常支持两种存储方式一种是存原始向量加元数据另一种是存向量加原始文本。我建议两种都存因为检索回来之后你往往需要展示原始文本给用户看如果只存了向量还得回源查询多一次网络往返。元数据部分用JSON格式存可以放文档来源、页码、时间戳等信息方便做过滤。向量的精度也值得注意。很多嵌入模型默认输出float32但存储时可以用float16甚至int8量化来节省空间。float16能把存储空间减半检索精度损失很小通常在1%以内。int8量化能压缩到四分之一但精度损失会明显一些大概在3%到5%。如果你的向量库规模在百万级别以下用float32就行没必要折腾量化。上了千万级别再考虑float16或int8。import numpy as np # 模拟一个嵌入向量 embedding np.random.rand(1536).astype(np.float32) # 转为float16存储 embedding_fp16 embedding.astype(np.float16) print(ffloat32大小: {embedding.nbytes} bytes) print(ffloat16大小: {embedding_fp16.nbytes} bytes)注意事项向量归一化是一个容易被忽略的细节。有些嵌入模型输出的向量已经归一化了有些没有。如果你用余弦相似度做检索归一化不影响结果但如果你用内积dot product做检索没归一化的向量会导致相似度分数不可比。最稳妥的做法是在存入向量库之前统一做L2归一化。2.4 结构化输出格式Function Calling与JSON Mode让LLM输出结构化数据是另一个高频需求。比如你让模型从一段文本里抽取实体希望它输出JSON格式的结果。但LLM天生是生成自由文本的你不加约束的话它可能在JSON前后加一堆解释性文字导致你的解析脚本直接崩溃。目前有两种主流方案。第一种是JSON Mode你在调用API时设置response_format{type: json_object}模型会保证输出合法的JSON。但JSON Mode只保证格式合法不保证字段名和字段类型符合你的预期。第二种是Function Calling也叫Tool Use你定义一个函数签名模型会按照签名来输出参数。这个约束更强字段名和类型都会被校验。我一般优先用Function Calling因为它的约束更明确。举个例子你要从用户评论中抽取情感倾向和关键词{ name: extract_sentiment, parameters: { type: object, properties: { sentiment: {type: string, enum: [positive, negative, neutral]}, keywords: {type: array, items: {type: string}}, confidence: {type: number, minimum: 0, maximum: 1} }, required: [sentiment, keywords] } }模型输出的时候会严格按照这个schema来你拿到结果直接json.loads就行不需要做额外的清洗。但要注意不是所有模型都支持Function Calling而且不同模型对schema的支持程度也不一样。有的模型不支持enum约束有的模型对嵌套对象的支持不好。你在选型的时候要先测试一下。3. 实操从原始数据到训练格式的完整转换流程3.1 数据清洗与字段标准化假设你手头有一批原始数据格式五花八门有的是CSV有的是JSON有的是纯文本。第一步是统一清洗。我通常写一个pipeline分三步走读取、清洗、输出。读取阶段要根据文件扩展名选择不同的解析器。CSV用pandasJSON用json库纯文本直接按行读。清洗阶段要做的事情包括去除HTML标签、去除多余空白字符、过滤掉过短或过长的样本、检查必填字段是否为空。输出阶段统一写成JSONL格式字段名标准化。import pandas as pd import json import re def clean_text(text): if not isinstance(text, str): return text re.sub(r[^], , text) # 去HTML标签 text re.sub(r\s, , text).strip() # 合并空白 return text def process_csv(filepath): df pd.read_csv(filepath) samples [] for _, row in df.iterrows(): sample { instruction: clean_text(row.get(question, )), input: clean_text(row.get(context, )), output: clean_text(row.get(answer, )) } if len(sample[instruction]) 5 or len(sample[output]) 1: continue samples.append(sample) return samples这段代码里有两个细节值得说。第一个是clean_text函数对非字符串类型的处理直接返回空字符串。因为pandas读取CSV时空单元格会变成NaN浮点数如果你不处理后续的字符串操作会报错。第二个是过滤逻辑指令少于5个字符或者输出为空的样本直接丢弃。这种样本通常是脏数据留着只会干扰训练。3.2 格式转换中的token长度控制清洗完之后下一步是控制token长度。LLM有最大上下文限制比如4096或8192个token。如果你的样本超长了要么截断要么丢弃。我的策略是instruction和output尽量保留完整input部分如果超长就截断。计算token数需要用目标模型对应的tokenizer。不要用字符数估算中英文的token比例差异很大。一个中文字通常对应1到2个token一个英文单词通常对应1到1.5个token。但具体数字取决于tokenizer的实现。from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(model_name) def truncate_sample(sample, max_length2048): instruction_tokens tokenizer.encode(sample[instruction]) output_tokens tokenizer.encode(sample[output]) input_budget max_length - len(instruction_tokens) - len(output_tokens) - 10 if input_budget 0: return None # 指令和输出本身就超长了丢弃 input_tokens tokenizer.encode(sample[input])[:input_budget] sample[input] tokenizer.decode(input_tokens) return sample这里的-10是留给特殊token的余量。不同模型的特殊token数量不一样有的多有的少留10个是比较安全的做法。如果你不确定可以留20个。实操心得截断input的时候尽量在句子边界截断不要从句子中间切断。你可以用tokenizer.decode之后找到最后一个句号或换行符的位置把后面的内容去掉。这样截断后的文本更自然对训练效果的影响更小。3.3 训练格式的最终组装最后一步是把清洗好的数据组装成训练框架需要的格式。如果你用Hugging Face的Trainer通常需要把数据转成datasets.Dataset对象。如果你用LLaMA-Factory这类微调框架它们通常支持直接读取JSONL文件你只需要在配置里指定字段映射就行。我以LLaMA-Factory的配置为例dataset: - name: my_dataset file: data/normalized.jsonl format: alpaca columns: instruction: instruction input: input output: output这里的format: alpaca指的是Alpaca格式它要求数据包含instruction、input、output三个字段。LLaMA-Factory会自动把这些字段拼接到模型对应的chat template里。你不需要自己拼接框架会处理。但要注意不同框架对格式的要求不一样。有的框架要求sharegpt格式数据里要有conversations字段每个元素包含from和value。你在准备数据之前先确定用哪个框架然后按照它的文档来准备。不要先准备数据再选框架那样很可能要返工。4. 常见问题与排查技巧实录4.1 编码问题乱码与BOM头编码问题是LLM数据处理中最常见的问题之一。典型症状是数据里出现\ufeff、é、’这样的乱码。\ufeff是BOMByte Order Mark头通常出现在Windows系统创建的UTF-8文件里。Python的open()函数默认不会去掉BOM你需要用encodingutf-8-sig来读取。# 错误写法 with open(data.jsonl, r, encodingutf-8) as f: for line in f: print(line[:20]) # 可能输出 \ufeff{instruction... # 正确写法 with open(data.jsonl, r, encodingutf-8-sig) as f: for line in f: print(line[:20]) # 输出 {instruction...é和’这类乱码通常是UTF-8被错误地按Latin-1解码导致的。解决办法是先用chardet库检测文件编码然后用正确的编码读取。import chardet with open(data.txt, rb) as f: raw f.read() detected chardet.detect(raw) print(detected) # {encoding: utf-8, confidence: 0.99}4.2 JSON解析失败尾逗号与转义字符JSON解析失败的原因主要有两个尾逗号和未转义的特殊字符。尾逗号是指{a: 1, b: 2,}这种最后一个元素后面多了逗号标准JSON不允许这样。未转义的特殊字符是指字符串里出现了双引号或反斜杠但没有转义。排查这类问题我一般用json.loads逐行解析捕获异常并打印行号import json with open(data.jsonl, r, encodingutf-8-sig) as f: for i, line in enumerate(f, 1): try: json.loads(line) except json.JSONDecodeError as e: print(f第{i}行解析失败: {e}) print(f内容: {line[:100]})如果数据量很大逐行解析太慢可以用jq命令行工具快速检查jq . data.jsonl /dev/nulljq会在遇到第一个非法JSON时报错并退出错误信息里会包含行号。4.3 字段缺失与类型不匹配字段缺失的问题在合并多个数据源时特别常见。A数据源用question字段B数据源用query字段你合并之后发现一半的数据没有instruction。解决办法是在合并之前先做字段映射映射表要显式定义不要靠猜。类型不匹配的问题也很常见。比如某个字段在A数据源里是字符串在B数据源里是数组。你直接合并的话训练脚本在处理时会报类型错误。我的做法是写一个schema校验函数对每个字段做类型检查不符合的样本直接丢弃或者转换。def validate_sample(sample): if not isinstance(sample.get(instruction), str): return False if not isinstance(sample.get(output), str): return False if len(sample[instruction]) 0 or len(sample[output]) 0: return False return True4.4 常见问题速查表问题现象可能原因排查方法解决方案文件开头出现\ufeffUTF-8 BOM头用hexdump查看文件头读取时用utf-8-sig编码中文显示为é编码解码不匹配用chardet检测编码用检测到的编码重新读取JSON解析报错尾逗号或未转义字符逐行json.loads捕获异常用正则清洗或手动修复字段缺失多数据源字段名不统一检查各数据源的字段名写字段映射脚本统一命名训练时loss为NaN数据中有空样本或超长样本检查样本长度分布过滤空样本截断超长样本模型输出重复训练数据中output有重复内容统计output的重复率去重或降低重复样本权重避坑技巧在处理大规模数据之前先拿100条样本跑一遍完整流程。确认格式转换、token计算、训练加载都没问题之后再全量处理。这样能避免跑了几小时才发现格式错误的情况。5. 不同场景下的格式选型建议5.1 微调场景JSONL ChatML是首选微调场景下我强烈建议用JSONL存储数据对话格式用ChatML。原因有三第一JSONL支持流式读取内存友好第二ChatML是业界通用标准主流模型和框架都支持第三JSONL方便做数据版本管理和增量更新你可以在文件末尾追加新数据不需要重写整个文件。如果你做的是指令微调而非对话微调Alpaca格式instruction/input/output也够用。但如果你后续要扩展到多轮对话还是ChatML更灵活。我的建议是一步到位直接用ChatML格式省得后面转换。5.2 RAG场景原始文本 向量 元数据三件套RAG场景的数据格式要复杂一些因为涉及文本、向量、元数据三种数据。我的做法是用一个JSON对象把三者打包在一起{ id: doc_001_chunk_003, text: 原始文本内容..., embedding: [0.123, -0.456, ...], metadata: { source: 某文档.pdf, page: 5, chunk_index: 3, timestamp: 2024-01-15T10:30:00Z } }存储的时候向量部分存入向量数据库文本和元数据存入关系数据库或文档数据库。检索时先用向量相似度找到候选ID再用ID回查文本和元数据。这种分离存储的方式比把所有数据都塞进向量数据库更灵活也更容易做元数据过滤。5.3 推理API场景请求与响应的格式约定调用LLM推理API时请求和响应的格式也需要约定好。请求体通常是JSON包含model、messages、temperature、max_tokens等参数。响应体也是JSON包含choices数组每个choice里有message字段。我建议在应用层定义一个统一的请求响应格式屏蔽不同API提供商的差异。这样你切换模型提供商的时候只需要改适配层不需要改业务代码。class LLMRequest: def __init__(self, messages, temperature0.7, max_tokens1024): self.messages messages self.temperature temperature self.max_tokens max_tokens def to_dict(self): return { messages: self.messages, temperature: self.temperature, max_tokens: self.max_tokens } class LLMResponse: def __init__(self, content, finish_reason, usage): self.content content self.finish_reason finish_reason self.usage usage这种封装方式的好处是你的业务代码只依赖LLMRequest和LLMResponse这两个类不直接依赖任何API提供商的SDK。切换提供商时只需要写一个新的适配器把LLMRequest转成目标API的格式再把响应转回LLMResponse。5.4 格式选型的决策流程面对一个具体项目怎么选数据格式我通常按这个流程走第一步确定数据用途。是训练、推理还是存储训练用JSONL推理用JSON存储看具体数据库。第二步确定模型类型。是对话模型还是基座模型对话模型用ChatML基座模型用纯文本加分隔符。第三步确定框架要求。你用的训练框架或推理框架对格式有什么硬性要求先查文档按文档来。第四步考虑扩展性。未来会不会加多轮对话会不会加多模态数据选一个容易扩展的格式避免后期大改。这套流程走下来基本上不会选错格式。我自己的项目里微调数据统一用JSONL ChatMLRAG数据用JSON 向量分离存储推理请求用统一的JSON封装。这套组合用了两年多没出过什么大问题。最后分享一个小技巧不管你用什么格式都写一个格式校验脚本在数据进入训练或推理流程之前跑一遍。校验内容包括JSON合法性、必填字段是否存在、字段类型是否正确、样本长度是否在范围内。这个脚本花不了多少时间但能帮你省下大量排查格式问题的时间。我在每个项目里都会放一个validate_data.py每次数据更新后自动跑一遍已经成为标准流程了。