尧图网页设计 
某天上午领导临时通知下午三点前要交一份销售汇总分区域、分销售、分产品还要把异常订单单独标出来。很多人的第一反应是打开 Excel 手工筛选、复制、求和再凭经验找几条金额异常的数据。真正做过的人都知道这项工作本身并不难难的是明细数据不干净、统计口径不统一、时间又紧。这个场景非常适合用 AI 辅助完成让代码先做确定性的读取、清洗和聚合让大模型负责生成汇总口径、识别并解释异常记录最后输出一份带汇总表和异常清单的 Excel 文件。本文会完整演示这样一套小工具的开发过程包括环境准备、核心代码、运行验证、常见问题排查和工程化建议读者可以照着实现也可以把思路迁移到月度经营分析、库存异常检查等场景。1. 销售汇总的真正难点数据杂、口径乱、时间紧1.1 手工汇总为什么慢慢在哪里大多数销售明细表会包含订单号、销售日期、区域、销售员、客户、产品、数量、单价、金额等字段。表面上看用 Excel 数据透视表几分钟就能完成统计但真实业务中至少有三类问题会让手工汇总速度明显下降。第一类是数据不干净。日期列存在2024/6/1、2024-06-01、2024.06.01混用金额列出现文本或负数客户名称留有空格甚至整行记录缺失。数据透视表遇到脏数据会直接报错或者合计结果明显不对操作者只能逐行排查。第二类是统计口径不一致。领导可能要求“按区域和销售员双维度汇总”也可能要求“把退货订单剔除”“不计入未回款的订单”“单笔超过 50 万的单独列出来”。每次需求变化手工操作都要重新调整筛选条件容易遗漏。第三类是时间紧。下午就要数据时往往没有时间慢慢核对。手工做出来的结果如果被人问一句“这个数怎么算出来的”回答不上来就很容易返工。1.2 AI 在销售汇总链路中的边界在动手设计工具之前先要明确 AI 在这里承担什么角色。大模型擅长的是语言理解、结构化文本生成、按规则描述异常它并不擅长对大规模数据做精确计算。如果把十万行明细直接丢给模型让它“计算一下总销售额”结果不可控算错了也没有可靠的校验方式。更合理的方式是把任务拆成两层数据层由代码做确定性强的事情包括读取文件、字段校验、缺失值检查、重复值检查、分组聚合、金额求和。判断层由大模型做需要业务经验的事情包括把聚合结果整理成易读的汇总说明、根据预设阈值筛选重点异常、为异常记录补充可读的原因描述。这样分工之后AI 不负责算数只负责“看数说话”。算数的部分由 pandas 完成结果可复现、可核对AI 输出的部分用于辅助决策即使某个判断不够准确也不会导致基础数据错误。1.3 一条可行的自动化处理主线整套流程按下面这条主线设计读取销售明细文件只保留需要的列统一日期格式。用代码做确定性校验找出日期缺失、金额为负、订单号重复等异常候选。对清洗后的明细按指定维度分组聚合得到汇总数据。把汇总数据和异常候选整理成紧凑的 JSON 结构放入提示词。调用大模型 API让模型输出结构化结果包括汇总分组说明和异常清单。将结果写入 Excel同时生成一个 Markdown 摘要方便直接粘贴进消息。这个流程的好处是数据问题有迹可循每个人的判断都记录在案领导追问时可以解释每一步的逻辑。2. 环境准备与项目结构设计2.1 技术选型pandas 做计算大模型 API 做判断示例采用 Python 作为主语言理由并不是 Python 本身有多特殊而是它在数据处理生态上最省事。pandas 可以处理 Excel、CSV 的读写和分组聚合openpyxl 负责 Excel 底层写入大模型 API 负责生成结构化判断结果。大模型调用部分推荐使用 OpenAI 兼容接口。当前很多大模型服务商都提供 OpenAI 兼容格式代码只需要关注base_url、api_key、model三个配置替换服务商时改动成本很低。下面的代码会以这种通用接口为例实际部署时替换成自己使用的模型名称和地址即可。2.2 环境依赖与版本确认建议使用 Python 3.10 或更高版本。依赖使用 pip 安装pip install pandas openpyxl openai python-dotenv安装完成后可以执行下面命令确认版本python -c import pandas; print(pandas.__version__) python -c import openai; print(openai.__version__)如果组织内部无法直接安装依赖可以考虑使用离线安装包或者把环境切换到 conda 等本地方案。这里要特别提醒不要只装了库就直接进入代码先确认 pandas 版本和 openpyxl 版本匹配。实际项目里遇到过 pandas 2.x 搭配老版本 openpyxl 写入 Excel 时样式异常的问题先统一版本能省很多事。2.3 项目目录与数据样例推荐按下面的目录结构组织项目sales_ai_summary/ ├── .env ├── config.json ├── data/ │ ├── sales_detail.xlsx │ └── output/ └── src/ ├── prompts.py └── summary_runner.py.env文件存放 API 密钥不进入版本库LLM_API_KEY你的密钥config.json存放模型配置和业务规则。下面是示例结构{ api: { base_url: https://your-llm-endpoint.example.com/v1, model: your-model-name, temperature: 0.1, max_tokens: 3000 }, fields: { order_id: 订单号, date: 销售日期, region: 区域, sales: 销售, customer: 客户, product: 产品, quantity: 数量, amount: 金额 }, group_by: [区域, 销售], thresholds: { amount_upper: 500000, amount_lower: 0, top_n: 5 } }参数说明如下参数含义建议配置调大影响调小影响temperature模型生成随机性0 到 0.2输出更发散异常描述更灵活输出更稳定但语言略显生硬max_tokens单次回答最多 token 数3000 左右能输出更多内容成本上升输出可能被截断JSON 不完整group_by汇总维度区域、销售、产品表格更细文件更大颗粒度粗信息量下降amount_upper单笔金额异常上限由业务定异常更少异常更多data/sales_detail.xlsx建议使用符合真实业务的样例数据至少包含下面这些列订单号 销售日期 区域 销售 客户 产品 数量 单价 金额 SO20240601001 2024-06-01 华东 张伟 客户A 办公椅 10 680 6800 SO20240601002 2024-06-01 华东 李娜 客户B 办公桌 5 1200 6000为了测试异常识别最好在样例中故意放几条问题数据例如日期为空、金额为负数、订单号重复、客户名称为空。3. 核心实现明细读取、汇总计算与 AI 异常识别3.1 用 pandas 读取销售明细先做确定性校验工具的第一步是读取文件。支持 Excel 和 CSV 两种格式根据文件后缀自动选择读取方式。import pandas as pd def load_details(file_path): if file_path.endswith(.csv): df pd.read_csv(file_path) else: df pd.read_excel(file_path) print(f[info] 读取明细 {len(df)} 条) return df读取之后不要急着发给 AI先做字段完整性检查和数据类型统一。def validate_details(df, cfg): fields cfg[fields] required_cols [fields[order_id], fields[date], fields[amount]] missing_cols [c for c in required_cols if c not in df.columns] if missing_cols: raise ValueError(f明细缺少必填列: {missing_cols}) df df.copy() # 日期统一解析无法解析的置为 NaT df[fields[date]] pd.to_datetime(df[fields[date]], errorscoerce) # 金额统一转成数值无法转换的置为 NaN df[fields[amount]] pd.to_numeric(df[fields[amount]], errorscoerce) anomalies [] date_col fields[date] amount_col fields[amount] order_col fields[order_id] customer_col fields[customer] upper cfg[thresholds][amount_upper] lower cfg[thresholds][amount_lower] date_bad df[df[date_col].isna()] for idx in date_bad.index: anomalies.append({ row: int(idx) 2, type: 日期无效, order_id: _safe_value(df, idx, order_col), reason: 销售日期为空或格式无法解析 }) amount_bad df[df[amount_col].isna()] for idx in amount_bad.index: anomalies.append({ row: int(idx) 2, type: 金额缺失, order_id: _safe_value(df, idx, order_col), reason: 金额为空或非数值 }) negative df[df[amount_col] lower] for idx in negative.index: anomalies.append({ row: int(idx) 2, type: 金额异常, order_id: _safe_value(df, idx, order_col), reason: f金额小于下限 {lower} }) high df[df[amount_col] upper] for idx in high.index: anomalies.append({ row: int(idx) 2, type: 金额偏高, order_id: _safe_value(df, idx, order_col), reason: f金额超过上限 {upper} }) dup_mask df.duplicated(subset[order_col], keepFalse) dup_df df[dup_mask].sort_values(order_col) for idx in dup_df.index: anomalies.append({ row: int(idx) 2, type: 订单重复, order_id: _safe_value(df, idx, order_col), reason: 同一订单号出现多次 }) if customer_col and customer_col in df.columns: empty_customer df[df[customer_col].isna() | (df[customer_col].astype(str).str.strip() )] for idx in empty_customer.index: anomalies.append({ row: int(idx) 2, type: 客户缺失, order_id: _safe_value(df, idx, order_col), reason: 客户名称为空 }) print(f[info] 确定性校验发现 {len(anomalies)} 条异常候选) return df, anomaliesdef _safe_value(df, idx, col): if col in df.columns and pd.notna(df.loc[idx, col]): return str(df.loc[idx, col]) return 这段代码背后有一个关键设计异常分两类。一类是绝对错误比如日期无法解析、金额为负另一类是业务重点关注比如单笔金额超过阈值、订单号重复。绝对错误后续要回归修正重点关注则会交给 AI 编排成可读的异常清单。3.2 构造提示词把数据切片传给模型很多第一次做 AI 汇总的人会陷入一个误区把整个明细 DataFrame 转成字符串塞进提示词希望模型自己找出所有问题。十万行数据对应的 token 量可能高达几百万既耗时又费钱模型也无法在超长上下文里稳定聚焦。正确做法是传给模型两样东西分组聚合后的汇总数据一般不超过 100 行。确定性校验发现的异常候选一般不超过几十条。这两样数据量很小模型可以快速做出判断。import json def build_prompt(df, anomalies, cfg): group_cols cfg[group_by] amount_col cfg[fields][amount] grouped ( df.groupby(group_cols, dropnaFalse)[amount_col] .agg([sum, count, mean]) .reset_index() ) grouped.columns group_cols [总金额, 订单数, 平均金额] grouped grouped.sort_values(总金额, ascendingFalse) # 控制传给模型的条数避免上下文过长 grouped_block grouped.head(100).to_json(orientrecords, force_asciiFalse) # 异常候选去掉 DataFrame 无关信息只保留可读内容 anomaly_block json.dumps(anomalies[:100], ensure_asciiFalse, indent2) user_prompt f 销售明细已经完成数据读取和初步清洗。以下内容用于辅助生成汇总说明和异常清单。 一、按 {group_cols} 汇总后的数据 {grouped_block} 二、代码检测出的异常候选 {anomaly_block} 请基于以上信息完成工作输出 JSON。 .strip() return user_prompt, grouped这个设计把“精确计算”和“智能解读”分离了。聚合结果由 pandas 计算模型只是站在聚合结果之上描述现象省 token结果也更可靠。系统提示词单独放在prompts.py中方便后续调优SYSTEM_PROMPT 你是一名销售运营分析助手。你会收到一份经过清洗的销售汇总数据和一份异常候选清单。 你的任务根据这些信息生成便于管理者阅读的汇总总结和异常清单。 要求 1. 只输出 JSON不要输出 JSON 以外的解释。 2. JSON 格式必须严格如下 { summary: [ { name: 华东-张伟, 总金额: 68200, 订单数: 12, 平均金额: 5683.33, 解读: 本组销售额最高主要贡献来源... } ], anomaly_summary: 整体数据中发现了3条需要关注的异常记录主要集中在日期和金额字段。, anomalies: [ { row: 5, type: 金额偏高, order_id: SO20240601005, reason: 单笔金额78万元明显高于同组平均水平 } ] } 3. summary 中的 name 用维度拼接例如 区域-销售。 4. anomaly_summary 不超过 120 字。 5. 如果没有任何异常anomalies 为空数组anomaly_summary 写“未发现明显异常”。 .strip()3.3 调用模型并解析结构化输出调用模型使用 OpenAI 兼容接口。密钥从环境变量读取避免硬编码到代码里。import os from openai import OpenAI def call_llm(system_prompt, user_prompt, cfg): client OpenAI( base_urlcfg[api][base_url], api_keyos.getenv(LLM_API_KEY), ) resp client.chat.completions.create( modelcfg[api][model], messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], temperaturecfg[api][temperature], max_tokenscfg[api][max_tokens], ) return resp.choices[0].message.content模型返回的内容不一定每次都是干净 JSON。常见表现是首尾多出 json 标记、解释性文字、甚至因为达到 max_tokens 被截断。解析时需要做容错处理。import json import re def parse_llm_json(text): if not text: raise ValueError(模型返回内容为空) text text.strip() # 去掉可能的 Markdown 代码块标记 text re.sub(r^(?:json)?|$, , text.strip()).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个 { 到最后一个 } 之间的内容 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: candidate text[start : end 1] try: return json.loads(candidate) except json.JSONDecodeError as e: raise ValueError(f模型输出不是合法 JSON原因为: {e}) from e raise注意解析容错只能解决边界标记问题解决不了模型真正算错的问题。因此summary 里的金额不要直接作为最终报表结果最终报表的合计金额必须以 pandas 聚合结果为准。3.4 生成汇总表 Excel 与异常清单拿到了模型输出之后把结果写入 Excel。这里有一个原则Excel 里的汇总数据使用 pandas 计算的精确结果模型的解读放在附加列中。这样既能保证数字准确又能保留 AI 的分析价值。def save_outputs(grouped, llm_data, anomalies, output_dir): import pathlib output_dir pathlib.Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) summary_text llm_data.get(summary, []) if summary_text: model_df pd.DataFrame(summary_text) # 合并先按 name 关联再把模型解读补到汇总表 group_keys list(grouped.columns[:len(grouped.columns)-3]) grouped[name] grouped[group_keys].astype(str).agg(-.join, axis1) merged grouped.merge( model_df[[name, 解读]], onname, howleft ) merged merged.drop(columns[name]) else: merged grouped anomaly_df pd.DataFrame(anomalies, columns[row, type, order_id, reason]) llm_anomalies llm_data.get(anomalies, []) if llm_anomalies: llm_anomaly_df pd.DataFrame(llm_anomalies) anomaly_df pd.concat([anomaly_df, llm_anomaly_df], ignore_indexTrue) sheet_name 汇总表 out_path output_dir / summary_output.xlsx with pd.ExcelWriter(out_path, engineopenpyxl) as writer: merged.to_excel(writer, sheet_namesheet_name, indexFalse) anomaly_df.to_excel(writer, sheet_name异常清单, indexFalse) md_lines [# 销售汇总摘要, ] md_lines.append(f### 汇总说明) md_lines.append(llm_data.get(anomaly_summary, )) md_lines.append() md_lines.append(### 分组汇总) for _, row in merged.head(10).iterrows(): md_lines.append(f- {row[group_keys[0]]}-{row[group_keys[1]]}: 订单数 {row[订单数]}, 总金额 {row[总金额]}) md_path output_dir / summary.md md_path.write_text(\n.join(md_lines), encodingutf-8) print(f[info] 输出汇总文件: {out_path}) print(f[info] 输出摘要文件: {md_path}) return str(out_path)这里要说明一下anomaly_df同时写入两类异常一类是代码确定性检测出来的另一类是模型根据业务经验补充的。两类的来源在导出时应保留type或备注字段区分避免后续误判责任方。实际生产项目中还可以加一列source值为rule或llm。4. 运行验证从命令行产出两份结果文件4.1 主入口脚本把上述函数串起来通过命令行参数控制输入输出。import argparse import json import os from dotenv import load_dotenv from prompts import SYSTEM_PROMPT def main(): load_dotenv() parser argparse.ArgumentParser(descriptionAI 销售汇总工具) parser.add_argument(--input, requiredTrue, help销售明细 Excel/CSV 路径) parser.add_argument(--config, defaultconfig.json, help配置文件路径) parser.add_argument(--output, defaultdata/output, help输出目录) args parser.parse_args() with open(args.config, r, encodingutf-8) as f: cfg json.load(f) df load_details(args.input) df, anomalies validate_details(df, cfg) user_prompt, grouped build_prompt(df, anomalies, cfg) llm_text call_llm(SYSTEM_PROMPT, user_prompt, cfg) llm_data parse_llm_json(llm_text) output_path save_outputs(grouped, llm_data, anomalies, args.output) print(f[done] {output_path}) if __name__ __main__: main()运行命令cd sales_ai_summary python src/summary_runner.py --input data/sales_detail.xlsx --config config.json --output data/output正常的输出大致如下[info] 读取明细 120 条 [info] 确定性校验发现 4 条异常候选 [info] 调用大模型接口完成 [info] 输出汇总文件: sales_ai_summary/data/output/summary_output.xlsx [info] 输出摘要文件: sales_ai_summary/data/output/summary.md [done] sales_ai_summary/data/output/summary_output.xlsx4.2 检查汇总表和异常清单打开 Excel 后至少检查三件事汇总表的总金额是否和明细列求和一致。可以新增一列用sum()交叉验证。异常清单里的row是否对应明细文件里的真实行号。注意 pandas 从 0 开始索引Excel 从 1 开始所以代码里做了2偏移其中 1 表示标题行1 表示索引从 0 到 Excel 行号从 1 的转换。模型生成的“解读”是否与数字矛盾。例如某组总金额最高解读却写成“金额较低”这种情况通常是提示词里没有说明排序规则需要补充。summary.md可以直接复制到 IM 工具里作为给领导的第一版汇报素材。4.3 边界场景验证除了正常数据建议用三类数据测试工具场景说明预期结果空文件Excel 只有表头代码提示读取明细 0 条不会调用模型字段缺失没有金额列抛出“明细缺少必填列”全量异常所有日期都解析失败异常清单很长模型会压缩成总体结论空文件场景需要在main()里加一个提前退出逻辑避免向模型发送空提示词。字段缺失场景依赖第 3.1 节的列校验。5. 必须掌握的排查链路AI 汇总脚本报错看哪里5.1 读不到文件或者 Excel 报错现象运行脚本后提示文件不存在或者Sheet2不存在或者编码报错。排查顺序确认文件路径是相对路径还是绝对路径当前工作目录是不是sales_ai_summary。确认文件扩展名和实际格式一致。有些同事把 CSV 内容另存为.xlsxpandas 会读取失败。确认 Excel 里是否存在多个 Sheet。默认读取第一个 Sheet如果目标数据在第二个 Sheet需要传入sheet_name。解决方案df pd.read_excel(file_path, sheet_name0) df pd.read_csv(file_path, encodingutf-8)Windows 环境下建议显式指定encodingutf-8否则 CSV 文件可能因为本地编码而乱码。5.2 模型返回内容无法解析现象脚本报模型输出不是合法 JSON或者输出的 Excel 里缺少某些字段。常见原因有三个模型没有遵守“只输出 JSON”的指令额外输出了解释性文字。max_tokens太小JSON 被截断。模型自身能力有限多行 JSON 里出现换行或转义错误。处理建议按顺序执行增加max_tokens例如从 3000 调到 5000。在系统提示词里追加“不要使用 Markdown 代码块包裹 JSON”。把temperature调到 0降低随机性。在parse_llm_json里增加日志输出保存原始响应便于分析。with open(data/last_llm_response.txt, w, encodingutf-8) as f: f.write(llm_text)5.3 汇总结果与明细对不上现象分组汇总的金额和 Excel 数据透视表不一致。优先检查数据清洗步骤。最常见原因是金额列里混入了文本例如6,800、6800元、6800 pd.to_numeric会把它们变成 NaN直接导致汇总金额偏小。需要先做文本替换df[amount_col] ( df[amount_col] .astype(str) .str.replace(,, , regexFalse) .str.replace(元, , regexFalse) .str.strip() ) df[amount_col] pd.to_numeric(df[amount_col], errorscoerce)如果样本里包含了被排除的订单也会导致对不上。建议在导出汇总表时额外保留“清洗前记录数”和“清洗后记录数”方便核对。5.4 API 调用超时或费用失控现象明细数据量一大脚本长时间不返回或者一次运行消耗了大量 token。这里有两个层面要注意时间层面API 调用需要增加超时配置client OpenAI( base_urlcfg[api][base_url], api_keyos.getenv(LLM_API_KEY), timeoutcfg[api].get(timeout, 60), )成本层面严格控制放入提示词的数据条数。grouped_block限制前 100 行异常候选限制前 100 条而不是把整个明细全部传进去。如果业务必须分析完整数据更好的做法是分页处理把 10 万条明细拆成多个批次每个批次只做局部汇总最后合并。不过对于销售汇总场景先做 pandas 聚合再传给模型通常已经足够。注意大模型 API 是按 token 计费的。提示词越长单次成本越高。不要让模型重复读取原始明细它并不需要这些原始记录也不需要重新计算总额。6. 让这套工具真正用得住的工程化建议6.1 提示词版本管理提示词是这套 AI 工具效果的核心但提示词变更非常频繁。建议把系统提示词单独放在prompts.py或独立文本文件中并在文件头部记录变更信息。# prompts.py - 版本 1.2 # 2025-01-15: 增加异常清单中保留 row 字段的约束 # 2025-01-16: 修复 summary.name 拼接规则改为 区域-销售每次修改提示词后至少用同一份样例数据回归一次对比新增输出和旧版输出。不要在生产环境中直接改提示词容易引入不可控变化。6.2 把确定性计算和生成式判断分离这套工具最核心的工程原则是能用规则解决的问题不要交给模型模型只做规则难以覆盖的判断和文本生成。在销售汇总场景里求和、计数、平均值、排序全部由 pandas 处理。日期格式、缺失值、负金额、重复订单号全部由规则检查。模型只负责“从多个异常候选中挑选最值得关注的几项”“把汇总结果改写成领导能看懂的话”。如果完全依赖模型输出汇总数字一旦模型“发挥不稳定”整个报表的数据可信度都会受影响。而数字一旦不可信AI 辅助办公这个工具就失去了价值。6.3 生产环境还需要补哪些能力学习和开发环境跑通只是开始真正部署到日常使用时建议补齐以下能力能力说明配置外置API 地址、模型名、阈值不要写死在代码里统一放配置中心或环境变量日志记录记录读取行数、异常候选数、API 请求耗时、输出路径权限校验销售明细可能包含敏感数据执行工具的人要有对应数据权限模型降级如果 API 调用失败至少保留 pandas 聚合结果不阻塞手工分析结果回滚每次输出保留带时间戳的目录方便对比不同版本数据脱敏如果模型服务是外部 API建议先脱敏客户名称和销售姓名学习环境可以全流程快速跑通但生产环境要把“模型不可用”当作常态设计。6.4 后续扩展定时任务、消息推送、Web 页面在沉淀成稳定工具后可以往三个方向扩展方向一接入定时任务。使用 cron 或 CI 计划任务每周定时读取最新明细自动生成汇总表发送到指定邮箱或 IM 群。方向二做成 Web 页面。上传 Excel 后自动调用处理流程页面展示汇总表和异常清单用户可以下载结果文件。需要额外增加文件上传安全校验和任务状态管理。方向三从“销售汇总”扩展到“经营日报”。同一套“明细读取 确定性校验 模型解读 结构化输出”的架构可以复用到库存异常、渠道对账、客户回款提醒等场景。只要把config.json里的字段映射和规则阈值换掉就能适配新的业务。对刚接触 AI 应用开发的读者建议不要一上来就写复杂平台先用这个项目跑通一条最小链路。把一个 Excel 明细变成汇总表和异常清单的过程中你会接触到数据清洗、提示词工程、结构化输出、容错解析、文件导出这些在实际项目中都会用到的能力。把一个场景做扎实比一次性堆很多功能更有价值。 AI辅助销售汇总：从数据清洗到自动化报告生成的完整实践
某天上午领导临时通知下午三点前要交一份销售汇总分区域、分销售、分产品还要把异常订单单独标出来。很多人的第一反应是打开 Excel 手工筛选、复制、求和再凭经验找几条金额异常的数据。真正做过的人都知道这项工作本身并不难难的是明细数据不干净、统计口径不统一、时间又紧。这个场景非常适合用 AI 辅助完成让代码先做确定性的读取、清洗和聚合让大模型负责生成汇总口径、识别并解释异常记录最后输出一份带汇总表和异常清单的 Excel 文件。本文会完整演示这样一套小工具的开发过程包括环境准备、核心代码、运行验证、常见问题排查和工程化建议读者可以照着实现也可以把思路迁移到月度经营分析、库存异常检查等场景。1. 销售汇总的真正难点数据杂、口径乱、时间紧1.1 手工汇总为什么慢慢在哪里大多数销售明细表会包含订单号、销售日期、区域、销售员、客户、产品、数量、单价、金额等字段。表面上看用 Excel 数据透视表几分钟就能完成统计但真实业务中至少有三类问题会让手工汇总速度明显下降。第一类是数据不干净。日期列存在2024/6/1、2024-06-01、2024.06.01混用金额列出现文本或负数客户名称留有空格甚至整行记录缺失。数据透视表遇到脏数据会直接报错或者合计结果明显不对操作者只能逐行排查。第二类是统计口径不一致。领导可能要求“按区域和销售员双维度汇总”也可能要求“把退货订单剔除”“不计入未回款的订单”“单笔超过 50 万的单独列出来”。每次需求变化手工操作都要重新调整筛选条件容易遗漏。第三类是时间紧。下午就要数据时往往没有时间慢慢核对。手工做出来的结果如果被人问一句“这个数怎么算出来的”回答不上来就很容易返工。1.2 AI 在销售汇总链路中的边界在动手设计工具之前先要明确 AI 在这里承担什么角色。大模型擅长的是语言理解、结构化文本生成、按规则描述异常它并不擅长对大规模数据做精确计算。如果把十万行明细直接丢给模型让它“计算一下总销售额”结果不可控算错了也没有可靠的校验方式。更合理的方式是把任务拆成两层数据层由代码做确定性强的事情包括读取文件、字段校验、缺失值检查、重复值检查、分组聚合、金额求和。判断层由大模型做需要业务经验的事情包括把聚合结果整理成易读的汇总说明、根据预设阈值筛选重点异常、为异常记录补充可读的原因描述。这样分工之后AI 不负责算数只负责“看数说话”。算数的部分由 pandas 完成结果可复现、可核对AI 输出的部分用于辅助决策即使某个判断不够准确也不会导致基础数据错误。1.3 一条可行的自动化处理主线整套流程按下面这条主线设计读取销售明细文件只保留需要的列统一日期格式。用代码做确定性校验找出日期缺失、金额为负、订单号重复等异常候选。对清洗后的明细按指定维度分组聚合得到汇总数据。把汇总数据和异常候选整理成紧凑的 JSON 结构放入提示词。调用大模型 API让模型输出结构化结果包括汇总分组说明和异常清单。将结果写入 Excel同时生成一个 Markdown 摘要方便直接粘贴进消息。这个流程的好处是数据问题有迹可循每个人的判断都记录在案领导追问时可以解释每一步的逻辑。2. 环境准备与项目结构设计2.1 技术选型pandas 做计算大模型 API 做判断示例采用 Python 作为主语言理由并不是 Python 本身有多特殊而是它在数据处理生态上最省事。pandas 可以处理 Excel、CSV 的读写和分组聚合openpyxl 负责 Excel 底层写入大模型 API 负责生成结构化判断结果。大模型调用部分推荐使用 OpenAI 兼容接口。当前很多大模型服务商都提供 OpenAI 兼容格式代码只需要关注base_url、api_key、model三个配置替换服务商时改动成本很低。下面的代码会以这种通用接口为例实际部署时替换成自己使用的模型名称和地址即可。2.2 环境依赖与版本确认建议使用 Python 3.10 或更高版本。依赖使用 pip 安装pip install pandas openpyxl openai python-dotenv安装完成后可以执行下面命令确认版本python -c import pandas; print(pandas.__version__) python -c import openai; print(openai.__version__)如果组织内部无法直接安装依赖可以考虑使用离线安装包或者把环境切换到 conda 等本地方案。这里要特别提醒不要只装了库就直接进入代码先确认 pandas 版本和 openpyxl 版本匹配。实际项目里遇到过 pandas 2.x 搭配老版本 openpyxl 写入 Excel 时样式异常的问题先统一版本能省很多事。2.3 项目目录与数据样例推荐按下面的目录结构组织项目sales_ai_summary/ ├── .env ├── config.json ├── data/ │ ├── sales_detail.xlsx │ └── output/ └── src/ ├── prompts.py └── summary_runner.py.env文件存放 API 密钥不进入版本库LLM_API_KEY你的密钥config.json存放模型配置和业务规则。下面是示例结构{ api: { base_url: https://your-llm-endpoint.example.com/v1, model: your-model-name, temperature: 0.1, max_tokens: 3000 }, fields: { order_id: 订单号, date: 销售日期, region: 区域, sales: 销售, customer: 客户, product: 产品, quantity: 数量, amount: 金额 }, group_by: [区域, 销售], thresholds: { amount_upper: 500000, amount_lower: 0, top_n: 5 } }参数说明如下参数含义建议配置调大影响调小影响temperature模型生成随机性0 到 0.2输出更发散异常描述更灵活输出更稳定但语言略显生硬max_tokens单次回答最多 token 数3000 左右能输出更多内容成本上升输出可能被截断JSON 不完整group_by汇总维度区域、销售、产品表格更细文件更大颗粒度粗信息量下降amount_upper单笔金额异常上限由业务定异常更少异常更多data/sales_detail.xlsx建议使用符合真实业务的样例数据至少包含下面这些列订单号 销售日期 区域 销售 客户 产品 数量 单价 金额 SO20240601001 2024-06-01 华东 张伟 客户A 办公椅 10 680 6800 SO20240601002 2024-06-01 华东 李娜 客户B 办公桌 5 1200 6000为了测试异常识别最好在样例中故意放几条问题数据例如日期为空、金额为负数、订单号重复、客户名称为空。3. 核心实现明细读取、汇总计算与 AI 异常识别3.1 用 pandas 读取销售明细先做确定性校验工具的第一步是读取文件。支持 Excel 和 CSV 两种格式根据文件后缀自动选择读取方式。import pandas as pd def load_details(file_path): if file_path.endswith(.csv): df pd.read_csv(file_path) else: df pd.read_excel(file_path) print(f[info] 读取明细 {len(df)} 条) return df读取之后不要急着发给 AI先做字段完整性检查和数据类型统一。def validate_details(df, cfg): fields cfg[fields] required_cols [fields[order_id], fields[date], fields[amount]] missing_cols [c for c in required_cols if c not in df.columns] if missing_cols: raise ValueError(f明细缺少必填列: {missing_cols}) df df.copy() # 日期统一解析无法解析的置为 NaT df[fields[date]] pd.to_datetime(df[fields[date]], errorscoerce) # 金额统一转成数值无法转换的置为 NaN df[fields[amount]] pd.to_numeric(df[fields[amount]], errorscoerce) anomalies [] date_col fields[date] amount_col fields[amount] order_col fields[order_id] customer_col fields[customer] upper cfg[thresholds][amount_upper] lower cfg[thresholds][amount_lower] date_bad df[df[date_col].isna()] for idx in date_bad.index: anomalies.append({ row: int(idx) 2, type: 日期无效, order_id: _safe_value(df, idx, order_col), reason: 销售日期为空或格式无法解析 }) amount_bad df[df[amount_col].isna()] for idx in amount_bad.index: anomalies.append({ row: int(idx) 2, type: 金额缺失, order_id: _safe_value(df, idx, order_col), reason: 金额为空或非数值 }) negative df[df[amount_col] lower] for idx in negative.index: anomalies.append({ row: int(idx) 2, type: 金额异常, order_id: _safe_value(df, idx, order_col), reason: f金额小于下限 {lower} }) high df[df[amount_col] upper] for idx in high.index: anomalies.append({ row: int(idx) 2, type: 金额偏高, order_id: _safe_value(df, idx, order_col), reason: f金额超过上限 {upper} }) dup_mask df.duplicated(subset[order_col], keepFalse) dup_df df[dup_mask].sort_values(order_col) for idx in dup_df.index: anomalies.append({ row: int(idx) 2, type: 订单重复, order_id: _safe_value(df, idx, order_col), reason: 同一订单号出现多次 }) if customer_col and customer_col in df.columns: empty_customer df[df[customer_col].isna() | (df[customer_col].astype(str).str.strip() )] for idx in empty_customer.index: anomalies.append({ row: int(idx) 2, type: 客户缺失, order_id: _safe_value(df, idx, order_col), reason: 客户名称为空 }) print(f[info] 确定性校验发现 {len(anomalies)} 条异常候选) return df, anomaliesdef _safe_value(df, idx, col): if col in df.columns and pd.notna(df.loc[idx, col]): return str(df.loc[idx, col]) return 这段代码背后有一个关键设计异常分两类。一类是绝对错误比如日期无法解析、金额为负另一类是业务重点关注比如单笔金额超过阈值、订单号重复。绝对错误后续要回归修正重点关注则会交给 AI 编排成可读的异常清单。3.2 构造提示词把数据切片传给模型很多第一次做 AI 汇总的人会陷入一个误区把整个明细 DataFrame 转成字符串塞进提示词希望模型自己找出所有问题。十万行数据对应的 token 量可能高达几百万既耗时又费钱模型也无法在超长上下文里稳定聚焦。正确做法是传给模型两样东西分组聚合后的汇总数据一般不超过 100 行。确定性校验发现的异常候选一般不超过几十条。这两样数据量很小模型可以快速做出判断。import json def build_prompt(df, anomalies, cfg): group_cols cfg[group_by] amount_col cfg[fields][amount] grouped ( df.groupby(group_cols, dropnaFalse)[amount_col] .agg([sum, count, mean]) .reset_index() ) grouped.columns group_cols [总金额, 订单数, 平均金额] grouped grouped.sort_values(总金额, ascendingFalse) # 控制传给模型的条数避免上下文过长 grouped_block grouped.head(100).to_json(orientrecords, force_asciiFalse) # 异常候选去掉 DataFrame 无关信息只保留可读内容 anomaly_block json.dumps(anomalies[:100], ensure_asciiFalse, indent2) user_prompt f 销售明细已经完成数据读取和初步清洗。以下内容用于辅助生成汇总说明和异常清单。 一、按 {group_cols} 汇总后的数据 {grouped_block} 二、代码检测出的异常候选 {anomaly_block} 请基于以上信息完成工作输出 JSON。 .strip() return user_prompt, grouped这个设计把“精确计算”和“智能解读”分离了。聚合结果由 pandas 计算模型只是站在聚合结果之上描述现象省 token结果也更可靠。系统提示词单独放在prompts.py中方便后续调优SYSTEM_PROMPT 你是一名销售运营分析助手。你会收到一份经过清洗的销售汇总数据和一份异常候选清单。 你的任务根据这些信息生成便于管理者阅读的汇总总结和异常清单。 要求 1. 只输出 JSON不要输出 JSON 以外的解释。 2. JSON 格式必须严格如下 { summary: [ { name: 华东-张伟, 总金额: 68200, 订单数: 12, 平均金额: 5683.33, 解读: 本组销售额最高主要贡献来源... } ], anomaly_summary: 整体数据中发现了3条需要关注的异常记录主要集中在日期和金额字段。, anomalies: [ { row: 5, type: 金额偏高, order_id: SO20240601005, reason: 单笔金额78万元明显高于同组平均水平 } ] } 3. summary 中的 name 用维度拼接例如 区域-销售。 4. anomaly_summary 不超过 120 字。 5. 如果没有任何异常anomalies 为空数组anomaly_summary 写“未发现明显异常”。 .strip()3.3 调用模型并解析结构化输出调用模型使用 OpenAI 兼容接口。密钥从环境变量读取避免硬编码到代码里。import os from openai import OpenAI def call_llm(system_prompt, user_prompt, cfg): client OpenAI( base_urlcfg[api][base_url], api_keyos.getenv(LLM_API_KEY), ) resp client.chat.completions.create( modelcfg[api][model], messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], temperaturecfg[api][temperature], max_tokenscfg[api][max_tokens], ) return resp.choices[0].message.content模型返回的内容不一定每次都是干净 JSON。常见表现是首尾多出 json 标记、解释性文字、甚至因为达到 max_tokens 被截断。解析时需要做容错处理。import json import re def parse_llm_json(text): if not text: raise ValueError(模型返回内容为空) text text.strip() # 去掉可能的 Markdown 代码块标记 text re.sub(r^(?:json)?|$, , text.strip()).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个 { 到最后一个 } 之间的内容 start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: candidate text[start : end 1] try: return json.loads(candidate) except json.JSONDecodeError as e: raise ValueError(f模型输出不是合法 JSON原因为: {e}) from e raise注意解析容错只能解决边界标记问题解决不了模型真正算错的问题。因此summary 里的金额不要直接作为最终报表结果最终报表的合计金额必须以 pandas 聚合结果为准。3.4 生成汇总表 Excel 与异常清单拿到了模型输出之后把结果写入 Excel。这里有一个原则Excel 里的汇总数据使用 pandas 计算的精确结果模型的解读放在附加列中。这样既能保证数字准确又能保留 AI 的分析价值。def save_outputs(grouped, llm_data, anomalies, output_dir): import pathlib output_dir pathlib.Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) summary_text llm_data.get(summary, []) if summary_text: model_df pd.DataFrame(summary_text) # 合并先按 name 关联再把模型解读补到汇总表 group_keys list(grouped.columns[:len(grouped.columns)-3]) grouped[name] grouped[group_keys].astype(str).agg(-.join, axis1) merged grouped.merge( model_df[[name, 解读]], onname, howleft ) merged merged.drop(columns[name]) else: merged grouped anomaly_df pd.DataFrame(anomalies, columns[row, type, order_id, reason]) llm_anomalies llm_data.get(anomalies, []) if llm_anomalies: llm_anomaly_df pd.DataFrame(llm_anomalies) anomaly_df pd.concat([anomaly_df, llm_anomaly_df], ignore_indexTrue) sheet_name 汇总表 out_path output_dir / summary_output.xlsx with pd.ExcelWriter(out_path, engineopenpyxl) as writer: merged.to_excel(writer, sheet_namesheet_name, indexFalse) anomaly_df.to_excel(writer, sheet_name异常清单, indexFalse) md_lines [# 销售汇总摘要, ] md_lines.append(f### 汇总说明) md_lines.append(llm_data.get(anomaly_summary, )) md_lines.append() md_lines.append(### 分组汇总) for _, row in merged.head(10).iterrows(): md_lines.append(f- {row[group_keys[0]]}-{row[group_keys[1]]}: 订单数 {row[订单数]}, 总金额 {row[总金额]}) md_path output_dir / summary.md md_path.write_text(\n.join(md_lines), encodingutf-8) print(f[info] 输出汇总文件: {out_path}) print(f[info] 输出摘要文件: {md_path}) return str(out_path)这里要说明一下anomaly_df同时写入两类异常一类是代码确定性检测出来的另一类是模型根据业务经验补充的。两类的来源在导出时应保留type或备注字段区分避免后续误判责任方。实际生产项目中还可以加一列source值为rule或llm。4. 运行验证从命令行产出两份结果文件4.1 主入口脚本把上述函数串起来通过命令行参数控制输入输出。import argparse import json import os from dotenv import load_dotenv from prompts import SYSTEM_PROMPT def main(): load_dotenv() parser argparse.ArgumentParser(descriptionAI 销售汇总工具) parser.add_argument(--input, requiredTrue, help销售明细 Excel/CSV 路径) parser.add_argument(--config, defaultconfig.json, help配置文件路径) parser.add_argument(--output, defaultdata/output, help输出目录) args parser.parse_args() with open(args.config, r, encodingutf-8) as f: cfg json.load(f) df load_details(args.input) df, anomalies validate_details(df, cfg) user_prompt, grouped build_prompt(df, anomalies, cfg) llm_text call_llm(SYSTEM_PROMPT, user_prompt, cfg) llm_data parse_llm_json(llm_text) output_path save_outputs(grouped, llm_data, anomalies, args.output) print(f[done] {output_path}) if __name__ __main__: main()运行命令cd sales_ai_summary python src/summary_runner.py --input data/sales_detail.xlsx --config config.json --output data/output正常的输出大致如下[info] 读取明细 120 条 [info] 确定性校验发现 4 条异常候选 [info] 调用大模型接口完成 [info] 输出汇总文件: sales_ai_summary/data/output/summary_output.xlsx [info] 输出摘要文件: sales_ai_summary/data/output/summary.md [done] sales_ai_summary/data/output/summary_output.xlsx4.2 检查汇总表和异常清单打开 Excel 后至少检查三件事汇总表的总金额是否和明细列求和一致。可以新增一列用sum()交叉验证。异常清单里的row是否对应明细文件里的真实行号。注意 pandas 从 0 开始索引Excel 从 1 开始所以代码里做了2偏移其中 1 表示标题行1 表示索引从 0 到 Excel 行号从 1 的转换。模型生成的“解读”是否与数字矛盾。例如某组总金额最高解读却写成“金额较低”这种情况通常是提示词里没有说明排序规则需要补充。summary.md可以直接复制到 IM 工具里作为给领导的第一版汇报素材。4.3 边界场景验证除了正常数据建议用三类数据测试工具场景说明预期结果空文件Excel 只有表头代码提示读取明细 0 条不会调用模型字段缺失没有金额列抛出“明细缺少必填列”全量异常所有日期都解析失败异常清单很长模型会压缩成总体结论空文件场景需要在main()里加一个提前退出逻辑避免向模型发送空提示词。字段缺失场景依赖第 3.1 节的列校验。5. 必须掌握的排查链路AI 汇总脚本报错看哪里5.1 读不到文件或者 Excel 报错现象运行脚本后提示文件不存在或者Sheet2不存在或者编码报错。排查顺序确认文件路径是相对路径还是绝对路径当前工作目录是不是sales_ai_summary。确认文件扩展名和实际格式一致。有些同事把 CSV 内容另存为.xlsxpandas 会读取失败。确认 Excel 里是否存在多个 Sheet。默认读取第一个 Sheet如果目标数据在第二个 Sheet需要传入sheet_name。解决方案df pd.read_excel(file_path, sheet_name0) df pd.read_csv(file_path, encodingutf-8)Windows 环境下建议显式指定encodingutf-8否则 CSV 文件可能因为本地编码而乱码。5.2 模型返回内容无法解析现象脚本报模型输出不是合法 JSON或者输出的 Excel 里缺少某些字段。常见原因有三个模型没有遵守“只输出 JSON”的指令额外输出了解释性文字。max_tokens太小JSON 被截断。模型自身能力有限多行 JSON 里出现换行或转义错误。处理建议按顺序执行增加max_tokens例如从 3000 调到 5000。在系统提示词里追加“不要使用 Markdown 代码块包裹 JSON”。把temperature调到 0降低随机性。在parse_llm_json里增加日志输出保存原始响应便于分析。with open(data/last_llm_response.txt, w, encodingutf-8) as f: f.write(llm_text)5.3 汇总结果与明细对不上现象分组汇总的金额和 Excel 数据透视表不一致。优先检查数据清洗步骤。最常见原因是金额列里混入了文本例如6,800、6800元、6800 pd.to_numeric会把它们变成 NaN直接导致汇总金额偏小。需要先做文本替换df[amount_col] ( df[amount_col] .astype(str) .str.replace(,, , regexFalse) .str.replace(元, , regexFalse) .str.strip() ) df[amount_col] pd.to_numeric(df[amount_col], errorscoerce)如果样本里包含了被排除的订单也会导致对不上。建议在导出汇总表时额外保留“清洗前记录数”和“清洗后记录数”方便核对。5.4 API 调用超时或费用失控现象明细数据量一大脚本长时间不返回或者一次运行消耗了大量 token。这里有两个层面要注意时间层面API 调用需要增加超时配置client OpenAI( base_urlcfg[api][base_url], api_keyos.getenv(LLM_API_KEY), timeoutcfg[api].get(timeout, 60), )成本层面严格控制放入提示词的数据条数。grouped_block限制前 100 行异常候选限制前 100 条而不是把整个明细全部传进去。如果业务必须分析完整数据更好的做法是分页处理把 10 万条明细拆成多个批次每个批次只做局部汇总最后合并。不过对于销售汇总场景先做 pandas 聚合再传给模型通常已经足够。注意大模型 API 是按 token 计费的。提示词越长单次成本越高。不要让模型重复读取原始明细它并不需要这些原始记录也不需要重新计算总额。6. 让这套工具真正用得住的工程化建议6.1 提示词版本管理提示词是这套 AI 工具效果的核心但提示词变更非常频繁。建议把系统提示词单独放在prompts.py或独立文本文件中并在文件头部记录变更信息。# prompts.py - 版本 1.2 # 2025-01-15: 增加异常清单中保留 row 字段的约束 # 2025-01-16: 修复 summary.name 拼接规则改为 区域-销售每次修改提示词后至少用同一份样例数据回归一次对比新增输出和旧版输出。不要在生产环境中直接改提示词容易引入不可控变化。6.2 把确定性计算和生成式判断分离这套工具最核心的工程原则是能用规则解决的问题不要交给模型模型只做规则难以覆盖的判断和文本生成。在销售汇总场景里求和、计数、平均值、排序全部由 pandas 处理。日期格式、缺失值、负金额、重复订单号全部由规则检查。模型只负责“从多个异常候选中挑选最值得关注的几项”“把汇总结果改写成领导能看懂的话”。如果完全依赖模型输出汇总数字一旦模型“发挥不稳定”整个报表的数据可信度都会受影响。而数字一旦不可信AI 辅助办公这个工具就失去了价值。6.3 生产环境还需要补哪些能力学习和开发环境跑通只是开始真正部署到日常使用时建议补齐以下能力能力说明配置外置API 地址、模型名、阈值不要写死在代码里统一放配置中心或环境变量日志记录记录读取行数、异常候选数、API 请求耗时、输出路径权限校验销售明细可能包含敏感数据执行工具的人要有对应数据权限模型降级如果 API 调用失败至少保留 pandas 聚合结果不阻塞手工分析结果回滚每次输出保留带时间戳的目录方便对比不同版本数据脱敏如果模型服务是外部 API建议先脱敏客户名称和销售姓名学习环境可以全流程快速跑通但生产环境要把“模型不可用”当作常态设计。6.4 后续扩展定时任务、消息推送、Web 页面在沉淀成稳定工具后可以往三个方向扩展方向一接入定时任务。使用 cron 或 CI 计划任务每周定时读取最新明细自动生成汇总表发送到指定邮箱或 IM 群。方向二做成 Web 页面。上传 Excel 后自动调用处理流程页面展示汇总表和异常清单用户可以下载结果文件。需要额外增加文件上传安全校验和任务状态管理。方向三从“销售汇总”扩展到“经营日报”。同一套“明细读取 确定性校验 模型解读 结构化输出”的架构可以复用到库存异常、渠道对账、客户回款提醒等场景。只要把config.json里的字段映射和规则阈值换掉就能适配新的业务。对刚接触 AI 应用开发的读者建议不要一上来就写复杂平台先用这个项目跑通一条最小链路。把一个 Excel 明细变成汇总表和异常清单的过程中你会接触到数据清洗、提示词工程、结构化输出、容错解析、文件导出这些在实际项目中都会用到的能力。把一个场景做扎实比一次性堆很多功能更有价值。