★ FEATURED ARTICLE 1. 数据管道里塞进大模型到底在解决什么麻烦很多团队第一次把大模型接进数据管道脑子里想的还是给数仓加个聊天窗口。这个思路从根上就跑偏了。聊天窗口是给人看的而数据管道是给下游系统、报表、特征工程、风控规则喂数据的。两者对输出的要求完全不同聊天可以容忍模糊、可以来回追问但数据管道要的是结构化、可校验、可回溯、可批量重跑的字段。我见过太多项目卡在同一个地方日志、工单、客服对话、合同扫描件、邮件正文这些非结构化文本明明躺在对象存储里却进不了数仓。传统做法是写正则、写规则、堆关键词字典维护成本高得离谱业务稍微换个说法就全废。大模型的出现本质上是把从自由文本里抽取结构化字段这件事从规则工程变成了语义工程。但这里有个关键认知大模型不是数据管道的目的它只是管道里的一个算子。就像你不会把整个ETL都写成Python UDF一样你也不该让大模型承担管道的调度、重试、幂等、血缘这些职责。它只负责一件事——把一段非结构化文本映射成一组符合schema的字段。剩下的交给成熟的数据工程体系。这篇文章想聊的就是怎么把这个算子稳稳地嵌进数据管道让非结构化数据真正落地成数仓里可用的表。适合已经会用大模型API、但还没想清楚怎么把它工程化的数据开发、数据平台、以及做AI应用落地的同学。我会把踩过的坑、选型的逻辑、以及能直接抄的配置都摊开讲。2. 为什么直接调API写进表这条路走不通2.1 一次性调用的三个致命假设最朴素的实现是这样的读一条文本拼个prompt调一次模型解析JSONinsert进表。跑通demo没问题但一上生产就崩。因为它隐含了三个假设而这三个假设在真实数据管道里全都不成立。第一个假设是模型输出永远合法。实际上大模型会返回多余的解释文字、会漏字段、会把数字写成中文、会在JSON外面套一层markdown代码块。你如果直接json.loads失败率能到两位数百分比。第二个假设是每条数据独立处理就够了。但数仓要的是分区、要的是批次一致性。你今天跑一批、明天跑一批同一份源数据可能因为模型版本更新而抽出不同结果下游对不上账。第三个假设是调用是免费的、无限速的。真实场景里你要考虑token成本、并发限流、超时重试。一条文本抽8个字段和抽3个字段成本差一大截。2.2 把大模型当函数而不是服务正确的定位是把大模型封装成一个纯函数式的抽取算子。输入是文本schema定义输出是结构化记录或明确的失败标记。它不碰数据库、不管调度、不做重试——这些交给外层的管道框架。这个定位一旦确立很多设计就顺了。比如幂等性同一条文本、同一个prompt版本、同一个模型版本必须产出同一个结果。做不到怎么办把prompt_version和model_version作为字段写进结果表重跑时按版本比对。再比如可回溯每条抽取结果都要能追到原始文本的存储路径和偏移量否则出了问题你连原文都找不到。提示在管道里大模型算子的输出必须是确定性可复现的。如果模型本身有随机性temperature0要么设成0要么把随机种子固定下来否则你的数仓表每天都在变。2.3 一个反直觉的结论先做schema再选模型大部分人顺序反了——先挑个最强的模型再想抽什么。正确顺序是先定义目标表的schema再倒推prompt和模型选型。原因很简单数仓表的字段是下游契约。你抽出来的字段名、类型、是否可空直接决定BI能不能用、特征工程能不能接。如果schema没定死模型抽得再花哨也是白搭。我通常的做法是先和下游对齐一张目标表DDL把字段、类型、枚举值范围全部写清楚然后才去设计抽取逻辑。举个例子假设你要从客服对话里抽工单信息目标表可能长这样字段名类型是否可空说明ticket_idstring否源系统工单号抽不到则整条丢弃categorystring否枚举退款/物流/质量/其他urgencyint否1-5抽不到默认3summarystring是一句话摘要限200字extracted_attimestamp否抽取时间prompt_versionstring否用于回溯有了这张表prompt要抽什么、哪些字段必填、哪些可以兜底一目了然。模型选型也有了依据如果只是分类短摘要小模型足够如果要抽多轮对话里的隐含诉求才需要上大模型。3. 抽取算子的核心设计从prompt到结构化输出3.1 prompt不是写作文是写接口契约很多人写prompt像写需求文档一大段自然语言。这在抽取场景里是灾难。抽取prompt应该像接口文档明确输入、明确输出格式、明确边界情况。我的模板结构固定为四段角色定义、任务说明、输出schema、few-shot示例。角色定义一句话带过任务说明讲清楚从什么文本抽什么重点是输出schema——用JSON Schema描述字段名、类型、枚举、必填全部写死。{ type: object, properties: { category: { type: string, enum: [退款, 物流, 质量, 其他] }, urgency: { type: integer, minimum: 1, maximum: 5 }, summary: { type: string, maxLength: 200 } }, required: [category, urgency] }把这个schema直接塞进prompt并明确告诉模型只输出JSON不要任何解释。实测下来配合few-shot示例合法输出率能从60%提到95%以上。3.2 输出解析必须做三层防御即便prompt写得再好也不能假设输出一定合法。解析层要做三层防御缺一不可。第一层是清洗。模型经常在JSON外面包json代码块或者前面加一句好的以下是结果。用正则把代码块标记剥掉再定位第一个{和最后一个}截取中间部分。第二层是校验。用JSON Schema校验器Python里用jsonschema库逐字段检查类型、枚举、范围。校验失败的字段单独标记不要整条丢弃。第三层是兜底。必填字段缺失时按业务规则给默认值枚举值不在范围内时映射到其他数字超范围时截断到边界。兜底策略要写进配置不能硬编码。import json import re from jsonschema import validate, ValidationError def parse_llm_output(raw, schema, defaults): # 第一层清洗 text re.sub(rjson|, , raw).strip() start, end text.find({), text.rfind(}) if start -1 or end -1: return None, no_json_found try: data json.loads(text[start:end1]) except json.JSONDecodeError: return None, json_decode_error # 第二层校验 try: validate(instancedata, schemaschema) except ValidationError as e: # 第三层兜底 for field, default in defaults.items(): data.setdefault(field, default) return data, ok3.3 批处理与并发别让限流拖垮整条管道单条调用在生产里毫无意义。真实场景是几百万条文本要抽。这时候要考虑并发和限流。我的做法是用生产者-消费者模型一个读取器从源存储拉文本塞进队列多个worker从队列取任务调模型结果写入一个中间层比如Parquet文件或临时表。worker数量根据API的QPS限制来定通常设成限流的70%留余量。关键细节是超时和重试。模型调用超时设30秒失败重试3次每次退避2倍。重试仍失败的写入死信队列不要阻塞主流程。死信队列每天人工或自动重跑一次。注意并发不是越高越好。我试过把并发拉到限流上限结果大量请求被拒反而比70%并发慢。留余量是为了应对突发和重试。4. 让结果真正进数仓落地层的工程细节4.1 中间层设计为什么不能直接写目标表抽取结果直接insert目标表看起来省事实则埋雷。一旦抽取逻辑要改、要重跑目标表就被污染了。正确做法是加一个中间层——通常是对象存储上的Parquet分区文件或者一张带batch_id的临时表。中间层的好处有三个一是可重放原始抽取结果保留重跑只覆盖中间层二是可对比新旧prompt版本的结果可以diff三是可审计每条记录都能追到源文本。中间层的分区键我一般用dt抽取日期source数据源标识。文件格式用Parquet压缩用snappy兼顾体积和读取速度。4.2 从中间层到目标表清洗、去重、合并中间层到目标表这一步才是真正的数仓工程。要做三件事清洗、去重、合并。清洗是处理抽取层的兜底残留比如summary超长截断、category空值填充。去重按业务主键比如ticket_id做同一主键保留最新extracted_at的记录。合并用MERGE INTO或INSERT OVERWRITE按分区覆盖。这里有个容易忽略的点主键冲突。如果源文本里同一个工单号出现多次抽取结果可能不一致。我的处理是保留置信度最高的那条——置信度可以由模型返回也可以用规则算比如字段完整度。MERGE INTO dwd_ticket_info AS t USING ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY ticket_id ORDER BY extracted_at DESC, completeness DESC ) AS rn FROM stg_ticket_extract WHERE dt ${bizdate} ) AS s ON t.ticket_id s.ticket_id WHEN MATCHED AND s.rn 1 THEN UPDATE SET ... WHEN NOT MATCHED AND s.rn 1 THEN INSERT ...4.3 血缘与版本出了问题能倒查大模型抽取最怕的是结果不对但不知道哪一步错了。所以血缘必须做全源文本路径、prompt版本、模型版本、抽取时间、解析状态全部作为字段落表。我还会额外记录一个raw_output字段存模型原始返回。虽然占空间但排查问题时能救命。存储成本可以用冷分区控制比如只保留最近30天的raw_output更早的归档到低频存储。版本管理上prompt和模型版本用语义化版本号如v1.2.0。每次改prompt必须升版本不能原地改。这样重跑时能精确指定用哪个版本结果可复现。5. 实测中绕不开的五个坑5.1 坑一模型自作聪明补全不存在的字段模型有个坏习惯文本里没提到的字段它会根据上下文猜一个填上。比如文本没提紧急程度它给你填个3。这在抽取场景里是致命的——你分不清哪些是抽出来的哪些是编的。解决办法是在prompt里明确要求文本未提及的字段返回null不要猜测。同时在解析层做校验如果某字段在原文里找不到对应证据标记为低置信度。更严格的做法是让模型返回每个字段的原文片段作为证据解析时校验片段是否真的在原文里。5.2 坑二长文本截断导致信息丢失模型有上下文长度限制长文本必须截断。但截断位置很讲究——截掉的部分可能正好包含关键信息。我的策略是分段抽取合并。把长文本按段落切成多个chunk每个chunk独立抽取最后按字段合并。合并规则分类类字段取多数投票数值类字段取加权平均摘要类字段拼接后二次摘要。分段时注意不要切断语义单元。按句号、换行符切比按固定长度切效果好得多。5.3 坑三成本失控——token消耗远超预期很多人算成本只算输入token忽略了输出token和重试消耗。实测下来输出token往往占大头因为模型喜欢啰嗦。控制成本的手段有三个一是prompt精简去掉冗余说明二是限制输出长度用max_tokens卡死三是缓存相同文本相同prompt版本直接命中缓存不重复调用。缓存用Redis或本地KV都行key用文本hashprompt版本。5.4 坑四模型版本升级导致结果漂移模型服务方升级模型是常事但对你来说可能是灾难——同一份文本昨天抽出来是退款今天变成其他。应对方式是锁定模型版本。如果服务方支持指定版本号就写死如果不支持就定期做回归测试用一批标注好的样本跑一遍对比新旧结果。漂移超过阈值就告警人工介入。5.5 坑五下游不认账——字段语义对不齐技术上都跑通了下游却说这个字段不能用。原因往往是语义没对齐。比如你抽的urgency是1-5下游以为是1-3你抽的category是中文枚举下游要英文。这类问题只能在schema定义阶段解决。我的做法是拉上下游一起评审目标表DDL每个字段的语义、取值范围、空值含义全部确认签字。评审通过再动手能省掉后期大量返工。6. 一套可复用的落地清单把上面的经验浓缩成一份清单新项目可以直接照着走。阶段一定义契约与下游对齐目标表DDL字段名、类型、枚举、空值含义全部确认确定业务主键和去重规则确定哪些字段必填、哪些可兜底阶段二设计抽取算子用JSON Schema定义输出结构prompt按角色任务schemafew-shot四段式写解析层做清洗、校验、兜底三层防御记录prompt版本和模型版本阶段三搭建管道中间层用Parquet分区存储保留原始输出并发按限流70%设置超时30秒重试3次死信队列单独处理不阻塞主流程阶段四落地数仓中间层到目标表做清洗、去重、合并血缘字段全量落表raw_output保留30天版本号语义化管理改prompt必升版本阶段五运维定期回归测试监控结果漂移成本监控token消耗按天统计死信队列每日重跑这套流程我在几个项目里跑下来非结构化数据进数仓的成功率能稳定在95%以上剩下的5%进死信队列人工处理。关键是每一步都有据可查出了问题能快速定位。最后分享一个小心得别追求100%的抽取准确率。非结构化数据本身就有模糊性强行追求完美只会让成本爆炸。把准确率目标定在业务可接受的阈值比如90%剩下的用人工审核或下游容错兜住整体ROI反而更高。数据管道要的是稳定和可维护不是学术意义上的最优。
阅读完成 · 觉得有帮助？
