diff --git a/README.md b/README.md index 22f4e10..4437ef6 100644 --- a/README.md +++ b/README.md @@ -10,13 +10,15 @@ pip install -r requirements.txt ``` -`pyodbc` 需要系统已安装对应的 ODBC 驱动（如 "ODBC Driver 17 for SQL Server"）。 +`pyodbc` 需要系统已安装对应的 ODBC 驱动。本项目 `config.yaml` 中使用的版本为 +"ODBC Driver 18 for SQL Server"，请按服务器实际安装的驱动版本填写 `database.driver`。 若服务器上已有 SQL Server 管理工具/客户端环境，通常已包含该驱动；否则需自行 安装 Microsoft 官方 ODBC Driver。 ## 配置 -编辑 `config.yaml`，填入以下三部分（模板中的 `CHANGE_ME` 必须替换）： +编辑 `config.yaml`，填入以下三部分（首次使用需替换为真实值，**请勿将含真实 +数据库密码 / API Key 的配置文件提交到版本库**）： - `database`：SQL Server 连接信息（含 `schema`）、表名、字段名 - `llm`：OpenAI 兼容接口的 `base_url`、`api_key`、`model` @@ -49,6 +51,9 @@ python main.py --id 26B742,26B743 --pretty # 允许模型使用"其他"兜底类目 python main.py --id 26B742 --mode fine --enable-other + +# 额外在 stderr 打印本批运行汇总（耗时/token/缓存命中率）；stdout 仍只输出干净 JSON Lines +python main.py --id 26B742,26B743 --summary ``` ## 分类粒度：coarse / fine @@ -94,11 +99,14 @@ python main.py --id 26B742 --mode fine --enable-other 默认逐行输出 JSON（JSON Lines），便于管道处理和逐条消费： ```json -{"zong_pai_hao": "26B742", "status": "ok", "has_attachment": true, "types": ["资料"]} +{"zong_pai_hao": "26B742", "status": "ok", "has_attachment": true, "types": ["资料"], "meta": {"elapsed_ms": 1476.6, "attempts": 1, "llm": {"model": "deepseek-v4-flash", "elapsed_ms": 1472.8, "usage": {"prompt_tokens": 1297, "completion_tokens": 15, "total_tokens": 1312, "prompt_cache_hit_tokens": 1280, "prompt_cache_miss_tokens": 17, "cache_hit_rate": 0.9869}}}} {"zong_pai_hao": "26B744", "status": "ok", "has_attachment": true, "types": ["配件:针型阀", "配件:表弯管"]} -{"zong_pai_hao": "26B999", "status": "not_found", "has_attachment": null, "types": []} +{"zong_pai_hao": "26B999", "status": "not_found", "has_attachment": null, "types": [], "meta": {"elapsed_ms": null, "attempts": 0, "llm": null}} ``` +每条结果都额外带一个 `meta` 运行统计对象（结构见下方"meta 字段说明"）。加上 `--summary` +后还会在 **stderr** 额外打印整批汇总（JSON），而 stdout 仍只输出上述干净的 JSON Lines，便于管道/重定向。 + `types` 是唯一的类型字段，不再有单独的 `fine_types`： - `coarse` 模式下，`types` 是大类列表，如 `["资料", "配件"]`。 - `fine` 模式下，`types` 里每一项都是"大类:细分类目"，如 @@ -119,6 +127,30 @@ python main.py --id 26B742 --mode fine --enable-other `status` 非 `ok` 时，`has_attachment` 为 `null`，`types` 为空数组， 不会有猜测性的默认值混入结果。 +### meta 字段说明（运行统计） + +每条结果的 `meta` 提供调用层面的性能与用量统计，便于核算耗时与 token 成本： + +- `elapsed_ms`：本条从进入分类到出结果的总耗时（毫秒，含格式校验重试与解析）。未发起 LLM 的情况（`not_found` / `empty_param` / `db_error`）为 `null`。 +- `attempts`：实际尝试次数（含格式校验重试）；调用彻底失败时为重试上限。 +- `llm`：本次成功调用 LLM 的统计块；未调用或调用失败（如 `not_found`、`llm_call_error`）时为 `null`。 + - `model`：实际服务的模型名（来自接口返回）。 + - `elapsed_ms`：单次 LLM 调用的耗时（毫秒）。 + - `usage`：token 用量与缓存统计： + - `prompt_tokens`：输入 token 总量（= 缓存命中 + 未命中）。 + - `completion_tokens` / `total_tokens`：输出 / 总 token。 + - `prompt_cache_hit_tokens` / `prompt_cache_miss_tokens`：输入中命中 / 未命中硬盘缓存的 token 数（DeepSeek 上下文缓存特性，相同前缀的后续请求会命中，见下）。 + - `cache_hit_rate`：缓存命中率 = `prompt_cache_hit_tokens / prompt_tokens`，取值 0~1；首次冷调用为 0.0，缓存预热后显著升高。 + +成功调用示例： +```json +"meta": {"elapsed_ms": 1476.6, "attempts": 1, "llm": {"model": "deepseek-v4-flash", "elapsed_ms": 1472.8, "usage": {"prompt_tokens": 1297, "completion_tokens": 15, "total_tokens": 1312, "prompt_cache_hit_tokens": 1280, "prompt_cache_miss_tokens": 17, "cache_hit_rate": 0.9869}}} +``` + +`--summary` 打印的整批汇总字段与 `meta.llm.usage` 对应：`count` / `total_elapsed_ms` / `llm_calls` / +各类 `total_*_tokens` / `total_cache_hit_tokens` / `total_cache_miss_tokens` / `avg_cache_hit_rate` +（按 token 加权的整体命中率）。 + ## 对话日志 每个实际发起 LLM 调用的总排号，都会在日志目录（默认 `logs/`，可用 @@ -132,6 +164,7 @@ logs/26B742_20260723_153012_123456.log - 发给模型的完整对话（system prompt + few-shot 示例 + 实际用户消息） - 模型的每一次原始回复——**包括被格式校验判定无效、触发重试的那些** - 每次尝试的格式校验结果（通过/失败及原因） +- 每次尝试的 `[调用统计]`：实际模型名、单次调用耗时、token 用量（prompt/completion/total）与缓存命中情况（hit/miss/命中率），调用失败时相应字段显示"无" - 最终的解析结果和 status 同一总排号被重复处理不会覆盖旧日志（文件名带精确到微秒的时间戳），方便对比 @@ -149,8 +182,10 @@ logs/26B742_20260723_153012_123456.log 如果思考阶段把预算用完，正文就会被截断成空字符串。 本工具通过两处配置应对： -- `llm.max_tokens`：调大到能覆盖"思考+正文"的总量（默认 1024），而不是只按 - 正文那两三行的长度来设。 +- `llm.max_tokens`：控制单次回复长度上限。本项目已开启 `llm.disable_thinking: true` + （关闭推理链），因此 `max_tokens` 设为 `200` 即可覆盖"大类+细分"的短输出；若把 + `disable_thinking` 改回 `false` 启用思考，则需调大到能覆盖"思考+正文"的总量（如 + 1024），否则思考阶段可能耗尽预算、正文被截断成空。 - `llm.disable_thinking`：设为 `true` 时会通过 `extra_body={"thinking": {"type": "disabled"}}` 关闭推理链——分类任务规则清晰、不需要模型思考，关闭 后响应更快、更省 token，也从根源上避免"思考耗尽预算"的问题。仅 DeepSeek-V4 @@ -181,20 +216,70 @@ JSON 的括号、引号、转义更容易被模型写错。约定 LLM 只输出" `enable_other` 参数控制（对应 `--enable-other` 或配置文件 `business.enable_other_category`），并且同一次调用里，提示词（`prompts.py`） 和格式校验（`parser.py`）用的 `enable_other` 必须一致，否则会出现"提示词 -允许但校验拒绝"的不一致——这层一致性由 `classifier.py` 统一负责传递。 + 允许但校验拒绝"的不一致——这层一致性由 `classifier.py` 统一负责传递。 + +## 写入落库（落库到 Common.Attachment） + +分类结果除在控制台以 JSON Lines 输出外，还可通过写入入口落库到生产库的 +`Common.Attachment` 表，供下游系统按总排号查询该订单的附件类目。 + +```bash +# 全量写入（按总排号遍历源表全部记录） +python write_attachments.py + +# 仅前 N 个总排号（测试用，避免一次性消耗大量 LLM 额度） +python write_attachments.py --limit 10 + +# 指定总排号文件（每行一个） +python write_attachments.py --ids-file ids.txt + +# 直接指定总排号：单个，或逗号分隔的多个（与 --ids-file 可合并，提供后不再全表扫描） +python write_attachments.py --id 26B742 +python write_attachments.py --id 26B742,26B743,26B744 + +# 粗分类写入（小类统一填占位值）；不指定 --mode 则用配置文件 business.default_mode +python write_attachments.py --mode coarse + +# 先预览将写入/跳过的行，不真正落库 +python write_attachments.py --dry-run + +# 允许"其他"兜底类目 +python write_attachments.py --mode fine --enable-other +``` + +运行结束后，除上面的 `ok/empty_param/not_found/error` 与"将写入行数/跳过数"统计外，还会打印一行 +`[汇总]`：本批总耗时、LLM 调用次数、token 总量（prompt/completion/total）与加权缓存命中率，便于核算成本。 + +落库约定（与 `db.py` 常量、`write_attachments.py` 的转换逻辑保持一致）： + +- **有附件**（`has_attachment=true`，`status=ok`）：每个 `type` 写一行。 + - `fine` 模式：`type` 形如 `大类:细分`，按首个冒号拆分为 `(NS, 大类, 细分)`。 + - `coarse` 模式：`type` 是大类，小类统一填占位值 `无` → `(NS, 大类, '无')`。 +- **无附件**（`has_attachment=false`，含 `empty_param` 与模型判无附件）：写哨兵行 + `(NS, '无附件', '无')`，下游用 `WHERE MajorCategory <> '无附件'` 取真实附件。 +- **无法确定/失败**（`has_attachment=null`，含 `not_found` / `llm_*_error` / + `db_error`）：一律不写，既不当作无附件，也不留脏数据。 +- **幂等**：写入时对同一总排号先删除旧行再插入本次结果（依赖 `NS, MajorCategory, + MinorCategory` 唯一索引），重跑安全。 + +目标表 `Common.Attachment` 字段：`NS`（nvarchar(30)，总排号/关联键）、 +`MajorCategory`（nvarchar(40)，附件大类）、`MinorCategory`（nvarchar(40)，附件小类）， +三者均 `NOT NULL`。 ## 项目结构 ``` ├── config.yaml # 配置文件 -├── main.py # 命令行入口（直接 import 同目录各模块） +├── main.py # 命令行入口：分类并输出 JSON Lines（支持 --summary 在 stderr 打印批汇总；直接 import 同目录各模块） +├── write_attachments.py # 写入入口：分类结果落库到 Common.Attachment（全量/--limit/--ids-file/--mode/--dry-run/--enable-other/--config；运行后打印 token/缓存汇总） ├── requirements.txt ├── config_loader.py # YAML 配置读取与校验 -├── db.py # SQL Server 查询 (pyodbc) +├── db.py # SQL Server 查询与落库（fetch_params_by_ids / fetch_all_ids / upsert_attachments，pyodbc） ├── prompts.py # 提示词与细分类目枚举（唯一需要改分类边界时编辑的文件） ├── llm_client.py # LLM 调用 (OpenAI 兼容接口) ├── parser.py # LLM 输出格式校验与清洗 → 结构化数据 ├── order_logger.py # 每个总排号一份的完整对话日志 ├── classifier.py # 编排：查库 -> 调LLM -> 校验解析 -> 记日志 -> 组装结果 -└── attachment_classifier.py # 历史版本（旧 Excel 版，已弃用，保留仅供参考） +├── attachment_classifier.py # 历史版本（旧 Excel 版，已弃用，保留仅供参考） +└── docs/ # 补充文档 ``` diff --git a/classifier.py b/classifier.py index 75c3da7..6cb561a 100644 --- a/classifier.py +++ b/classifier.py @@ -28,12 +28,13 @@ classify_batch/classify_single 一路透传到 llm_client.classify_raw（决定 from __future__ import annotations import logging +import time 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 llm_client import LLMCallResult, LLMClient from order_logger import OrderLogger from parser import FormatError, ParsedResult, parse_llm_output @@ -46,9 +47,55 @@ def _empty_result(zong_pai_hao: str, status: str) -> dict[str, Any]: "status": status, "has_attachment": None, "types": [], + # 未发起 LLM 调用的结果（not_found/db_error），meta 给出一致的空壳， + # 便于下游统一读取（elapsed_ms 为 None，统计时按 0 处理）。 + "meta": {"elapsed_ms": None, "attempts": 0, "llm": None}, } +def _build_usage(u: dict[str, Any]) -> dict[str, Any]: + """把原始 usage dict 整理成对外结构，并算出缓存率。 + + 缓存率 = prompt_cache_hit_tokens / prompt_tokens（prompt_tokens 为 0 时按 + 0.0 处理，避免除零）。保留 4 位小数。 + """ + prompt = u.get("prompt_tokens", 0) or 0 + completion = u.get("completion_tokens", 0) or 0 + total = u.get("total_tokens", 0) or 0 + hit = u.get("prompt_cache_hit_tokens", 0) or 0 + miss = u.get("prompt_cache_miss_tokens", 0) or 0 + cache_hit_rate = round(hit / prompt, 4) if prompt else 0.0 + return { + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": total, + "prompt_cache_hit_tokens": hit, + "prompt_cache_miss_tokens": miss, + "cache_hit_rate": cache_hit_rate, + } + + +def _make_meta( + call_result: LLMCallResult | None, start: float, attempt: int +) -> dict[str, Any]: + """组装单条结果的 meta 对象。 + + elapsed_ms 为从 classify_param_text 入口到此处的总耗时（含重试+解析）； + 若本次有成功调用，则 llm 块记录模型名、单次调用耗时与整理后的 usage； + 否则（调用失败/未调用）llm 为 None。 + """ + elapsed_ms = (time.perf_counter() - start) * 1000 + if call_result is not None and call_result.error is None and call_result.usage is not None: + llm_block = { + "model": call_result.model, + "elapsed_ms": call_result.elapsed_ms, + "usage": _build_usage(call_result.usage), + } + else: + llm_block = None + return {"elapsed_ms": elapsed_ms, "attempts": attempt, "llm": llm_block} + + def classify_param_text( llm_client: LLMClient, param_text: str | None, @@ -56,7 +103,7 @@ def classify_param_text( format_retry: int = 2, order_logger: OrderLogger | None = None, enable_other: bool = False, -) -> tuple[str, bool | None, list[str]]: +) -> tuple[str, bool | None, list[str], dict[str, Any]]: """对单条"新参数"文本做分类，返回 (status, has_attachment, types)。 mode 决定分类粒度："coarse"只输出大类(资料/配件/耗材，启用 enable_other @@ -79,32 +126,36 @@ def classify_param_text( order_logger 若提供，会记录本次分类过程中的每一次LLM往返（无论成败）。 """ + start = time.perf_counter() + 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, [] + return "empty_param", False, [], _make_meta(None, start, 0) if order_logger is not None: order_logger.log_param_text(param_text) + last_call_result: LLMCallResult | None = None 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) + last_call_result = call_result 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, [] + return "llm_call_error", None, [], _make_meta(call_result, start, attempt) 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 + return "ok", parsed.has_attachment, parsed.types, _make_meta(call_result, start, attempt) except FormatError as e: last_format_err = e logger.warning( @@ -117,7 +168,7 @@ def classify_param_text( 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, [] + return "llm_format_error", None, [], _make_meta(last_call_result, start, format_retry) def classify_batch( @@ -162,7 +213,7 @@ def classify_batch( 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( + status, has_attachment, types, meta = classify_param_text( llm_client, param_map[zph], mode=mode, order_logger=order_logger, enable_other=enable_other, ) @@ -171,6 +222,7 @@ def classify_batch( "status": status, "has_attachment": has_attachment, "types": types, + "meta": meta, } if ids_to_classify: @@ -197,3 +249,42 @@ def classify_single( db_cfg, llm_cfg, [zong_pai_hao], max_workers=1, mode=mode, log_dir=log_dir, enable_other=enable_other, )[0] + + +def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: + """汇总一批 classify_batch 结果的运行统计。 + + 从每条结果的 meta 里累加：总耗时、实际 LLM 调用次数、各类 token 总量、 + 以及整体缓存命中率（按 prompt_cache_hit_tokens / prompt_tokens 加权）。 + 供 CLI 打印批统计用，不改变 classify_batch 的返回结构。 + """ + total_elapsed = 0.0 + llm_calls = 0 + sum_prompt = sum_completion = sum_total = sum_hit = sum_miss = 0 + for r in results: + meta = r.get("meta") or {} + e = meta.get("elapsed_ms") + if e: + total_elapsed += e + llm = meta.get("llm") + if isinstance(llm, dict): + u = llm.get("usage") + if isinstance(u, dict): + llm_calls += 1 + sum_prompt += u.get("prompt_tokens", 0) or 0 + sum_completion += u.get("completion_tokens", 0) or 0 + sum_total += u.get("total_tokens", 0) or 0 + sum_hit += u.get("prompt_cache_hit_tokens", 0) or 0 + sum_miss += u.get("prompt_cache_miss_tokens", 0) or 0 + avg_cache_hit_rate = round(sum_hit / sum_prompt, 4) if sum_prompt else 0.0 + return { + "count": len(results), + "total_elapsed_ms": round(total_elapsed, 1), + "llm_calls": llm_calls, + "total_prompt_tokens": sum_prompt, + "total_completion_tokens": sum_completion, + "total_tokens": sum_total, + "total_cache_hit_tokens": sum_hit, + "total_cache_miss_tokens": sum_miss, + "avg_cache_hit_rate": avg_cache_hit_rate, + } diff --git a/db.py b/db.py index 6fe44be..8ac0d02 100644 --- a/db.py +++ b/db.py @@ -86,3 +86,89 @@ def fetch_params_by_ids( 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) + + +# --------------------------------------------------------------------------- +# 附件分类落库（目标表 Common.Attachment，与源表 productionContractData 分属不同 schema） +# --------------------------------------------------------------------------- +# 落库约定（与项目记忆 MEMORY.md 中写入约定一致，请勿在此处用字面量以外的值）： +# - 无附件（has_attachment=false）：写哨兵行 (NS, '无附件', '无') +# - 无小类的行（coarse 真实附件 / 哨兵行）：MinorCategory 统一填 '无' +ATTACHMENT_SCHEMA = "Common" +ATTACHMENT_TABLE = "Attachment" + +SENTINEL_MAJOR = "无附件" +SENTINEL_MINOR = "无" +NO_MINOR = "无" + + +def fetch_all_ids(db_cfg: dict[str, Any], limit: int | None = None) -> list[str]: + """拉取源表全部总排号（去重、排除 NULL），用于全量分类写入。 + + limit 用于测试时只取前 N 个（按总排号排序）。返回的总排号列表可直接 + 喂给 classify_batch。 + """ + schema = db_cfg["schema"] + table = db_cfg["table"] + id_col = db_cfg["id_column"] + qualified = f"[{schema}].[{table}]" + sql = f"SELECT DISTINCT [{id_col}] FROM {qualified} WHERE [{id_col}] IS NOT NULL" + params: list[Any] = [] + if limit is not None: + sql += f" ORDER BY [{id_col}] OFFSET 0 ROWS FETCH NEXT ? ROWS ONLY" + params.append(int(limit)) + conn_str = _build_conn_str(db_cfg) + try: + with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn: + conn.timeout = db_cfg["query_timeout"] + cursor = conn.cursor() + cursor.execute(sql, params) + return [row[0] for row in cursor.fetchall()] + except pyodbc.Error as e: + raise DatabaseError(f"查询总排号失败: {e}") from e + + +def upsert_attachments( + db_cfg: dict[str, Any], + rows: list[tuple[str, str, str]], +) -> tuple[int, int]: + """幂等写入 Common.Attachment。 + + rows: 本次要写入的 (NS, MajorCategory, MinorCategory) 列表。 + 对出现的每个 NS 先 DELETE 其旧行，再批量 INSERT——保证重跑总是反映 + 最新分类结果，不会因唯一索引 (NS, 大类, 小类) 冲突而失败。 + + 返回 (deleted_rows, inserted_rows) 计数。 + """ + if not rows: + return (0, 0) + + qualified = f"[{ATTACHMENT_SCHEMA}].[{ATTACHMENT_TABLE}]" + distinct_ns = sorted({r[0] for r in rows}) + + conn_str = _build_conn_str(db_cfg) + try: + with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn: + conn.timeout = db_cfg["query_timeout"] + cursor = conn.cursor() + # 1) 删除本批所有 NS 的旧行 + del_ph = ",".join("?" for _ in distinct_ns) + cursor.execute( + f"DELETE FROM {qualified} WHERE [NS] IN ({del_ph})", distinct_ns + ) + deleted = cursor.rowcount + # 2) 插入新行 + cursor.executemany( + f"INSERT INTO {qualified} ([NS], [MajorCategory], [MinorCategory]) " + f"VALUES (?, ?, ?)", + rows, + ) + inserted = len(rows) + conn.commit() + logger.info( + "upsert_attachments: 删除 %d 行, 插入 %d 行, 涉及 %d 个 NS", + deleted, inserted, len(distinct_ns), + ) + return (deleted, inserted) + except pyodbc.Error as e: + raise DatabaseError(f"写入附件表失败: {e}") from e diff --git a/llm_client.py b/llm_client.py index b4114f5..a291667 100644 --- a/llm_client.py +++ b/llm_client.py @@ -40,6 +40,51 @@ class LLMCallResult: messages: list[dict[str, str]] # 实际发送给模型的完整消息列表(含system+few-shot+用户) raw_response: str | None # 模型原始回复；调用失败时为 None error: str | None # 调用失败时的错误描述；成功时为 None + # 以下字段为性能/用量统计，默认 None（调用失败时无有效值）： + usage: dict[str, Any] | None = None # resp.usage 转成的 dict（含 prompt/completion/total + # tokens 及 DeepSeek 的 prompt_cache_hit/miss_tokens），失败时为 None + elapsed_ms: float | None = None # 单次成功调用的耗时（毫秒，含网络+推理），失败时为 None + model: str | None = None # 实际服务的模型名（来自 resp.model），失败时为 None + attempt: int | None = None # 本次返回对应的是第几次尝试（成功那次；全失败则为 max_retry） + + +def _extract_usage(resp: Any) -> dict[str, Any] | None: + """从 chat.completions 响应里稳健地取出 usage 字典。 + + DeepSeek 在标准 OpenAI usage 之上，额外返回 `prompt_cache_hit_tokens` / + `prompt_cache_miss_tokens`（与 prompt_tokens 同级）。openai SDK 未必为这些 + 厂商字段建模，所以先按属性访问，取不到再回退到 pydantic 的 model_dump() + （extra 字段通常会被带出来），两层都试，避免漏掉缓存统计。 + """ + if resp is None or getattr(resp, "usage", None) is None: + return None + u = resp.usage + out: dict[str, Any] = {} + for key in ( + "prompt_tokens", + "completion_tokens", + "total_tokens", + "prompt_cache_hit_tokens", + "prompt_cache_miss_tokens", + ): + val = getattr(u, key, None) + if val is not None: + out[key] = val + if not out: + try: + dumped = u.model_dump() # type: ignore[attr-defined] + except Exception: + dumped = {} + for key in ( + "prompt_tokens", + "completion_tokens", + "total_tokens", + "prompt_cache_hit_tokens", + "prompt_cache_miss_tokens", + ): + if key in dumped and dumped[key] is not None: + out[key] = dumped[key] + return out or None class LLMClient: @@ -96,6 +141,7 @@ class LLMClient: extra_body["thinking"] = {"type": "disabled"} for attempt in range(1, max_retry + 1): + call_start = time.perf_counter() try: resp = self._client.chat.completions.create( model=self._cfg["model"], @@ -114,7 +160,18 @@ class LLMClient: f"模型返回内容为空 (content={content!r})，" f"若模型支持思维链，可能是推理耗尽了max_tokens预算" ) - return LLMCallResult(messages=messages, raw_response=content, error=None) + # 统计：单次调用耗时 + token 用量 + 实际服务模型 + call_elapsed_ms = (time.perf_counter() - call_start) * 1000 + usage = _extract_usage(resp) + return LLMCallResult( + messages=messages, + raw_response=content, + error=None, + usage=usage, + elapsed_ms=call_elapsed_ms, + model=getattr(resp, "model", None), + attempt=attempt, + ) except (APIError, APITimeoutError) as e: last_err = e logger.warning( @@ -134,4 +191,8 @@ class LLMClient: messages=messages, raw_response=None, error=f"LLM 调用重试 {max_retry} 次后仍失败: {last_err}", + usage=None, + elapsed_ms=None, + model=None, + attempt=max_retry, ) diff --git a/main.py b/main.py index c038a92..2ef3764 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ python main.py --id 26B742 --pretty # 格式化输出JSON(默认单行紧凑) python main.py --id 26B742 --log-dir /tmp/logs # 覆盖配置文件里的日志目录 python main.py --id 26B742 --enable-other # 允许模型使用"其他"兜底类目 + python main.py --id 26B742 --summary # 额外在 stderr 打印本批运行汇总统计 分类模式(--mode): coarse (默认) - 只判断大类: 资料/配件/耗材(启用 --enable-other 时还有"其他") @@ -57,7 +58,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) -from classifier import classify_batch # noqa: E402 +from classifier import classify_batch, summarize_results # noqa: E402 from config_loader import ConfigError, load_config # noqa: E402 @@ -111,6 +112,11 @@ def main() -> None: help="允许模型使用'其他'兜底类目。不指定则使用配置文件 business.enable_other_category" "（缺省为关闭）；本开关只能从命令行打开，无法用命令行强制关闭配置文件里已打开的设置", ) + parser.add_argument( + "--summary", action="store_true", + help="在 stderr 额外打印本批运行的汇总统计（耗时/token/缓存命中率）；" + "stdout 仍只输出干净的 JSON Lines 结果，便于管道/重定向", + ) args = parser.parse_args() if not args.id and not args.ids_file: @@ -155,6 +161,11 @@ def main() -> None: for r in results: print(json.dumps(r, ensure_ascii=False)) + if args.summary: + summary = summarize_results(results) + print("--- 运行汇总 ---", file=sys.stderr) + print(json.dumps(summary, ensure_ascii=False, indent=2), file=sys.stderr) + if __name__ == "__main__": main() diff --git a/order_logger.py b/order_logger.py index 48d7654..d7b647a 100644 --- a/order_logger.py +++ b/order_logger.py @@ -103,6 +103,33 @@ class OrderLogger: for line in (call_result.raw_response or "").splitlines(): self._lines.append(f" {line}") + # 调用统计：模型名、单次调用耗时、token 用量与缓存命中情况。 + # 失败调用（未拿到 usage）时相应字段显示"无"，保证排查信息完整。 + self._lines.append("") + self._lines.append("[调用统计]") + self._lines.append(f" 模型: {call_result.model}") + if call_result.elapsed_ms is not None: + self._lines.append(f" 单次调用耗时: {call_result.elapsed_ms:.1f} ms") + else: + self._lines.append(" 单次调用耗时: 无（调用失败）") + u = call_result.usage + if isinstance(u, dict): + self._lines.append( + f" tokens: prompt={u.get('prompt_tokens')} " + f"completion={u.get('completion_tokens')} " + f"total={u.get('total_tokens')}" + ) + prompt = u.get("prompt_tokens") or 0 + hit = u.get("prompt_cache_hit_tokens") or 0 + rate = round(hit / prompt, 4) if prompt else 0.0 + self._lines.append( + f" 缓存: hit={u.get('prompt_cache_hit_tokens')} " + f"miss={u.get('prompt_cache_miss_tokens')} " + f"命中率={rate}" + ) + else: + self._lines.append(" 用量统计: 无（调用失败或未返回 usage）") + self._lines.append("") if format_error is not None: self._lines.append("[格式校验] 失败") diff --git a/write_attachments.py b/write_attachments.py new file mode 100644 index 0000000..7763123 --- /dev/null +++ b/write_attachments.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""基于源表总排号，调用 classify_batch 分类，将结果按约定写入 Common.Attachment。 + +落库约定（详见项目记忆 MEMORY.md，与 db.py 中的常量保持一致）： + - has_attachment=true (status=ok): + 每个 type 一行。fine 模式 type 形如 "大类:细分"，按首个冒号拆成 (大类, 小类)； + coarse 模式 type 是大类，小类统一填 '无'。 + - has_attachment=false (无附件): 写哨兵行 (NS, '无附件', '无')。 + - has_attachment=null (not_found / llm_error / db_error): 不写（属"无法确定/失败"）。 + +用法: + python write_attachments.py # 全量写入 + python write_attachments.py --limit 10 # 仅前 10 个总排号（测试） + python write_attachments.py --ids-file ids.txt # 指定总排号（文件，每行一个） + python write_attachments.py --id 26B742 # 单个总排号写入 + python write_attachments.py --id 26B742,26B743 # 多个总排号（逗号分隔）写入 + python write_attachments.py --mode coarse # 粗分类写入 + python write_attachments.py --dry-run # 只打印将写入的行，不落库 + python write_attachments.py --config other.yaml +""" +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) + +from classifier import classify_batch, summarize_results # noqa: E402 +from config_loader import ConfigError, load_config # noqa: E402 +from db import ( # noqa: E402 + ATTACHMENT_SCHEMA, + ATTACHMENT_TABLE, + DatabaseError, + SENTINEL_MAJOR, + SENTINEL_MINOR, + NO_MINOR, + fetch_all_ids, + upsert_attachments, +) + + +def convert_results_to_rows(results: list[dict], mode: str) -> tuple[list[tuple[str, str, str]], list[str]]: + """将 classify_batch 的结果列表转成可写入 Common.Attachment 的行。 + + 返回 (rows, skipped)： + rows - list[(NS, MajorCategory, MinorCategory)] + skipped- list[总排号]，has_attachment 为 None（not_found/失败）未写入的 + """ + rows: list[tuple[str, str, str]] = [] + skipped: list[str] = [] + for r in results: + zph = r["zong_pai_hao"] + has = r["has_attachment"] + if has is None: + skipped.append(zph) + continue + if has is True: + for t in r["types"]: + t = t.strip() + if not t: + continue + if mode == "fine" and ":" in t: + major, minor = t.split(":", 1) + major = major.strip() + minor = minor.strip() or NO_MINOR + else: + major = t + minor = NO_MINOR + rows.append((zph, major, minor)) + else: # has is False -> 无附件 + rows.append((zph, SENTINEL_MAJOR, SENTINEL_MINOR)) + return rows, skipped + + +def _parse_ids(args: argparse.Namespace) -> list[str]: + """从 --id（逗号分隔，单个或多个）或 --ids-file 解析出总排号列表，去除空白项。 + + 两个来源可同时提供，合并后返回（顺序：先 --id，后 --ids-file）。 + --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="将订单附件分类结果写入 Common.Attachment", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument("--config", type=str, default="config.yaml") + parser.add_argument( + "--id", type=str, default=None, + help="总排号，单个或逗号分隔的多个，例如: 26B742 或 26B742,26B743；" + "提供后直接按指定总排号写入，不再全表扫描（--limit 此时无效）", + ) + parser.add_argument( + "--limit", type=int, default=None, + help="仅处理前 N 个总排号（按总排号排序），测试用；与 --id/--ids-file 互斥", + ) + parser.add_argument( + "--ids-file", type=str, default=None, + help="包含总排号的文本文件路径，每行一个总排号；可单独使用或与 --id 合并", + ) + parser.add_argument( + "--mode", type=str, default=None, choices=["coarse", "fine"], + help="分类粒度，不指定则用配置文件 business.default_mode", + ) + parser.add_argument("--log-dir", type=str, default=None) + parser.add_argument("--enable-other", action="store_true") + parser.add_argument( + "--dry-run", action="store_true", + help="只打印将要写入/跳过的统计，不连接目标表写入", + ) + args = parser.parse_args() + + 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, + ) + logger = logging.getLogger("write_attachments") + + 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) + ) + + # 1) 取总排号列表 + if args.id or args.ids_file: + ids = _parse_ids(args) + if not ids: + print("[错误] --id / --ids-file 未解析到任何有效的总排号", file=sys.stderr) + sys.exit(1) + logger.info("从 --id / --ids-file 读取 %d 个总排号", len(ids)) + else: + ids = fetch_all_ids(cfg["database"], limit=args.limit) + logger.info( + "源表读取 %d 个总排号%s", len(ids), + f"（limit={args.limit}）" if args.limit else "", + ) + + if not ids: + print("[提示] 没有可处理的总排号，退出") + return + + # 2) 分类 + logger.info("开始分类 (mode=%s, enable_other=%s)", mode, enable_other) + 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, + ) + + # 3) 转换 + rows, skipped = convert_results_to_rows(results, mode) + + # 统计各状态 + n_ok = sum(1 for r in results if r["status"] == "ok") + n_empty = sum(1 for r in results if r["status"] == "empty_param") + n_notfound = sum(1 for r in results if r["status"] == "not_found") + n_err = sum(1 for r in results if r["status"].endswith("error")) + + print( + f"[统计] 输入 {len(results)} 个总排号: " + f"ok={n_ok} empty_param={n_empty} not_found={n_notfound} error={n_err}" + ) + print( + f"[统计] 将写入 {len(rows)} 行, " + f"跳过(无附件判定=null) {len(skipped)} 个" + ) + + # 运行汇总：总耗时/token 总量/缓存命中率，从每条结果的 meta 累加。 + # 走 stdout 与上面的 [统计] 一致；不属于落库动作本身，仅作诊断输出。 + summary = summarize_results(results) + print( + f"[汇总] 总耗时 {summary['total_elapsed_ms']} ms, " + f"LLM 调用 {summary['llm_calls']} 次, " + f"token: prompt={summary['total_prompt_tokens']} " + f"completion={summary['total_completion_tokens']} " + f"total={summary['total_tokens']}, " + f"缓存命中率 {summary['avg_cache_hit_rate']} " + f"(hit={summary['total_cache_hit_tokens']} " + f"miss={summary['total_cache_miss_tokens']})" + ) + + if args.dry_run: + print("[dry-run] 不落库。示例行（前 10）:") + for row in rows[:10]: + print(" ", row) + return + + # 4) 写入 + target = f"{ATTACHMENT_SCHEMA}.{ATTACHMENT_TABLE}" + try: + deleted, inserted = upsert_attachments(cfg["database"], rows) + print(f"[完成] 已写入 {target}: 删除旧行 {deleted}, 插入新行 {inserted}") + except DatabaseError as e: + print(f"[写入失败] {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main()