✨ feat: add order-attachment LLM classifier
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
168
attachment_classifier.py
Normal file
168
attachment_classifier.py
Normal file
@@ -0,0 +1,168 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
布莱迪压力表 - 订单附件识别批处理脚本
|
||||
读取 Excel 中"新参数"字段，调用大模型判断每条订单是否携带附件，
|
||||
并对附件类型(资料/配件)做分类，结果写回 Excel。
|
||||
|
||||
用法:
|
||||
pip install openpyxl anthropic
|
||||
export ANTHROPIC_API_KEY=sk-...
|
||||
python attachment_classifier.py 输入.xlsx 输出.xlsx
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import concurrent.futures as cf
|
||||
import openpyxl
|
||||
from anthropic import Anthropic
|
||||
|
||||
# ============ 配置 ============
|
||||
MODEL = "claude-sonnet-5" # 分类任务用 Sonnet 足够，性价比高
|
||||
MAX_WORKERS = 8 # 并发线程数，按你的 API 速率上限调整
|
||||
MAX_RETRY = 3 # 单条失败重试次数
|
||||
PARAM_COL_NAME = "新参数" # 参数所在列名(合同工作表)
|
||||
CONTRACT_SHEET = None # None=自动取第一个含"新参数"列的表；也可写死表名
|
||||
|
||||
client = Anthropic() # 自动读取 ANTHROPIC_API_KEY
|
||||
|
||||
# ============ 提示词 ============
|
||||
SYSTEM_PROMPT = """你是布莱迪压力表公司的订单参数审核助手。你的任务是判断一条订单的"技术参数"文本中，除了压力表产品本体之外，是否还要求交付"附件"。
|
||||
|
||||
【背景】公司生产压力表（用于化工行业）。参数文本里大部分内容是描述产品本体的规格，这些都不是附件。
|
||||
|
||||
【什么是产品本体（不算附件）】
|
||||
表壳直径/材质、精度等级、防护等级(IP)、过压保护、弹性元件材质、接头材质、接头螺纹、安装方式、表盘材质/刻度/颜色、量程、充油、玻璃、指针、调零装置、缓冲钉/阻尼钉、结构代号(如513/531)、卡盘尺寸、以及各种打标/示值/红线标识要求。
|
||||
|
||||
【附件 = 产品本体之外，随货交付的东西，分两类】
|
||||
|
||||
一、资料类（交付给客户的技术文件）：
|
||||
合格证、检测报告、检验报告、出厂报告、材质证明/材质单/材质报告、说明书(使用/安装/维护)、标定证书/校验证书、制造记录/检验记录、图纸、含技术信息的二维码/标签。
|
||||
※ 注意：装箱单、送货单、贴箱单、发货章、包装/箱体标注等，属于物流包装指令，不算资料。
|
||||
|
||||
二、配件类（随货的独立实物零件）：
|
||||
针型阀、二/三阀组、冷凝管/冷凝圈/冷凝弯、表弯管/缓冲弯、虹吸管/存水弯、过压保护器/保护罐、散热器、焊接接头/连接接头、法兰隔膜/隔膜、活接、卡箍、支架、单独提供的垫片、转接件等，凡是随表一起出货的独立零件都算。
|
||||
※ 注意：位号牌、铭牌、标牌、号牌属于标识件，不算配件。缓冲钉/阻尼钉是打进接头的工艺，不算配件。
|
||||
|
||||
【判断步骤】
|
||||
1. 逐项扫描文本，剔除所有"产品本体"描述。
|
||||
2. 看剩下的内容里是否出现资料类或配件类的东西。
|
||||
3. 只要出现任一类，has_attachment 就是"是"；两类都没有就是"否"。
|
||||
|
||||
【输出要求】
|
||||
只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块：
|
||||
{"has_attachment":"是或否","types":["资料"或"配件"或两者],"items":["具体识别到的附件名称"],"reason":"一句话依据"}
|
||||
- 若 has_attachment 为"否"，types 和 items 都为空数组 []。
|
||||
- types 只能包含 "资料" 和/或 "配件"。"""
|
||||
|
||||
# few-shot：正例 + 陷阱负例，显著提升边界准确率
|
||||
FEWSHOT = [
|
||||
("表壳直径:100mm 表壳材质:304SS 精度等级:1.6级 接头螺纹:M20×1.5 带缓冲钉 513结构",
|
||||
'{"has_attachment":"否","types":[],"items":[],"reason":"全部为产品本体规格，缓冲钉是工艺不算配件"}'),
|
||||
("表壳直径:63mm 精度等级:2.5级 带不锈钢位号牌",
|
||||
'{"has_attachment":"否","types":[],"items":[],"reason":"位号牌为标识件，不算配件"}'),
|
||||
("表壳直径:100mm 张贴好装箱单并注明订单号 送货单盖发货章",
|
||||
'{"has_attachment":"否","types":[],"items":[],"reason":"装箱单送货单属物流指令，不算资料"}'),
|
||||
("表壳直径:100mm 提供出厂检测报告 合格证随货",
|
||||
'{"has_attachment":"是","types":["资料"],"items":["出厂检测报告","合格证"],"reason":"要求交付检测报告和合格证"}'),
|
||||
("表壳直径:100mm 针型阀材质:304SS 配316L表弯管",
|
||||
'{"has_attachment":"是","types":["配件"],"items":["针型阀","表弯管"],"reason":"随货提供针型阀和表弯管独立零件"}'),
|
||||
("出厂检验报告 仪表使用说明书 配冷凝圈 过压保护器",
|
||||
'{"has_attachment":"是","types":["资料","配件"],"items":["检验报告","使用说明书","冷凝圈","过压保护器"],"reason":"既有技术文件又有独立零件"}'),
|
||||
]
|
||||
|
||||
|
||||
def build_messages(text: str):
|
||||
msgs = []
|
||||
for u, a in FEWSHOT:
|
||||
msgs.append({"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{u}\n"""'})
|
||||
msgs.append({"role": "assistant", "content": a})
|
||||
msgs.append({"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{text}\n"""'})
|
||||
return msgs
|
||||
|
||||
|
||||
def parse_json(raw: str) -> dict:
|
||||
raw = raw.strip()
|
||||
if raw.startswith("```"):
|
||||
raw = raw.strip("`")
|
||||
if raw.startswith("json"):
|
||||
raw = raw[4:]
|
||||
s, e = raw.find("{"), raw.rfind("}")
|
||||
if s != -1 and e != -1:
|
||||
raw = raw[s:e + 1]
|
||||
return json.loads(raw)
|
||||
|
||||
|
||||
def classify(text: str) -> dict:
|
||||
if not text or not str(text).strip():
|
||||
return {"has_attachment": "否", "types": [], "items": [], "reason": "空参数"}
|
||||
last_err = None
|
||||
for _ in range(MAX_RETRY):
|
||||
try:
|
||||
resp = client.messages.create(
|
||||
model=MODEL,
|
||||
max_tokens=400,
|
||||
system=SYSTEM_PROMPT,
|
||||
messages=build_messages(str(text)),
|
||||
)
|
||||
return parse_json(resp.content[0].text)
|
||||
except Exception as ex: # noqa
|
||||
last_err = ex
|
||||
time.sleep(1.5)
|
||||
return {"has_attachment": "错误", "types": [], "items": [],
|
||||
"reason": f"调用失败: {last_err}"}
|
||||
|
||||
|
||||
def find_sheet_and_col(wb):
|
||||
for ws in wb.worksheets:
|
||||
if CONTRACT_SHEET and ws.title != CONTRACT_SHEET:
|
||||
continue
|
||||
headers = [c.value for c in ws[1]]
|
||||
if PARAM_COL_NAME in headers:
|
||||
return ws, headers.index(PARAM_COL_NAME) + 1
|
||||
raise RuntimeError(f'未找到含"{PARAM_COL_NAME}"列的工作表')
|
||||
|
||||
|
||||
def main(in_path: str, out_path: str):
|
||||
wb = openpyxl.load_workbook(in_path)
|
||||
ws, col = find_sheet_and_col(wb)
|
||||
max_row = ws.max_row
|
||||
print(f"工作表: {ws.title} 参数列: 第{col}列 数据行: {max_row - 1}")
|
||||
|
||||
# 追加结果列
|
||||
base = ws.max_column
|
||||
ws.cell(row=1, column=base + 1, value="是否携带附件")
|
||||
ws.cell(row=1, column=base + 2, value="附件类型")
|
||||
ws.cell(row=1, column=base + 3, value="识别明细")
|
||||
ws.cell(row=1, column=base + 4, value="判断依据")
|
||||
|
||||
rows = list(range(2, max_row + 1))
|
||||
results = {}
|
||||
|
||||
with cf.ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
|
||||
fut = {pool.submit(classify, ws.cell(row=r, column=col).value): r for r in rows}
|
||||
done = 0
|
||||
for f in cf.as_completed(fut):
|
||||
r = fut[f]
|
||||
results[r] = f.result()
|
||||
done += 1
|
||||
if done % 50 == 0:
|
||||
print(f" 进度 {done}/{len(rows)}")
|
||||
|
||||
for r in rows:
|
||||
d = results[r]
|
||||
ws.cell(row=r, column=base + 1, value=d.get("has_attachment", ""))
|
||||
ws.cell(row=r, column=base + 2, value="、".join(d.get("types", [])))
|
||||
ws.cell(row=r, column=base + 3, value="、".join(d.get("items", [])))
|
||||
ws.cell(row=r, column=base + 4, value=d.get("reason", ""))
|
||||
|
||||
wb.save(out_path)
|
||||
n_yes = sum(1 for d in results.values() if d.get("has_attachment") == "是")
|
||||
print(f"完成 -> {out_path} 携带附件 {n_yes}/{len(rows)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 3:
|
||||
print("用法: python attachment_classifier.py 输入.xlsx 输出.xlsx")
|
||||
sys.exit(1)
|
||||
main(sys.argv[1], sys.argv[2])
|
||||
Reference in New Issue
Block a user