资讯动态

大模型+ pandas 实现销售明细自动汇总与异常检测

发布时间:2026/9/7 8:55:32 来源:尧图企业网站定制
下午要交销售汇总明细表还乱成一锅粥靠手工透视表加肉眼找异常时间根本不够用。这次我们直接换一个思路用大模型 AI 把销售明细自动拆成“汇总表 异常清单”明细进去结果出来中间不需要你在 Excel 里反复拖字段。这个方案的重点不是概念多复杂而是能不能在普通办公电脑上跑通。它的核心能力可以归纳成五点输入只要是 CSV 或 Excel 格式的销售明细就能自动生成按区域、按产品、按销售员的汇总统计大模型负责解读表格结构和生成汇总规则pandas 负责计算结果稳定可控自动识别异常数据比如负金额、空客户、异常折扣、销量暴涨暴跌汇总表和异常清单以 Excel 文件输出直接可以发给领导支持批量任务几十个明细文件可以排队处理也支持通过 API 接入现有办公系统。本文会带读者完成从环境准备、数据读取、大模型调用、汇总生成、异常检测到批量输出的完整流程全部用可复现代码演示。适合做销售运营、数据分析、财务对账以及所有经常被“领导下午就要”这类需求追着跑的人。1. 销售汇总 AI 方案核心能力速览能力项说明输入格式CSV、Excelxlsx / xls输出格式汇总表 Excel 异常清单 Excel核心功能按维度汇总、异常检测、AI 生成分析摘要运行环境Windows / macOS / LinuxPython 3.9硬件要求普通办公电脑即可无需 GPU调用模型主流大模型 API如通义千问、DeepSeek、Kimi 等是否支持 API支持可封装成 HTTP 接口是否支持批量任务支持按目录批量处理适合场景销售周报、月度汇总、区域业绩统计、异常订单排查使用边界涉及客户信息、金额、内部经营数据时需脱敏并确认授权从材料看这个方案最有价值的地方不是“让 AI 替代 Excel”而是把 AI 放在“规则理解”和“异常解释”这两个环节计算本身还是交给 pandas避免大模型算错数。2. 适用场景与使用边界2.1 适合谁运营和销售助理每天或每周需要整理销售明细输出汇总表格。数据分析师处理多来源、多格式的销售数据时先做快速汇总和异常筛查。财务人员核对订单金额、折扣、退款记录找异常条目。中小企业管理者没有专门 BI 系统只能靠 Excel 手工汇总。2.2 能解决什么问题手工透视表操作慢尤其字段多、数据量大时。肉眼找异常数据不全面容易漏掉负金额、空值、超低折扣。汇总口径不统一每个人做的表都不一样。领导临时要数据来不及做完整分析。2.3 不适合什么场景数据量极大千万行以上时建议先用数据库或专业 BI 工具。对数据准确性要求极度严格的财务审计场景AI 只能辅助不能替代人工复核。涉及核心商业机密且不允许外发到第三方 API 的环境需要私有化部署模型。2.4 合规与安全边界销售明细通常包含客户名称、联系方式、产品价格、销售额等敏感信息。使用 AI 工具时要注意调用云端大模型 API 前对客户姓名、手机号、地址等敏感字段进行脱敏。获取明确的内部数据使用授权不要私自把公司数据传到未授权的平台。不要用真实数据做公开测试。最终报表发布前必须人工复核确认汇总口径正确。3. 环境准备与前置条件3.1 软件环境Python 3.9 或更高版本pip 包管理工具一个可调用的 LLM API通义千问、DeepSeek、Kimi 等均可建议创建独立虚拟环境避免依赖冲突。python -m venv sales_envWindows 激活sales_env\Scripts\activatemacOS / Linux 激活source sales_env/bin/activate3.2 安装依赖pip install pandas openpyxl requestspandas处理数据表做分组汇总。openpyxl生成 Excel 文件。requests调用大模型 API。3.3 API Key 准备以通义千问、DeepSeek、Kimi 为例你只需要去对应的开放平台创建一个 API Key并确认账户有可用额度。由于不同平台的接口地址和模型名称不同建议把 Key 和接口地址写到环境变量里不要硬编码到脚本中。Windows PowerShell 设置环境变量$env:LLM_API_KEY 你的API Key $env:LLM_BASE_URL https://api.example.com/v1 $env:LLM_MODEL your-model-namemacOS / Linux 设置环境变量export LLM_API_KEY你的API Key export LLM_BASE_URLhttps://api.example.com/v1 export LLM_MODELyour-model-name这里不指定具体平台是因为各家 API 的模型名称、请求格式略有差异代码中使用兼容 OpenAI Chat Completions 风格的接口通用于大多数平台。4. 销售明细自动汇总实现思路整体流程分四个步骤读取销售明细文件对数据做基础清洗让大模型识别表结构并生成汇总规则用 pandas 执行汇总并检测异常输出 Excel。4.1 读取销售明细先定义一个读取函数支持 CSV 和 Excel。import pandas as pd from pathlib import Path def load_sales_data(file_path): file_path Path(file_path) suffix file_path.suffix.lower() if suffix .csv: # 尝试常见编码保证中文不乱码 for encoding in [utf-8, gbk, gb18030]: try: return pd.read_csv(file_path, encodingencoding) except UnicodeDecodeError: continue raise ValueError(f无法识别CSV编码: {file_path}) elif suffix in [.xlsx, .xls]: return pd.read_excel(file_path) else: raise ValueError(f不支持的文件格式: {suffix})4.2 数据基础清洗销售明细常见的脏数据包括空行、金额列有千分位逗号、负金额表示退款、日期格式不统一。这里做一层基础清理def clean_sales_data(df): df df.dropna(howall) # 去除列名首尾空格 df.columns [str(col).strip() for col in df.columns] # 去重完全相同的行 df df.drop_duplicates() return df这一步解决的是“能不能算”的问题。更复杂的清洗规则比如金额列解析、日期标准化可以根据实际表格字段调整。如果发现某列明明是数字却是文本类型可以用pd.to_numeric强制转换。4.3 让大模型识别表结构和汇总维度销售明细表的字段名千奇百怪可能是“业务员”“销售员”“销售人员”可能是“销售额”“金额”“成交额”可能是“区域”“大区”“地区”。直接写死字段名换一张表就失效。所以这里让大模型根据列名和样例数据推理出维度列、指标列、日期列并输出结构化 JSON。import json import requests def generate_summary_schema(df): columns list(df.columns) sample_rows df.head(5).to_dict(orientrecords) prompt f 你是一名销售数据分析助手。请分析下面这个销售明细表的字段结构确定汇总方案。 字段列表 {columns} 前5行样例数据 {json.dumps(sample_rows, ensure_asciiFalse, defaultstr)} 请返回 JSON {{ dimension_columns: [按什么字段分组汇总如区域、产品、销售员], metric_columns: [哪些是数值指标如销售额、数量], date_column: 日期字段名没有则填空字符串, date_format: 日期格式如 %Y-%m-%d没有则填空字符串, summary_description: 用一句话描述这个表做什么汇总 }} 只返回 JSON不要返回其他内容。 # 这里使用兼容 OpenAI Chat Completions 的请求格式 response requests.post( urlbase_url /chat/completions, headers{ Authorization: fBearer {api_key}, Content-Type: application/json }, json{ model: model_name, messages: [{role: user, content: prompt}], temperature: 0.1, response_format: {type: json_object} }, timeout60 ) response.raise_for_status() content response.json()[choices][0][message][content] return json.loads(content)注意上面的base_url、api_key、model_name需要从环境变量读取下面给出完整读取方式。import os api_key os.environ.get(LLM_API_KEY, ) base_url os.environ.get(LLM_BASE_URL, ) model_name os.environ.get(LLM_MODEL, )如果某些平台不支持response_format参数返回的是普通文本 JSON可以用json.loads直接解析也可以先用字符串截取再解析。4.4 校验大模型返回的字段名大模型生成的字段名未必和原表完全一致很可能出现“销售员”和“销售人员”这种差异。所以必须做一次字段匹配只保留真实存在于 DataFrame 中的列。def validate_schema(df, schema): valid_dimensions [col for col in schema.get(dimension_columns, []) if col in df.columns] valid_metrics [col for col in schema.get(metric_columns, []) if col in df.columns] valid_date schema.get(date_column, ) if valid_date and valid_date not in df.columns: valid_date return { dimension_columns: valid_dimensions, metric_columns: valid_metrics, date_column: valid_date, date_format: schema.get(date_format, ), summary_description: schema.get(summary_description, ) }这一步很重要。大模型的输出只能作为“建议”最终执行权要回到真实数据上否则列名对不上pandas 直接报 KeyError。4.5 执行汇总计算使用 pandas 的groupby做分组汇总。指标可以选求和、平均值、计数默认都做一遍输出到不同的列。def build_summary_table(df, schema): dimensions schema[dimension_columns] metrics schema[metric_columns] if not dimensions: # 没有合适的维度列则全表汇总 summary df[metrics].sum().to_frame().T summary.insert(0, 汇总范围, 全表) else: agg_dict {} for metric in metrics: agg_dict[metric _sum] (metric, sum) agg_dict[metric _avg] (metric, mean) summary df.groupby(dimensions).agg(**agg_dict).reset_index() return summary如果日期列存在还可以增加按月份汇总的 Sheetdef add_monthly_summary(writer, df, schema): date_col schema.get(date_column, ) date_format schema.get(date_format, ) if not date_col: return if date_format: df[date_col] pd.to_datetime(df[date_col], formatdate_format, errorscoerce) else: df[date_col] pd.to_datetime(df[date_col], errorscoerce) df df.dropna(subset[date_col]) df[月份] df[date_col].dt.to_period(M).astype(str) metrics schema[metric_columns] agg_dict {metric _sum: (metric, sum) for metric in metrics} monthly df.groupby(月份).agg(**agg_dict).reset_index() monthly.to_excel(writer, indexFalse, sheet_name月度汇总)5. 异常清单自动检测汇总表解决“总数是多少”异常清单解决“哪些数据有问题”。异常检测规则分为固定规则和 AI 辅助规则两类。5.1 固定规则这些规则直接用 pandas 实现速度快且结果可复现金额为负数或零数量为负关键字段为空折扣小于 0 或大于 1单价异常偏高或偏低用分位数判断销售额相比同组均值偏离超过 3 倍标准差。def detect_anomalies(df, schema): metrics schema[metric_columns] anomalies [] # 规则1数值列为负 for metric in metrics: if metric in df.columns: neg df[df[metric] 0] for idx, row in neg.iterrows(): anomalies.append({ 行号: idx 2, # 考虑表头占一行 异常类型: 负值, 字段: metric, 异常值: row[metric], 说明: f{metric}为负数可能表示退款或数据错误 }) # 规则2关键字段为空 dims schema.get(dimension_columns, []) for col in dims: if col in df.columns: empty df[df[col].isna()] for idx, row in empty.iterrows(): anomalies.append({ 行号: idx 2, 异常类型: 关键字段为空, 字段: col, 异常值: None, 说明: f维度字段{col}为空 }) # 规则3同组均值的极端偏离 for metric in metrics: if metric in df.columns and dims: grouped df.groupby(dims)[metric].transform(lambda x: (x - x.mean()).abs() 3 * x.std() 1e-9) outliers df[grouped] for idx, row in outliers.iterrows(): anomalies.append({ 行号: idx 2, 异常类型: 极端偏离, 字段: metric, 异常值: row[metric], 说明: f{metric}与同组均值偏差过大 }) if not anomalies: return pd.DataFrame(columns[行号, 异常类型, 字段, 异常值, 说明]) return pd.DataFrame(anomalies)5.2 AI 辅助异常解释固定规则能找出“数值不对”的行但找不出“业务逻辑不对”的情况。例如某个新客户首单金额异常大某个产品的折扣突然从 0.9 降到 0.3某个区域销量连续下滑但整体数据没有负值。这类问题更适合交给大模型做文本摘要。可以把汇总表转成文本让 AI 输出“需要人工关注的业务点”。def generate_ai_insight(summary_df, df, schema): summary_text summary_df.head(20).to_string(indexFalse) anomaly_count len(df) prompt f 以下是销售数据汇总结果和前若干行原始数据。 请分析并输出业务上需要重点关注的问题点包括 1. 业绩突出或严重下滑的团队/产品 2. 折扣、单价明显异常的情况 3. 数据质量可能存在的问题。 汇总表 {summary_text} 要求用简洁的中文输出每条用 - 开头不要超过8条。 response requests.post( urlbase_url /chat/completions, headers{ Authorization: fBearer {api_key}, Content-Type: application/json }, json{ model: model_name, messages: [{role: user, content: prompt}], temperature: 0.2 }, timeout60 ) response.raise_for_status() return response.json()[choices][0][message][content]6. 完整流程与输出 Excel把上面几个函数串起来形成主流程def run_sales_summary(input_path, output_path): print(f[1/5] 读取文件{input_path}) df load_sales_data(input_path) print(f数据行数{len(df)}列{list(df.columns)}) print([2/5] 基础清洗) df clean_sales_data(df) print([3/5] AI 识别表结构) schema_raw generate_summary_schema(df) schema validate_schema(df, schema_raw) print(识别维度, schema[dimension_columns]) print(识别指标, schema[metric_columns]) print([4/5] 生成汇总表和异常清单) summary build_summary_table(df, schema) anomalies detect_anomalies(df, schema) insight generate_ai_insight(summary, anomalies, schema) print([5/5] 写入 Excel) with pd.ExcelWriter(output_path, engineopenpyxl) as writer: summary.to_excel(writer, indexFalse, sheet_name汇总表) anomalies.to_excel(writer, indexFalse, sheet_name异常清单) # 如果存在日期字段增加月度汇总 if schema[date_column]: add_monthly_summary(writer, df, schema) # AI 洞察写入单独 Sheet insight_df pd.DataFrame({AI分析: [line for line in insight.split(\n) if line.strip()]}) insight_df.to_excel(writer, indexFalse, sheet_nameAI业务洞察) print(f完成输出文件{output_path}) if __name__ __main__: input_file sales_detail.xlsx output_file sales_summary_output.xlsx run_sales_summary(input_file, output_file)输出的 Excel 文件包含至少两个核心 Sheet汇总表按维度分组包含求和和平均值异常清单列出问题行号、异常类型、异常值、说明月度汇总如果存在日期字段则自动生成AI 业务洞察大模型生成的文字分析。这里的“行号”对应的是原表的真实位置方便回到明细表核对。7. 批量任务处理日常场景中很可能不是给一个文件而是给一个文件夹里面有几十个门店或区域的销售明细。批量处理时要注意每个文件单独生成一个输出文件失败的文件不能中断整个任务要记录错误日志汇总结果可以额外合并成一个总表。from pathlib import Path def batch_process(input_dir, output_dir): input_dir Path(input_dir) output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) supported_suffix {.csv, .xlsx, .xls} files [p for p in input_dir.iterdir() if p.suffix.lower() in supported_suffix] all_summaries [] error_log [] for file_path in files: try: output_path output_dir / f{file_path.stem}_汇总.xlsx run_sales_summary(file_path, output_path) # 合并汇总到总表 df load_sales_data(file_path) df clean_sales_data(df) all_summaries.append({文件名: file_path.name, 行数: len(df)}) print(f成功{file_path.name}) except Exception as e: error_log.append({文件名: file_path.name, 错误: str(e)}) print(f失败{file_path.name}错误{e}) # 输出处理日志 log_df pd.DataFrame(error_log) if error_log else pd.DataFrame(columns[文件名, 错误]) log_df.to_excel(output_dir / _处理日志.xlsx, indexFalse) # 输出文件清单 manifest pd.DataFrame(all_summaries) manifest.to_excel(output_dir / _文件清单.xlsx, indexFalse) print(f批量处理完成共 {len(files)} 个文件失败 {len(error_log)} 个)批量处理的关键是“失败隔离”。单个文件报错不应该影响其他文件最后统一看处理日志再回头排查问题文件。8. 资源占用与性能观察8.1 运行时长整个流程中耗时主要在大模型 API 调用本地计算部分非常快。以几万行销售明细为例本地读取和清洗秒级分组汇总秒级大模型识别表结构约 3 到 10 秒大模型生成 AI 业务洞察约 5 到 15 秒。实际耗时取决于上游模型服务的响应速度网络不稳定时可能更久。可以在代码中打印每个步骤的耗时方便排查。import time start time.time() run_sales_summary(sales_detail.xlsx, output.xlsx) print(f总耗时{time.time() - start:.2f}秒)8.2 显存和硬件占用这个过程不涉及本地大模型推理不需要 GPUCPU 内存占用也主要集中在 pandas 读取数据阶段。普通办公电脑即可运行。8.3 如何降低延迟减少发送给大模型的数据量样例数据只发送前 5 行而不是全表。控制汇总表长度AI 业务洞察只取前 20 行汇总数据。并发调用批量处理时可以用线程池同时调用大模型 API但要注意上游 API 的限流。9. 常见问题与排查方法问题现象可能原因排查方式解决方案CSV 读取后中文乱码文件编码不是 UTF-8打印前几行检查改用 gbk 或 gb18030 编码读取Excel 文件打不开文件正在被 Excel 占用检查文件是否被锁定关闭 Excel 后重试大模型返回的不是 JSON部分平台不支持 response_format打印原始返回内容用字符串截取或正则提取 JSON大模型返回的列名在表中不存在模型对列名理解偏差打印 schema 原始输出增加字段匹配映射或重新描述列名金额列有逗号无法计算千分位格式检查该列 dtype用pd.to_numeric(str.replace(,, ))汇总表出现 NaN分组字段有空值查看空值分布在清洗阶段填充或删除空值API 调用超时网络问题或模型响应慢查看错误日志增大 timeout设置重试机制批量任务中部分文件失败文件格式不同或缺少字段查看处理日志单独测试失败文件修改清洗逻辑生成的汇总数字和手工透视表对不上字段识别错误打印维度列和指标列手动指定字段名不依赖 AI 推理9.1 API 调用失败重试调用云模型接口时网络抖动或限流很常见。建议加入指数退避重试import time def call_llm_with_retry(payload, max_retries3): for attempt in range(max_retries): try: response requests.post( urlbase_url /chat/completions, headers{ Authorization: fBearer {api_key}, Content-Type: application/json }, jsonpayload, timeout60 ) response.raise_for_status() return response.json() except Exception as e: print(f请求失败第 {attempt 1} 次重试{e}) time.sleep(2 ** attempt) raise RuntimeError(大模型 API 请求失败次数过多)9.2 提示词优化方向如果大模型识别字段不准可以从两个方向优化提示词在样例数据中标注“金额在 500 到 5000 之间的列是销额”这类特征直接把列名和业务含义的映射表放进 prompt例如“销售员也叫业务员或销售代表”。这种方法比反复调 temperature 更有效。10. 最佳实践与使用建议10.1 先小数据验证第一次运行时不要直接处理几十万行的大文件。先截取 100 行数据测试确认字段识别、汇总计算、异常检测都符合预期再跑全量。不然一旦字段识别错输出结果会整体错掉。10.2 保留字段映射配置对于固定格式的销售明细建议把 AI 识别结果手动保存为 JSON 配置后续直接读取配置不再调用大模型识别表结构速度快而且稳定。schema_config { dimension_columns: [区域, 产品, 销售员], metric_columns: [销售额, 数量], date_column: 订单日期, date_format: %Y-%m-%d, summary_description: 按区域、产品、销售员汇总销售数据 } with open(schema_config.json, w, encodingutf-8) as f: json.dump(schema_config, f, ensure_asciiFalse, indent2)10.3 输入输出分目录管理建议目录结构如下sales_project/ ├── input/ # 原始销售明细 ├── output/ # 汇总结果 ├── config/ # 字段映射配置 ├── logs/ # 处理日志 └── scripts/ # Python 脚本10.4 数据脱敏调用外部大模型 API 前建议对客户手机号、邮箱、详细地址等敏感字段做替换处理def mask_sensitive_data(df, columns): for col in columns: if col in df.columns: df[col] df[col].astype(str).apply(lambda x: x[:3] *** x[-2:]) return df10.5 人工复核AI 生成的业务洞察只能作为参考。最终发给领导的报表必须由懂业务的人确认一遍。特别是“数据异常”的判断要结合业务背景比如电商大促期间的销量暴涨可能是正常现象而不是异常。这次推荐的方案最值得尝试的点是把“字段理解”交给大模型、把“数值计算”交给 pandas两者各干各擅长的事。很建议代码写完后先用一张真实的历史销售明细做一次全流程测试重点看三件事大模型是否正确识别了维度列和指标列、异常清单是否把明显有问题的行都找出来、汇总数字和手工透视表是否一致。最容易踩的坑有两个一个是字段名匹配失败解决方案是加一层 validate_schema 校验另一个是数据编码问题尤其是 CSV 文件中文乱码。建议把清洗和校验逻辑前置确认没问题后再接大模型否则排查问题会很痛苦。后续可以继续扩展的方向包括接入企业微信或钉钉机器人每天定时自动跑汇总并推送报告把脚本封装成 FastAPI 服务让同事通过网页上传文件就能拿到结果针对固定报表格式做模板化减少对大模型的重复调用。整套流程跑通之后下午要汇报这种临时需求基本可以控制在十分钟内出结果。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价