手把手教你用 Agent 做数据治理实战

导读：做数据治理的同学，大概都有过这样的崩溃时刻：

• 面对几十万条杂乱数据，手动清洗空值、去重、修正格式，熬到深夜还做不完；
• 给数据打标签全靠人工分类，不仅费时间，还容易出错；
• 排查数据问题时，不知道数据从哪来、经过哪些处理，血缘追溯全靠翻文档……

而现在，有了 AI Agent（智能体），这一切都能自动化搞定！

今天就手把手教大家，用 Agent 实现数据治理三大核心场景——自动化清洗、智能标注、全链路血缘追踪，全程附实战代码，新手也能跟着做！

本文核心：不聊复杂理论，只讲落地实操，从环境搭建到 Agent 部署，再到效果验证，一步步带你搞定 Agent 数据治理实战。

---

一、先搞懂：为什么用 Agent 做数据治理？

在动手之前，先跟大家说清楚 Agent 做数据治理的核心优势，避免大家盲目跟风。

• ✅ 解放双手：把人工重复工作（清洗、标注、追溯）全部自动化，数据人员不用再做“数据搬运工”，专注做高价值决策；
• ✅ 高效精准：Agent 能 24 小时不间断工作，处理海量数据的速度是人工的 10 倍以上，还能减少人工操作的误差；
• ✅ 自主迭代：Agent 能记录处理过程，学习业务规则，后续遇到类似数据，能自动优化处理方案，越用越聪明；
• ✅ 全流程覆盖：从数据接入后的清洗，到标注分类，再到血缘追踪，Agent 能打通全链路，不用再切换多个工具。

划重点：本文用「大模型+Agent 框架」实现，选用 Python 语言（新手友好），核心用到 LangChain（Agent 框架）、Pandas（数据处理）、Neo4j（血缘存储），全程实战，跟着步骤走就能跑通！

---

二、前置准备：环境搭建

先搭建好运行环境，后续所有实战代码都基于这个环境运行，直接复制命令执行即可。

1. 安装核心依赖包

打开终端，执行以下命令（建议先用 venv 创建虚拟环境）：

# 安装核心依赖pip install langchain langchain-openai pandas openai python-dotenv neo4j py2neo# 安装数据处理辅助包pip install numpy matplotlib

2. 配置关键参数

需要配置两个核心参数：大模型 API 密钥 和 Neo4j 数据库连接信息。

步骤：

1. 在项目根目录新建 .env 文件，写入以下内容（替换为自己的信息）：```plaintext

   大模型 API 配置（OpenAI 示例，也可替换为国内大模型如通义千问等）OPENAI_API_KEY=sk-your-openai-api-keyOPENAI_MODEL=gpt-3.5-turbo# Neo4j 数据库配置（本地或云端均可）NEO4J_URI=bolt://localhost:7687NEO4J_USER=neo4jNEO4J_PASSWORD=your-neo4j-password

2. 启动 Neo4j 数据库（本地安装直接启动服务，云端获取 Bolt 地址），确保能正常连接。

3. 准备测试数据

新建 data 文件夹，创建 test_data.csv 文件，内容如下（包含各种典型脏数据）：

customer_id,name,phone,order_amount,order_date,customer_type1001,张三，13812345678,299.9,2026-01-15，普通客户1002,李四，,599.0,2026/02/20，会员客户1003,王五，13987654321,abc,2026-03-10，普通客户1004,赵六，1361234567,899.5,2026-04-05,1005,孙七，13787654321,-100.0,2026-05-20，会员客户1001,张三，13812345678,399.0,2026-06-12，普通客户

**数据痛点，**包含空值、重复行、非数字金额、负数金额、日期格式不统一、手机号位数缺失等问题。

---

三、实战第一步：用 Agent 实现自动化数据清洗

数据清洗是最繁琐的一步。我们让 Agent 自主完成：识别问题 → 制定规则 → 执行清洗。

1. 定义 Agent 清洗逻辑

新建 agent_clean.py，代码如下：

import osimport pandas as pdfrom dotenv import load_dotenvfrom langchain.agents import initialize_agent, Toolfrom langchain_openai import ChatOpenAIfrom langchain.prompts import PromptTemplatefrom langchain.chains import LLMChain# 加载环境变量load_dotenv()# 1. 初始化大模型llm = ChatOpenAI(    api_key=os.getenv("OPENAI_API_KEY"),    model_name=os.getenv("OPENAI_MODEL"),    temperature=0.3# 低温度保证逻辑严谨)# 2. 定义数据探查工具def data_profiling(file_path):    """数据探查：识别数据中的空值、重复值、异常值、格式问题"""    df = pd.read_csv(file_path)    report = f"""    数据基本信息：    - 行数：{len(df)}, 列数：{len(df.columns)}    - 空值统计：{df.isnull().sum().to_dict()}    - 重复行数量：{df.duplicated().sum()}    - 数据类型：{df.dtypes.to_dict()}    """    # 检查金额异常    if"order_amount"in df.columns:        non_numeric = df[pd.to_numeric(df["order_amount"], errors="coerce").isna()].shape[0]        negative = df[pd.to_numeric(df["order_amount"], errors="coerce") < 0].shape[0]        report += f"\n订单金额异常：非数字={non_numeric}, 负数={negative}"        # 检查日期异常    if"order_date"in df.columns:        invalid_date = pd.to_datetime(df["order_date"], errors="coerce").isna().sum()        report += f"\n订单日期异常：格式无效={invalid_date}"        print("🔍 数据探查报告生成完毕")    return df, report# 3. 定义数据清洗工具def data_cleaning(df_str, cleaning_rules):    """数据清洗：根据规则执行清洗 (注意：实际生产中建议传递文件路径而非 DataFrame 字符串)"""    # 这里为了演示简化，重新读取原始数据并应用规则    df = pd.read_csv("data/test_data.csv")        rules_list = eval(cleaning_rules) if isinstance(cleaning_rules, str) else cleaning_rules        if"去重"in str(rules_list):        df = df.drop_duplicates()        print("✅ 已执行去重")        if"空值处理"in str(rules_list):        df["phone"] = df["phone"].fillna("未知")        df["customer_type"] = df["customer_type"].fillna("普通客户")        print("✅ 已执行空值填充")        if"金额异常处理"in str(rules_list):        df["order_amount"] = pd.to_numeric(df["order_amount"], errors="coerce")        avg_val = df["order_amount"].mean()        df["order_amount"] = df["order_amount"].fillna(avg_val).apply(lambda x: abs(x))        print("✅ 已执行金额修正")        if"日期格式统一"in str(rules_list):        df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce").dt.strftime("%Y-%m-%d")        print("✅ 已执行日期格式化")            if"手机号标准化"in str(rules_list):        df["phone"] = df["phone"].astype(str).apply(lambda x: x.zfill(11) if x.isdigit() and len(x)<11else x)        print("✅ 已执行手机号补全")    df.to_csv("data/cleaned_data.csv", index=False)    return"清洗完成，文件已保存至 data/cleaned_data.csv"# 4. 定义提示词prompt_template = """你是一名数据治理专家。请根据以下数据探查报告，制定具体的清洗规则列表。只需输出 Python 列表格式的字符串，例如：["去重", "空值处理", "金额异常处理", "日期格式统一", "手机号标准化"]。不要输出其他解释。数据探查报告：{report}"""prompt = PromptTemplate(template=prompt_template, input_variables=["report"])rule_chain = LLMChain(llm=llm, prompt=prompt)# 5. 注册工具tools = [    Tool(name="DataProfiling", func=lambda x: data_profiling("data/test_data.csv"), description="探查数据问题"),    Tool(name="DataCleaning", func=lambda x: data_cleaning(*eval(x)), description="执行清洗，输入格式为 '(df_placeholder, rules_list)'")]# 6. 初始化并运行 Agentif __name__ == "__main__":    # 第一步：探查    df, report = data_profiling("data/test_data.csv")        # 第二步：LLM 制定规则    cleaning_rules = rule_chain.run(report=report)    print(f"🤖 Agent 制定的规则：{cleaning_rules}")        # 第三步：执行清洗 (模拟 Agent 调用工具)    # 在实际 LangChain Agent 中，这步会自动由 ReAct 逻辑完成，这里为了稳定性直接调用    result = data_cleaning("dummy_df", cleaning_rules)    print(f"🎉 结果：{result}")

2. 效果验证

运行后，data/cleaned_data.csv 将发生神奇变化：

• 空值 ➡️ 被填充为“未知”或“普通客户”。
• 重复行 ➡️ 被删除。
• 金额 “abc” ➡️ 变为平均值；**-100** ➡️ 变为 100。
• 日期 ➡️ 统一为 YYYY-MM-DD。
• 手机号 ➡️ 自动补全为 11 位。

---

四、实战第二步：用 Agent 实现智能数据标注

清洗完的数据需要打标才能产生业务价值。Agent 可以根据复杂的业务逻辑自动打标。

1. 定义 Agent 标注逻辑

新建 agent_label.py：

import pandas as pdfrom dotenv import load_dotenvfrom langchain_openai import ChatOpenAIfrom langchain.prompts import PromptTemplatefrom langchain.chains import LLMChainimport osimport astload_dotenv()llm = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"), model_name=os.getenv("OPENAI_MODEL"), temperature=0.2)# 加载清洗后的数据df = pd.read_csv("data/cleaned_data.csv")# 定义标注逻辑提示词label_prompt = """你是一个数据标注助手。请根据以下业务规则，为每一行数据生成 'customer_level' 和 'customer_activity' 标签。规则：1. customer_level: 金额>=800->'高价值', 300<=金额<800->'中价值', 其他->'普通'2. customer_activity: 日期在 2026-06 -> '高活跃', 2026-04~05 -> '中活跃', 其他 -> '低活跃'请返回一个 JSON 列表，包含每行的 customer_id 和对应的两个新标签。数据样本：{data_sample}"""chain = LLMChain(llm=llm, prompt=PromptTemplate(template=label_prompt, input_variables=["data_sample"]))if __name__ == "__main__":    # 取前 5 行让 LLM 理解格式并生成逻辑（实际生产可分批处理或让 LLM 生成 Pandas 代码）    # 这里为了演示稳定性，我们让 LLM 生成代码逻辑，或者直接应用 Pandas 逻辑（模拟 Agent 决策后的执行）        # 模拟 Agent 决策后的执行代码（实际场景中 Agent 可生成此代码并 exec）    df["order_date"] = pd.to_datetime(df["order_date"])        def get_level(amount):        if amount >= 800: return"高价值客户"        elif amount >= 300: return"中价值客户"        else: return"普通客户"    def get_activity(date):        if date >= pd.to_datetime("2026-06-01"): return"高活跃"        elif date >= pd.to_datetime("2026-04-01"): return"中活跃"        else: return"低活跃"    df["customer_level"] = df["order_amount"].apply(get_level)    df["customer_activity"] = df["order_date"].apply(get_activity)        df.to_csv("data/labeled_data.csv", index=False)    print("✅ 智能标注完成！预览：")    print(df[["customer_id", "order_amount", "customer_level", "customer_activity"]].head())

2. 标注结果

生成的 labeled_data.csv 将自动新增两列，标准统一，无需人工逐行判断。

---

五、实战第三步：用 Agent 实现数据血缘追踪

这是最酷的部分！让 Agent 自动记录“谁生了谁”，并存入图数据库 Neo4j。

1. 定义 Agent 血缘逻辑

新建 agent_lineage.py：

from py2neo import Graph, Node, Relationshipfrom dotenv import load_dotenvimport osload_dotenv()# 连接 Neo4jgraph = Graph(    os.getenv("NEO4J_URI"),    user=os.getenv("NEO4J_USER"),    password=os.getenv("NEO4J_PASSWORD"))# 清空旧数据（演示用）graph.delete_all()# 定义节点和关系source = Node("DataSet", name="test_data.csv", type="Raw")step1 = Node("Process", name="DataCleaning", operation="去重/填充/格式化")intermediate = Node("DataSet", name="cleaned_data.csv", type="Cleaned")step2 = Node("Process", name="DataLabeling", operation="打标签：等级/活跃度")target = Node("DataSet", name="labeled_data.csv", type="Labeled")# 构建关系链rel1 = Relationship(source, "GENERATES_AFTER", step1)rel2 = Relationship(step1, "PRODUCES", intermediate)rel3 = Relationship(intermediate, "GENERATES_AFTER", step2)rel4 = Relationship(step2, "PRODUCES", target)# 写入数据库graph.create(source | step1 | intermediate | step2 | target | rel1 | rel2 | rel3 | rel4)print("🕸️ 血缘关系已成功存入 Neo4j！")print("👉 请访问 http://localhost:7474 查看可视化图谱")

2. 效果验证

打开 Neo4j Browser，输入 MATCH (n) RETURN n，你将看到一张清晰的流程图：test_data.csv ➡️ (清洗) ➡️ cleaned_data.csv ➡️ (标注) ➡️ labeled_data.csv。 以后数据出问题，顺着线就能找到根源！

---

六、总结与进阶

实战总结

今天我们完成了三个壮举：

1. 自动化清洗：Agent 像侦探一样发现数据问题，并像工匠一样修复它。
2. 智能标注：Agent 理解业务规则，批量完成人工耗时的工作。
3. 血缘追踪：Agent 自动记录全过程，让数据流转透明化。

进阶建议

• 国产化替代：将 ChatOpenAI 替换为 ChatZhipuAI 或 ChatTongyi，适配国内环境。
• 增加反馈环：让业务人员对标注结果点赞/点踩，Agent 自动微调规则。
• 可视化界面：用 Streamlit 封装上述代码，做成一个“一键治理”的小工具。

最后花荣想说： Agent 不是为了替代数据工程师，而是为了把我们从**“洗数据、贴标签、查文档”的泥潭中解放出来，让我们有更多时间去思考数据背后的业务价值**。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】