别急着教 Agent 思考，先喂它吃口干净的：ETL 入门

别急着教 Agent 思考先喂它吃口干净的ETL 入门摘要很多人一上来就盯着 Agent 的规划、决策、反思却把更底下那层忘了数据到底干不干净。ETL 在 Agent 时代一点都不过时反而更要命。传统程序吃到脏数据常见反应是报错停下Agent 吃到脏数据更容易顺着错数据继续往下编而且还说得像那么回事。本文用一个“摸鱼新闻 Agent”的例子把Extract / Transform / Load三步拆开讲明白数据怎么采、怎么洗、怎么存以及它和 Agent 感知模块到底差在哪。目录先说结论Agent 也得先吃干净数据ETL 是什么三十秒说清楚一个例子贯穿全文摸鱼新闻 Agent第一步Extract先把数据搬回来第二步Transform把脏数据洗干净第三步Load把干净数据放到该放的地方串起来一条完整的 ETL 管道最重要的一件事ETL 不是 Agent 的感知模块最后一句话先说结论Agent 也得先吃干净数据你花了两周调通一个 Agent接了大模型配了工具调用画了决策流程图prompt 改了三十遍。Demo 演示时全场鼓掌。上线第一天用户传来一个 Excel。日期列里同时出现2024/5/1、2024年5月1日、5-1-2024手机号列混着13800138000和138-0013-8000销售额里还掺了一条面议。你的 Agent 读完这些数据自信地给出一个完全错误的结论。不是因为它笨也不是因为模型不够贵只是因为它吃进去的东西本来就不干净。Garbage in, garbage out在 AI 时代不但没过时反而更狠了。以前的程序吃到垃圾数据顶多报错停下现在的 Agent 吃到垃圾数据更可能一本正经地胡说八道而且语气还特别自信。问题不在 Agent 的脑子而在它前面那道备料工序。这道工序的名字就叫 ETL。ETL 是什么三十秒说清楚ETL 是三个字母的缩写Extract抽取、Transform转换、Load加载。说得直白一点就是三个动作去菜市场、洗菜切菜、装盘上桌。你做的 Agent 像个厨师。厨师手艺再好给他一棵没洗的白菜、一块带血水的肉、一把混着沙子的米他也很难炒出像样的菜。ETL 干的就是这个活在下锅之前先把食材收拾利索。这不是什么新概念数据工程早就讲了很多年。只是到了 Agent 时代它一下又变得扎眼了。传统程序碰到脏数据往往直接报错Agent 碰到脏数据反而可能脑补出一个看起来挺顺的答案继续跑。传统程序吃坏肚子会喊疼Agent 吃坏肚子会笑着说“味道不错”。一个例子贯穿全文摸鱼新闻 Agent假设你要做一个“摸鱼新闻 Agent”。它每天从几个科技网站抓新闻整理干净后存进数据库再由 Agent 读取这些数据生成一份“今天值得看的五条技术新闻”。这条 ETL 线其实很简单Extract从网站抓新闻标题、链接、发布时间Transform去 HTML 标签、统一日期格式、去重、过滤广告Load把干净结果写进 MySQL或其它数据库等 Agent 来读下面就顺着这条线往下拆。每个阶段都有能直接抄的代码顺手也把几个很容易踩的坑拎出来。第一步Extract先把数据搬回来Extract 就记一条先搬回来别急着在这一层做复杂加工。你去菜市场不会站在摊位前洗菜切菜通常都是先买回家。数据采集也一样先把原始数据拿到手清洗和规整留到 Transform 再做。从 API 搬API 算是最省心的数据源返回的是结构化 JSON字段通常也比较稳定。不过这里有个坑最好单独拎出来说永远加timeout。importrequestsdefextract_from_api(url,paramsNone,headersNone):responserequests.get(url,paramsparams,headersheaders,timeout10)response.raise_for_status()returnresponse.json()timeout10这几个字符真的能救命。不加 timeout网络一出问题requests可能就一直挂在那里不回来。流程既不报错也不结束就那么卡住。raise_for_status()也别省。它会把404、500这类 HTTP 错误抛成异常方便你在上层统一处理。否则程序很可能拿着一个错误页面继续往下跑。从数据库搬在企业场景里这往往才是最常见的数据源。订单、日志、用户信息最后基本都绕不开数据库。importpymysqlfromcontextlibimportcontextmanagercontextmanagerdefget_db_connection(config):connpymysql.connect(**config)try:yieldconnfinally:conn.close()defextract_from_db(config,sql,paramsNone):withget_db_connection(config)asconn:withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute(sql,paramsor())returncursor.fetchall()这里有两个地方最好别省用contextmanager保证连接一定关闭。否则一旦中途抛异常连接泄漏起来会非常难查。SQL 参数用%s占位符传值不要自己拼字符串。这不是编码风格问题是最基本的防注入要求。从文件搬Agent 经常要读 CSV、Excel、PDF。这里有个有点反直觉、但很好用的建议先把所有列都按字符串读进来。importpandasaspddefextract_from_csv(file_path):returnpd.read_csv(file_path,dtypestr)为什么要这么干因为pandas的自动类型推断有时会“帮倒忙”手机号可能被识别成数字最后变成13800138000.0身份证、工号这类带前导零的字段前面的零可能被吞掉日期被自动转成时间对象后面和别的源合并时反而更麻烦先原样读进来后面再自己决定怎么转通常更稳。从网页搬网页采集是最脆的一种方式。网站一改版选择器就可能全失效。但有时候没 API也只能硬着头皮爬。importtimeimportrequestsfrombs4importBeautifulSoupdefextract_from_webpage(url,selector,delay1):headers{User-Agent:Mozilla/5.0}time.sleep(delay)resprequests.get(url,headersheaders,timeout10)resp.raise_for_status()resp.encodingresp.apparent_encoding soupBeautifulSoup(resp.text,html.parser)return[el.get_text(stripTrue)forelinsoup.select(selector)]这里我把time.sleep()直接写进示例里因为很多人明明知道“要加延时”复制代码时第一个删掉的还是它。别高频请求别人的服务器轻则被封 IP重则把采集源直接搞没。第二步Transform把脏数据洗干净ETL 最耗时间的通常不是采也不是存而是 Transform。原因不复杂数据源越多、格式越乱清洗规则就越多。更麻烦的是每个数据源的脏法还不一样。你很难写一套规则把所有场景一锅端。去噪把明显垃圾先扔掉importreimportpandasaspddefremove_html_tags(text):returnre.compile(r[^]).sub(,text)defremove_outliers(df,column):q1,q3df[column].quantile(0.25),df[column].quantile(0.75)iqrq3-q1 lower,upperq1-1.5*iqr,q31.5*iqrreturndf[(df[column]lower)(df[column]upper)]去噪的关键不是“会不会写代码”而是你到底知不知道什么才算噪声。温度1000在室温传感器里是异常在工业熔炉里可能完全正常。规则不跟业务对一下最后最容易删掉的往往偏偏是最值钱的数据。标准化让同一种数据只剩一种写法fromdatetimeimportdatetimedefstandardize_date(date_str):forfmtin[%Y-%m-%d,%Y/%m/%d,%Y年%m月%d日,%m/%d/%Y,%m-%d-%Y,]:try:returndatetime.strptime(date_str.strip(),fmt).strftime(%Y-%m-%d)exceptValueError:continuereturnNone标准化这件事看起来很枯燥但特别值钱。一个2024/05/01、一个2024年5月1日、一个5-1-2024人一眼就知道是同一天程序和数据库可未必认。很多数据清洗里的 bug追到最后都很像一句话同一个东西被写成了两种甚至三种表示。结构化让 Agent 能直接吃这一步在 Agent 项目里尤其常见。用户输入、网页正文、客服记录本来都是非结构化文本可 Agent 真正吃起来更喜欢结构化 JSON。在已经有 LLM 的前提下很多场景里用它做结构化提取会比手搓正则稳一些importjsonfromopenaiimportOpenAI clientOpenAI()defstructurize(text,schema):promptf把下面文本抽取成 JSONschema 如下\n{schema}\n\n文本{text}\n只返回 JSON。respclient.chat.completions.create(modelgpt-4o-mini,messages[{role:user,content:prompt}],response_format{type:json_object},)returnjson.loads(resp.choices[0].message.content)response_format{type: json_object}很关键。没有它模型很可能在 JSON 前面先来一句“好的以下是结果”你的json.loads()当场就炸。这种任务一般也没必要上最贵的模型。结构化提取多数时候是体力活不是哲学辩论gpt-4o-mini这类小一档的模型往往就够了。去重同一盘菜别端两次defdeduplicate(data_list,key_field):seenset()result[]foritemindata_list:keyitem[key_field]ifkeynotinseen:seen.add(key)result.append(item)returnresult去重麻烦的地方通常不在代码而在“唯一键到底选什么”。用标题去重同一篇新闻在不同网站上标题可能不同用 URL 去重http和https可能被算成两篇用内容哈希最稳但也最重所以去重真正考验的不是语法而是你到底懂不懂这份数据。第三步Load把干净数据放到该放的地方数据洗干净之后就该装盘上桌了。存去哪儿不看你顺不顺手主要看下游怎么用。关系型数据库适合结构化数据和精确查询比如订单、审批、报表CSV / JSON 文件适合临时交换、离线分析、批处理向量数据库适合语义检索、长期记忆、文档问答如果你要落到 MySQL最该先记住的通常就两件事批量写入和显式提交。importpymysqldefload_to_db(config,table_name,rows):ifnotrows:returnsqlf INSERT INTO{table_name}(title, source) VALUES (%s, %s) values[(row[title],row[source])forrowinrows]connpymysql.connect(**config)try:withconn.cursor()ascursor:cursor.executemany(sql,values)conn.commit()finally:conn.close()这里顺手提醒一句上面这个例子里table_name应该来自你自己可控的配置别直接拿用户输入去拼。业务里如果表名真是动态的最好先做一层白名单校验。executemany()比一条一条插快得多量一上来差距会很明显。conn.commit()也别忘不然很多“我明明写进去了怎么表里没有”的排查最后都只是因为没提交。串起来一条完整的 ETL 管道把前面的步骤串起来一个最小可用版本大概长这样DB_CONFIG{host:127.0.0.1,user:root,password:your_password,database:agent_data,charset:utf8mb4,}defrun_pipeline():print([Extract] 开始采集...)raw_titlesextract_from_webpage(https://tech.example.com/,.news-title a)print(f 搬回来{len(raw_titles)}条原始新闻)print([Transform] 开始清洗...)cleaned[]seenset()foriteminraw_titles:titleremove_html_tags(item).strip()ifnottitleortitleinseen:continueseen.add(title)cleaned.append({title:title,source:tech_web})print(f 洗干净剩{len(cleaned)}条扔掉{len(raw_titles)-len(cleaned)}条脏数据或重复)print([Load] 写入数据库...)load_to_db(DB_CONFIG,tech_news,cleaned)print(f 上桌{len(cleaned)}条)print(食堂备料完成厨师可以下锅了。)这条管道丢进定时任务里每天跑一次就够了。后面的 Agent 直接读tech_news这张表不用再管数据从哪来、原始格式有多乱拿来就能用。最重要的一件事ETL 不是 Agent 的感知模块很多人学到这里会很自然地把 ETL 和 Agent 的“感知模块”混在一起。表面上看它们都在处理外部数据但底层逻辑不是一回事。ETL 是单向管道数据进来洗干净存好结束它不关心“谁来用这些数据”也不关心“用完之后发生了什么”。就像食堂把菜洗好切好放在那里厨师什么时候来炒、炒成什么菜不归备料的人管。Agent 的感知模块不是这样。它在闭环里。Agent 感知完要做决策、执行动作动作执行完返回结果又会变成新的感知输入。举个例子Agent 调了一个订单查询 API返回“该用户没有订单”。这条返回值本身就是新的环境信息。Agent 下一步可能会追问用户是不是输错了订单号换一个时间范围重查直接告诉用户没查到结果关键就在这儿执行结果会反过来影响下一轮判断。而 ETL 没有这个闭环。它负责的是“备料”不是“边做边看边改主意”。所以更准确的分工是ETL把数据从各处搬来洗干净存好Agent 感知模块在交互闭环里理解环境、读取反馈、支持下一步决策如果你现在刚开始做 Agent最务实的路线不是一头扎进“规划、反思、多轮推理”而是先把 ETL 学明白。因为很多看起来像“模型不行”的事故最后根本不是推理出了问题而是输入从一开始就是脏的。最后一句话别跳过备料直接学炒菜。我见过太多人把 Agent 框架学得滚瓜烂熟prompt 也写得飞起结果上线第一天就被一个日期格式问题干趴下。不是 Agent 不行是你喂进去的菜没洗干净。如果你在做 Agent先把这三个动作练熟会稳定地采数据会有边界地洗数据会把干净数据存到对的地方等这些都顺了再去谈感知、规划、反思和自治。顺序别反。版权声明本文为原创整理仅作学习与交流使用。如果这篇文章对你有帮助欢迎点赞、收藏也欢迎关注专栏后续关于 Agent 开发与 AI 应用落地的内容。专栏还有如下博客Token 到底是什么在Claude使用中为什么同样的字数计费能差 6 倍不同模型还不同AI模型都这么强了为什么提示词工程仍然重要6组数据讲透 Prompt 还有没有用LoRA 微调实战手册别再被几十条数据就能训骗了AI这缸中之脑如何触碰现实AI 的脑机接口Function CallTransformer当初凭什么一统天下又将如何被颠覆AI不是百度是伙伴搞懂Harness机制把文字接龙大师榨干成赛博牛马Agent 是什么解决什么问题6 组数据看懂它的真实价值多模态 AI 架构原理解析它是怎么同时看懂图文音视频的