数据清洗智能体怎么搭建？用白名单操作、预览和变更日志约束执行

搭建数据清洗智能体时，可以让模型理解用户提出的目标，但不能把模型生成的任意代码直接执行。更容易复核的结构是：模型只输出有限的“动作+列名”计划，确定性的执行器逐项验证，先返回预览与变更日志，最后由实际审批状态决定是否保存。

定义允许动作和拒绝条件

教学计划只允许 trim_whitespace 清理 city 列首尾空格，以及 drop_exact_duplicates 删除内容完全相同的行。输入计划多字段、未知动作或非白名单列都报错。这里的“智能体计划”由人写在脚本中模拟，没有调用线上模型。

删除重复行前要区分“完全相同”与“业务主键重复”：两个同编号订单可能有不同状态，绝不能只按一个 ID 自动删掉。这个例子里的两行 u2/上海 完全相同，才用整体行去重。

运行受控预览

安装 pandas，将代码存为 cleaning_agent.py 并运行。source 是人工构造表；preview 是执行后的候选结果，提交前仍保留原表。

import json
from pathlib import Path
import pandas as pd

# plans 可由模型提出；本例是人工编写的候选计划，没有调用模型。
source=pd.DataFrame({'id':['u1','u2','u2'],'city':['北京 ','上海','上海']})
agent_proposal={'source_id':'teaching-table-v1','plans':[{'action':'trim_whitespace','column':'city'}, {'action':'drop_exact_duplicates','column':None}]}
plans=agent_proposal['plans']
allowed={'trim_whitespace','drop_exact_duplicates'}
def execute(frame,plan):
    if set(plan)!={'action','column'} or plan['action'] not in allowed:raise ValueError('operation_not_allowed')
    result=frame.copy()
    if plan['action']=='trim_whitespace':
        if plan['column'] not in {'city'}:raise ValueError('column_not_allowed')
        result[plan['column']]=result[plan['column']].str.strip()
    else:result=result.drop_duplicates()
    return result
preview=source.copy();log=[]
for plan in plans:
    before=len(preview);preview=execute(preview,plan)
    log.append({'action':plan['action'],'column':plan['column'],'before':before,'after':len(preview)})
print('preview=\n'+preview.to_string(index=False))
print('change_log=',log)
approval_record={'source_id':'teaching-table-v1','approved':True,'reviewer':'demo-reviewer'}
if approval_record['source_id']!=agent_proposal['source_id']:raise ValueError('source_changed')
saved=preview.copy() if approval_record['approved'] else source.copy()
assert len(saved)==2 and saved['city'].str.startswith(' ').sum()==0
log.append({'action':'commit','approved':approval_record['approved'],'reviewer':approval_record['reviewer'],'after':len(saved)})
saved_path=Path(__file__).with_name('03_cleaned.csv')
log_path=Path(__file__).with_name('03_change_log.json')
if approval_record['approved']:
    saved.to_csv(saved_path,index=False,encoding='utf-8-sig')
    log_path.write_text(json.dumps(log,ensure_ascii=False,indent=2),encoding='utf8')
    reloaded=pd.read_csv(saved_path,encoding='utf-8-sig')
    result_checked=(len(reloaded)==2 and reloaded['city'].tolist()==['北京','上海'])
else:result_checked=False
print('saved_rows=',len(saved),'source_rows=',len(source),'result_checked=',result_checked)
print('saved_file=',saved_path.name,'log_file=',log_path.name)
print('final_log=',log)
try:execute(source,{'action':'delete_everything','column':None})
except ValueError as error:print('rejected=',error)

检查预览、日志与拒绝结果

预览中城市空格被去掉，三行变两行；日志逐步写明每个动作前后的行数。示例的审批记录与提案绑定同一个 source_id，确认后把结果写入 03_cleaned.csv、日志写入 03_change_log.json，再回读 CSV 核对行数和城市值。delete_everything 会触发 operation_not_allowed。脚本中的审批记录是教学样例，生产系统必须从真实审批服务读取，而非硬编码为真。

preview=
id city
u1   北京
u2   上海
change_log= [{'action': 'trim_whitespace', 'column': 'city', 'before': 3, 'after': 3}, {'action': 'drop_exact_duplicates', 'column': None, 'before': 3, 'after': 2}]
saved_rows= 2 source_rows= 3 result_checked= True
saved_file= 03_cleaned.csv log_file= 03_change_log.json
final_log= [{'action': 'trim_whitespace', 'column': 'city', 'before': 3, 'after': 3}, {'action': 'drop_exact_duplicates', 'column': None, 'before': 3, 'after': 2}, {'action': 'commit', 'approved': True, 'reviewer': 'demo-reviewer', 'after': 2}]
rejected= operation_not_allowed

pandas 的重复行判断文档解释了整行与指定列的区别；DataFrame 基础指南可用于核对复制、选择与字符串列操作。

实际接入模型时的边界

对真实工具调用，模型只负责生成符合预定 JSON 结构的 plans，工具负责校验与执行，审批系统负责给出 approved；此处没有接入真实 LLM 或审批服务，只把教学结果写到脚本同目录。上线还要限制文件路径、最大行数、可写目录和日志中的敏感数据。每次建议都应展示原值、候选值和受影响行数；没有可靠的去重依据就停在预览。