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

下午要交销售汇总，明细表还乱成一锅粥，靠手工透视表加肉眼找异常，时间根本不够用。这次我们直接换一个思路：用大模型 AI 把销售明细自动拆成“汇总表 + 异常清单”，明细进去，结果出来，中间不需要你在 Excel 里反复拖字段。

这个方案的重点不是概念多复杂，而是能不能在普通办公电脑上跑通。它的核心能力可以归纳成五点：

• 输入只要是 CSV 或 Excel 格式的销售明细，就能自动生成按区域、按产品、按销售员的汇总统计；
• 大模型负责解读表格结构和生成汇总规则，pandas 负责计算，结果稳定可控；
• 自动识别异常数据，比如负金额、空客户、异常折扣、销量暴涨暴跌；
• 汇总表和异常清单以 Excel 文件输出，直接可以发给领导；
• 支持批量任务，几十个明细文件可以排队处理，也支持通过 API 接入现有办公系统。

本文会带读者完成从环境准备、数据读取、大模型调用、汇总生成、异常检测到批量输出的完整流程，全部用可复现代码演示。适合做销售运营、数据分析、财务对账，以及所有经常被“领导下午就要”这类需求追着跑的人。

1. 销售汇总 AI 方案核心能力速览

从材料看，这个方案最有价值的地方不是“让 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_env

Windows 激活：

sales_env\Scripts\activate

macOS / Linux 激活：

source sales_env/bin/activate

3.2 安装依赖

pip install pandas openpyxl requests

• pandas：处理数据表，做分组汇总。
• 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-name"

macOS / Linux 设置环境变量：

export LLM_API_KEY="你的API Key" export LLM_BASE_URL="https://api.example.com/v1" export LLM_MODEL="your-model-name"

这里不指定具体平台，是因为各家 API 的模型名称、请求格式略有差异，代码中使用兼容 OpenAI Chat Completions 风格的接口，通用于大多数平台。

4. 销售明细自动汇总实现思路

整体流程分四个步骤：

1. 读取销售明细文件；
2. 对数据做基础清洗；
3. 让大模型识别表结构并生成汇总规则；
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, encoding=encoding) 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(how="all") # 去除列名首尾空格 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(orient="records") prompt = f""" 你是一名销售数据分析助手。请分析下面这个销售明细表的字段结构，确定汇总方案。 字段列表： {columns} 前5行样例数据： {json.dumps(sample_rows, ensure_ascii=False, default=str)} 请返回 JSON： {{ "dimension_columns": ["按什么字段分组汇总，如区域、产品、销售员"], "metric_columns": ["哪些是数值指标，如销售额、数量"], "date_column": "日期字段名，没有则填空字符串", "date_format": "日期格式，如 %Y-%m-%d，没有则填空字符串", "summary_description": "用一句话描述这个表做什么汇总" }} 只返回 JSON，不要返回其他内容。 """ # 这里使用兼容 OpenAI Chat Completions 的请求格式 response = requests.post( url=base_url + "/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json={ "model": model_name, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "response_format": {"type": "json_object"} }, timeout=60 ) 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

如果日期列存在，还可以增加按月份汇总的 Sheet：

def 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], format=date_format, errors="coerce") else: df[date_col] = pd.to_datetime(df[date_col], errors="coerce") 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, index=False, 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(index=False) anomaly_count = len(df) prompt = f""" 以下是销售数据汇总结果和前若干行原始数据。 请分析并输出业务上需要重点关注的问题点，包括： 1. 业绩突出或严重下滑的团队/产品； 2. 折扣、单价明显异常的情况； 3. 数据质量可能存在的问题。 汇总表： {summary_text} 要求：用简洁的中文输出，每条用 - 开头，不要超过8条。 """ response = requests.post( url=base_url + "/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json={ "model": model_name, "messages": [{"role": "user", "content": prompt}], "temperature": 0.2 }, timeout=60 ) 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, engine="openpyxl") as writer: summary.to_excel(writer, index=False, sheet_name="汇总表") anomalies.to_excel(writer, index=False, 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, index=False, sheet_name="AI业务洞察") 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(parents=True, exist_ok=True) 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", index=False) # 输出文件清单 manifest = pd.DataFrame(all_summaries) manifest.to_excel(output_dir / "_文件清单.xlsx", index=False) 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 显存和硬件占用

这个过程不涉及本地大模型推理，不需要 GPU，CPU 内存占用也主要集中在 pandas 读取数据阶段。普通办公电脑即可运行。

8.3 如何降低延迟

• 减少发送给大模型的数据量：样例数据只发送前 5 行，而不是全表。
• 控制汇总表长度：AI 业务洞察只取前 20 行汇总数据。
• 并发调用：批量处理时，可以用线程池同时调用大模型 API，但要注意上游 API 的限流。

9. 常见问题与排查方法

9.1 API 调用失败重试

调用云模型接口时，网络抖动或限流很常见。建议加入指数退避重试：

import time def call_llm_with_retry(payload, max_retries=3): for attempt in range(max_retries): try: response = requests.post( url=base_url + "/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json=payload, timeout=60 ) 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", encoding="utf-8") as f: json.dump(schema_config, f, ensure_ascii=False, indent=2)

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 df

10.5 人工复核

AI 生成的业务洞察只能作为参考。最终发给领导的报表，必须由懂业务的人确认一遍。特别是“数据异常”的判断，要结合业务背景，比如电商大促期间的销量暴涨可能是正常现象，而不是异常。

这次推荐的方案，最值得尝试的点是把“字段理解”交给大模型、把“数值计算”交给 pandas，两者各干各擅长的事。很建议代码写完后，先用一张真实的历史销售明细做一次全流程测试，重点看三件事：大模型是否正确识别了维度列和指标列、异常清单是否把明显有问题的行都找出来、汇总数字和手工透视表是否一致。

最容易踩的坑有两个：一个是字段名匹配失败，解决方案是加一层 validate_schema 校验；另一个是数据编码问题，尤其是 CSV 文件中文乱码。建议把清洗和校验逻辑前置，确认没问题后再接大模型，否则排查问题会很痛苦。

后续可以继续扩展的方向包括：接入企业微信或钉钉机器人，每天定时自动跑汇总并推送报告；把脚本封装成 FastAPI 服务，让同事通过网页上传文件就能拿到结果；针对固定报表格式做模板化，减少对大模型的重复调用。整套流程跑通之后，下午要汇报这种临时需求，基本可以控制在十分钟内出结果。