Excel数据治理新思路:引入智能体实现自动纠错【Python+Agent】

简介: 本文介绍如何利用智能体与Python代码批量处理Excel中的脏数据,解决人工录入导致的格式混乱、逻辑错误等问题。通过构建具备数据校验、异常标记及自动修正功能的系统,将数小时的人工核查任务缩短至分钟级,大幅提升数据一致性和办公效率。

👩🏻‍💻 在企业日常办公中,Excel表格承载着大量核心数据,但人工录入导致的格式混乱、逻辑错误、信息矛盾等问题却常常成为效率的“隐形杀手”。手动逐行核查不仅耗时费力,还容易因疏漏引发二次风险。

本文以真实场景为例,构建了一个具备数据校验、异常标记、自动修正功能的智能体,并通过代码演示了如何批量处理表格中脏数据。将原本需要数小时的人工核对任务缩短至分钟级,同时通过标准化逻辑规则提升数据一致性。

如下表格我们可以看到,有很多错误数据。案例数据Excel下载,📎input.xlsx

上面这些错误的数据,涉及不同字段和不同类型的错误,人工判断起来比较困难,如何通过大模型来批量处理表格当中的脏数据呢?

大致思路:编写 Python 代码,逐行读取数据 → 将每一行的数据发送给相应的智能体 → 智能体处理完数据后,会将结果返回到 Python 程序中 → 再将这些结果写入 Excel 文件中。

具体流程如下:

1. 搭建智能体用于处理对应表格当中的数据

1.1 智能体实现逻辑

为了方便处理并返回对应的数据我们处理数据的时候,数据的格式统一为JSON格式。

例如输入的数据格式为

{
  "name":"陈晓明",
  "birthday":"1995/8/14",
  "age":"28",
  "gender":"男",
  "height","175"
  "weight":"68",
  "birth_place":"浙江杭州",
  "abnormalities","否"
}

其中name,birthday,age,gender,height,weight,birth_place,abnormalities和对应Excel的表头对应的。数据是人工录入批量输入的,可能存在录入错误,需要搭建一个智能体来完成数据清洗与标注任务。


最终输出的数据格式应与输入的JSON格式保持一致。

格式如下:

{
  "name":"陈晓明",
  "birthday":"1997/8/14",
  "age":"28(异常)",
  "gender":"男",
  "height","175"
  "weight":"68",
  "birth_place":"浙江广州(异常)",
  "abnormalities","是"
}

1.2 智能体搭建

选择对应的模型,并填入提示词。

提示词如下:

# 角色
你是一个数据清洗和打标智能体,专门负责处理批量输入的数据,确保数据的准确性和一致性。
## 技能
### 技能 1: 数据验证与清洗
- **任务**:对输入的JSON格式数据进行验证和清洗。
- **具体步骤**:
  - **出生年月日与年龄验证**:
    - 根据`birthday`字段计算实际年龄。
    - 如果计算出的年龄与`age`字段不符,在`age`字段后面标注“(异常)”,并更新`age`字段为正确值。
    - 在`age`字段后面添加“(年龄有误请核对)”。
  - **性别验证**:
    - 检查`gender`字段是否为“男”或“女”。
    - 如果不是,则在`gender`字段后面标注“(性别有误)”。
  - **身高和体重验证**:
    - 根据`age`字段推断身高和体重是否在正常范围内。
    - 如果不在正常范围内,在`height`或`weight`字段后面标注“(异常)”。
    - 检查身高和体重的比例是否正常,如果不正常,在`height`或`weight`字段后面标注“(异常)”。
  - **籍贯验证**:
    - 检查`birth_place`字段是否填写正确。
    - 如果填写有误,在`birth_place`字段后面标注“(籍贯异常)”。
    - birthday字段数据格式统一为yyyy/MM/dd 如果不是这个格式转换为这个格式
    如果时间有错误, birthday字段后面标注“(异常)”。
### 技能 2: 数据打标
- **任务**:根据验证结果对`abnormalities`字段进行标注。
- **具体步骤**:
  - 如果上述任何一项验证结果存在异常,则将`abnormalities`字段设置为“是”。
  - 如果所有验证结果均无异常,则将`abnormalities`字段设置为“否”。
## 限制
- 只处理输入的JSON格式数据。
- 保持输出数据格式与输入数据格式一致。
- 输出数据为存JSON格式的文本,不要有其他内容,不要有代码块。
- 在标注异常时,确保注释清晰且易于理解。
- 不引入个人观点或偏见,仅基于数据进行验证和标注。
## 示例
输入数据:
```json
{
  "name":"陈晓明",
  "birthday":"1993/8/14",
  "age":"28",
  "gender":"男",
  "height":"175",
  "weight":"68",
  "birth_place":"浙江杭州",
  "abnormalities":"否"
}
```
输出数据:
```json
{
  "name":"陈晓明",
  "birthday":"1993/8/14",
  "age":"28(年龄有误请核对)",
  "gender":"男",
  "height":"175",
  "weight":"68",
  "birth_place":"浙江杭州",
  "abnormalities":"是"
}
```
如果输入数据:
```json
{
  "name":"李华",
  "birthday":"1990/1/1",
  "age":"30",
  "gender":"其他",
  "height":"180",
  "weight":"70",
  "birth_place":"浙江上海",
  "abnormalities":"否"
}
```
输出数据:
```json
{
  "name":"李华",
  "birthday":"1990/1/1",
  "age":"30",
  "gender":"其他(性别有误)",
  "height":"180",
  "weight":"70",
  "birth_place":"浙江上海(籍贯异常)",
  "abnormalities":"是"
}


2. 编写Python程序实现数据清洗和标注的功能

2.1 代码业务逻辑说明

我们需要借助Python逐行读取Excel表格中的内容,并将其转换为对应的JSON格式数据。

{
  "name":"陈晓明",
  "birthday":"1993/8/14",
  "age":"28",
  "gender":"男",
  "height":"175",
  "weight":"68",
  "birth_place":"浙江杭州",
  "abnormalities":"否"
}

随后,将这些数据发送给智能体进行处理,再将返回的结果插入到表格的新列中。为了便于对比,我们不会覆盖原有数据,而是新增几列存储处理后的结果。如果需要筛选出正确的数据,可以通过“abnormalities”字段进行过滤,从而精准定位并提取目标数据。

2.2 创建项目并安装依赖

1、安装依赖:openpyxl,用于处理Excel。

安装命令:

pip install openpyxl

2、安装依赖:pandas,用于数据分析和处理(如读取/写入数据、数据清洗等)。

安装命令:

pip install pandas

3、dashscope,调用 DashScope 的 API。

安装命令:

pip install dashscope

2.3 编写代码实现功能

import pandas as pd
import json
from datetime import datetime
import logging
# 配置区域 - 根据实际情况修改
EXCEL_INPUT = "input.xlsx"  # 输入Excel文件名
EXCEL_OUTPUT = "output.xlsx"  # 输出Excel文件名
SHEET_NAME = "Sheet1"  # 工作表名
API_KEY = "sk-API_KEY"
APP_ID = "APP_ID_APP_ID"
from http import HTTPStatus
from dashscope import Application
def process_excel():
    """主处理函数:读取Excel、处理数据、保存结果"""
    # 读取Excel数据,将所有列作为字符串处理
    try:
        df = pd.read_excel(EXCEL_INPUT, sheet_name=SHEET_NAME, dtype=str)
        # 替换可能的NaN值
        df = df.fillna("")
    except Exception as e:
        print(f"读取Excel失败: {str(e)}")
        return
    # 检查必要列是否存在
    required_columns = ["name", "birthday", "age","gender", "height", "weight", "birth_place"]
    for col in required_columns:
        if col not in df.columns:
            logging.error(
                f"错误:缺少必要列 '{col}'"
            )
            return
    # 结果列初始化
    result_columns = ['agent_name', 'agent_birthday', 'agent_age', 'agent_gender', 'agent_height', 'agent_weight', 'agent_birth_place','agent_abnormalities']
    for col in result_columns:
        df[col] = ""
    # 处理每一行数据
    for index, row in df.iterrows():
        # 构建JSON数据
        person_data = {
            "name": str(row.get("name", "")),
            "birthday": str(row.get("birthday", "")),
            "age": str(row.get("age")),
            "gender": str(row.get("gender", "")),
            "height": str(row.get("height", "")),
            "weight": str(row.get("weight", "")),
            "birth_place": str(row.get("birth_place", ""))
        }
        response = Application.call(
            api_key=API_KEY,
            app_id=APP_ID,
            prompt=person_data.__str__() +" 当前系统时间:"+datetime.now().__str__())
        if response.status_code != HTTPStatus.OK:
            logging.error(
                f'request_id={response.request_id}',
                f'code={response.status_code}',
                f'message={response.message}'
            )
        else:
            try:
                print(response.output.text)
                dataJson = json.loads(response.output.text)
                df.at[index, 'agent_name'] = dataJson["name"]
                df.at[index, 'agent_birthday'] = dataJson["birthday"]
                df.at[index, 'agent_age'] = dataJson["age"]
                df.at[index, 'agent_gender'] = dataJson["gender"]
                df.at[index, 'agent_height'] = dataJson["height"]
                df.at[index, 'agent_weight'] = dataJson["weight"]
                df.at[index, 'agent_birth_place'] = dataJson["birth_place"]
                df.at[index, 'agent_abnormalities'] = dataJson["abnormalities"]
                print(dataJson)
            except json.JSONDecodeError as e:
                logging.error(f"JSON解析失败: {str(e)}")
    # 保存处理结果
    try:
        df.to_excel(EXCEL_OUTPUT, index=False)
        print(f"处理完成!结果已保存到: {EXCEL_OUTPUT}")
    except Exception as e:
        logging.error(f"保存结果失败: {str(e)}")
if __name__ == "__main__":
    process_excel()
  1. Excel数据处理:读取包含人员基础信息(姓名、生日、性别、身高体重等)的Excel文件,并验证必要字段完整性。
  2. AI代理API集成:通过DashScope平台调用大模型API,将原始数据构造成JSON格式作为prompt输入,获取增强后的代理数据(如修正后的个人信息)。
  3. 异常处理与日志记录:采用模块化设计,对API调用失败、JSON解析异常等情况进行捕获和记录,支持通过环境变量配置敏感信息提升安全性。
  4. 结果输出:将处理结果写入新Excel文件,自动筛选并统计无异常数据条目。
    系统通过分离配置参数、增加日志模块、优化错误处理机制,提升了代码的可维护性和健壮性,适用于需要批量校验和增强结构化数据的场景。

2.4 运行测试,查看最后Excel当中数据清洗的结果

将刚刚需要处理的Excel复制到项目根目录下面,也就是当前代码文件的傍边即可,运行代码进行测试。

测试成功。

最终结果:

3. 说重点,如何快速适配自己的数据文件

3.1 比如你的Excel文件长这样

换一个体检的场景,你的excel只有"name", "mmhg", "heartrate","ml", "bloodtype" 这几个字段,要怎么快速的把上面的代码适配到自己的数据呢?📎体检信息.xlsx

3.2 你要修改下面这些代码

修改1:检查必要列是否存在

# 检查必要列是否存在
required_columns = ["name", "birthday", "age","gender", "height", "weight", "birth_place"]

修改为

# 检查必要列是否存在
    required_columns = ["name", "mmhg", "heartrate","ml", "bloodtype"]

修改2:结果列初始化

# 结果列初始化
    result_columns = ['agent_name', 'agent_birthday', 'agent_age', 'agent_gender', 'agent_height', 'agent_weight', 'agent_birth_place','agent_abnormalities']

修改为

# 结果列初始化
    result_columns = ['agent_name', 'agent_mmhg', 'agent_heartrate', 'agent_ml', 'agent_bloodtype']

修改3:构建JSON数据

# 构建JSON数据
        person_data = {
            "name": str(row.get("name", "")),
            "birthday": str(row.get("birthday", "")),
            "age": str(row.get("age")),
            "gender": str(row.get("gender", "")),
            "height": str(row.get("height", "")),
            "weight": str(row.get("weight", "")),
            "birth_place": str(row.get("birth_place", ""))
        }

修改为

# 构建JSON数据
        person_data = {
            "name": str(row.get("name", "")),
            "mmhg": str(row.get("mmhg", "")),
            "heartrate": str(row.get("heartrate")),
            "ml": str(row.get("ml", "")),
            "bloodtype": str(row.get("bloodtype", ""))
        }

上述内容要和对应的表头所对应。

修改4:结果构建

try:
                print(response.output.text)
                dataJson = json.loads(response.output.text)
                df.at[index, 'agent_name'] = dataJson["name"]
                df.at[index, 'agent_birthday'] = dataJson["birthday"]
                df.at[index, 'agent_age'] = dataJson["age"]
                df.at[index, 'agent_gender'] = dataJson["gender"]
                df.at[index, 'agent_height'] = dataJson["height"]
                df.at[index, 'agent_weight'] = dataJson["weight"]
                df.at[index, 'agent_birth_place'] = dataJson["birth_place"]
                df.at[index, 'agent_abnormalities'] = dataJson["abnormalities"]
                print(dataJson)
            except json.JSONDecodeError as e:
                logging.error(f"JSON解析失败: {str(e)}")

修改为

try:
                print(response.output.text)
                dataJson = json.loads(response.output.text)
                df.at[index, 'agent_name'] = dataJson["name"]
                df.at[index, 'agent_mmhg'] = dataJson["birthday"]
                df.at[index, 'agent_heartrate'] = dataJson["heartrate"]
                df.at[index, 'agent_ml'] = dataJson["ml"]
                df.at[index, 'agent_bloodtype'] = dataJson["bloodtype"]
                print(dataJson)
            except json.JSONDecodeError as e:
                logging.error(f"JSON解析失败: {str(e)}")

上述内容要和Agent里面输出的JSON对应。

修改5:修改智能体信息

Agent 智能体的对应提示词当中输出的JSON要修改为

{
  "name":"张三",
  "birthday":"110/70",
  "heartrate":"75",
  "ml":"3000",
  "bloodtype":"B"
}

参照以上步骤就可以实现最终的数据清洗的效果啦,可以一步一步跟着步骤操作哦~~


小编在这里给小伙伴们提供了阿里云百炼便捷的进入入口,可直接点击链接进入哈:

👉阿里云百炼详情了解可点击此官网链接:阿里云百炼官网介绍

👉阿里云百炼控制台页面可点击此链接直接进入阿里云百炼控制台


如果在创建过程中有任何的疑问都可以在评论区中留言探讨或是加入我们的官方支持群(群号:77600022533)进行交流反馈!

image.png

作者介绍
目录