大模型数据筛选、分类、生成任务，满足并发速度和处理准确要求
抽象出来，整个流程我觉得是这样的：
数据清洗与增强流程
一、目标
本阶段旨在系统提升原始数据质量，消除噪声与不一致性，并通过增强手段丰富数据信息，为模型训练与推理提供高质量数据支持，确保生成、分类、筛选等任务的准确性、鲁棒性和可靠性。
二、文本数据清洗与增强流程
原则：所有清洗操作需保持"无损"或"可逆"，即保留原始数据备份并完整记录处理步骤。
1. 编码统一与乱码处理
目标：确保文本文件编码一致性，避免乱码问题。
操作：
- 统一将文本文件(.txt/.csv/.json等)转换为UTF-8编码
- 使用Python的chardet库自动检测未知文件编码
- 示例：
- 问题：Windows导出的CSV文件可能为GBK编码
- 处理：检测到GB2312编码后，使用iconv或codecs库转换为UTF-8
- 记录：保存文件名、原始编码、转换后编码至日志
2. 空白字符标准化
目标：消除来源不同导致的空白字符差异。
操作：
- 将全角空格、制表符等替换为单个半角空格
- 移除每行首尾空白字符
- 示例：
- 处理前：" 这是一段 包含 不规则 空格的文本。 "
- 处理后："这是一段 包含 不规则 空格的文本。"
- 记录：保存处理文件列表及空白字符替换规则
3. 特殊字符处理
目标：移除干扰模型处理的无效字符。
操作：
- 定义允许字符集(中英文数字+基本标点)
- 根据后续流程需求处理(如文件路径需去除中文)
- 示例：
- 场景：包含文件路径的文本
- 处理前："数据/图片/北京风景.jpg"
- 处理后："data/images/scenery_bj.jpg"
- 记录：明确特殊字符处理策略及影响数据量
4. 文本智能分块
目标：将长文本分割为适合模型处理的片段。
操作：
- 固定长度重叠分块：滑动窗口分割(如512 tokens)，设置重叠长度(如50 tokens)
- 语义分块：利用文本结构或NLP工具在自然边界处分割
- 示例：
- 任务：总结长论文
- 策略：先按章节分割，过长章节再按段落二次分块
- 记录：保存分块策略及前后文本平均长度
5. 语义增强
目标：为数据添加上下文信息。
操作：
- 添加元信息：来源、时间戳、作者等
- 上下文扩展：附加对话背景或相关事件
- 示例：
- 增强前："今天销量增长了15%。"
- 增强后："[财经报告-2024Q1] 今天销量增长了15%。"
- 记录：保存元数据来源和增强规则
三、图像数据清洗与增强流程
1. 尺度归一化
目标：调整图像至模型所需尺寸。
操作：
- 等比例缩放：长边缩至目标尺寸(如224px)，短边等比缩放后填充灰色
- 中心裁剪：直接裁剪得到目标尺寸
- 示例：
- 处理1920x1080图像：缩至224x126后上下各填充49像素灰色
- 记录：保存原始尺寸、目标尺寸及处理参数
2. 图像质量增强
目标：提升图像视觉质量。
操作：
- 对比度/亮度调整：使用直方图均衡化
- 锐化：应用边缘增强滤波器
- 去噪：使用高斯/中值滤波
- 注意：避免过度增强产生伪影
- 示例：
- 处理暗光模糊商品图：先去噪→提高对比度→轻微锐化
- 记录：保存增强算法参数及处理前后对比图
提示词工程优化指南
一、核心目标与概念
目标：通过系统化方法设计高质量提示词，精准引导大模型输出预期结果，确保生成、分类、筛选等任务的稳定性和准确性。
二、核心原则一：编写清晰具体的指令
模糊指令会导致随机输出。清晰具体的指令能有效约束模型行为，提高结果的可预测性。
使用明确分隔符
目标：清晰界定指令、用户输入和上下文，防止混淆
操作：
- 使用三重引号（"""）、XML标签（<tag>）或节标题（### 指令 ###）分隔内容
- 选择文本中极少出现的字符组合作为分隔符
优化示例：
请完成以下两个步骤：
1. 翻译：将三重引号内的文本译为英文
2. 总结：用一句话概括译文核心观点文本：\"\"\"人工智能是当今最具变革性的技术之一，正在重塑各行各业的发展格局。\"\"\"
要求结构化输出
目标：便于程序解析和处理输出结果
操作：
- 指定JSON、XML或Markdown表格等输出格式
- 明确字段结构和内容要求
示例：
请分析以下评论的情感倾向，按指定JSON格式输出：评论：「产品质量很好，但物流速度太慢了。」输出格式：
{"sentiment": "positive/negative/neutral","confidence": 0.9,"aspects": ["质量", "物流"],"summary": "简要总结"
}
设定条件约束
目标：增强系统鲁棒性，妥善处理边界情况
操作：
- 使用"如果...那么..."逻辑引导条件判断
- 明确异常处理机制
示例：
请按以下逻辑响应用户查询：
1. 判断查询意图：- 技术支持问题 → 进入步骤二- 产品咨询 → 提供功能说明和价格范围- 无关问题 → 礼貌说明职责范围2. 技术支持问题处理：- 账号问题 → 提供密码重置指南- 功能问题 → 引导查看帮助文档用户查询：「我忘记密码了，怎么办？」
正反约束法
目标：多角度约束要求，提升输出质量
操作：
- 明确期望结果的特征
- 列出需要避免的情况
示例：
判断图片是否显示有意义的室内空间：返回true的情况：
✅ 任何功能空间（卧室、厨房等）
✅ 过渡区域（走廊、楼梯间等）
✅ 虚拟空间（3D效果图）返回false的情况：
❌ 纯俯视图/抽象图案
❌ 建筑外观/室外场景
❌ 无法辨认的图像判断标准：能否识别空间功能
不确定时默认返回false
少样本提示
目标：通过示例提高生成质量
操作：
- 提供1-3个完整示例
- 保持示例格式与预期输出一致
示例：
### 示例 ###
问题：篮子里有5个苹果和3个梨，吃掉2个苹果又放入4个梨，现在有多少水果？
推理：5苹果+3梨=8 → 吃掉2苹果剩3 → 3苹果+3梨=6 → 加4梨=7梨 → 3+7=10
答案：10问题：10元钱买笔花3元，买本子是笔的2倍，剩多少钱？
推理：笔3元 → 本子6元 → 共花9元 → 10-9=1
答案：1
### 结束示例 ###请用相同方法解答：房间有4盏灯，关1盏开2盏，现在亮几盏？
三、核心原则二：引导模型思考过程
让模型模拟推理而非直接作答，可显著提升复杂任务准确性。
逐步推理引导
目标：分解复杂问题，降低决策难度
示例：
请逐步解决以下问题：问题：长方形周长36cm，长是宽的2倍，求面积？解答格式：
1. 设宽为w，则长为2w
2. 周长公式：2×(2w+w)=6w
3. 方程：6w=36
4. 解得：w=6cm
5. 长=2w=12cm
6. 面积=12×6=72cm²最终答案：72平方厘米
自我验证机制
目标：减少信息幻觉，提高可靠性
示例：
根据以下文本回答问题。若信息不足，请回复"无法确定"。文本：「火星是太阳系第四颗行星，属于类地行星。」
问题：「火星的平均温度是多少？」
四、迭代优化流程
提示词开发是数据驱动的迭代过程：
- 构思假设：明确目标，设计初版提示词
- 测试验证：使用代表性样本测试
- 分析评估：识别错误模式（格式错误/答非所问等）
- 优化重构：针对性改进提示词
迭代示例：
v1：总结这篇文章
v2：200字内总结主要观点、论据和结论
v3：
- 用1句话概括核心论点
- 列出3个支撑证据
- 指出1个局限性
- 总字数150-200字
建议先在小规模数据集测试优化，确认效果后再扩展应用。
并发处理大模型 API 请求优化方案
一、核心目标
本阶段旨在通过并发技术高效调度大模型 API 请求，在遵守服务配额限制、保障系统稳定性的前提下，最大化处理吞吐量，满足大规模数据处理需求。
二、并发模式选择
模式分析
针对大模型 API 调用的 I/O 密集型特征，选择资源利用率高、实现复杂度适中的并发方案：
多进程 (Multiprocessing)
- 优势：完全绕过 GIL 限制，适合 CPU 密集型任务
- 劣势：创建开销大，进程间数据共享复杂
多线程 (Multithreading)
- 优势：线程切换成本低，I/O 操作时释放 GIL
- 劣势：受 GIL 限制，无法实现真正的 CPU 并行
异步 (Asynchronous)
- 优势：资源开销极小，支持高并发连接数
- 劣势：需要事件循环和协程支持
决策建议
- 异步模式：最佳选择，完美匹配 I/O 密集型场景
- 多线程：可靠备选方案，实现直观
三、实现方案
多线程实现
import requests
from concurrent.futures import ThreadPoolExecutor, as_completeddef call_model_api(item):"""执行单个 API 调用"""try:response = requests.post("https://api.example.com/v1/chat/completions",json={"messages": [{"role": "user", "content": item["prompt"]}]},headers={"Authorization": "Bearer KEY"},timeout=60)response.raise_for_status()return {"status": "success", "result": response.json(), "id": item["id"]}except Exception as e:return {"status": "error", "error": str(e), "id": item["id"]}def process_concurrently_with_threads(items, max_workers=20):"""线程池并发处理"""results = []with ThreadPoolExecutor(max_workers=max_workers) as executor:future_to_item = {executor.submit(call_model_api, item): item for item in items}for future in as_completed(future_to_item):results.append(future.result())return results
异步实现
import asyncio
import aiohttpasync def call_model_api_async(session, item, semaphore):"""异步 API 调用"""async with semaphore:try:async with session.post("https://api.example.com/v1/chat/completions",json={"messages": [{"role": "user", "content": item["prompt"]}]},headers={"Authorization": "Bearer KEY"},timeout=aiohttp.ClientTimeout(total=60)) as response:response.raise_for_status()data = await response.json()return {"status": "success", "result": data, "id": item["id"]}except Exception as e:return {"status": "error", "error": str(e), "id": item["id"]}async def process_concurrently_async(items, max_concurrent=50):"""异步并发处理主函数"""semaphore = asyncio.Semaphore(max_concurrent)async with aiohttp.ClientSession() as session:tasks = [asyncio.create_task(call_model_api_async(session, item, semaphore)) for item in items]return await asyncio.gather(*tasks, return_exceptions=False)
四、批次处理与重试
批次处理
def process_in_batches(all_items, batch_size=100, use_async=True):"""分批次并发处理"""all_results = []total_batches = (len(all_items) + batch_size - 1) // batch_sizefor batch_index in range(total_batches):start_idx = batch_index * batch_sizebatch_items = all_items[start_idx: start_idx + batch_size]print(f"Processing batch {batch_index+1}/{total_batches}")if use_async:batch_results = asyncio.run(process_concurrently_async(batch_items))else:batch_results = process_concurrently_with_threads(batch_items)all_results.extend(batch_results)if batch_index < total_batches - 1:time.sleep(60) # 遵守 API 速率限制return all_results
重试机制
async def call_model_api_with_retry(session, item, semaphore, max_retries=3):"""带重试机制的 API 调用"""for attempt in range(max_retries + 1):try:result = await call_model_api_async(session, item, semaphore)if result["status"] == "success":return resultelif attempt < max_retries:wait = 2 ** attemptawait asyncio.sleep(wait)except (aiohttp.ClientError, asyncio.TimeoutError) as e:if attempt < max_retries:wait = 2 ** attemptawait asyncio.sleep(wait)else:return {"status": "error", "error": f"重试{max_retries}次后失败: {str(e)}", "id": item["id"]}
结果处理方案
一、核心原则
- 数据完整性保障
- 实现全有或全无写入机制，杜绝数据中间状态
- 核心操作需具备事务回滚能力
- 处理可靠性
- 支持断点精确续传（自动跳过已处理项）
- 异常数据自动隔离机制，确保流程畅通
- 全链路可追溯
- 详细记录处理日志（状态、耗时、错误详情）
- 异常数据分类统计与归档管理
二、关键处理流程
- 原子写入机制
# 伪代码实现
def atomic_write(data, output_path):temp_file = create_temp_file(output_path) # 在目标目录创建临时文件try:write_data(temp_file, data) # 写入数据rename(temp_file, output_path) # 原子替换目标文件log_success(output_path) # 记录成功日志except Exception as e:delete(temp_file) # 清理临时文件log_error(e) # 记录错误详情
- 断点续传处理
# 伪代码实现
class ResumableProcessor:def __init__(self, checkpoint_file):self.progress = load_checkpoint() # 加载进度：{completed: [id1,id2], last_id: "xxx"}def process(self, items):pending_items = filter_processed(items) # 过滤已完成项for item in pending_items:try:result = call_model(item) # 调用模型处理atomic_write(result, item.id) # 原子写入结果mark_completed(item.id) # 标记为已完成save_checkpoint() # 更新进度文件except Exception as e:classify_error(e, item.id) # 错误分类统计log_error(e, item.id) # 记录错误日志
三、日志与监控
- 处理流程监控
- 每条消息处理过程完整记录
- 实时统计处理结果，便于质量检查
- 处理主循环
while 有待处理数据:读取批量数据（自动跳过已处理项）for 每条数据:try:执行模型任务 → 原子写入结果 → 更新进度catch 错误:错误分类 → 记录日志 → 异常归档每处理10条保存进度（优化IO性能）