1. 项目概述一个被误解的“ChatGPT”仓库在GitHub上搜索“ChatGPT”你会得到成千上万个结果其中有一个仓库名为saschaschramm/chatgpt。乍一看你可能会以为这是一个官方客户端、一个逆向工程接口或者一个功能强大的封装库。但点进去之后你大概率会感到一丝困惑甚至有些失望。因为这个仓库的内容与我们所熟知的、那个能进行深度对话的AI模型似乎关联不大。它更像是一个特定场景下的、高度定制化的工具脚本集合。这个项目本质上是一个基于OpenAI API的、用于自动化处理特定格式文本特别是对话记录的命令行工具。它的核心价值不在于提供一个炫酷的聊天界面而在于解决一个非常具体且常见的工程问题如何将结构混乱或特定格式的原始对话文本比如从某些平台导出的日志批量、自动地转换成符合OpenAI Chat Completion API要求的结构化格式例如JSONL并进行后续的模型微调或批量对话任务。我最初接触这个项目是因为需要处理大量来自不同渠道的客服对话记录用于训练一个垂直领域的助手。手动清洗和格式化这些数据是噩梦级的任务。saschaschramm/chatgpt虽然没有提供开箱即用的聊天机器人但它提供了一套清晰的思路和可复用的代码框架教会了我如何用脚本将“脏数据”管道化地处理成AI的“营养餐”。对于需要处理非标准对话数据、进行批量API调用或数据预处理的开发者来说这个仓库的价值被严重低估了。它更像是一把瑞士军刀中的专业工具而非那把主刀。2. 核心需求与设计思路拆解2.1 真实需求从混乱日志到结构化提示词在AI应用开发中尤其是涉及微调Fine-tuning或构建复杂对话系统时我们面临的原始数据往往是五花八门的。它们可能来自客服系统导出通常是CSV或TXT包含用户ID、时间戳、对话内容但不同客服的发言可能混在一起没有清晰的“用户/助理”角色划分。社区论坛或评论数据帖子与回复的嵌套结构需要被扁平化成线性的对话轮次。自定义的聊天记录可能是某种特殊的日志格式字段分隔符不统一包含大量元信息如IP、设备类型需要剥离。多轮对话的拼接需要将单轮问答组合成符合GPT多轮对话上下文格式的数据。saschaschramm/chatgpt项目敏锐地捕捉到了这个痛点。它的设计思路不是再造一个聊天前端而是专注于数据格式的转换与管道处理。其核心设计可以概括为“输入任意格式的文本流 - 通过可配置的解析器提取对话轮次 - 映射为标准的OpenAI消息格式 - 输出为JSONL或直接调用API”。这种设计将复杂性封装在解析规则里而将通用处理流程如API调用、错误重试、结果收集标准化。开发者只需要关心如何为自己的数据源编写一个“解析器”或“适配器”剩下的批量处理、格式校验、并发请求等脏活累活都可以复用项目提供的框架。2.2 技术选型考量为什么是Python脚本项目选择了Python作为实现语言这是一个非常务实且高效的选择。生态丰富用于文本处理的re正则表达式、json、csv库是Python标准库的一部分。对于更复杂的解析pandas可以轻松集成。HTTP请求有成熟的requests库异步处理有asyncio和aiohttp。快速原型Python的脚本特性非常适合这种数据处理任务。开发者可以快速编写、测试不同的解析规则而不需要编译和复杂的项目结构。与OpenAI SDK无缝集成OpenAI官方提供了Python SDK (openai库)使得API调用变得异常简单。项目可以直接利用无需自己封装HTTP客户端。命令行友好通过argparse或click库可以轻松构建功能强大的命令行界面CLI方便集成到自动化流水线如CI/CD或由其他脚本调用。注意这个项目通常不会是一个需要长期运行的服务它的生命周期是“任务型”的准备数据时运行一次。因此轻量级的脚本形式比构建一个完整的Web服务或桌面应用更合适。3. 核心模块解析与实操要点虽然原始仓库的具体实现可能比较简单但我们可以基于其思想构建一个更健壮、更通用的处理管道。一个完整的解决方案通常包含以下核心模块3.1 输入处理器这个模块负责读取原始数据。它需要支持多种输入源本地文件读取.txt,.csv,.json,.jsonl等。目录批量处理遍历一个文件夹下的所有指定格式文件。标准输入支持管道操作例如cat chat.log | python script.py。数据库查询直接从MySQL、PostgreSQL或MongoDB中拉取对话记录。实操要点使用with open(...) as f来确保文件句柄正确关闭。对于大文件考虑流式读取逐行避免一次性加载到内存。为不同的文件格式编写对应的读取函数并通过一个统一的接口调用。import json import csv from pathlib import Path def read_file(file_path): 根据文件后缀名选择读取器 suffix Path(file_path).suffix.lower() if suffix .jsonl: return read_jsonl(file_path) elif suffix .csv: return read_csv(file_path) elif suffix .txt: return read_txt(file_path) else: raise ValueError(fUnsupported file format: {suffix}) def read_jsonl(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: yield json.loads(line.strip()) def read_csv(file_path): with open(file_path, r, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: yield row3.2 对话解析器这是整个项目的灵魂所在也是最需要定制开发的部分。解析器的任务是将一行原始文本或一个数据对象解析成一组具有“角色”和“内容”的对话消息。常见解析模式正则表达式匹配适用于有固定模式的日志。示例日志格式为[2023-10-27 10:00:00] USER: 你好吗和[2023-10-27 10:00:05] AGENT: 我很好。解析思路用正则提取角色和内容按时间戳排序。分隔符分割用特定的分隔符如\n\n、划分对话轮次再根据每轮的首行或特定标记判断角色。基于关键字的角色推断例如包含“用户说”的行是用户包含“系统回复”的行是助理。上下文关联解析对于论坛数据需要根据“回复给XX楼”来重建对话树。实操要点与避坑指南角色标准化最终必须映射到OpenAI API支持的角色通常是system,user,assistant。你需要定义自己的映射规则。处理多行消息一条消息可能包含换行符。解析时要注意贪婪匹配确保将属于同一条消息的多行文本合并。清洗无用信息原始数据中的时间戳、ID、表情符号代码等可能需要移除或单独存储为元数据。编写单元测试为你的解析器编写测试用例至关重要。准备一小段典型的原始数据和一个你期望的解析结果确保解析器在各种边界情况下都能正常工作。import re class RegexLogParser: 一个基于正则表达式的简单日志解析器示例 def __init__(self): # 假设日志格式[时间] 角色(大写): 内容 self.pattern re.compile(r^\[(.*?)\] (\w): (.*)$, re.MULTILINE) def parse(self, text): messages [] matches self.pattern.findall(text) for timestamp, role_raw, content in matches: # 将角色映射为标准角色 role self._map_role(role_raw) if role: messages.append({role: role, content: content.strip()}) return messages def _map_role(self, raw_role): role_map { USER: user, AGENT: assistant, SYSTEM: system, BOT: assistant, CUSTOMER: user } return role_map.get(raw_role.upper())3.3 消息格式化器与批处理器解析出原始消息后需要根据你的目标任务进行格式化。用于微调的格式OpenAI的聊天模型微调需要特定的JSONL格式每行是一个对象包含一个messages列表。列表中的每个元素就是上面解析出来的{role: ..., content: ...}。用于批量推理的格式你可能想用同一个系统提示词搭配不同的用户输入进行批量问答。这时需要将每条数据格式化为一个独立的请求消息列表。批处理器则负责将格式化后的数据以合理的批次大小发送给OpenAI API。这里需要考虑API速率限制OpenAI对不同模型有每分钟请求数RPM和每分钟令牌数TPM的限制。错误处理与重试网络错误、速率限制错误429需要加入指数退避重试机制。并发控制使用asyncio或线程池来提高大批量处理的效率但要小心控制并发数避免触发速率限制。3.4 输出处理器处理结果需要妥善保存。保存为JSONL这是最通用的格式方便后续直接用于微调或其它分析。保存为CSV如果结果结构简单CSV便于用Excel打开查看。直接入库将API返回的结果写回数据库。生成报告统计处理成功率、失败原因、消耗的token数量等。4. 构建你自己的“ChatGPT”数据处理管道实战步骤下面我将带你从零开始构建一个简化但功能完整的数据处理脚本其思想正是源于saschaschramm/chatgpt这类项目。4.1 环境准备与依赖安装首先创建一个新的项目目录并设置虚拟环境。mkdir chatgpt-data-pipeline cd chatgpt-data-pipeline python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖。我们主要需要openai库来调用APItqdm来显示进度条aiohttp用于异步请求可选用于高级并发。pip install openai tqdm # 如果需要高级异步处理可以安装 aiohttp 和 asyncio # pip install aiohttp设置你的OpenAI API密钥。永远不要将密钥硬编码在代码中# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here或者在代码中通过环境变量读取import os from openai import OpenAI client OpenAI(api_keyos.environ.get(OPENAI_API_KEY))4.2 编写核心解析器我们假设要处理一种简单的自定义对话格式每轮对话由一行“U:”或“A:”开头。原始数据 (sample_chat.txt):U: 你好我想咨询一下产品的保修政策。 A: 您好我们的产品提供一年有限保修。请问您购买多久了呢 U: 我买了大概8个月。 A: 那还在保修期内。请问产品具体出现了什么问题目标将上述内容解析成OpenAI的消息格式。代码实现 (parser.py):import re from typing import List, Dict, Any class SimpleChatParser: 解析以 U:/A: 开头的简单对话格式 def parse_file(self, file_path: str) - List[List[Dict[str, str]]]: 解析文件返回一个列表每个元素代表一段对话messages列表 with open(file_path, r, encodingutf-8) as f: content f.read() return self.parse_text(content) def parse_text(self, text: str) - List[List[Dict[str, str]]]: 解析文本内容 conversations [] # 首先按空行分割可能有多段对话 raw_conversations re.split(r\n\s*\n, text.strip()) for raw_conv in raw_conversations: messages [] lines raw_conv.strip().split(\n) for line in lines: line line.strip() if not line: continue # 匹配 U: 或 A: 开头的行 if line.startswith(U:): role user content line[2:].strip() elif line.startswith(A:): role assistant content line[2:].strip() else: # 如果不是以U:/A:开头可以视为上一行内容的延续处理多行消息 if messages: messages[-1][content] \n line continue messages.append({role: role, content: content}) if messages: # 避免添加空对话 # 可选为对话添加一个系统提示词 # system_message {role: system, content: 你是一个专业的客服助手。} # conversations.append([system_message] messages) conversations.append(messages) return conversations # 测试一下 if __name__ __main__: parser SimpleChatParser() convs parser.parse_file(sample_chat.txt) for idx, msg_list in enumerate(convs): print(f对话 {idx 1}:) for msg in msg_list: print(f {msg[role]}: {msg[content]}) print(- * 20)4.3 构建格式化与API调用模块接下来我们编写模块来处理解析后的对话并调用OpenAI的API。这里我们实现两个功能1) 将对话格式化为微调用的JSONL2) 使用最新的模型进行批量补全例如为每段对话生成一个总结。代码实现 (processor.py):import json import os from typing import List, Dict, Any, Iterator from openai import OpenAI from tqdm import tqdm import time class OpenAIChatProcessor: def __init__(self, api_key: str None, model: str gpt-3.5-turbo): self.client OpenAI(api_keyapi_key or os.environ.get(OPENAI_API_KEY)) self.model model def save_for_finetuning(self, conversations: List[List[Dict]], output_path: str): 将对话保存为微调所需的JSONL格式 with open(output_path, w, encodingutf-8) as f: for messages in conversations: # 每条对话就是一行JSON record {messages: messages} f.write(json.dumps(record, ensure_asciiFalse) \n) print(f已保存 {len(conversations)} 段对话到 {output_path}) def batch_complete(self, conversations: List[List[Dict]], system_prompt: str None, max_tokens: int 500, temperature: float 0.7) - List[str]: 批量调用Chat Completion API为每段对话生成一个助理回复例如总结 results [] for messages in tqdm(conversations, desc调用API中): # 为本次请求构建消息列表 request_messages [] if system_prompt: request_messages.append({role: system, content: system_prompt}) # 将历史对话作为上下文传入 request_messages.extend(messages) # 最后附加一个user消息指示模型进行总结或其他任务 request_messages.append({role: user, content: 请总结一下这段对话的核心内容。}) try: response self.client.chat.completions.create( modelself.model, messagesrequest_messages, max_tokensmax_tokens, temperaturetemperature ) summary response.choices[0].message.content results.append(summary) except Exception as e: print(fAPI调用出错: {e}) results.append(None) # 简单的延迟避免触发RPM限制根据你的套餐调整 time.sleep(0.5) return results4.4 组装主程序并运行最后我们创建一个主脚本来串联整个流程。代码实现 (main.py):import argparse from parser import SimpleChatParser from processor import OpenAIChatProcessor def main(): parser argparse.ArgumentParser(descriptionChatGPT 对话数据处理管道) parser.add_argument(input_file, help输入的对话文本文件路径) parser.add_argument(--mode, choices[format, summarize], defaultformat, help处理模式format-格式化为JSONLsummarize-调用API进行总结) parser.add_argument(--output, defaultoutput.jsonl, help输出文件路径) parser.add_argument(--model, defaultgpt-3.5-turbo, help使用的OpenAI模型) args parser.parse_args() # 1. 解析对话 print(f正在解析文件: {args.input_file}) chat_parser SimpleChatParser() conversations chat_parser.parse_file(args.input_file) print(f解析完成共发现 {len(conversations)} 段对话。) # 2. 初始化处理器 processor OpenAIChatProcessor(modelargs.model) # 3. 根据模式处理 if args.mode format: # 格式化为JSONL用于微调 processor.save_for_finetuning(conversations, args.output) print(f数据已格式化为 {args.output}可用于微调。) elif args.mode summarize: # 调用API进行批量总结 print(f开始使用 {args.model} 模型进行批量总结...) summaries processor.batch_complete( conversations, system_prompt你是一个擅长总结的助手。, max_tokens150 ) # 保存总结结果 with open(args.output, w, encodingutf-8) as f: for idx, (conv, summary) in enumerate(zip(conversations, summaries)): f.write(f对话 {idx1} 总结:\n) f.write(f{summary}\n) f.write(-*40 \n) print(f总结完成结果已保存到 {args.output}) # 打印一些示例 print(\n--- 示例总结 ---) for i in range(min(2, len(summaries))): if summaries[i]: print(f对话{i1}: {summaries[i][:100]}...) if __name__ __main__: main()运行示例格式化模式用于微调数据准备python main.py sample_chat.txt --mode format --output fine-tuning-data.jsonl总结模式批量调用APIpython main.py sample_chat.txt --mode summarize --output summaries.txt --model gpt-45. 高级技巧与生产环境考量当你需要处理成千上万条对话时基础脚本就需要升级了。5.1 实现健壮的异步批量处理同步请求会非常慢。使用asyncio和aiohttp可以大幅提升吞吐量。核心是创建一个信号量来控制最大并发数并实现指数退避重试。import aiohttp import asyncio from tenacity import retry, stop_after_attempt, wait_exponential class AsyncOpenAIProcessor: def __init__(self, api_key, max_concurrent10): self.api_key api_key self.semaphore asyncio.Semaphore(max_concurrent) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) async def _make_request(self, session, messages): async with self.semaphore: # 控制并发 url https://api.openai.com/v1/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: gpt-3.5-turbo, messages: messages, max_tokens: 500 } async with session.post(url, jsonpayload, headersheaders) as resp: if resp.status 429: raise Exception(Rate limit exceeded) resp.raise_for_status() result await resp.json() return result[choices][0][message][content] async def process_batch_async(self, all_messages): async with aiohttp.ClientSession() as session: tasks [self._make_request(session, msg) for msg in all_messages] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果和异常 return results5.2 数据处理的质量控制长度过滤过滤掉过短可能无意义或过长可能超出上下文窗口的对话。角色比例检查确保对话中用户和助理的轮次大致平衡避免全是单轮问答或单方面发言。敏感信息脱敏使用正则表达式或关键词列表自动检测并替换电话号码、邮箱、身份证号等个人信息。去重对高度相似例如仅有一两个词不同的对话进行去重提高微调数据质量。5.3 成本监控与优化批量调用API成本可能迅速增加。估算Token使用tiktoken库在发送请求前估算token消耗对超长内容进行截断或分割。选择合适模型根据任务难度选择模型。简单的文本格式化或分类任务gpt-3.5-turbo比gpt-4成本低得多。缓存结果对于相同的输入提示将API响应缓存到本地数据库或文件避免重复调用。6. 常见问题与排查技巧实录在实际操作中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方案问题1解析器在处理不规则数据时崩溃或输出错误。排查总是从一小段最具代表性也最混乱的数据开始测试你的解析器。使用print语句或调试器查看每一步解析的中间结果。技巧编写“防御性”代码。假设数据是脏的使用try...except包裹可能出错的解析步骤记录错误行并跳过而不是让整个程序崩溃。最终生成一个错误报告文件。问题2API调用频繁返回429速率限制错误。排查确认你的请求是否符合OpenAI的速率限制RPM和TPM。免费用户和不同付费等级的账户限制不同。解决方案降低并发减少同时发起的请求数max_concurrent。增加延迟在请求间加入随机延迟例如time.sleep(random.uniform(0.5, 1.5))。使用指数退避如上文代码所示当遇到429错误时等待一段时间如2秒再重试如果继续失败等待时间加倍。监控TPM如果你的提示词或补全结果很长可能先触发了TPM限制。需要估算token并控制批次大小。问题3处理大量文件时内存占用过高程序变慢甚至崩溃。排查检查是否一次性将所有数据读入内存如json.load()整个文件。解决方案流式处理使用迭代器逐行或逐块读取和处理文件如for line in open(big.jsonl)。分批次处理将大任务分成多个小批次处理完一批就写入磁盘并释放内存。使用数据库对于超大规模数据考虑使用SQLite或PostgreSQL作为中间存储进行分页查询和处理。问题4生成的JSONL文件在OpenAI微调工具中验证失败。排查使用OpenAI提供的官方数据验证工具或简单的格式检查。常见原因消息角色错误角色只能是system,user,assistant。检查你的映射是否正确。消息顺序错误通常应以system(可选) -user-assistant-user... 的顺序交替。避免连续两个相同角色的消息某些情况允许但最好遵循交替模式。内容为空某条消息的content字段是空字符串。JSON格式错误确保每行都是有效的JSON没有尾随逗号字符串引号正确。问题5批量调用API时部分请求失败导致数据缺失。解决方案实现一个可靠的错误处理和工作队列机制。将所有待处理的任务放入一个队列。worker从队列中取出任务执行。如果成功标记任务完成并保存结果。如果失败非永久性错误如429将任务重新放回队列末尾等待重试。记录所有失败的任务ID和原因最后生成一个“待处理任务”清单方便手动复查或重跑。这个基于saschaschramm/chatgpt思想扩展而来的数据处理管道其价值在于它将一个模糊的需求——“处理聊天数据”——变成了一个可自动化、可扩展、可监控的工程流程。它可能没有华丽的界面但却是构建可靠AI应用背后不可或缺的基石。当你下次再看到类似的“不起眼”的工具仓库时不妨深入看看它的设计思路很可能就能解决你手头那个棘手的、重复性的数据处理难题。
从混乱日志到结构化提示词：基于OpenAI API的数据处理管道实战
1. 项目概述一个被误解的“ChatGPT”仓库在GitHub上搜索“ChatGPT”你会得到成千上万个结果其中有一个仓库名为saschaschramm/chatgpt。乍一看你可能会以为这是一个官方客户端、一个逆向工程接口或者一个功能强大的封装库。但点进去之后你大概率会感到一丝困惑甚至有些失望。因为这个仓库的内容与我们所熟知的、那个能进行深度对话的AI模型似乎关联不大。它更像是一个特定场景下的、高度定制化的工具脚本集合。这个项目本质上是一个基于OpenAI API的、用于自动化处理特定格式文本特别是对话记录的命令行工具。它的核心价值不在于提供一个炫酷的聊天界面而在于解决一个非常具体且常见的工程问题如何将结构混乱或特定格式的原始对话文本比如从某些平台导出的日志批量、自动地转换成符合OpenAI Chat Completion API要求的结构化格式例如JSONL并进行后续的模型微调或批量对话任务。我最初接触这个项目是因为需要处理大量来自不同渠道的客服对话记录用于训练一个垂直领域的助手。手动清洗和格式化这些数据是噩梦级的任务。saschaschramm/chatgpt虽然没有提供开箱即用的聊天机器人但它提供了一套清晰的思路和可复用的代码框架教会了我如何用脚本将“脏数据”管道化地处理成AI的“营养餐”。对于需要处理非标准对话数据、进行批量API调用或数据预处理的开发者来说这个仓库的价值被严重低估了。它更像是一把瑞士军刀中的专业工具而非那把主刀。2. 核心需求与设计思路拆解2.1 真实需求从混乱日志到结构化提示词在AI应用开发中尤其是涉及微调Fine-tuning或构建复杂对话系统时我们面临的原始数据往往是五花八门的。它们可能来自客服系统导出通常是CSV或TXT包含用户ID、时间戳、对话内容但不同客服的发言可能混在一起没有清晰的“用户/助理”角色划分。社区论坛或评论数据帖子与回复的嵌套结构需要被扁平化成线性的对话轮次。自定义的聊天记录可能是某种特殊的日志格式字段分隔符不统一包含大量元信息如IP、设备类型需要剥离。多轮对话的拼接需要将单轮问答组合成符合GPT多轮对话上下文格式的数据。saschaschramm/chatgpt项目敏锐地捕捉到了这个痛点。它的设计思路不是再造一个聊天前端而是专注于数据格式的转换与管道处理。其核心设计可以概括为“输入任意格式的文本流 - 通过可配置的解析器提取对话轮次 - 映射为标准的OpenAI消息格式 - 输出为JSONL或直接调用API”。这种设计将复杂性封装在解析规则里而将通用处理流程如API调用、错误重试、结果收集标准化。开发者只需要关心如何为自己的数据源编写一个“解析器”或“适配器”剩下的批量处理、格式校验、并发请求等脏活累活都可以复用项目提供的框架。2.2 技术选型考量为什么是Python脚本项目选择了Python作为实现语言这是一个非常务实且高效的选择。生态丰富用于文本处理的re正则表达式、json、csv库是Python标准库的一部分。对于更复杂的解析pandas可以轻松集成。HTTP请求有成熟的requests库异步处理有asyncio和aiohttp。快速原型Python的脚本特性非常适合这种数据处理任务。开发者可以快速编写、测试不同的解析规则而不需要编译和复杂的项目结构。与OpenAI SDK无缝集成OpenAI官方提供了Python SDK (openai库)使得API调用变得异常简单。项目可以直接利用无需自己封装HTTP客户端。命令行友好通过argparse或click库可以轻松构建功能强大的命令行界面CLI方便集成到自动化流水线如CI/CD或由其他脚本调用。注意这个项目通常不会是一个需要长期运行的服务它的生命周期是“任务型”的准备数据时运行一次。因此轻量级的脚本形式比构建一个完整的Web服务或桌面应用更合适。3. 核心模块解析与实操要点虽然原始仓库的具体实现可能比较简单但我们可以基于其思想构建一个更健壮、更通用的处理管道。一个完整的解决方案通常包含以下核心模块3.1 输入处理器这个模块负责读取原始数据。它需要支持多种输入源本地文件读取.txt,.csv,.json,.jsonl等。目录批量处理遍历一个文件夹下的所有指定格式文件。标准输入支持管道操作例如cat chat.log | python script.py。数据库查询直接从MySQL、PostgreSQL或MongoDB中拉取对话记录。实操要点使用with open(...) as f来确保文件句柄正确关闭。对于大文件考虑流式读取逐行避免一次性加载到内存。为不同的文件格式编写对应的读取函数并通过一个统一的接口调用。import json import csv from pathlib import Path def read_file(file_path): 根据文件后缀名选择读取器 suffix Path(file_path).suffix.lower() if suffix .jsonl: return read_jsonl(file_path) elif suffix .csv: return read_csv(file_path) elif suffix .txt: return read_txt(file_path) else: raise ValueError(fUnsupported file format: {suffix}) def read_jsonl(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: yield json.loads(line.strip()) def read_csv(file_path): with open(file_path, r, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: yield row3.2 对话解析器这是整个项目的灵魂所在也是最需要定制开发的部分。解析器的任务是将一行原始文本或一个数据对象解析成一组具有“角色”和“内容”的对话消息。常见解析模式正则表达式匹配适用于有固定模式的日志。示例日志格式为[2023-10-27 10:00:00] USER: 你好吗和[2023-10-27 10:00:05] AGENT: 我很好。解析思路用正则提取角色和内容按时间戳排序。分隔符分割用特定的分隔符如\n\n、划分对话轮次再根据每轮的首行或特定标记判断角色。基于关键字的角色推断例如包含“用户说”的行是用户包含“系统回复”的行是助理。上下文关联解析对于论坛数据需要根据“回复给XX楼”来重建对话树。实操要点与避坑指南角色标准化最终必须映射到OpenAI API支持的角色通常是system,user,assistant。你需要定义自己的映射规则。处理多行消息一条消息可能包含换行符。解析时要注意贪婪匹配确保将属于同一条消息的多行文本合并。清洗无用信息原始数据中的时间戳、ID、表情符号代码等可能需要移除或单独存储为元数据。编写单元测试为你的解析器编写测试用例至关重要。准备一小段典型的原始数据和一个你期望的解析结果确保解析器在各种边界情况下都能正常工作。import re class RegexLogParser: 一个基于正则表达式的简单日志解析器示例 def __init__(self): # 假设日志格式[时间] 角色(大写): 内容 self.pattern re.compile(r^\[(.*?)\] (\w): (.*)$, re.MULTILINE) def parse(self, text): messages [] matches self.pattern.findall(text) for timestamp, role_raw, content in matches: # 将角色映射为标准角色 role self._map_role(role_raw) if role: messages.append({role: role, content: content.strip()}) return messages def _map_role(self, raw_role): role_map { USER: user, AGENT: assistant, SYSTEM: system, BOT: assistant, CUSTOMER: user } return role_map.get(raw_role.upper())3.3 消息格式化器与批处理器解析出原始消息后需要根据你的目标任务进行格式化。用于微调的格式OpenAI的聊天模型微调需要特定的JSONL格式每行是一个对象包含一个messages列表。列表中的每个元素就是上面解析出来的{role: ..., content: ...}。用于批量推理的格式你可能想用同一个系统提示词搭配不同的用户输入进行批量问答。这时需要将每条数据格式化为一个独立的请求消息列表。批处理器则负责将格式化后的数据以合理的批次大小发送给OpenAI API。这里需要考虑API速率限制OpenAI对不同模型有每分钟请求数RPM和每分钟令牌数TPM的限制。错误处理与重试网络错误、速率限制错误429需要加入指数退避重试机制。并发控制使用asyncio或线程池来提高大批量处理的效率但要小心控制并发数避免触发速率限制。3.4 输出处理器处理结果需要妥善保存。保存为JSONL这是最通用的格式方便后续直接用于微调或其它分析。保存为CSV如果结果结构简单CSV便于用Excel打开查看。直接入库将API返回的结果写回数据库。生成报告统计处理成功率、失败原因、消耗的token数量等。4. 构建你自己的“ChatGPT”数据处理管道实战步骤下面我将带你从零开始构建一个简化但功能完整的数据处理脚本其思想正是源于saschaschramm/chatgpt这类项目。4.1 环境准备与依赖安装首先创建一个新的项目目录并设置虚拟环境。mkdir chatgpt-data-pipeline cd chatgpt-data-pipeline python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖。我们主要需要openai库来调用APItqdm来显示进度条aiohttp用于异步请求可选用于高级并发。pip install openai tqdm # 如果需要高级异步处理可以安装 aiohttp 和 asyncio # pip install aiohttp设置你的OpenAI API密钥。永远不要将密钥硬编码在代码中# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here或者在代码中通过环境变量读取import os from openai import OpenAI client OpenAI(api_keyos.environ.get(OPENAI_API_KEY))4.2 编写核心解析器我们假设要处理一种简单的自定义对话格式每轮对话由一行“U:”或“A:”开头。原始数据 (sample_chat.txt):U: 你好我想咨询一下产品的保修政策。 A: 您好我们的产品提供一年有限保修。请问您购买多久了呢 U: 我买了大概8个月。 A: 那还在保修期内。请问产品具体出现了什么问题目标将上述内容解析成OpenAI的消息格式。代码实现 (parser.py):import re from typing import List, Dict, Any class SimpleChatParser: 解析以 U:/A: 开头的简单对话格式 def parse_file(self, file_path: str) - List[List[Dict[str, str]]]: 解析文件返回一个列表每个元素代表一段对话messages列表 with open(file_path, r, encodingutf-8) as f: content f.read() return self.parse_text(content) def parse_text(self, text: str) - List[List[Dict[str, str]]]: 解析文本内容 conversations [] # 首先按空行分割可能有多段对话 raw_conversations re.split(r\n\s*\n, text.strip()) for raw_conv in raw_conversations: messages [] lines raw_conv.strip().split(\n) for line in lines: line line.strip() if not line: continue # 匹配 U: 或 A: 开头的行 if line.startswith(U:): role user content line[2:].strip() elif line.startswith(A:): role assistant content line[2:].strip() else: # 如果不是以U:/A:开头可以视为上一行内容的延续处理多行消息 if messages: messages[-1][content] \n line continue messages.append({role: role, content: content}) if messages: # 避免添加空对话 # 可选为对话添加一个系统提示词 # system_message {role: system, content: 你是一个专业的客服助手。} # conversations.append([system_message] messages) conversations.append(messages) return conversations # 测试一下 if __name__ __main__: parser SimpleChatParser() convs parser.parse_file(sample_chat.txt) for idx, msg_list in enumerate(convs): print(f对话 {idx 1}:) for msg in msg_list: print(f {msg[role]}: {msg[content]}) print(- * 20)4.3 构建格式化与API调用模块接下来我们编写模块来处理解析后的对话并调用OpenAI的API。这里我们实现两个功能1) 将对话格式化为微调用的JSONL2) 使用最新的模型进行批量补全例如为每段对话生成一个总结。代码实现 (processor.py):import json import os from typing import List, Dict, Any, Iterator from openai import OpenAI from tqdm import tqdm import time class OpenAIChatProcessor: def __init__(self, api_key: str None, model: str gpt-3.5-turbo): self.client OpenAI(api_keyapi_key or os.environ.get(OPENAI_API_KEY)) self.model model def save_for_finetuning(self, conversations: List[List[Dict]], output_path: str): 将对话保存为微调所需的JSONL格式 with open(output_path, w, encodingutf-8) as f: for messages in conversations: # 每条对话就是一行JSON record {messages: messages} f.write(json.dumps(record, ensure_asciiFalse) \n) print(f已保存 {len(conversations)} 段对话到 {output_path}) def batch_complete(self, conversations: List[List[Dict]], system_prompt: str None, max_tokens: int 500, temperature: float 0.7) - List[str]: 批量调用Chat Completion API为每段对话生成一个助理回复例如总结 results [] for messages in tqdm(conversations, desc调用API中): # 为本次请求构建消息列表 request_messages [] if system_prompt: request_messages.append({role: system, content: system_prompt}) # 将历史对话作为上下文传入 request_messages.extend(messages) # 最后附加一个user消息指示模型进行总结或其他任务 request_messages.append({role: user, content: 请总结一下这段对话的核心内容。}) try: response self.client.chat.completions.create( modelself.model, messagesrequest_messages, max_tokensmax_tokens, temperaturetemperature ) summary response.choices[0].message.content results.append(summary) except Exception as e: print(fAPI调用出错: {e}) results.append(None) # 简单的延迟避免触发RPM限制根据你的套餐调整 time.sleep(0.5) return results4.4 组装主程序并运行最后我们创建一个主脚本来串联整个流程。代码实现 (main.py):import argparse from parser import SimpleChatParser from processor import OpenAIChatProcessor def main(): parser argparse.ArgumentParser(descriptionChatGPT 对话数据处理管道) parser.add_argument(input_file, help输入的对话文本文件路径) parser.add_argument(--mode, choices[format, summarize], defaultformat, help处理模式format-格式化为JSONLsummarize-调用API进行总结) parser.add_argument(--output, defaultoutput.jsonl, help输出文件路径) parser.add_argument(--model, defaultgpt-3.5-turbo, help使用的OpenAI模型) args parser.parse_args() # 1. 解析对话 print(f正在解析文件: {args.input_file}) chat_parser SimpleChatParser() conversations chat_parser.parse_file(args.input_file) print(f解析完成共发现 {len(conversations)} 段对话。) # 2. 初始化处理器 processor OpenAIChatProcessor(modelargs.model) # 3. 根据模式处理 if args.mode format: # 格式化为JSONL用于微调 processor.save_for_finetuning(conversations, args.output) print(f数据已格式化为 {args.output}可用于微调。) elif args.mode summarize: # 调用API进行批量总结 print(f开始使用 {args.model} 模型进行批量总结...) summaries processor.batch_complete( conversations, system_prompt你是一个擅长总结的助手。, max_tokens150 ) # 保存总结结果 with open(args.output, w, encodingutf-8) as f: for idx, (conv, summary) in enumerate(zip(conversations, summaries)): f.write(f对话 {idx1} 总结:\n) f.write(f{summary}\n) f.write(-*40 \n) print(f总结完成结果已保存到 {args.output}) # 打印一些示例 print(\n--- 示例总结 ---) for i in range(min(2, len(summaries))): if summaries[i]: print(f对话{i1}: {summaries[i][:100]}...) if __name__ __main__: main()运行示例格式化模式用于微调数据准备python main.py sample_chat.txt --mode format --output fine-tuning-data.jsonl总结模式批量调用APIpython main.py sample_chat.txt --mode summarize --output summaries.txt --model gpt-45. 高级技巧与生产环境考量当你需要处理成千上万条对话时基础脚本就需要升级了。5.1 实现健壮的异步批量处理同步请求会非常慢。使用asyncio和aiohttp可以大幅提升吞吐量。核心是创建一个信号量来控制最大并发数并实现指数退避重试。import aiohttp import asyncio from tenacity import retry, stop_after_attempt, wait_exponential class AsyncOpenAIProcessor: def __init__(self, api_key, max_concurrent10): self.api_key api_key self.semaphore asyncio.Semaphore(max_concurrent) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) async def _make_request(self, session, messages): async with self.semaphore: # 控制并发 url https://api.openai.com/v1/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: gpt-3.5-turbo, messages: messages, max_tokens: 500 } async with session.post(url, jsonpayload, headersheaders) as resp: if resp.status 429: raise Exception(Rate limit exceeded) resp.raise_for_status() result await resp.json() return result[choices][0][message][content] async def process_batch_async(self, all_messages): async with aiohttp.ClientSession() as session: tasks [self._make_request(session, msg) for msg in all_messages] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果和异常 return results5.2 数据处理的质量控制长度过滤过滤掉过短可能无意义或过长可能超出上下文窗口的对话。角色比例检查确保对话中用户和助理的轮次大致平衡避免全是单轮问答或单方面发言。敏感信息脱敏使用正则表达式或关键词列表自动检测并替换电话号码、邮箱、身份证号等个人信息。去重对高度相似例如仅有一两个词不同的对话进行去重提高微调数据质量。5.3 成本监控与优化批量调用API成本可能迅速增加。估算Token使用tiktoken库在发送请求前估算token消耗对超长内容进行截断或分割。选择合适模型根据任务难度选择模型。简单的文本格式化或分类任务gpt-3.5-turbo比gpt-4成本低得多。缓存结果对于相同的输入提示将API响应缓存到本地数据库或文件避免重复调用。6. 常见问题与排查技巧实录在实际操作中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方案问题1解析器在处理不规则数据时崩溃或输出错误。排查总是从一小段最具代表性也最混乱的数据开始测试你的解析器。使用print语句或调试器查看每一步解析的中间结果。技巧编写“防御性”代码。假设数据是脏的使用try...except包裹可能出错的解析步骤记录错误行并跳过而不是让整个程序崩溃。最终生成一个错误报告文件。问题2API调用频繁返回429速率限制错误。排查确认你的请求是否符合OpenAI的速率限制RPM和TPM。免费用户和不同付费等级的账户限制不同。解决方案降低并发减少同时发起的请求数max_concurrent。增加延迟在请求间加入随机延迟例如time.sleep(random.uniform(0.5, 1.5))。使用指数退避如上文代码所示当遇到429错误时等待一段时间如2秒再重试如果继续失败等待时间加倍。监控TPM如果你的提示词或补全结果很长可能先触发了TPM限制。需要估算token并控制批次大小。问题3处理大量文件时内存占用过高程序变慢甚至崩溃。排查检查是否一次性将所有数据读入内存如json.load()整个文件。解决方案流式处理使用迭代器逐行或逐块读取和处理文件如for line in open(big.jsonl)。分批次处理将大任务分成多个小批次处理完一批就写入磁盘并释放内存。使用数据库对于超大规模数据考虑使用SQLite或PostgreSQL作为中间存储进行分页查询和处理。问题4生成的JSONL文件在OpenAI微调工具中验证失败。排查使用OpenAI提供的官方数据验证工具或简单的格式检查。常见原因消息角色错误角色只能是system,user,assistant。检查你的映射是否正确。消息顺序错误通常应以system(可选) -user-assistant-user... 的顺序交替。避免连续两个相同角色的消息某些情况允许但最好遵循交替模式。内容为空某条消息的content字段是空字符串。JSON格式错误确保每行都是有效的JSON没有尾随逗号字符串引号正确。问题5批量调用API时部分请求失败导致数据缺失。解决方案实现一个可靠的错误处理和工作队列机制。将所有待处理的任务放入一个队列。worker从队列中取出任务执行。如果成功标记任务完成并保存结果。如果失败非永久性错误如429将任务重新放回队列末尾等待重试。记录所有失败的任务ID和原因最后生成一个“待处理任务”清单方便手动复查或重跑。这个基于saschaschramm/chatgpt思想扩展而来的数据处理管道其价值在于它将一个模糊的需求——“处理聊天数据”——变成了一个可自动化、可扩展、可监控的工程流程。它可能没有华丽的界面但却是构建可靠AI应用背后不可或缺的基石。当你下次再看到类似的“不起眼”的工具仓库时不妨深入看看它的设计思路很可能就能解决你手头那个棘手的、重复性的数据处理难题。
