commit 2110e6f38c5440c74fc0f16f4b54921293c1b9bb Author: Misaka_Company Date: Fri Jul 24 13:35:28 2026 +0800 ✨ feat: add order-attachment LLM classifier Co-Authored-By: Claude diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..63b9a20 --- /dev/null +++ b/.gitignore @@ -0,0 +1,41 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +*.egg +.eggs/ +build/ +dist/ + +# Virtual environments +venv/ +.venv/ +env/ +ENV/ + +# Logs & runtime output +logs/ +*.log + +# Environment / secrets +.env +.env.local +.env.*.local +config.yaml + +# Test & type caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +# Editor / OS (kept in case this folder is used standalone) +.idea/ +.vscode/ +*.swp +.DS_Store + +# Project-local agent data (uncomment if you do NOT want to track it) +.workbuddy/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..ec983c6 --- /dev/null +++ b/README.md @@ -0,0 +1,200 @@ +# 布莱迪压力表 - 订单附件识别工具 + +根据"总排号"从 SQL Server 查询"新参数"字段，调用大语言模型判断该订单是否携带 +附件，并以 JSON 输出结果。支持粗分类（资料/配件/耗材）和精分类（"大类:细分 +类目"，具体到针型阀、说明书等）两种粒度。 + +## 安装 + +```bash +pip install -r requirements.txt +``` + +`pyodbc` 需要系统已安装对应的 ODBC 驱动（如 "ODBC Driver 17 for SQL Server"）。 +若服务器上已有 SQL Server 管理工具/客户端环境，通常已包含该驱动；否则需自行 +安装 Microsoft 官方 ODBC Driver。 + +## 配置 + +编辑 `config.yaml`，填入以下三部分（模板中的 `CHANGE_ME` 必须替换）： + +- `database`：SQL Server 连接信息（含 `schema`）、表名、字段名 +- `llm`：OpenAI 兼容接口的 `base_url`、`api_key`、`model` +- `business`：并发数、日志级别、默认分类模式(`default_mode`)、日志目录(`log_dir`)、 + 是否启用"其他"兜底类目(`enable_other_category`，默认关闭) + +## 使用 + +```bash +# 单个总排号，默认粗分类 +python main.py --id 26B742 + +# 批量，逗号分隔 +python main.py --id 26B742,26B743,26B744 + +# 批量，文件输入（每行一个总排号） +python main.py --ids-file ids.txt + +# 精分类：输出"大类:细分类目"组合 +python main.py --id 26B742 --mode fine + +# 指定其他配置文件 +python main.py --id 26B742 --config other_config.yaml + +# 临时覆盖日志目录 +python main.py --id 26B742 --log-dir /tmp/debug_logs + +# 格式化输出（默认是紧凑的 JSON Lines，每行一条） +python main.py --id 26B742,26B743 --pretty + +# 允许模型使用"其他"兜底类目 +python main.py --id 26B742 --mode fine --enable-other +``` + +## 分类粒度：coarse / fine + +由 `--mode` 参数或配置文件 `business.default_mode` 指定，`--mode` 优先级更高。 + +- **coarse（默认）**：只判断三个大类——资料 / 配件 / 耗材 +- **fine**：在同一个 `types` 字段里直接给出"大类:细分类目"，例如 + "配件:针型阀""资料:说明书" + +两种模式对应的提示词都定义在 `src/prompts.py` 中，修改分类边界或细分类目 +枚举，只需要改这一个文件。 + +### 分类体系（已与业务方确认边界） + +**资料类**（12 个细分）：出厂检测检验报告、材质证明类、说明书、 +检验记录过程性、图纸类、标定校验证书、质量证明书、原产地证明、检验合格证书 +综合、其他交工文件、营业执照、型式检验报告 + +**配件类**（25 个细分）：针型阀、球阀、截止阀、旋塞阀、角阀、阀组、冷凝圈、 +冷凝管、冷凝弯、虹吸管、表弯管、缓冲管缓冲弯、法兰隔膜、隔离器、过压保护器、 +散热器散热片、铅封、转换接头、焊接接头短节短管、卡箍抱箍、紧固件、活接头、 +接线盒、电缆插头、变送器 + +**耗材类**（1 个细分）：垫片（四氟/紫铜/缠绕/密封圈等各种材质，单列为 +第三大类，不算在配件里，避免大量小垫片稀释"配件"标签的信息量） + +**其他类（可选，默认关闭）**：由 `--enable-other` 或配置文件 +`business.enable_other_category` 开启。开启后新增第四大类"其他"，只有 +唯一细分值"其他"，作为兜底——遇到确实是随货附件、但不属于以上任何一个 +具体细分类目的情况才使用。关闭时模型和格式校验都不知道这个选项的存在， +行为与不支持"其他"之前完全一致。 + +**明确不算配件**：位号牌/铭牌/标牌（标识件）；缓冲钉/阻尼钉/阻尼帽（均为 +工艺处理，不算随货实物配件） + +**明确不算资料**：装箱单/送货单/贴箱等物流包装指令；合格证（无论参数文本 +是否提到，一律忽略——既不影响"是否携带附件"的判断，也不会出现在细分 +类目里） + +## 输出格式 + +默认逐行输出 JSON（JSON Lines），便于管道处理和逐条消费： + +```json +{"zong_pai_hao": "26B742", "status": "ok", "has_attachment": true, "types": ["资料"]} +{"zong_pai_hao": "26B744", "status": "ok", "has_attachment": true, "types": ["配件:针型阀", "配件:表弯管"]} +{"zong_pai_hao": "26B999", "status": "not_found", "has_attachment": null, "types": []} +``` + +`types` 是唯一的类型字段，不再有单独的 `fine_types`： +- `coarse` 模式下，`types` 是大类列表，如 `["资料", "配件"]`。 +- `fine` 模式下，`types` 里每一项都是"大类:细分类目"，如 + `["资料:说明书", "配件:针型阀"]`，同一个订单可能同时出现多个大类、多个细分。 +- 若启用了"其他"兜底类目，对应项会体现为 `"其他:其他"`，格式与其余项保持一致。 + +### status 字段说明 + +| status | 含义 | +|---|---| +| `ok` | 识别成功 | +| `not_found` | 数据库中查不到该总排号 | +| `empty_param` | 总排号存在，但"新参数"字段为空，未调用 LLM | +| `llm_call_error` | LLM 调用失败（网络/接口错误），重试耗尽 | +| `llm_format_error` | LLM 输出内容不符合约定格式，重试耗尽 | +| `db_error` | 数据库连接/查询失败，影响整批请求 | + +`status` 非 `ok` 时，`has_attachment` 为 `null`，`types` 为空数组， +不会有猜测性的默认值混入结果。 + +## 对话日志 + +每个实际发起 LLM 调用的总排号，都会在日志目录（默认 `logs/`，可用 +`--log-dir` 或配置文件 `business.log_dir` 指定）下生成一个独立的日志文件： + +``` +logs/26B742_20260723_153012_123456.log +``` + +日志内容为纯文本，完整记录： +- 发给模型的完整对话（system prompt + few-shot 示例 + 实际用户消息） +- 模型的每一次原始回复——**包括被格式校验判定无效、触发重试的那些** +- 每次尝试的格式校验结果（通过/失败及原因） +- 最终的解析结果和 status + +同一总排号被重复处理不会覆盖旧日志（文件名带精确到微秒的时间戳），方便对比 +"调整提示词前后，同一条记录的判断有没有变化"。 + +`not_found`（查不到）和 `empty_param`（参数为空）的总排号不会生成日志文件， +因为它们本就没有与 LLM 的对话内容可记。 + +## 故障排查：模型回复为空（推理型模型特有） + +如果日志里出现"[调用结果] 已收到模型回复"但"[模型原始回复]"下面是空的（或 +`status` 为 `llm_call_error`，错误信息提到"推理耗尽了max_tokens预算"），这是 +带思维链（thinking/reasoning）能力的模型的已知行为——像 DeepSeek-V4 系列， +默认会先打一段思考草稿再给正文，而 `max_tokens` 限制的是"思考+正文"的总量。 +如果思考阶段把预算用完，正文就会被截断成空字符串。 + +本工具通过两处配置应对： +- `llm.max_tokens`：调大到能覆盖"思考+正文"的总量（默认 1024），而不是只按 + 正文那两三行的长度来设。 +- `llm.disable_thinking`：设为 `true` 时会通过 `extra_body={"thinking": + {"type": "disabled"}}` 关闭推理链——分类任务规则清晰、不需要模型思考，关闭 + 后响应更快、更省 token，也从根源上避免"思考耗尽预算"的问题。仅 DeepSeek-V4 + 系列等支持该参数的模型有效；换用其他不支持该参数的模型时改回 `false` + （多数接口对不认识的字段会直接忽略，但不保证所有厂商都是如此）。 + +## 设计说明 + +**为什么 LLM 不直接输出 JSON？** +JSON 的括号、引号、转义更容易被模型写错。约定 LLM 只输出"标签: 值"格式的 +纯文本（coarse 模式 2 行，fine 模式 3 行），程序侧用严格的正则做格式校验—— +校验通过才提取字段、组装成本工具自己定义的 JSON 结构；校验失败会触发重试 +（次数由配置文件 `business.format_retry` 控制，与 LLM 网络层重试 `llm.max_retry` +分开计数），重试仍失败则标记 `llm_format_error`，绝不把不确定的内容硬凑进 +最终结果。 + +**fine 模式的交叉自洽校验：** +第三行（细分类目）和第二行（大类）必须自洽——比如模型选了"针型阀"，第二行 +就必须包含"配件"，否则视为模型输出自相矛盾，同样触发格式校验失败重试。 +校验通过后，`parser.py` 会把每个细分类目和它所属的大类拼接成"大类:细分类目" +（如"配件:针型阀"），作为最终对外的 `types`；程序内部不再单独保留"大类列表" +和"细分类目列表"两份数据。 + +**"其他"兜底类目是怎么接入这套校验体系的：** +"其他"被当成第四个大类，且细分类目固定只有"其他"自身一个值（`其他:其他`）， +不允许模型编造具体名称——这样既给了"枚举之外的附件"一个去处，又不破坏 +"细分类目必须严格匹配枚举"这条不信任 LLM 输出的核心原则。是否启用由 +`enable_other` 参数控制（对应 `--enable-other` 或配置文件 +`business.enable_other_category`），并且同一次调用里，提示词（`prompts.py`） +和格式校验（`parser.py`）用的 `enable_other` 必须一致，否则会出现"提示词 +允许但校验拒绝"的不一致——这层一致性由 `classifier.py` 统一负责传递。 + +## 项目结构 + +``` +├── config.yaml # 配置文件 +├── main.py # 命令行入口 +├── requirements.txt +└── src/ + ├── config_loader.py # YAML 配置读取与校验 + ├── db.py # SQL Server 查询 (pyodbc) + ├── prompts.py # 提示词与细分类目枚举（唯一需要改分类边界时编辑的文件） + ├── llm_client.py # LLM 调用 (OpenAI 兼容接口) + ├── parser.py # LLM 输出格式校验与清洗 → 结构化数据 + ├── order_logger.py # 每个总排号一份的完整对话日志 + └── classifier.py # 编排：查库 -> 调LLM -> 校验解析 -> 记日志 -> 组装结果 +``` diff --git a/attachment_classifier.py b/attachment_classifier.py new file mode 100644 index 0000000..1d59906 --- /dev/null +++ b/attachment_classifier.py @@ -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]) diff --git a/classifier.py b/classifier.py new file mode 100644 index 0000000..75c3da7 --- /dev/null +++ b/classifier.py @@ -0,0 +1,199 @@ +# -*- coding: utf-8 -*- +"""编排层：串联 数据库查询 -> LLM调用 -> 格式校验解析 -> 组装最终JSON。 + +单条记录的处理逻辑（classify_param_text）: + 1. 若新参数为空/None -> 直接判定无附件，不调用LLM(省成本，也没有可判断内容)。 + 2. 调用 LLM 拿到调用结果(LLMCallResult)。 + - 若调用本身失败(网络/接口错误,重试耗尽) -> 记日志,标记 llm_call_error。 + 3. 用 parser.parse_llm_output(mode=..., enable_other=...) 做严格格式校验。 + - 校验通过 -> 组装结果,记日志,结束。 + - 校验失败(FormatError) -> 记日志,重新调用LLM重试。 + 4. 格式校验重试全部耗尽 -> 标记 status="llm_format_error"，不让脏数据进入最终结果。 + +无论上述哪一步，只要 OrderLogger 存在，每一次尝试都会被完整记录——包括发给 +模型的完整对话、模型的原始回复（如果有）、格式校验是否通过。这是本模块与 +日志模块的核心约定：不因为"这次调用失败了"就跳过记录，失败的调用恰恰最 +需要被记下来供排查。 + +批量处理(classify_batch)在此基础上先查数据库拿到 {总排号: 新参数} 映射， +对查不到的总排号直接标记 status="not_found"，不发起LLM调用（也不生成日志， +因为根本没有"新参数"可供判断，没有对话可记）。 + +enable_other 控制是否允许模型使用"其他"兜底类目（默认 False），从 +classify_batch/classify_single 一路透传到 llm_client.classify_raw（决定 +提示词要不要包含"其他"）和 parser.parse_llm_output（决定校验要不要放行 +"其他"）。本模块是唯一同时持有这两个调用点的地方，因此由本模块负责把 +同一个值传给两边，不暴露给更上层去分别设置从而导致不一致。 +""" +from __future__ import annotations + +import logging +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from typing import Any + +from db import DatabaseError, fetch_params_by_ids +from llm_client import LLMClient +from order_logger import OrderLogger +from parser import FormatError, ParsedResult, parse_llm_output + +logger = logging.getLogger(__name__) + + +def _empty_result(zong_pai_hao: str, status: str) -> dict[str, Any]: + return { + "zong_pai_hao": zong_pai_hao, + "status": status, + "has_attachment": None, + "types": [], + } + + +def classify_param_text( + llm_client: LLMClient, + param_text: str | None, + mode: str = "coarse", + format_retry: int = 2, + order_logger: OrderLogger | None = None, + enable_other: bool = False, +) -> tuple[str, bool | None, list[str]]: + """对单条"新参数"文本做分类，返回 (status, has_attachment, types)。 + + mode 决定分类粒度："coarse"只输出大类(资料/配件/耗材，启用 enable_other + 时还有"其他")；"fine"输出"大类:细分类目"（如"配件:针型阀"）的列表。 + 不再有单独的 fine_types 返回值——fine 模式下细分类目已经内嵌在 types + 的每个元素里。 + + enable_other 控制是否允许模型使用"其他"兜底那些不在既有枚举范围内的 + 附件，默认关闭；关闭时的行为与不支持"其他"之前完全一致。 + + status 取值: + "ok" - 正常识别成功 + "empty_param" - 新参数字段为空，直接判无附件，未调用LLM + "llm_call_error" - LLM 网络/接口调用失败（重试耗尽） + "llm_format_error" - LLM 返回内容格式校验多次失败（重试耗尽） + + format_retry 控制"格式校验失败后重新请求LLM"的次数，与 LLMClient 内部的 + 网络层重试是两回事：网络重试解决的是"请求没成功"，这里解决的是 + "请求成功了但模型没按格式输出"。两层重试互不影响，各自独立计数。 + + order_logger 若提供，会记录本次分类过程中的每一次LLM往返（无论成败）。 + """ + if param_text is None or not str(param_text).strip(): + if order_logger is not None: + order_logger.log_param_text(param_text) + order_logger.log_final_result("empty_param", None) + return "empty_param", False, [] + + if order_logger is not None: + order_logger.log_param_text(param_text) + + last_format_err: Exception | None = None + for attempt in range(1, format_retry + 1): + call_result = llm_client.classify_raw(param_text, mode=mode, enable_other=enable_other) + + if call_result.error is not None: + logger.error("LLM 调用失败: %s", call_result.error) + if order_logger is not None: + order_logger.log_llm_attempt(attempt, call_result, None, None) + order_logger.log_final_result("llm_call_error", None) + return "llm_call_error", None, [] + + try: + parsed = parse_llm_output(call_result.raw_response, mode=mode, enable_other=enable_other) + if order_logger is not None: + order_logger.log_llm_attempt(attempt, call_result, parsed, None) + order_logger.log_final_result("ok", parsed) + return "ok", parsed.has_attachment, parsed.types + except FormatError as e: + last_format_err = e + logger.warning( + "LLM 输出格式校验失败 (第 %d/%d 次): %s | 原始输出: %r", + attempt, format_retry, e, call_result.raw_response, + ) + if order_logger is not None: + order_logger.log_llm_attempt(attempt, call_result, None, str(e)) + + logger.error("格式校验重试 %d 次后仍失败，放弃: %s", format_retry, last_format_err) + if order_logger is not None: + order_logger.log_final_result("llm_format_error", None) + return "llm_format_error", None, [] + + +def classify_batch( + db_cfg: dict[str, Any], + llm_cfg: dict[str, Any], + zong_pai_hao_list: list[str], + max_workers: int = 8, + mode: str = "coarse", + log_dir: str | Path | None = None, + enable_other: bool = False, +) -> list[dict[str, Any]]: + """批量分类入口：查库 -> 并发调用LLM -> 组装结果列表。 + + log_dir 若提供，会为每个实际发起LLM调用的总排号在该目录下生成一个日志 + 文件；查库失败(not_found)或参数为空(empty_param)的总排号不生成日志文件， + 因为它们本就没有与LLM的对话内容可记。 + + 返回结果顺序与输入 zong_pai_hao_list 一致，即使某些查询失败或格式错误也会 + 补全为对应 status 的占位结果，保证"输入N个总排号，输出N条结果"。 + """ + # 去重但保留顺序，避免用户传入重复总排号导致重复查询/调用 + seen: set[str] = set() + unique_ids: list[str] = [] + for zph in zong_pai_hao_list: + if zph not in seen: + seen.add(zph) + unique_ids.append(zph) + + try: + param_map = fetch_params_by_ids(db_cfg, unique_ids) + except DatabaseError as e: + logger.error("数据库查询失败，本批次全部标记为 db_error: %s", e) + return [_empty_result(zph, "db_error") for zph in zong_pai_hao_list] + + llm_client = LLMClient(llm_cfg) + results_by_id: dict[str, dict[str, Any]] = {} + + ids_to_classify = [zph for zph in unique_ids if param_map.get(zph) is not None] + for zph in unique_ids: + if param_map.get(zph) is None: + results_by_id[zph] = _empty_result(zph, "not_found") + + def _worker(zph: str) -> tuple[str, dict[str, Any]]: + order_logger = OrderLogger(log_dir, zph, mode) if log_dir is not None else None + status, has_attachment, types = classify_param_text( + llm_client, param_map[zph], mode=mode, order_logger=order_logger, + enable_other=enable_other, + ) + return zph, { + "zong_pai_hao": zph, + "status": status, + "has_attachment": has_attachment, + "types": types, + } + + if ids_to_classify: + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = [pool.submit(_worker, zph) for zph in ids_to_classify] + for fut in as_completed(futures): + zph, result = fut.result() + results_by_id[zph] = result + + # 按原始输入顺序（含重复项）展开最终结果 + return [results_by_id[zph] for zph in zong_pai_hao_list] + + +def classify_single( + db_cfg: dict[str, Any], + llm_cfg: dict[str, Any], + zong_pai_hao: str, + mode: str = "coarse", + log_dir: str | Path | None = None, + enable_other: bool = False, +) -> dict[str, Any]: + """单个总排号分类的便捷封装。""" + return classify_batch( + db_cfg, llm_cfg, [zong_pai_hao], max_workers=1, mode=mode, log_dir=log_dir, + enable_other=enable_other, + )[0] diff --git a/config_loader.py b/config_loader.py new file mode 100644 index 0000000..daf8500 --- /dev/null +++ b/config_loader.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +"""配置文件加载与校验。""" +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any + +import yaml + +_REQUIRED_KEYS = { + "database": [ + "driver", "server", "port", "database", "schema", "username", "password", + "table", "id_column", "param_column", "connect_timeout", "query_timeout", + ], + "llm": [ + "base_url", "api_key", "model", "temperature", "max_tokens", + "disable_thinking", "timeout", "max_retry", "retry_backoff_seconds", + ], + "business": ["max_workers", "log_level", "default_mode", "log_dir", "format_retry"], +} +_VALID_MODES = {"coarse", "fine"} + + +class ConfigError(Exception): + """配置文件缺失或格式错误。""" + + +def load_config(path: str | Path = "config.yaml") -> dict[str, Any]: + """读取 YAML 配置文件并做基础字段校验。 + + 校验失败会抛出 ConfigError 并说明缺哪个字段，避免程序运行到一半才因为 + 配置缺失而报出难以理解的异常。 + """ + p = Path(path) + if not p.exists(): + raise ConfigError(f"配置文件不存在: {p.resolve()}") + + with p.open("r", encoding="utf-8") as f: + try: + cfg = yaml.safe_load(f) + except yaml.YAMLError as e: + raise ConfigError(f"配置文件 YAML 解析失败: {e}") from e + + if not isinstance(cfg, dict): + raise ConfigError("配置文件顶层必须是一个字典 (database/llm/business)") + + missing_sections = [s for s in _REQUIRED_KEYS if s not in cfg] + if missing_sections: + raise ConfigError(f"配置文件缺少顶层配置块: {missing_sections}") + + for section, keys in _REQUIRED_KEYS.items(): + section_cfg = cfg[section] + if not isinstance(section_cfg, dict): + raise ConfigError(f"配置块 '{section}' 必须是字典") + missing = [k for k in keys if k not in section_cfg] + if missing: + raise ConfigError(f"配置块 '{section}' 缺少字段: {missing}") + + default_mode = cfg["business"]["default_mode"] + if default_mode not in _VALID_MODES: + raise ConfigError( + f"business.default_mode 必须是 {_VALID_MODES} 之一，实际为: {default_mode!r}" + ) + + # enable_other_category 是后加的可选字段，不放进 _REQUIRED_KEYS——避免 + # 旧配置文件因为没有这个字段就直接报"缺字段"。不写就按 False（关闭）处理； + # 一旦写了就必须是布尔值，防止 "false"(字符串) 这类误写被当成真值静默生效。 + if "enable_other_category" in cfg["business"]: + other_flag = cfg["business"]["enable_other_category"] + if not isinstance(other_flag, bool): + raise ConfigError( + f"business.enable_other_category 必须是布尔值 true/false，实际为: {other_flag!r}" + ) + + # 简单的占位符提醒，防止用户忘记改配置直接运行 + placeholders = [] + if cfg["database"]["password"] == "CHANGE_ME": + placeholders.append("database.password") + if cfg["llm"]["api_key"] == "CHANGE_ME": + placeholders.append("llm.api_key") + if placeholders: + print( + f"[警告] 以下配置项仍是占位符，请修改 config.yaml: {placeholders}", + file=sys.stderr, + ) + + return cfg diff --git a/db.py b/db.py new file mode 100644 index 0000000..6fe44be --- /dev/null +++ b/db.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +"""SQL Server 数据访问层，通过 pyodbc 按总排号批量查询新参数字段。""" +from __future__ import annotations + +import logging +from typing import Any + +import pyodbc + +logger = logging.getLogger(__name__) + + +class DatabaseError(Exception): + """数据库连接或查询失败。""" + + +def _build_conn_str(db_cfg: dict[str, Any]) -> str: + parts = [ + f"DRIVER={{{db_cfg['driver']}}};", + f"SERVER={db_cfg['server']},{db_cfg['port']};", + f"DATABASE={db_cfg['database']};", + f"UID={db_cfg['username']};", + f"PWD={db_cfg['password']};", + f"Connection Timeout={db_cfg['connect_timeout']};", + ] + # 数据库服务器使用自签名/不受信任证书时，跳过证书链校验（连接仍保持加密）。 + # ODBC Driver 17/18 默认 Encrypt=Yes，遇到自签证书会报"不受信任的颁发机构"， + # 加 TrustServerCertificate=Yes 即可信任该证书。设为 false 时不影响原有行为。 + if db_cfg.get("trust_server_certificate", False): + parts.append("TrustServerCertificate=Yes;") + return "".join(parts) + + +def fetch_params_by_ids( + db_cfg: dict[str, Any], zong_pai_hao_list: list[str] +) -> dict[str, str | None]: + """按总排号批量查询新参数字段。 + + 返回 dict：{总排号: 新参数文本}。数据库中不存在的总排号，其值为 None + （而不是直接从结果里省略该 key），方便调用方区分"没查到"和"查到但内容为空"。 + + 一个总排号只对应一条记录（业务已确认为一对一关系）；如果实际数据出现 + 重复总排号，取查询结果的第一条并记录一条 WARNING 日志，不中断整体流程。 + """ + if not zong_pai_hao_list: + return {} + + result: dict[str, str | None] = {zph: None for zph in zong_pai_hao_list} + + schema = db_cfg["schema"] + table = db_cfg["table"] + id_col = db_cfg["id_column"] + param_col = db_cfg["param_column"] + + # 表名必须带 schema 前缀（如 [dbo].[表名]），只写表名在 schema 不是默认dbo时 + # 会查到错误的表，甚至直接报"找不到对象"。schema 和表名分别加中括号转义， + # 不能写成 [schema.table]，那样会被当成一个整体标识符解析。 + qualified_table = f"[{schema}].[{table}]" + + # 用参数化查询防止总排号里混入特殊字符导致 SQL 注入或语法错误 + placeholders = ",".join("?" for _ in zong_pai_hao_list) + sql = f"SELECT [{id_col}], [{param_col}] FROM {qualified_table} WHERE [{id_col}] IN ({placeholders})" + + conn_str = _build_conn_str(db_cfg) + try: + with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn: + cursor = conn.cursor() + # 查询超时设在 Connection 上（pyodbc 的 timeout 是 Connection 属性， + # Cursor 没有该属性，设 cursor.timeout 会报 AttributeError）。 + conn.timeout = db_cfg["query_timeout"] + cursor.execute(sql, zong_pai_hao_list) + seen = set() + for row in cursor.fetchall(): + zph, param = row[0], row[1] + if zph in seen: + logger.warning("总排号 %s 存在重复记录，已取第一条", zph) + continue + seen.add(zph) + result[zph] = param + except pyodbc.Error as e: + raise DatabaseError(f"数据库查询失败: {e}") from e + + return result + + +def fetch_param_by_id(db_cfg: dict[str, Any], zong_pai_hao: str) -> str | None: + """单个总排号查询的便捷封装。""" + return fetch_params_by_ids(db_cfg, [zong_pai_hao]).get(zong_pai_hao) diff --git a/llm_client.py b/llm_client.py new file mode 100644 index 0000000..b4114f5 --- /dev/null +++ b/llm_client.py @@ -0,0 +1,137 @@ +# -*- coding: utf-8 -*- +"""LLM 调用层：OpenAI 兼容接口，要求模型输出固定的纯文本格式。 + +之所以不让模型直接输出 JSON，是因为 JSON 的括号/引号/转义更容易被模型写错； +"标签: 值"这种极简格式模型几乎不会出错，且比 json.loads 更容易做宽松解析。 +本模块只负责"调用模型、拿到原始文本"，格式校验和清洗交给 parser.py。 + +提示词内容（system prompt + few-shot）全部来自 prompts.py，本文件不内嵌任何 +提示词文字——修改分类边界/细分类目，只需要改 prompts.py。 + +为配合日志记录，classify_raw 返回时会连带一份完整的"本次调用消息列表"， +调用方（classifier.py）据此写日志，即使调用失败也能拿到"发出去的消息是什么"。 + +enable_other 原样透传给 prompts.get_prompt，决定这次调用的提示词里要不要 +包含"其他"兜底类目；调用方（classifier.py）必须把同一个值也传给 +parser.parse_llm_output，两边保持一致。 +""" +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any + +from openai import APIError, APITimeoutError, OpenAI + +from prompts import get_prompt + +logger = logging.getLogger(__name__) + + +class LLMClientError(Exception): + """LLM 调用失败（网络/超时/接口错误），重试耗尽后抛出。""" + + +@dataclass +class LLMCallResult: + """一次 LLM 调用的完整记录，无论成功与否都会产出，供日志模块使用。""" + + messages: list[dict[str, str]] # 实际发送给模型的完整消息列表(含system+few-shot+用户) + raw_response: str | None # 模型原始回复；调用失败时为 None + error: str | None # 调用失败时的错误描述；成功时为 None + + +class LLMClient: + """封装 OpenAI 兼容接口的调用，内置 few-shot 和重试。""" + + def __init__(self, llm_cfg: dict[str, Any]): + self._cfg = llm_cfg + self._client = OpenAI( + base_url=llm_cfg["base_url"], + api_key=llm_cfg["api_key"], + timeout=llm_cfg["timeout"], + ) + + def _build_messages( + self, param_text: str, mode: str, enable_other: bool = False + ) -> list[dict[str, str]]: + system_prompt, fewshot = get_prompt(mode, enable_other=enable_other) + messages: list[dict[str, str]] = [{"role": "system", "content": system_prompt}] + for user_text, assistant_text in fewshot: + messages.append( + {"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{user_text}\n"""'} + ) + messages.append({"role": "assistant", "content": assistant_text}) + messages.append( + {"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{param_text}\n"""'} + ) + return messages + + def classify_raw( + self, param_text: str, mode: str = "coarse", enable_other: bool = False + ) -> LLMCallResult: + """调用模型，返回完整调用记录（消息列表 + 原始回复/错误）。 + + 与旧版不同：即使调用最终失败（重试耗尽），也不再抛异常中断调用方， + 而是把失败信息装进 LLMCallResult 返回——这样 classifier.py 才能在 + "调用彻底失败"的情况下依然拿到"发出去的消息是什么"，写进日志方便排查 + （比如看是不是消息本身有问题导致接口一直拒绝）。 + + 调用方如果需要区分"成功"还是"失败"，检查 result.error is None 即可。 + """ + messages = self._build_messages(param_text, mode, enable_other=enable_other) + max_retry = self._cfg["max_retry"] + backoff = self._cfg["retry_backoff_seconds"] + last_err: Exception | None = None + + # extra_body 用于透传标准 OpenAI SDK 不识别的厂商专属参数。 + # disable_thinking=true 时关闭推理链——像 DeepSeek-V4 系列这类默认开启 + # thinking 的模型，max_tokens 限制的是"推理token+正文token"的总量； + # 分类任务规则清晰、不需要推理，关闭后能避免推理阶段耗尽token预算导致 + # 正文被截断为空（表现为调用"成功"但content是空字符串）。 + # 若接口不支持该字段，通常会被直接忽略而非报错，但仍建议按需关闭本配置。 + extra_body: dict[str, Any] = {} + if self._cfg.get("disable_thinking", False): + extra_body["thinking"] = {"type": "disabled"} + + for attempt in range(1, max_retry + 1): + try: + resp = self._client.chat.completions.create( + model=self._cfg["model"], + temperature=self._cfg["temperature"], + max_tokens=self._cfg["max_tokens"], + messages=messages, + extra_body=extra_body or None, + ) + content = resp.choices[0].message.content + # 空字符串和 None 同样视为"没拿到有效正文"——推理型模型在 + # max_tokens 预算被推理阶段耗尽时，常表现为 content="" 而非 + # None（HTTP层面调用是成功的），必须一并捕获，否则会把这种 + # 情况误判为"调用成功、只是格式不对"，掩盖了真正的原因。 + if not content: + raise LLMClientError( + f"模型返回内容为空 (content={content!r})，" + f"若模型支持思维链，可能是推理耗尽了max_tokens预算" + ) + return LLMCallResult(messages=messages, raw_response=content, error=None) + except (APIError, APITimeoutError) as e: + last_err = e + logger.warning( + "LLM 调用失败 (第 %d/%d 次): %s", attempt, max_retry, e + ) + if attempt < max_retry: + time.sleep(backoff * attempt) + except Exception as e: # noqa: BLE001 - 捕获SDK未明确分类的异常，统一包装 + last_err = e + logger.warning( + "LLM 调用出现未预期异常 (第 %d/%d 次): %s", attempt, max_retry, e + ) + if attempt < max_retry: + time.sleep(backoff * attempt) + + return LLMCallResult( + messages=messages, + raw_response=None, + error=f"LLM 调用重试 {max_retry} 次后仍失败: {last_err}", + ) diff --git a/main.py b/main.py new file mode 100644 index 0000000..c038a92 --- /dev/null +++ b/main.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""布莱迪压力表 - 订单附件识别 命令行入口。 + +用法: + python main.py --id 26B742 # 单个总排号(粗分类,默认) + python main.py --id 26B742,26B743,26B744 # 批量，逗号分隔 + python main.py --ids-file ids.txt # 批量，文件每行一个总排号 + python main.py --id 26B742 --mode fine # 精分类(输出"大类:细分"格式) + python main.py --id 26B742 --config other.yaml # 指定其他配置文件 + python main.py --id 26B742 --pretty # 格式化输出JSON(默认单行紧凑) + python main.py --id 26B742 --log-dir /tmp/logs # 覆盖配置文件里的日志目录 + python main.py --id 26B742 --enable-other # 允许模型使用"其他"兜底类目 + +分类模式(--mode): + coarse (默认) - 只判断大类: 资料/配件/耗材(启用 --enable-other 时还有"其他") + fine - 输出"大类:细分类目"，如"配件:针型阀""资料:说明书" + +"其他"兜底类目(--enable-other 或配置文件 business.enable_other_category): + 默认关闭。关闭时，模型和格式校验都不知道"其他"这个选项存在，行为与 + 未支持"其他"之前完全一致。开启后，遇到确实是随货附件、但不属于任何 + 已列出细分类目的情况，模型可以归为"其他"，避免这类附件被漏判、或被 + 模型勉强套进不准确的类目里。命令行只能把它从"关"打开成"开"，无法 + 反过来用命令行强制关闭配置文件里已经打开的设置。 + +输出: + 每个总排号一行 JSON（JSON Lines 格式），例如: + {"zong_pai_hao": "26B742", "status": "ok", "has_attachment": true, "types": ["资料"]} + {"zong_pai_hao": "26B744", "status": "ok", "has_attachment": true, "types": ["配件:针型阀", "配件:表弯管"]} + {"zong_pai_hao": "XXXXXX", "status": "not_found", "has_attachment": null, "types": []} + + types 是唯一的类型字段，不再有单独的 fine_types： + coarse模式下 types 是大类列表；fine模式下每一项都是"大类:细分类目"。 + + status 字段含义: + ok - 识别成功 + not_found - 数据库中查不到该总排号 + empty_param - 该总排号存在，但新参数字段为空 + llm_call_error - LLM 调用失败(网络/接口错误,重试耗尽) + llm_format_error - LLM 返回内容格式不符合约定(重试耗尽) + db_error - 数据库连接/查询失败(影响整批) + +日志: + 每次实际发起LLM调用的总排号，都会在日志目录下生成一个独立的日志文件 + (文件名含总排号和时间戳)，完整记录发给模型的对话内容、模型的每一次 + 原始回复(含被判定格式错误、触发重试的)、以及最终解析结果。查不到 + (not_found)或参数为空(empty_param)的总排号不产生日志文件。 + 日志目录默认读取配置文件 business.log_dir，可用 --log-dir 临时覆盖。 +""" +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) + +from classifier import classify_batch # noqa: E402 +from config_loader import ConfigError, load_config # noqa: E402 + + +def _parse_ids(args: argparse.Namespace) -> list[str]: + """从 --id 或 --ids-file 解析出总排号列表，去除空白项。""" + ids: list[str] = [] + if args.id: + ids.extend(s.strip() for s in args.id.split(",") if s.strip()) + if args.ids_file: + p = Path(args.ids_file) + if not p.exists(): + print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr) + sys.exit(1) + with p.open("r", encoding="utf-8") as f: + ids.extend(line.strip() for line in f if line.strip()) + return ids + + +def main() -> None: + parser = argparse.ArgumentParser( + description="布莱迪压力表订单附件识别工具", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument( + "--id", type=str, default=None, + help="总排号，单个或逗号分隔的多个，例如: 26B742 或 26B742,26B743", + ) + parser.add_argument( + "--ids-file", type=str, default=None, + help="包含总排号的文本文件路径，每行一个总排号", + ) + parser.add_argument( + "--config", type=str, default="config.yaml", + help="配置文件路径 (默认: config.yaml)", + ) + parser.add_argument( + "--mode", type=str, default=None, choices=["coarse", "fine"], + help="分类粒度: coarse(粗分类,默认) 或 fine(精分类)。不指定则使用配置文件 business.default_mode", + ) + parser.add_argument( + "--log-dir", type=str, default=None, + help="LLM对话日志存放目录。不指定则使用配置文件 business.log_dir", + ) + parser.add_argument( + "--pretty", action="store_true", + help="以缩进格式输出JSON数组，而非默认的紧凑JSON Lines格式", + ) + parser.add_argument( + "--enable-other", action="store_true", + help="允许模型使用'其他'兜底类目。不指定则使用配置文件 business.enable_other_category" + "（缺省为关闭）；本开关只能从命令行打开，无法用命令行强制关闭配置文件里已打开的设置", + ) + args = parser.parse_args() + + if not args.id and not args.ids_file: + parser.error("必须提供 --id 或 --ids-file 其中之一") + + ids = _parse_ids(args) + if not ids: + print("[错误] 未解析到任何有效的总排号", file=sys.stderr) + sys.exit(1) + + try: + cfg = load_config(args.config) + except ConfigError as e: + print(f"[配置错误] {e}", file=sys.stderr) + sys.exit(1) + + logging.basicConfig( + level=getattr(logging, cfg["business"]["log_level"], logging.INFO), + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + stream=sys.stderr, # 日志走stderr，保证stdout只有干净的JSON结果，便于管道/重定向 + ) + + mode = args.mode or cfg["business"]["default_mode"] + log_dir = args.log_dir or cfg["business"]["log_dir"] + # enable_other_category 是后加的可选配置项，缺省按"关闭"处理；--enable-other + # 只能从"关"打开成"开"，不支持反过来用命令行强制关闭配置文件里已打开的设置。 + enable_other = bool(args.enable_other) or bool(cfg["business"].get("enable_other_category", False)) + + results = classify_batch( + db_cfg=cfg["database"], + llm_cfg=cfg["llm"], + zong_pai_hao_list=ids, + max_workers=cfg["business"]["max_workers"], + mode=mode, + log_dir=log_dir, + enable_other=enable_other, + ) + + if args.pretty: + print(json.dumps(results, ensure_ascii=False, indent=2)) + else: + for r in results: + print(json.dumps(r, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/order_logger.py b/order_logger.py new file mode 100644 index 0000000..48d7654 --- /dev/null +++ b/order_logger.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +"""对话日志模块。 + +设计目标：每个总排号一个日志文件，纯文本格式，完整记录"这次调用到底发生了 +什么"——system prompt、few-shot、实际发给模型的用户消息、模型每一次原始 +回复（含被格式校验拒绝、触发重试的那些），以及最终的解析结果。成功失败都写， +不因为调用失败就少记东西；恰恰是失败的调用最需要完整日志去排查。 + +文件命名: {log_dir}/{总排号}_{时间戳}.log +同一个总排号如果被反复处理（比如手动重跑），不会覆盖旧日志，每次生成新文件， +方便对比"改了提示词前后，同一条记录的判断有没有变化"。 +""" +from __future__ import annotations + +import re +from datetime import datetime +from pathlib import Path + +from llm_client import LLMCallResult +from parser import ParsedResult + +# 总排号可能含有对文件名不友好的字符（如斜杠），做一次保守清理 +_UNSAFE_FILENAME_CHARS = re.compile(r'[\\/:*?"<>|]') + + +def _safe_filename_part(zong_pai_hao: str) -> str: + return _UNSAFE_FILENAME_CHARS.sub("_", zong_pai_hao) + + +class OrderLogger: + """负责单个总排号在一次分类过程中的完整日志记录。 + + 典型用法（在 classifier.py 里）： + logger = OrderLogger(log_dir, zong_pai_hao, mode) + logger.log_param_text(param_text) + # ... 每次调用LLM后: + logger.log_llm_attempt(attempt_no, call_result, parsed_or_none, format_error_or_none) + # ... 最终: + logger.log_final_result(status, parsed_result_or_none) + logger.close() + """ + + def __init__(self, log_dir: str | Path, zong_pai_hao: str, mode: str): + self._log_dir = Path(log_dir) + self._log_dir.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + filename = f"{_safe_filename_part(zong_pai_hao)}_{timestamp}.log" + self._path = self._log_dir / filename + self._lines: list[str] = [] + + self._write_header(zong_pai_hao, mode) + + @property + def path(self) -> Path: + return self._path + + def _write_header(self, zong_pai_hao: str, mode: str) -> None: + self._lines.append("=" * 70) + self._lines.append(f"总排号: {zong_pai_hao}") + self._lines.append(f"分类模式: {mode}") + self._lines.append(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + self._lines.append("=" * 70) + self._flush() + + def log_param_text(self, param_text: str | None) -> None: + self._lines.append("") + self._lines.append("--- 新参数原文 ---") + self._lines.append(param_text if param_text else "(空)") + self._flush() + + def log_llm_attempt( + self, + attempt_no: int, + call_result: LLMCallResult, + parsed: ParsedResult | None, + format_error: str | None, + ) -> None: + """记录一次 LLM 调用的完整往返，无论成功还是失败都调用本方法。 + + call_result.error 非 None 表示网络/接口层面调用失败（对话都没发起成功 + 或没拿到回复）；call_result.error 为 None 但 format_error 非 None 表示 + 调用成功但模型输出格式不合规；两者都为 None 表示本次调用完全成功。 + """ + self._lines.append("") + self._lines.append(f"--- 第 {attempt_no} 次尝试 ---") + + self._lines.append("") + self._lines.append("[发送给模型的完整对话]") + for msg in call_result.messages: + role = msg["role"] + content = msg["content"] + self._lines.append(f" [{role}]") + for line in content.splitlines(): + self._lines.append(f" {line}") + + self._lines.append("") + if call_result.error is not None: + self._lines.append("[调用结果] 网络/接口调用失败") + self._lines.append(f" 错误信息: {call_result.error}") + else: + self._lines.append("[调用结果] 已收到模型回复") + self._lines.append("[模型原始回复]") + for line in (call_result.raw_response or "").splitlines(): + self._lines.append(f" {line}") + + self._lines.append("") + if format_error is not None: + self._lines.append("[格式校验] 失败") + self._lines.append(f" 原因: {format_error}") + else: + self._lines.append("[格式校验] 通过") + if parsed is not None: + self._lines.append(f" has_attachment = {parsed.has_attachment}") + self._lines.append(f" types = {parsed.types}") + self._flush() + + def log_final_result(self, status: str, parsed: ParsedResult | None) -> None: + self._lines.append("") + self._lines.append("=" * 70) + self._lines.append(f"最终状态: {status}") + if parsed is not None: + self._lines.append(f"最终结果: has_attachment={parsed.has_attachment}, types={parsed.types}") + self._lines.append(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + self._lines.append("=" * 70) + self._flush() + + def _flush(self) -> None: + """每次追加内容后立即落盘，避免进程中途异常退出导致日志丢失。""" + with self._path.open("a", encoding="utf-8") as f: + f.write("\n".join(self._lines) + "\n") + self._lines = [] + + def close(self) -> None: + """预留的显式收尾方法；当前实现下 _flush 已即时写盘，close 是空操作， + 但保留该接口方便未来切换为缓冲写入而不改动调用方代码。""" + pass diff --git a/parser.py b/parser.py new file mode 100644 index 0000000..5180669 --- /dev/null +++ b/parser.py @@ -0,0 +1,186 @@ +# -*- coding: utf-8 -*- +"""解析与清洗层。 + +职责边界很明确： +- LLM 只负责"识别"，输出约定好的纯文本（coarse模式2行，fine模式3行）。 +- 本模块只负责"验证格式是否正确 + 提取字段 + 组装成程序自己定义的结构"。 + +不信任 LLM 输出的具体体现： +1. 逐行校验，任何一行不匹配预期正则就判定为格式错误，绝不"猜测式"兜底解析。 +2. 枚举值必须完全匹配 prompts.py 里定义的枚举，出现模型编造的第三种值也判格式错误。 +3. fine模式下，细分类目与大类的归属关系必须自洽（如"针型阀"必须伴随"配件"）。 +4. 格式错误交给上层去做重试，而不是在这里勉强凑出一个结果。 + +enable_other 参数贯穿本模块的校验函数：是否把"其他"当作合法值，取决于这次 +分类调用传入的 enable_other，必须和生成该输出时 prompts.get_prompt 用的 +取值保持一致——由 classifier.py 负责把同一个值同时传给两边。 + +types 字段说明（对外唯一的类型字段，不再单独暴露 fine_types）： +- coarse 模式：大类列表，如 ["资料", "配件"]。 +- fine 模式：每一项都是"大类:细分类目"，如 ["资料:说明书", "配件:针型阀"]， + 由第二行（大类）和第三行（细分类目）校验通过后拼接而成；启用"其他"时， + 兜底项会体现为 "其他:其他"，格式与其余项保持一致。 +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from prompts import all_fine_types, fine_type_to_category, valid_categories + +_LINE1_RE = re.compile(r"^是否携带附件[:：]\s*(是|否)\s*$") +_LINE2_RE = re.compile(r"^附件类型[:：]\s*(.*)$") +_LINE3_RE = re.compile(r"^具体分类[:：]\s*(.*)$") + + +class FormatError(Exception): + """LLM 原始输出不符合约定格式，调用方应据此触发重试。""" + + +@dataclass +class ParsedResult: + """分类结果。 + + types 是对外唯一的类型字段：coarse 模式下是大类列表；fine 模式下每一项 + 都已经是"大类:细分类目"的组合字符串。不再有单独的 fine_types 字段—— + fine 模式的"大类"信息已经内嵌在每个元素里，无需再查第二份数据。 + """ + + has_attachment: bool + types: list[str] = field(default_factory=list) + + +def _strip_markdown_fence(raw_text: str) -> str: + cleaned = raw_text.strip() + return re.sub(r"^```[a-zA-Z]*\n?|```$", "", cleaned).strip() + + +def _parse_line1(line: str) -> bool: + m = _LINE1_RE.match(line) + if not m: + raise FormatError(f"第一行格式不匹配 '是否携带附件: 是/否': {line!r}") + return m.group(1) == "是" + + +def _parse_line2(line: str, has_attachment: bool, categories: set[str]) -> list[str]: + m = _LINE2_RE.match(line) + if not m: + raise FormatError(f"第二行格式不匹配 '附件类型: ...': {line!r}") + types_raw = m.group(1).strip() + + if not has_attachment: + if types_raw: + raise FormatError( + f"第一行为'否'，但第二行附件类型非空: {types_raw!r}（模型输出自相矛盾）" + ) + return [] + + if not types_raw: + raise FormatError("第一行为'是'，但第二行附件类型为空（模型输出自相矛盾）") + + types = [t.strip() for t in types_raw.split(",") if t.strip()] + invalid = [t for t in types if t not in categories] + if invalid: + raise FormatError(f"附件类型出现非法值 {invalid}，合法值仅为 {categories}") + if len(types) != len(set(types)): + raise FormatError(f"附件类型出现重复: {types}") + return types + + +def parse_llm_output_coarse(raw_text: str, enable_other: bool = False) -> ParsedResult: + """校验并解析 coarse 模式的两行输出。 + + enable_other 必须和生成该输出时 prompts.get_prompt 使用的取值一致， + 否则"其他"要么被误判为非法值拒绝，要么被误放行。 + """ + if raw_text is None: + raise FormatError("LLM 返回内容为 None") + + categories = valid_categories(enable_other) + cleaned = _strip_markdown_fence(raw_text) + lines = [ln.strip() for ln in cleaned.splitlines() if ln.strip()] + + if len(lines) != 2: + raise FormatError(f"coarse模式预期恰好2行有效内容，实际得到{len(lines)}行: {lines!r}") + + has_attachment = _parse_line1(lines[0]) + types = _parse_line2(lines[1], has_attachment, categories) + return ParsedResult(has_attachment=has_attachment, types=types) + + +def parse_llm_output_fine(raw_text: str, enable_other: bool = False) -> ParsedResult: + """校验并解析 fine 模式的三行输出，并把"大类+细分"合并进最终的 types。 + + 在 coarse 校验规则的基础上，额外校验： + - 第三行每个细分类目都必须在生效的细分类目全集内（是否含"其他"取决于 + enable_other）。 + - 若第一行为"否"，第三行必须为空；若为"是"，第三行至少要有一项。 + - 第三行不能有重复项。 + - 交叉自洽：第三行每个细分类目所属的大类，必须出现在第二行的大类列表里 + （例如选了"针型阀"但第二行没有"配件"，视为模型输出自相矛盾）。 + + 全部校验通过后，把第三行的每个细分类目和它所属的大类拼成 + "大类:细分类目"，作为最终 types 返回——不再单独暴露"大类列表"和 + "细分类目列表"两份数据。 + """ + if raw_text is None: + raise FormatError("LLM 返回内容为 None") + + categories = valid_categories(enable_other) + fine_types_enum = all_fine_types(enable_other) + fine_to_category = fine_type_to_category(enable_other) + + cleaned = _strip_markdown_fence(raw_text) + lines = [ln.strip() for ln in cleaned.splitlines() if ln.strip()] + + if len(lines) != 3: + raise FormatError(f"fine模式预期恰好3行有效内容，实际得到{len(lines)}行: {lines!r}") + + has_attachment = _parse_line1(lines[0]) + line2_types = _parse_line2(lines[1], has_attachment, categories) + + m3 = _LINE3_RE.match(lines[2]) + if not m3: + raise FormatError(f"第三行格式不匹配 '具体分类: ...': {lines[2]!r}") + fine_raw = m3.group(1).strip() + + if not has_attachment: + if fine_raw: + raise FormatError( + f"第一行为'否'，但第三行具体分类非空: {fine_raw!r}（模型输出自相矛盾）" + ) + return ParsedResult(has_attachment=False, types=[]) + + if not fine_raw: + raise FormatError("第一行为'是'，但第三行具体分类为空（模型输出自相矛盾）") + + fine_types = [t.strip() for t in fine_raw.split(",") if t.strip()] + invalid_fine = [t for t in fine_types if t not in fine_types_enum] + if invalid_fine: + raise FormatError( + f"具体分类出现非法值 {invalid_fine}，不在枚举内（enable_other={enable_other}）" + ) + if len(fine_types) != len(set(fine_types)): + raise FormatError(f"具体分类出现重复: {fine_types}") + + # 交叉自洽校验：每个细分类目对应的大类，必须出现在第二行里 + implied_categories = {fine_to_category[t] for t in fine_types} + missing_categories = implied_categories - set(line2_types) + if missing_categories: + raise FormatError( + f"具体分类 {fine_types} 隐含大类 {implied_categories}，" + f"但第二行附件类型只有 {line2_types}，缺少 {missing_categories}（模型输出自相矛盾）" + ) + + # 合并为最终的 "大类:细分类目" 列表，替代原来分开暴露的 types/fine_types + combined = [f"{fine_to_category[t]}:{t}" for t in fine_types] + return ParsedResult(has_attachment=True, types=combined) + + +def parse_llm_output(raw_text: str, mode: str = "coarse", enable_other: bool = False) -> ParsedResult: + """按模式分派到对应的解析函数。mode 只接受 'coarse' 或 'fine'。""" + if mode == "coarse": + return parse_llm_output_coarse(raw_text, enable_other=enable_other) + if mode == "fine": + return parse_llm_output_fine(raw_text, enable_other=enable_other) + raise ValueError(f"不支持的分类模式: {mode!r}，只能是 'coarse' 或 'fine'") diff --git a/prompts.py b/prompts.py new file mode 100644 index 0000000..58df1ad --- /dev/null +++ b/prompts.py @@ -0,0 +1,339 @@ +# -*- coding: utf-8 -*- +"""提示词集中管理模块。 + +设计目的：把"模型该怎么判断"和"怎么调用模型"分开——本文件只关心提示词内容， +llm_client.py 只负责拿着提示词去调接口。以后要调整分类边界、增删细分类目， +只需要改这一个文件，不涉及调用逻辑。 + +两种模式共用同一套背景知识（什么是产品本体、资料/配件/耗材的定义），只是 +模式为"fine"时，额外要求模型在第三行给出细分类目，且 few-shot 的回复也带 +上第三行。 + +细分类目枚举以布莱迪真实历史数据（新参数字段，约2.4万条）为基础人工归纳， +并与业务方逐项确认边界后定稿： +- 螺栓/螺母单列"紧固件"一类（不并入法兰隔膜等附属部件）。 +- 垫片单列第三大类"耗材/密封件"，不算在"配件"里，避免大量小垫片稀释"配件" + 标签的信息量（一张订单如果只要了一片垫片，和要了一整套阀组，不该被算作 + 同等权重的"携带配件"）。 +- 阻尼帽比照阻尼钉，视为工艺处理，不算配件、不进枚举。 +- 合格证明确不计入判断：既不算资料，也不算任何其他附件类型。无论参数文本 + 里是否提到合格证，一律忽略——不影响"是否携带附件"的判断结果，也不会 + 出现在细分类目里。这是与此前版本的主要差异之一，few-shot 里同步给出了 + "仅有合格证"的反例（否则模型容易在缺少监督信号时把合格证当成资料的一种）。 + +"其他"兜底类目（可选，默认关闭）： +遇到确实是随货交付的附件、但不属于以上任何一个具体细分类目的情况， +如果调用方传入 enable_other=True，模型可以把这类附件归为第四大类"其他"， +细分类目也固定填"其他"（不允许模型自己编造具体名称，保持枚举严格校验的 +风格不变）。enable_other=False（默认）时，"其他"完全不出现在提示词和 +校验枚举里，行为与不支持"其他"之前完全一致——这个参数必须和 +parser.parse_llm_output 使用的 enable_other 保持一致，否则会出现 +"提示词允许但校验拒绝"或反过来的不一致；classifier.py 负责保证两边一致。 +""" +from __future__ import annotations + +# ============================================================ +# 细分类目枚举（fine 模式专用，供 parser.py 做严格校验） +# ============================================================ + +DOC_FINE_TYPES: list[str] = [ + "出厂检测检验报告", + "材质证明类", + "说明书", + "检验记录过程性", + "图纸类", + "标定校验证书", + "质量证明书", + "原产地证明", + "检验合格证书综合", + "其他交工文件", + "营业执照", + "型式检验报告", +] + +PART_FINE_TYPES: list[str] = [ + "针型阀", + "球阀", + "截止阀", + "旋塞阀", + "角阀", + "阀组", + "冷凝圈", + "冷凝管", + "冷凝弯", + "虹吸管", + "表弯管", + "缓冲管缓冲弯", + "法兰隔膜", + "隔离器", + "过压保护器", + "散热器散热片", + "铅封", + "转换接头", + "焊接接头短节短管", + "卡箍抱箍", + "紧固件", + "活接头", + "接线盒", + "电缆插头", + "变送器", +] + +CONSUMABLE_FINE_TYPES: list[str] = [ + "垫片", +] + +# "其他"大类专用：整个大类只有这一个细分值，且值本身就是"其他"——不允许 +# 模型编造具体名称，兜底之外仍保持和三大类同样严格的枚举校验风格。 +OTHER_CATEGORY = "其他" +OTHER_FINE_TYPE = "其他" + +# 三大类固定存在；"其他"是否作为第四类生效，完全由调用方传入的 enable_other +# 决定，默认不生效。 +CATEGORY_NAMES = ("资料", "配件", "耗材") + +# 三大类下所有细分类目的全集（不含"其他"） +ALL_FINE_TYPES: set[str] = set(DOC_FINE_TYPES) | set(PART_FINE_TYPES) | set(CONSUMABLE_FINE_TYPES) + +# 细分类目 -> 所属大类（不含"其他"） +FINE_TYPE_TO_CATEGORY: dict[str, str] = { + **{t: "资料" for t in DOC_FINE_TYPES}, + **{t: "配件" for t in PART_FINE_TYPES}, + **{t: "耗材" for t in CONSUMABLE_FINE_TYPES}, +} + + +def valid_categories(enable_other: bool = False) -> set[str]: + """返回当前生效的大类合法值集合，enable_other=True 时多一个"其他"。""" + return set(CATEGORY_NAMES) | ({OTHER_CATEGORY} if enable_other else set()) + + +def all_fine_types(enable_other: bool = False) -> set[str]: + """返回当前生效的细分类目全集，enable_other=True 时多一个"其他"。""" + return ALL_FINE_TYPES | ({OTHER_FINE_TYPE} if enable_other else set()) + + +def fine_type_to_category(enable_other: bool = False) -> dict[str, str]: + """返回当前生效的"细分类目 -> 大类"映射，enable_other=True 时多"其他"->"其他"。""" + if not enable_other: + return dict(FINE_TYPE_TO_CATEGORY) + return {**FINE_TYPE_TO_CATEGORY, OTHER_FINE_TYPE: OTHER_CATEGORY} + + +# ============================================================ +# 背景知识（两种模式共用） +# ============================================================ + +_BACKGROUND = """你是布莱迪压力表公司的订单参数审核助手。你的任务是判断一条订单的"技术参数"文本中，除了压力表产品本体之外，是否还要求交付"附件"。 + +【背景】公司生产压力表（用于化工行业）。参数文本里大部分内容是描述产品本体的规格，这些都不是附件。 + +【什么是产品本体（不算附件）】 +表壳直径/材质、精度等级、防护等级(IP)、过压保护、弹性元件材质、接头材质、接头螺纹、安装方式、表盘材质/刻度/颜色、量程、充油、玻璃、指针、调零装置、缓冲钉/阻尼钉/阻尼帽（均属工艺处理，不算配件）、结构代号(如513/531)、卡盘尺寸、以及各种打标/示值/红线标识要求。 + +【附件 = 产品本体之外，随货交付的东西，分三大类】 + +一、资料类（交付给客户的技术文件）： +检测报告、检验报告、出厂报告、材质证明/材质单/材质报告、说明书(使用/安装/维护)、标定证书/校验证书、制造记录/检验记录、图纸、含技术信息的二维码/标签、营业执照、原产地证明、型式检验报告。 +※ 注意：装箱单、送货单、贴箱单、发货章、包装/箱体标注等，属于物流包装指令，不算资料。合格证也不算资料——无论参数文本里是否提到合格证，一律忽略，既不影响"是否携带附件"的判断，也绝不能出现在第三行的具体分类里。 + +二、配件类（随货的独立实物零件，不含垫片）： +针型阀、球阀、截止阀、旋塞阀、角阀、阀组（二/三/五阀组等）、冷凝管/冷凝圈/冷凝弯、表弯管/缓冲弯、虹吸管、过压保护器、散热器/散热片、焊接接头/短节/短管、法兰隔膜、隔离器、铅封、转换接头、卡箍/抱箍、紧固件（螺栓/螺母）、活接头、接线盒、电缆/插头、变送器等，凡是随表一起出货的独立零件都算。 +※ 注意：位号牌、铭牌、标牌、号牌属于标识件，不算配件。缓冲钉/阻尼钉/阻尼帽都是工艺处理，不算配件。 + +三、耗材/密封件类： +垫片（四氟垫片、紫铜垫片、缠绕垫、密封圈、O型圈等各种材质的垫片），单独列为耗材类，不算在配件里。 + +【判断步骤】 +1. 逐项扫描文本，剔除所有"产品本体"描述，并剔除合格证（合格证不计入判断，见上方资料类说明）。 +2. 看剩下的内容里是否出现资料类、配件类或耗材类的东西。 +3. 只要出现任一类，第一行就是"是"；三类都没有就是"否"。""" + +_OTHER_BACKGROUND_SUFFIX = """ + +四、其他类（兜底类目，仅在本次调用启用该选项时可用）： +如果确认订单确实随货交付了某个附件，但它明显不属于以上资料/配件/耗材已列出的任何一个具体细分类目（包括各大类里已有的"其他XX"兜底项，如"其他交工文件""检验合格证书综合"），也无法合理归入其中任何一类，才可以归为"其他"。"其他"是最后的兜底选项——凡是能归到前三类具体细分类目的，都必须归到那里，不要图省事就往"其他"里放。""" + + +def _background(enable_other: bool) -> str: + return _BACKGROUND + (_OTHER_BACKGROUND_SUFFIX if enable_other else "") + + +# ============================================================ +# coarse 模式：只输出大类 +# ============================================================ + +def _coarse_system_prompt(enable_other: bool) -> str: + ordered_categories = list(CATEGORY_NAMES) + ([OTHER_CATEGORY] if enable_other else []) + categories_str = "、".join(f'"{c}"' for c in ordered_categories) + return ( + _background(enable_other) + + f""" + +【输出格式 —— 必须严格按此格式，只输出这两行，不要多余文字、不要标点符号、不要解释】 +是否携带附件: 是 +附件类型: 资料,配件 + +【格式规则】 +- 第一行只能是"是否携带附件: 是"或"是否携带附件: 否"。 +- 第二行"附件类型: "后面，如果第一行是"否"，留空；如果是"是"，写{categories_str}中的一个或多个，用英文逗号分隔（不要空格）。 +- 不要输出 JSON、不要输出 markdown 代码块、不要输出任何解释性文字。""" + ) + + +FEWSHOT_COARSE: list[tuple[str, str]] = [ + ( + "表壳直径:100mm 表壳材质:304SS 精度等级:1.6级 接头螺纹:M20×1.5 带缓冲钉 513结构", + "是否携带附件: 否\n附件类型: ", + ), + ( + "表壳直径:63mm 精度等级:2.5级 带不锈钢位号牌", + "是否携带附件: 否\n附件类型: ", + ), + ( + "表壳直径:100mm 张贴好装箱单并注明订单号 送货单盖发货章", + "是否携带附件: 否\n附件类型: ", + ), + ( + "接头螺纹:ZG1/4 带阻尼帽 不要阻尼钉", + "是否携带附件: 否\n附件类型: ", + ), + ( + "表壳直径:100mm 表壳材质:304SS 合格证随货", + "是否携带附件: 否\n附件类型: ", + ), + ( + "表壳直径:100mm 提供出厂检测报告 合格证随货", + "是否携带附件: 是\n附件类型: 资料", + ), + ( + "表壳直径:100mm 针型阀材质:304SS 配316L表弯管", + "是否携带附件: 是\n附件类型: 配件", + ), + ( + "配四氟垫片 法兰专用四氟垫片", + "是否携带附件: 是\n附件类型: 耗材", + ), + ( + "出厂检验报告 仪表使用说明书 配冷凝圈 过压保护器 配四氟垫片", + "是否携带附件: 是\n附件类型: 资料,配件,耗材", + ), +] + +# 仅在 enable_other=True 时追加到 FEWSHOT_COARSE 后面 +FEWSHOT_COARSE_OTHER: list[tuple[str, str]] = [ + ( + "表壳直径:100mm 随货附带一个定制不锈钢防雨罩", + "是否携带附件: 是\n附件类型: 其他", + ), +] + + +# ============================================================ +# fine 模式：额外输出第三行细分类目 +# ============================================================ + +def _fine_system_prompt(enable_other: bool) -> str: + ordered_categories = list(CATEGORY_NAMES) + ([OTHER_CATEGORY] if enable_other else []) + categories_str = "、".join(f'"{c}"' for c in ordered_categories) + other_line = ( + f"\n其他类细分：{OTHER_FINE_TYPE}（仅当确认是随货附件、但资料/配件/耗材下所有具体" + f"细分类目都无法覆盖时才使用，不需要也不允许编造更具体的名称）" + if enable_other else "" + ) + return ( + _background(enable_other) + + f""" + +【细分类目枚举 —— 第三行必须只从下面这些词里选，不能自己编造新词】 +资料类细分：{"、".join(DOC_FINE_TYPES)} +配件类细分：{"、".join(PART_FINE_TYPES)} +耗材类细分：{"、".join(CONSUMABLE_FINE_TYPES)}{other_line} + +【输出格式 —— 必须严格按此格式，只输出这三行，不要多余文字、不要标点符号、不要解释】 +是否携带附件: 是 +附件类型: 资料,配件 +具体分类: 出厂检测检验报告,针型阀 + +【格式规则】 +- 第一行只能是"是否携带附件: 是"或"是否携带附件: 否"。 +- 第二行"附件类型: "后面，如果第一行是"否"，留空；如果是"是"，写{categories_str}中的一个或多个，用英文逗号分隔（不要空格）。 +- 第三行"具体分类: "后面，如果第一行是"否"，留空；如果是"是"，从上面枚举的细分类目中选出所有识别到的，用英文逗号分隔（不要空格），不能选枚举之外的词，也不能只写大类名称（如"资料"）。 +- 第三行选出的每个细分类目，都必须能对应到第二行里的某个大类（例如选了"针型阀"，第二行就必须包含"配件"）。 +- 不要输出 JSON、不要输出 markdown 代码块、不要输出任何解释性文字。""" + ) + + +FEWSHOT_FINE: list[tuple[str, str]] = [ + ( + "表壳直径:100mm 表壳材质:304SS 精度等级:1.6级 接头螺纹:M20×1.5 带缓冲钉 513结构", + "是否携带附件: 否\n附件类型: \n具体分类: ", + ), + ( + "表壳直径:63mm 精度等级:2.5级 带不锈钢位号牌", + "是否携带附件: 否\n附件类型: \n具体分类: ", + ), + ( + "表壳直径:100mm 张贴好装箱单并注明订单号 送货单盖发货章", + "是否携带附件: 否\n附件类型: \n具体分类: ", + ), + ( + "接头螺纹:ZG1/4 带阻尼帽 不要阻尼钉", + "是否携带附件: 否\n附件类型: \n具体分类: ", + ), + ( + "表壳直径:100mm 表壳材质:304SS 合格证随货", + "是否携带附件: 否\n附件类型: \n具体分类: ", + ), + ( + "表壳直径:100mm 提供出厂检测报告 合格证随货", + "是否携带附件: 是\n附件类型: 资料\n具体分类: 出厂检测检验报告", + ), + ( + "表壳直径:100mm 针型阀材质:304SS 配316L表弯管", + "是否携带附件: 是\n附件类型: 配件\n具体分类: 针型阀,表弯管", + ), + ( + "配四氟垫片 法兰专用四氟垫片", + "是否携带附件: 是\n附件类型: 耗材\n具体分类: 垫片", + ), + ( + "带不锈钢螺栓螺母固定 配304SS法兰隔膜", + "是否携带附件: 是\n附件类型: 配件\n具体分类: 紧固件,法兰隔膜", + ), + ( + "出厂检验报告 仪表使用说明书 配冷凝圈 过压保护器 配四氟垫片", + "是否携带附件: 是\n附件类型: 资料,配件,耗材\n具体分类: 出厂检测检验报告,说明书,冷凝圈,过压保护器,垫片", + ), +] + +# 仅在 enable_other=True 时追加到 FEWSHOT_FINE 后面 +FEWSHOT_FINE_OTHER: list[tuple[str, str]] = [ + ( + "表壳直径:100mm 随货附带一个定制不锈钢防雨罩", + "是否携带附件: 是\n附件类型: 其他\n具体分类: 其他", + ), +] + + +def get_prompt(mode: str, enable_other: bool = False) -> tuple[str, list[tuple[str, str]]]: + """按模式（以及是否启用"其他"兜底类目）返回 (system_prompt, few_shot_examples)。 + + mode 只接受 "coarse" 或 "fine"，其他值直接抛 ValueError——这里不做 + 宽松兼容，避免拼写错误的 mode 静默退化成某个默认行为。 + + enable_other 默认 False：不传该参数时，"其他"完全不出现在提示词里， + 模型和之后的 parser 校验都不知道这个选项的存在，行为与不支持"其他" + 之前完全一致。调用方（llm_client.py）必须把这里用的 enable_other 原样 + 传给 parser.parse_llm_output，两边必须一致，否则会出现"提示词允许但 + 校验拒绝"或反过来的不一致。 + """ + if mode == "coarse": + prompt = _coarse_system_prompt(enable_other) + fewshot = FEWSHOT_COARSE + (FEWSHOT_COARSE_OTHER if enable_other else []) + return prompt, fewshot + if mode == "fine": + prompt = _fine_system_prompt(enable_other) + fewshot = FEWSHOT_FINE + (FEWSHOT_FINE_OTHER if enable_other else []) + return prompt, fewshot + raise ValueError(f"不支持的分类模式: {mode!r}，只能是 'coarse' 或 'fine'") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b83aae8 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +pyyaml>=6.0 +pyodbc>=5.0 +openai>=1.0