把一段脏数据洗干净，真正需要 AI 的部分通常不到三成。去前后空格、统一日期格式、校验手机号、把 "1,299.00" 转成 1299.0，这些用 pandas 加正则就能做完，10 万行大概 3 秒，成本 0 元。同样的活全交给 AI，按次计费就是 3000 元起，还得等十几分钟。
AI 值钱的只有三类活：从一整段没人管格式的自由文本里抽出结构化字段；判断「北京市朝阳区建国路1号」和「北京朝阳建国路1号院」是不是同一个地方；把「一斤」「500g」「0.5公斤」归一成同一个数。这三件事要么写不出稳定的规则，要么写出来的规则三天就被新数据打穿。
| 清洗任务 | 用什么做 | 10 万行耗时 | 成本 |
|---|---|---|---|
| 去空格、去零宽字符 | str.strip() 加正则 | 约 2 秒 | 0 元 |
| 日期格式统一 | pandas.to_datetime | 约 4 秒 | 0 元 |
| 手机号、邮箱校验 | 正则 | 约 3 秒 | 0 元 |
| 克与千克换算 | 字典映射 | 约 2 秒 | 0 元 |
| 从商品描述抽 6 个字段 | AI API | 约 25 分钟 | 930 元起 |
| 判断两条地址是否同一地 | AI API | 约 25 分钟 | 930 元起 |
最后两行的成本不是笔误，第七节会把账一步步算清楚。
我手上的 12 万行商品表，规则清洗直接砍掉 71%：去重带走 3.4 万行，缺字段补齐 4.1 万行，格式统一顺手清掉一批。真正需要语义理解的只剩 3.5 万行。也就是说 AI 只碰了 29% 的数据，账单按 29% 算。顺序一旦反了，先上 AI 再清洗，等于给每个多余的空格付一次 0.031 元。
判断一条清洗逻辑该归谁，有个土办法：把这条逻辑写成规则，看它能覆盖多少数据。覆盖到九成五以上，它就是规则活，写死就完了；只能覆盖六七成，剩下三成还得人工兜，那它就是 AI 活。这条判断反过来也成立——某条 AI 抽取如果在抽检里准确率怎么都上不去，先别急着换模型，很可能这一步本来就该用规则做。
用 Excel 另存为 CSV，文件开头会多三个字节 EF BB BF。直接 pd.read_csv("raw.csv") 大概率抛：
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xbf in position 0: invalid start byte修法是把编码写成 utf-8-sig，它会先吃掉 BOM 再解析：
import pandas as pd
df = pd.read_csv(
"raw.csv",
encoding="utf-8-sig", # 吃掉 BOM，不是 utf-8
dtype=str, # 全部按字符串读，别让 pandas 猜类型
keep_default_na=False, # 空单元格给 ""，不给 float 的 nan
)如果报的是 'utf-8' codec can't decode byte 0xb5 in position 0，那文件是 GBK 存出来的，换 encoding="gb18030"。gb18030 是 gbk 的超集，用它比用 gbk 少踩生僻字。
pandas.errors.ParserError: Error tokenizing data. C error: Expected 4 fields in line 1572, saw 7原因基本两种：某个字段里有英文逗号，或者被引号包起来的字段里带换行。先用 on_bad_lines="warn" 把坏行翻出来：
df = pd.read_csv("raw.csv", encoding="utf-8-sig", dtype=str,
engine="python", on_bad_lines="warn")engine="python" 慢一些，但报错会说清是哪一行。定位完把这几行单独挑出来修，再换回默认引擎全量跑。
"￥1,299.00" 送进模型，抽出来的 净含量数值 可能是 1299.00，也可能是 1，模型看见逗号容易当成两个数。全角数字 １２３ 同理。这两类必须规则先过一遍，不给 AI 机会。
def to_halfwidth(s):
out = []
for ch in str(s):
c = ord(ch)
if c == 0x3000:
out.append(" ")
elif 0xFF01 <= c <= 0xFF5E:
out.append(chr(c - 0xFEE0))
else:
out.append(ch)
return "".join(out)读进来之后还有一件事值得先做：把列名也清一遍。Excel 表头经常带空格、带换行、带全角括号，列名在两份文件里看着一模一样，代码里却取不到。花两行把列名去掉首尾空白再统一成半角，能省掉后面大量找不到列名的排查时间。
还有个坑是分隔符。有些系统导出的文件用分号或者制表符分隔，pandas 自动嗅探偶尔会猜错，把整行塞进第一列，你会看到某一列里全是长字符串。读进来先打印前两行和列名看一眼，比出了问题再回头找快得多。
这个坑不盯代码发现不了。读 CSV 时没关掉默认空值识别，或者 fillna("") 之后又做了一次 astype(str)，一个空单元格就变成字符串 "nan"，一路写进 JSON，模型照着 "nan" 抽，最后你的 净含量数值 字段里全是 nan。
import re
RE_ZW = re.compile(r"[\u200b-\u200f\ufeff\u2060]")
NULLISH = {"nan", "NaN", "None", "NULL", "null", "无", "-", "--"}
def clean_text(s):
if s is None:
return ""
s = to_halfwidth(s)
s = RE_ZW.sub("", s) # 零宽字符，肉眼看不见
return re.sub(r"[ \t\u3000]+", " ", s).strip()
df = df.astype(str).map(clean_text) # pandas 2.1+；旧版写成 applymap(clean_text)
df = df.mask(df.isin(NULLISH)) # 统一成真正的空值df.mask(条件) 把条件为真的单元格换成 NaN，写进 JSON 时就是 null，不是字符串 "nan"。这一步放在所有规则之前，后面每一条判断都少一层意外。
df["下单时间"] = pd.to_datetime(df["下单时间"], errors="coerce", format="mixed")
df["下单时间"] = df["下单时间"].dt.strftime("%Y-%m-%d %H:%M:%S")errors="coerce" 把转不了的值变成 NaT，不会让整列崩掉。format="mixed" 是 pandas 2.0 之后的写法，让它逐行猜格式。转完立刻看一眼有多少 NaT，超过 5% 说明日期列里混了中文，比如「2025年3月1日」，得先做一次 str.replace("年", "-") 这类替换再转。
时区也要留意。带时区的字符串在转成本地时间之前先统一到同一个时区，否则同一批订单里跨天的记录会被分到两天去，按天出报表时数量对不上，还很难查出是哪几条的问题。
PHONE = re.compile(r"^(?:\+?86)?1[3-9]\d{9}$")
EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[A-Za-z]{2,}$")
bad_phone = df[~df["联系电话"].fillna("").str.match(PHONE)]
bad_phone.to_csv("bad_phone.csv", index=False, encoding="utf-8-sig")判不过的行单独导成一份人工看。别让这些行进 AI，模型会「帮你猜」一个格式完全合法的手机号出来，这比留空危险得多，你后面根本分不清哪些是真数据。
规则这一段写得越厚，后面的账单越薄。我的习惯是先写一个小报告函数，把每一列的空值比例、重复比例、格式不合规比例打印出来，看着这份报告决定清洗顺序。数据里哪一列最脏一眼就能看出来，不用凭感觉猜，也不会漏掉某一列。
调接口时加上这一行：
response_format={"type": "json_object"}它保证返回内容能被 json.loads 解析成一个对象。但它不管你字段名叫什么、值在不在枚举里、有没有少字段。字段层面的约束只能写在 prompt 里，这一点别抱幻想。
枚举是这个环节最关键的设计。写成自由文本，模型今天给你「食品」，明天给你「食品类」，后天给你「快消食品」，你的下游聚合全部报废。
SYSTEM = '''你是商品信息抽取器，只输出 JSON，不写解释。
返回 {"items": [...]}，数组长度必须等于输入行数，顺序一一对应。
每个元素只含这 6 个字段：
- 品类：只能取 食品/饮料/日化/母婴/数码/家居/其他 之一
- 品牌：原文出现过的品牌名，没有就 unknown
- 规格：包装规格原文，例如 "6瓶装"
- 净含量数值：纯数字，不带单位，例如 500
- 单位：只能取 g/kg/ml/L/片/粒/包 之一
- 适用人群：只能取 成人/儿童/婴幼儿/老人/通用 之一
只取单件净含量，包装数量不算。找不到的字段一律填 unknown，禁止推测。'''「禁止推测」这四个字必须留着。不加这句，模型会给每条数据都编一个品牌名，而且编得很像真的，你在抽检之前完全看不出来。
import os, json
from openai import OpenAI
client = OpenAI(api_key=os.environ["XYU_API_KEY"], base_url="https://xyuapi.top/v1")
def extract(desc: str) -> dict:
r = client.chat.completions.create(
model="gemini-2.5-pro",
temperature=0,
response_format={"type": "json_object"},
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": desc},
],
)
return json.loads(r.choices[0].message.content)["items"][0]小鱼API（xyuai.cc）是全系列 AI 大模型 API 接入平台，主入口 https://xyuapi.top/v1，备用入口 https://xyuai.cc/v1，把 base_url 换过去就行，代码一行不用改。协议是 OpenAI 兼容的，/v1/chat/completions 和 /v1/models 都通，所以直接用 openai 的 Python SDK 即可。想走 Gemini 原生协议，把路径换成 /v1beta。想先在客户端里手动试 prompt 效果，Chatbox、Cherry Studio、NextChat、LobeChat、Cline、RooCode、KiloCode 这些支持自定义 OpenAI 地址的客户端都能填。
prompt 里还有一个容易忽略的点：把输入行的编号写进去，并要求模型按编号顺序返回数组。合并多行请求时，模型偶尔会把两条相似的描述合成一条，或者干脆漏掉一条，有编号才能立刻对齐回来，不然你根本不知道抽出来的第二条到底对应哪一行原始数据。
模型的选择也别上来就挑贵的。固定 schema 抽取是个结构化任务，不需要长链推理，gemini-2.5-pro 这一类完全够用。真正需要推理模型的只有地址去重那种要「讲道理」的判断，量不到整体的十分之一。把贵模型留给必要的十分之一，账单差别很实在。
温度参数记得设成 0。抽取是确定性任务，不需要任何创造性，温度调高只会让同一行数据跑两次得到两个答案，你在抽检时更难判断到底哪个对。
response_format 也不是万无一失。部分模型在 json_object 模式下会先输出一句「好的，我来抽取：」，再给 JSON。这时 json.loads 直接抛 JSONDecodeError: Expecting value: line 1 column 1 (char 0)，整批数据全废。
RE_JSON = re.compile(r"\{[\s\S]*\}")
FENCE = chr(96) * 3 # 不要在图省事直接在源码里写三个反引号，会撑破外层代码围栏
def load_json_loose(text):
text = (text or "").strip()
if text.startswith(FENCE): # 顺手拆掉 markdown 围栏
text = re.sub("^" + FENCE + r"[a-zA-Z]*\s*|\s*" + FENCE + "$", "", text).strip()
try:
return json.loads(text)
except json.JSONDecodeError:
m = RE_JSON.search(text) # 贪婪：从第一个 { 到最后一个 }
if not m:
raise
return json.loads(m.group(0))\{[\s\S]*\} 是贪婪匹配。如果模型在 JSON 后面又补了一句带花括号的说明，它会把整段一起括进去导致解析失败；这种情况把 * 改成 *? 用非贪婪，代价是遇到嵌套的对象会截断。两种都试一遍、哪个成功用哪个，是最省事的写法。
| 输入样例 | json.loads | ast.literal_eval |
|---|---|---|
{"a": 1} | 通过 | 通过 |
单引号 {'a': 1} | 报错 | 通过 |
{"a": true} | 通过 | 抛 ValueError |
{"a": True} | 报错 | 通过 |
{"a": null} | 通过 | 抛 ValueError |
尾随逗号 {"a":1,} | 报错 | 报错 |
表达式 1+1 | 报错 | 报错 |
两个解析器都不认尾随逗号，别指望它们互相兜底。真正的坑在第三行和第五行：ast.literal_eval('{"a": true}') 抛的是 ValueError: malformed node or string，不是 JSONDecodeError。如果你的兜底只捕获了 json.JSONDecodeError，这个异常会直接穿出去，把整个线程的批次打断。捕获写全：
try:
data = load_json_loose(text)
except (json.JSONDecodeError, ValueError) as e:
data = {"_error": "%s: %s" % (type(e).__name__, e)}一是模型返回了空的 content，被安全策略拦了，或者流式返回没读完整；二是没加 response_format，模型回了一整段自然语言；三是拿到的是 Python 风格的 {'品类': '食品'}，单引号开头。前两种改参数，第三种改用 ast.literal_eval 试一次。对号入座，别上来就改 prompt。
兜底的分寸也要拿捏。解析失败的第一反应是重试一次，模型偶发的格式抖动重试一次基本就好了；连续两次还失败，就别再花次数了，直接把这一行写进失败文件留给人工处理。无限重试是贵的错误处理方式，按次计费下每一次重试都是真金白银。
还有一个细节：把模型返回的原始文本一并存下来，和解析后的结果分两列放。抽检时发现某一行不对，能立刻回看当时模型到底说了什么，是靠猜还是真读懂了。省这一列的空间，排查问题时要多花一倍时间，而且原始文本丢了你连它错在哪都不知道。
| 原写法 | 归一成克 | 说明 |
|---|---|---|
| 500g 或 500 克 | 500 | 直接取数 |
| 0.5kg 或 0.5 公斤 | 500 | 乘 1000 |
| 一斤 或 1 斤 | 500 | 市斤等于 500 克 |
| 半斤 | 250 | 0.5 斤 |
| 1 两 | 50 | 市两等于 50 克 |
| 2 磅 或 2lb | 907.18 | 1 磅约 453.59 克 |
| 1oz 或 1 盎司 | 28.35 | 常衡盎司 |
液体的体积不要默认按 1 比 1 折算成重量。500ml 水是 500 克，500ml 食用油只有 460 克左右，密度不一样。要让 AI 帮你判体积还是重量，就在 prompt 里补一句「遇到 ml、L 这类体积单位时照原样填，不要换算成克」，把折算权收回自己手里。
「一斤」「半斤」「两斤半」「3 斤 2 两」这类表达，正则能处理绝大多数：
UNIT_TO_G = {"g": 1, "克": 1, "kg": 1000, "公斤": 1000, "千克": 1000,
"斤": 500, "两": 50, "磅": 453.59, "lb": 453.59, "oz": 28.35}
RE_NUM = re.compile(r"(\d+(?:\.\d+)?)")
def normalize_weight(value, unit):
m = RE_NUM.search(str(value or ""))
if not m:
return None
return round(float(m.group(1)) * UNIT_TO_G.get(str(unit or "").strip().lower(), 1), 2)「半斤」这种不含阿拉伯数字的，模型抽 净含量数值 时会填 0.5、单位 填斤，上面这个函数算出来正好 250。规则和 AI 在这里是接力关系，不是二选一。凡是能用字典映射收口的换算，都别放进 prompt 让模型现场做算术，模型算 0.5 乘 1000 偶尔会给你 5000。
「北京市朝阳区建国路1号」和「北京朝阳建国路1号院」，用 difflib.SequenceMatcher 算相似度大概 0.68，阈值卡 0.85 就漏判了。「朝阳」和「朝阳区」指的是同一个区，「1号」和「1号院」在绝大多数业务场景里指同一块地。这件事相似度算法不知道，模型知道。给两两组合加一句「忽略行政区划后缀差异和门牌号书写差异」，同判率能上去一大截。
反例同样得交给它：「建国路1号」和「建国路1号院2号楼」可能是同一个园区的不同楼，地址只差一个后缀但确实不是一处。合不合并是业务决定，模型的价值在于告诉你「这两个很近但不确定」，比相似度甩给你一个 0.8 的数字有用得多。
PROMPT_DEDUP = '''判断下面每组地址是否指向同一地点。
返回 {"pairs": [{"i": 0, "same": true, "reason": "一句话理由"}]}
忽略省市区县后缀与院、号的书写差异；地址里的楼栋、楼层不同视为不同地点；不确定一律填 false。'''按 50 组一批送进去，一次请求判 100 条地址的两两关系，比逐对调用省下 99% 的次数。
「北京字节跳动科技有限公司」和「字节跳动」是同一家；「北京字节跳动科技有限公司」和「北京字节跳动网络技术有限公司」是两家不同法人。相似度在这两组上给出几乎一样的分，模型加一句「先判断是不是同一法人主体，简称与全称视为同一主体，名称后缀不同视为不同主体」就能把两组分开。这是整个清洗流程里最难写规则、也最值得花钱的一段。
去重的价值经常被低估。同一家公司的地址重复出现三次，下游做地域统计时这个城市的数字就被放大了三倍，这种错误在报表上看不出任何异常，但结论全是错的。花钱让 AI 判一遍地址和公司名，比事后发现统计错了再重跑一遍便宜得多。
单位归一化做完之后，建议再加一道范围校验：单件重量落在 1 克到 50 千克之外的行全部标红导出。这条规则能抓住绝大多数抽取错误，比如模型把「6瓶装」算进去得到 3000 克，或者干脆把价格当成了含量。规则和 AI 之间加一层互相校验，比只靠抽检可靠得多。
抽字段是纯 I/O 等待，理论上并发开 32 也不占 CPU。但撞限流的概率跟着并发数线性往上涨，上游网关通常按账号或按 IP 限速。4 到 8 是稳态区间：6 并发下 3 万行大约 25 分钟跑完，再往上加，省下的时间会被重试和退避吃掉。别一上来就开 32，然后怀疑平台不稳定。
import time, random
from concurrent.futures import ThreadPoolExecutor, as_completed
def call_with_backoff(fn, *args, max_retry=5, **kwargs):
for i in range(max_retry):
try:
return fn(*args, **kwargs)
except Exception as e:
msg = str(e)
if ("429" in msg or "rate limit" in msg.lower()) and i < max_retry - 1:
time.sleep(min(2 ** i, 30) + random.random())
continue
raise
raise RuntimeError("429 重试 5 次仍然失败")
def run_batches(batches, workers=6):
out = {}
with ThreadPoolExecutor(max_workers=workers) as ex:
fut2batch = {ex.submit(call_with_backoff, worker, b): b for b in batches}
for fut in as_completed(fut2batch):
out[id(fut2batch[fut])] = fut.result()
return out两个细节决定成败。random.random() 这个抖动不能省：不加抖动，所有线程会在同一秒一起重试，然后一起再撞一次 429。min(2 ** i, 30) 是封顶，第六次等 32 秒会把整批拖死，封到 30 秒止损。响应头里如果有 Retry-After，优先按它给的秒数等，那是上游明确的排队时长，比你自己猜准。
用 as_completed 而不是按提交顺序取结果，慢的批次不会卡住后面已经完成的批次落盘。写 JSONL 时每写一行 flush() 一次，进程被 Ctrl+C 打断也不丢已完成的部分。
并发数不是固定值，要按当天上游的实际表现调。一批数据跑下来如果 429 的比例超过 5%，把并发降到 4 再跑；如果整批一次都没撞上，下次可以试着加到 8。这个数放在配置里比写在代码里好，改的时候不用重新读一遍逻辑。
另一个常见问题是超时。个别请求卡住不动，整个线程池会被一个慢请求占满，看起来像程序死了。给每次请求设一个超时上限，超时的批次重新排回队列末尾，别让它一直占着线程不放。
还有个实战经验：先拿 100 行小样本把并发和批次跑通，确认代码没问题再上全量。直接拿 3 万行试错，最容易在第一分钟就撞满限流，然后整个晚上都在等退避。
小鱼API 的按次计费，是按「你发了一次 /v1/chat/completions 请求」收固定价。gemini-2.5-pro 是 0.031 元一次，输入 200 字还是 2000 字，这一次都是 0.031 元。这条规则直接决定省钱策略：把 5 行数据合并成 1 次请求，5 行只花 1 次的钱，账单直接掉 80%。
但别一次塞 50 行。输出会变长，模型漏字段、张冠李戴的概率明显上升，返工一次就把省下的钱赔回去。5 到 10 行一批是实测比较稳的区间。另外平台同时提供按量计费与无限卡套餐，如果你的输入是几万字的长合同、长报告，先按量计费算一遍再决定用哪种，别默认按次一定划算。最低充值 7 元，支付宝或微信付款，不需要海外信用卡。
按量计费和按次计费的差别，本质上是「短文本按次划算，长文档按量划算」。判断方法很直接：拿你平均的单条文本长度乘上次数，两种算法各算一遍，看哪个数字小就用哪个。数据清洗场景里单条通常只有一两百字，按次计费的优势非常明显，这也正是合并请求能省下八成的根本原因。
10 万行商品数据，先用规则清洗砍掉 70%，剩 3 万行交 AI：
gemini-2.5-prodeepseek-v3.2-thinking：30000 × 0.049 = 1470 元这四行数字放在一起看就明白了：真正决定账单的不是选哪个模型，而是有多少行数据被送到了模型面前。规则清洗砍掉的那 70%，等于直接省下 2170 元。
日更量小的场景是另一笔账：每天 100 次要抽取，100 × 0.031 = 3.1 元一天，一个月 93 元。这个量级不值得为省 80% 去写合并逻辑，直接单行单次最省心。
| 模型 | 单价 |
|---|---|
| gemini-2.5-pro | 0.031 元/次 |
| deepseek-v3.2-thinking | 0.049 元/次 |
| deepseek-r1-thinking | 0.049 元/次 |
| deepseek-v4-flash-thinking | 0.05 元/次 |
| grok-4.1 | 0.05 元/次 |
| gemini-3-pro-preview | 0.05 元/次 |
| grok-4.5 与 grok-4.6 | 0.09 元/次 |
| claude-sonnet-4-5-thinking | 0.09 元/次 |
| deepseek-v4-pro-thinking | 0.09 元/次 |
| kimi-k2.5 与 kimi-k2.6 | 0.09 元/次 |
| gemini-3.1-pro-preview | 0.09 元/次 |
| claude-opus-4-5-thinking | 0.12 元/次 |
| claude-sonnet-4-6-thinking 与 claude-sonnet-4-7-thinking | 0.2 元/次 |
| claude-opus-4-6-thinking 与 claude-opus-4-7-thinking | 0.25 元/次 |
| gpt-5.5 与 gpt-5.5-pro-thinking | 0.2 元/次 |
| gpt-5.4-pro-thinking | 0.15 元/次 |
| gpt-5.3-pro | 0.3 元/次 |
抽字段这种任务不需要顶级模型。gemini-2.5-pro 和 deepseek-v3.2-thinking 在固定 schema 抽取上的差距很小，单价差 58%。先用便宜的跑 200 行验证，准确率够了就别换。
| 方案 | 请求次数 | gemini-2.5-pro | deepseek-v3.2-thinking | 备注 |
|---|---|---|---|---|
| 单行单次 | 30000 | 930 元 | 1470 元 | 单条失败重跑代价小 |
| 5 行合并 | 6000 | 186 元 | 294 元 | 推荐起点 |
| 10 行合并 | 3000 | 93 元 | 147 元 | 输出变长，漏字段概率上升 |
合并的账很直观：5 行一批省掉 80%，10 行一批省 90%。但 10 行那一行我标了风险，省下的钱可能不够请人复核。真正稳的做法是先用 5 行跑 200 行做抽检，整行准确率到 95% 了再考虑加到 10 行。
还有一个容易被忽略的成本项是重试。按次计费下每次重试都是一次完整计费，如果某批数据的解析失败率是 10%，你的实际账单就会在理论值上多出十个点。把失败率压下去比抠单价更值钱，这一点在合并请求之后尤其明显——一批五行里有一行格式崩了，整批都得重来。
上游补一次数据，行号就全变了，按行号续跑等于从头再来。用业务主键（商品 ID）可以，但主键缺失时用整行内容哈希最稳：
import hashlib, json
def row_key(row: dict) -> str:
raw = json.dumps(row, ensure_ascii=False, sort_keys=True)
return hashlib.md5(raw.encode("utf-8")).hexdigest()结果写成 JSONL 追加，每行带一个 _key。重跑时先把已完成文件的 _key 读进集合，只处理不在集合里的行。
def load_done(path):
done = {}
if os.path.exists(path):
with open(path, encoding="utf-8") as f:
for line in f:
try:
rec = json.loads(line)
done[rec["_key"]] = rec
except (json.JSONDecodeError, KeyError):
continue # 上次崩在写一半的那行，丢掉重跑
return doneexcept 里那个 continue 是必须的。进程在写 JSONL 时被杀，最后一行大概率是半截，不跳过它，每次重跑都会在同一个地方崩。
全量跑之前先抽 200 行。抽样要随机，别只抽前 200 行，CSV 开头的部分通常最规整，拿它算准确率会虚高。
两个指标分开算：单字段准确率等于该字段抽对的行数除以 200；整行准确率等于 6 个字段全对的行数除以 200。经验线是整行准确率 95%，低于这个数先别全量跑。3 万行 × 0.031 = 930 元花出去再返工，比重跑 200 行贵得多。
先看错误集中在哪个字段，再动手。如果集中在 净含量数值，说明模型把包装数量算进去了，比如「6瓶装 500ml」被抽成 3000，在 prompt 里补一句「只取单件净含量，包装数量不算」。如果 单位 字段大量出现枚举外的值，比如瓶、袋、罐，那是枚举不够用，模型在硬凑，把枚举补全或者加一个「其他」。
第三个动作是加 3 到 5 个 few-shot 例子，尤其要放一条根本抽不出来的，期望输出 unknown。只给正例不给反例，模型会继续硬猜。一次只改一处，改完重跑同一批 200 行看指标动没动。三处一起改，你根本不知道是哪句话起了作用。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""dirty.csv -> rules_clean.csv + out.jsonl + clean.csv
规则清洗砍掉大部分行，只把需要语义理解的行交给 AI；重跑自动跳过已完成行。
依赖: pip install pandas openai
"""
import os, re, json, time, random, hashlib
from concurrent.futures import ThreadPoolExecutor, as_completed
import pandas as pd
from openai import OpenAI
BASE_URL = "https://xyuapi.top/v1" # 备用入口: https://xyuai.cc/v1
API_KEY = os.environ["XYU_API_KEY"]
MODEL = "gemini-2.5-pro"
BATCH = 5 # 每 5 行合并成 1 次请求
WORKERS = 6 # 稳态区间 4-8
OUT_JSONL, OUT_CSV = "out.jsonl", "clean.csv"
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
SYSTEM = '''你是商品信息抽取器，只输出 JSON，不写解释。
返回 {"items": [...]}，数组长度必须等于输入行数，顺序一一对应。
每个元素只含这 6 个字段：
- 品类：只能取 食品/饮料/日化/母婴/数码/家居/其他 之一
- 品牌：原文出现过的品牌名，没有就 unknown
- 规格：包装规格原文，例如 "6瓶装"
- 净含量数值：纯数字，不带单位，例如 500
- 单位：只能取 g/kg/ml/L/片/粒/包 之一
- 适用人群：只能取 成人/儿童/婴幼儿/老人/通用 之一
只取单件净含量，包装数量不算。找不到的字段一律填 unknown，禁止推测。'''
RE_ZW = re.compile(r"[\u200b-\u200f\ufeff\u2060]")
RE_WS = re.compile(r"[ \t\u3000]+")
RE_JSON = re.compile(r"\{[\s\S]*\}")
RE_NUM = re.compile(r"(\d+(?:\.\d+)?)")
FENCE = chr(96) * 3 # 三个反引号，写成 chr(96) 免得撑破外层围栏
NULLISH = {"nan", "NaN", "None", "NULL", "null", "无", "-", "--"}
UNIT_TO_G = {"g": 1, "克": 1, "kg": 1000, "公斤": 1000, "千克": 1000,
"斤": 500, "两": 50, "磅": 453.59, "lb": 453.59, "oz": 28.35}
def to_halfwidth(s):
out = []
for ch in str(s):
c = ord(ch)
if c == 0x3000:
out.append(" ")
elif 0xFF01 <= c <= 0xFF5E:
out.append(chr(c - 0xFEE0))
else:
out.append(ch)
return "".join(out)
def clean_text(s):
if s is None:
return ""
s = RE_ZW.sub("", to_halfwidth(s))
return RE_WS.sub(" ", s).strip()
def clean_rules(df):
df = df.astype(str).map(clean_text) # pandas 2.1+；旧版用 applymap
df = df.mask(df.isin(NULLISH)) # "nan" 这类字符串统一成真空值
for col in ("价格", "净含量"):
if col in df.columns:
df[col] = df[col].str.replace(",", "", regex=False)
keep = [c for c in ("商品标题", "商品描述") if c in df.columns]
return df.drop_duplicates(subset=keep) if keep else df
def normalize_weight(value, unit):
m = RE_NUM.search(str(value or ""))
if not m:
return None
return round(float(m.group(1)) * UNIT_TO_G.get(str(unit or "").strip().lower(), 1), 2)
def row_key(row):
raw = json.dumps(row, ensure_ascii=False, sort_keys=True)
return hashlib.md5(raw.encode("utf-8")).hexdigest()
def load_done(path):
done = {}
if os.path.exists(path):
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
done[rec["_key"]] = rec
except (json.JSONDecodeError, KeyError):
continue # 崩在写一半的那行，丢掉重跑
return done
def load_json_loose(text):
text = (text or "").strip()
if text.startswith(FENCE):
text = re.sub("^" + FENCE + r"[a-zA-Z]*\s*|\s*" + FENCE + "$", "", text).strip()
try:
return json.loads(text)
except json.JSONDecodeError:
m = RE_JSON.search(text)
if not m:
raise
return json.loads(m.group(0))
def call_batch(batch):
payload = "\n".join("%d. %s" % (i + 1, r.get("商品描述") or r.get("商品标题", ""))
for i, r in enumerate(batch))
for attempt in range(6):
try:
resp = client.chat.completions.create(
model=MODEL, temperature=0,
response_format={"type": "json_object"},
messages=[{"role": "system", "content": SYSTEM},
{"role": "user", "content": payload}],
)
items = load_json_loose(resp.choices[0].message.content).get("items", [])
if len(items) != len(batch):
raise ValueError("返回 %d 条 / 输入 %d 条" % (len(items), len(batch)))
return items
except Exception as e:
if attempt == 5:
return [{"_error": "%s: %s" % (type(e).__name__, e)}] * len(batch)
wait = min(2 ** attempt, 30) + random.random() if "429" in str(e) else 1 + attempt
time.sleep(wait)
def main():
df = pd.read_csv("dirty.csv", encoding="utf-8-sig",
dtype=str, keep_default_na=False)
df = clean_rules(df)
df.to_csv("rules_clean.csv", index=False, encoding="utf-8-sig")
print("规则清洗后 %d 行" % len(df))
done = load_done(OUT_JSONL)
todo = []
for row in df.to_dict("records"):
row["_key"] = row_key(row) # 先算哈希，再挂字段
if row["_key"] not in done:
todo.append(row)
print("已处理 %d 行，本次待处理 %d 行" % (len(done), len(todo)))
batches = [todo[i:i + BATCH] for i in range(0, len(todo), BATCH)]
if batches:
with open(OUT_JSONL, "a", encoding="utf-8") as fout:
with ThreadPoolExecutor(max_workers=WORKERS) as ex:
fut2batch = {ex.submit(call_batch, b): b for b in batches}
for n, fut in enumerate(as_completed(fut2batch), 1):
batch, items = fut2batch[fut], fut.result()
for row, item in zip(batch, items):
if isinstance(item, dict):
item["_key"] = row["_key"]
item["净含量克"] = normalize_weight(
item.get("净含量数值"), item.get("单位"))
fout.write(json.dumps(item, ensure_ascii=False) + "\n")
fout.flush()
if n % 50 == 0:
print("已完成 %d/%d 批" % (n, len(batches)))
ai = load_done(OUT_JSONL)
merged = df.copy()
merged["_key"] = [row_key(r) for r in merged.to_dict("records")]
extra = pd.DataFrame(list(ai.values()))
if not extra.empty:
merged = merged.merge(extra, on="_key", how="left")
merged.to_csv(OUT_CSV, index=False, encoding="utf-8-sig")
print("写出 %s，共 %d 行" % (OUT_CSV, len(merged)))
if __name__ == "__main__":
main()跑法就三步：把源文件命名成 dirty.csv 放同目录，配好密钥，然后跑脚本。中途断了再跑一次，脚本会打印已处理多少行、本次待处理多少行，直接续上。
pip install --upgrade pandas openai
export XYU_API_KEY="你的密钥" # Windows PowerShell 用 $env:XYU_API_KEY="你的密钥"
python clean.py写回 CSV 时用 encoding="utf-8-sig"，这样 Excel 双击打开不会乱码，而你要给下游程序读的话，out.jsonl 那份 UTF-8 更稳。两个文件都留着，人看的和机器读的分开，比互相迁就省事。
抽检这一步的花费小到可以忽略。200 行按 5 行合并是 40 次请求，用 gemini-2.5-pro 一共 1.24 元，一顿早饭的钱就能提前知道这套 prompt 行不行。跳过它直接全量跑，风险全落在自己身上，而且跑完你依然不知道结果准不准。