1. 为什么AI应用开发需要先把数据管道想清楚标题里的“Spring AI ELT”这两件事乍一看一个是大模型应用框架一个是数据工程方法论好像没什么交集。但做了几个月的AI功能开发之后我得说一句真正让AI应用卡壳的从来不是模型调用代码怎么写而是数据怎么喂进去、喂得干不干净、喂完还能不能持续更新。这也是我决定把这两个词放在一起写一篇的原因。先说我遇到的真实情况。很多团队上手Spring AI时第一反应是“把大模型API包一层Service然后让业务系统调用”。做Demo确实很快两三天的功夫就能跑通一个“智能问答”界面。可一旦进入生产环境问题就全冒出来了模型的上下文窗口装不下业务方提供的十几万字资料模型对你们公司内部的订单状态、客户等级、业务规则一无所知你以为把PDF传上去就行结果模型经常一本正经地胡说八道而且业务数据天天在变昨天还准确的回答今天就成了过时信息。这些问题本质上都不是“模型能力不行”而是“数据没准备好”。把数据库里的结构化记录、文件服务器里的文档、消息队列里的实时事件经过抽取、处理、组织之后变成模型能够理解和检索的形式——这个环节正好就是ELT干的事。ELT是提取、加载、转换Extract、Load、Transform的缩写和传统ETL最大的区别是先把数据原样加载到目标存储里再在存储内部做转换加工。所以这篇文章我会讲清楚两件事一件是把ELT这套思路用在Spring AI应用里解决“模型没有业务知识、数据没有更新机制”的问题另一件是反过来用AI能力增强ELT管道本身让数据清洗、打标签、实体抽取这些苦活自动完成。适合的人群是正在用Spring AI做实际项目的Java后端开发者以及准备评估AI应用落地可行性、但还没想清楚数据层怎么设计的架构师。2. 为什么ELT比ETL更适合AI应用的数据场景2.1 ETL和ELT的区别在哪里用最简单的话说ETL是“先洗菜再下锅”ELT是“先全部倒进锅再在锅里挑拣”。传统数据仓库时代数据量还没那么大团队倾向于在管道里先完成清洗、过滤、转换再把干净的数据加载到目标库这样做的好处是下游查询性能好、数据质量可控。但它的缺点也很明显转换逻辑一旦写死后面想加新的数据处理规则就得改管道重跑而且抽取的原始数据被丢掉后后续要用原始数据做新的分析就没了依据。ELT的思路是先把数据完完整整地加载到目标存储里原始数据保留下来转换动作在存储层面按需执行。这样做的好处有三个一是存储和计算分离之后管道变得简单直接二是原始数据不丢随时可以重新加工三是转换可以分批做、按需做、增量做不用一上来就把所有转换逻辑定死。AI应用的数据场景恰好特别吃ELT这套灵活性。因为大模型相关的数据处理流程经常变这周用三句话做摘要下周可能想改成五句话保留更多细节这周按固定大小切块下周发现按章节切效果更好。如果转换逻辑都写死在同步管道里每次调整都是一次重新抽数、重新入库的大工程。2.2 面向AI的ELT管道到底是什么样Spring AI应用的数据管道和传统数仓的ELT有一点本质差别传统ELT的“转换”是SQL聚合、字段映射、清洗去重面向AI的ELT转换环节多了一个关键动作就是调用嵌入模型把文本转成向量。这是AI应用能实现语义检索的前提。我习惯把这条管道分成四个环节来设计。抽取Extract从关系型数据库、文件系统、消息队列、第三方接口里把原始数据拿进来可能是订单记录可能是产品文档也可能是客户工单。加载Load把原始数据尽量原样写入目标存储。这里的存储通常是对象存储、关系型数据库或专门的知识库服务。加载阶段最重要的原则是“别丢字段”原始内容必须完整保留。转换Transform在目标存储之上做切块、清洗、生成摘要、调用嵌入模型做向量化、加索引元数据。这个阶段是AI管道特有的大头决定了下游检索质量。服务ServeSpring AI应用在用户提问时先到向量库里做相似度检索把命中的上下文拼进Prompt再调用大模型生成回答。这个结构里抽取和加载可以做成一个相对稳定的同步管道转换则做成一个可以随时手动触发、增量执行的任务模块。这个设计从骨架上就避免了“每次改切块策略都要重跑全量抽取”的窘境。3. 一条完整的Spring AI数据管道怎么拆解3.1 先分清楚数据源和目标存储设计管道的第一步不是写代码而是把所有数据源列明白。我经手的项目里数据源的种类通常逃不出这几类业务库里的结构化表比如用户信息、订单记录文件服务器上的半结构化正文比如合同PDF、产品手册、帮助文档还有消息通道里的实时变更事件。目标存储的选择取决于你要做什么。只做检索问答可以选一款向量数据库存切块后的文本和向量要做数据分析类的问答需要在关系型数据库里维护一份清洗后的宽表如果要同时支持关键词检索和语义检索那就要在存储里同时保留结构化字段和向量字段。这里我踩过一个坑一开始图省事把抽取和转换在同一个事务里做完也就是“读取一条、切块、向量化、写入”串成一个循环。结果数据量上来之后抽取慢、转换也慢而且每次改向量化策略所有数据都得重新拉一遍。后来老老实实把抽取和转换拆开先批量把原始文档和结构化记录拉进一张原始数据表再异步处理向量化速度一下就上来了。3.2 抽取阶段怎么做才不容易出错抽取阶段看起来简单无非是读数据源但实际操作里最容易出事的是对变更的感知。业务库里的数据天天变全量重抽太笨重增量抽取又得考虑怎么追踪变更。我在项目里一般用两套手段配合。关系型数据库的表如果有update_time这类字段就靠时间戳做增量抽取每次记录上次抽取的游标如果表结构里没有时间字段就加一张增量记录表用主键比对找出新增和变更的数据。文件类型的资料相对稳定一般用指纹比对处理完的文件记录一个哈希值下次遇到相同的哈希直接跳过。抽取阶段另一个容易忽略的问题是“原始数据必须留底”。很多人做AI管道时只保留切块后的文本觉得原始文档没用了。但等到要调整切块策略、要重新生成摘要、要排查某条回答到底引用了来源的哪一段时没有留底就只能干瞪眼。所以我在设计原始数据表时一定会包含这些字段数据源标识、业务主键、原始内容、采集时间、处理状态。3.3 加载和转换的边界怎么划分ELT里“先加载再转换”的边界放到Spring AI场景下我建议这样切加载阶段只做数据搬运和字段保留把原始内容完整写进存储不做任何语义层面的处理转换阶段统一处理切块、清洗、摘要和向量化。这样划分有一个直接好处转换任务可以“排队重跑”。比如某天你换了更好的嵌入模型旧向量不需要了那只要把状态重置成待处理转换任务就会把所有记录重新向量化加载阶段的原始数据完全不用动。加载阶段还有一个容易忽视的技术点写入要支持幂等。管道重跑是常态如果同一份数据被加载两次就产生重复记录后面的检索质量会受很大影响。我的做法是在原始数据表上建唯一索引用“数据源业务主键”作为唯一键写入时用数据库的冲突处理机制保证同一份数据只存在一份。转换阶段的重点工作是切块。切块策略直接影响检索效果我后面会单独展开说。这里只强调一点切块和向量化最好分开两个步骤。先基于原始文本切块把切块结果也存成一张表再对每个切块做向量化。这样一旦你想调整“用哪几个块拼上下文”或者想重新向量化都不用重新切块。4. 核心代码骨架Spring AI数据管道的落地实现4.1 项目结构和关键依赖这一段给一个可以直接抄作业的最小实现骨架。假设我们要实现一条管道从关系型数据库读取产品文档切块后调用嵌入接口向量化写入向量存储再通过Spring AI的问答接口对外提供服务。工程结构大致如下com.example.aieltpipeline ├── pipeline │ ├── extractor // 数据源抽取 │ ├── loader // 原始数据加载 │ ├── transformer // 切块、向量化 │ └── runner // 管道调度 ├── service │ └── ChatService // 面向业务的问答服务 └── controller └── ChatController依赖上Spring Boot项目里引入Spring AI核心包再加关系型数据库访问的依赖和一个向量数据库的客户端依赖。要注意的是Spring AI本身还提供了一个叫“文档读取器”的组件可以直接读取PDF、Word、Markdown这些文件类型它会负责把文件内容解析成文本。用这个组件能省掉不少文件解析的工程量。4.2 抽取和加载环节的实现要点抽取环节我用JdbcTemplate写了一个简单的数据读取器。最关键的一点是分页读取而不是一次性把所有数据load进内存。文档数据多了以后一次性读取很容易把内存撑爆。Component public class DocumentExtractor { private final JdbcTemplate jdbcTemplate; public DocumentExtractor(JdbcTemplate jdbcTemplate) { this.jdbcTemplate jdbcTemplate; } public ListRawDocument extract(int offset, int limit) { String sql SELECT id, doc_code, title, content, update_time FROM source_document ORDER BY id LIMIT ? OFFSET ? ; return jdbcTemplate.query(sql, (rs, rowNum) - new RawDocument( rs.getLong(id), rs.getString(doc_code), rs.getString(title), rs.getString(content), rs.getTimestamp(update_time).toLocalDateTime() ), limit, offset); } }加载环节的核心是幂等写入。建表时给doc_code加唯一索引入库用MERGE这种“有则更新、无则插入”的写法。这样管道重复跑多少次数据表里都不会出现重复记录。这个细节我反复跟人强调过很多人一开始不重视等向量库里出现大量重复块、问答结果乱套的时候才后悔。4.3 转换环节切块与向量化切块策略我一般用Spring AI里的TokenTextSplitter或者自己写一个基于段落分割器。直接上代码Component public class DocumentTransformer { private final EmbeddingModel embeddingModel; public DocumentTransformer(EmbeddingModel embeddingModel) { this.embeddingModel embeddingModel; } public ListDocumentChunk transform(RawDocument doc, int chunkSize, int overlap) { var splitter new TokenTextSplitter(chunkSize, overlap, null, null, null); Listorg.springframework.ai.document.Document chunks splitter.apply( List.of(new org.springframework.ai.document.Document(doc.content())) ); return chunks.stream().map(chunk - { float[] vector embeddingModel.embed(chunk.getText()); return new DocumentChunk(doc.id(), chunk.getText(), vector); }).toList(); } }这段代码里TokenTextSplitter负责把长文本按token数切块overlap参数控制相邻块之间的重叠量。embeddingModel是Spring AI里对嵌入模型的统一抽象底层调用哪家的接口由配置文件决定业务代码不需要关心。切块参数的选择我会在常见问题那一节详细说这里先给一个经验值一般生产环境里chunkSize取500到800overlap取50到100具体要根据你的文档类型和模型上下文窗口来调。技术文档、合同这种结构性强的文本overlap可以小一点对话记录这种没有明显边界的文本overlap需要大一些。4.4 管道调度和问答服务怎么衔接管道可以做成一个定时任务也可以手动触发。我实现了一个简单的PipelineRunner持有抽取器和转换器的引用一个方法跑完一个批次Component public class PipelineRunner { private static final int BATCH_SIZE 50; private final DocumentExtractor extractor; private final DocumentTransformer transformer; private final VectorStore vectorStore; public PipelineRunner(DocumentExtractor extractor, DocumentTransformer transformer, VectorStore vectorStore) { this.extractor extractor; this.transformer transformer; this.vectorStore vectorStore; } public void runBatch(int offset) { ListRawDocument docs extractor.extract(offset, BATCH_SIZE); for (RawDocument doc : docs) { ListDocumentChunk chunks transformer.transform(doc, 600, 80); Listorg.springframework.ai.document.Document aiDocs chunks.stream() .map(chunk - org.springframework.ai.document.Document.builder() .withText(chunk.text()) .withMetadata(sourceDocId, chunk.sourceDocId()) .build()) .toList(); vectorStore.add(aiDocs); } } }跑完管道之后业务侧的问答服务就简单了。直接用Spring AI的ChatClient把用户问题先丢给向量库检索拿命中的文本片段做上下文再让大模型基于上下文回答。Service public class ChatService { private final ChatClient chatClient; private final VectorStore vectorStore; public ChatService(ChatClient.Builder builder, VectorStore vectorStore) { this.chatClient builder.build(); this.vectorStore vectorStore; } public String answer(String question) { Listorg.springframework.ai.document.Document hits vectorStore.similaritySearch(question); String context hits.stream() .map(d - d.getText()) .reduce(, (a, b) - a \n---\n b); return chatClient.prompt() .system(你是一名擅长回答业务问题的助手。你必须只依据提供的资料回答资料中没有的信息要明确说不知道。) .user(资料\n context \n\n问题 question) .call() .content(); } }这里有个特别重要的参数系统Prompt里一定要约束模型“只依据提供的资料回答”。不写这句话大模型就会开始自由发挥把自己训练数据里的内容混进答案里检索增强就失去了意义。5. 用AI反向增强ELT管道让LLM做清洗和打标5.1 为什么管道本身也需要AI参与数据管道从业务库和文件系统里抽出来的数据质量往往一言难尽。合同文档可能是扫描件转文字结果字错误一大堆工单记录里客户把产品型号写错、把问题描述得模棱两可业务系统里的“地址”字段各种缩写格式混用。传统做法是写正则表达式、维护字典映射能解决一部分但遇到没有规律的文本就只能靠人工。这就是用AI增强ELT管道的价值。大模型的语义理解能力在文本清洗、实体抽取、自动摘要、分类打标这些任务上表现相当好可以直接在管道转换阶段调用。而且Spring AI给了一套不错的抽象把Prompt模板、输出解析这些东西都封装好了写起来不算费事。我最早是在一个工单分类项目里尝试这个思路。人工给工单分大类效率太低我写了个Spring AI的批处理程序每天凌晨把前一天的新工单捞出来调用大模型输出工单类别、紧急程度、关联产品线三个字段再回写到表里。跑了一个月分类准确率稳定在人工水平的九成以上成本也远低于招人专门做标注。5.2 用Spring AI做结构化输出的正确姿势用LLM做数据清洗和打标最怕的就是输出格式不稳定。同一个Prompt这轮给你JSON下轮给你一段话里面夹着JSON。解决办法是用Spring AI的输出解析器让模型严格按固定结构输出。看一段实际代码我要让模型给每一条工单输出一个结构化标签public record TicketLabel(String category, String urgency, String productLine) {} MapString, String ticket getTicket(id); String promptTemplate 你是一名工单分类助手。 请阅读下面这条客户工单输出分类结果。 工单标题%s 工单内容%s 要求 1. category必须是售后咨询、退换货、技术支持、投诉建议、其他五选一。 2. urgency必须是低、中、高三选一。 3. productLine写产品线名称不确定时写“未知”。 只输出JSON对象不要输出任何其他内容。 ; String promptText promptTemplate.formatted(ticket.get(title), ticket.get(content)); TicketLabel label chatClient.prompt() .user(promptText) .call() .entity(TicketLabel.class);核心就是最后那行.entity(TicketLabel.class)。Spring AI会把返回文本按类结构解析并且会自动处理模型偶尔在JSON前后加的多余文字。这个比自己在代码里写JSON解析靠谱得多我强烈建议别手写解析。5.3 批次处理、成本控制和失败重试用LLM做数据清洗和打标要面对三个实际问题。一是成本。每调一次接口都是花真金白银所以同样的任务能合并就合并。比如工单分类可以一次把5条工单放在一个Prompt里让模型一次性输出5个JSON对象成本能省很多。代价是单次Prompt变长后某些模型偶尔会漏输出所以要对结果数量做校验少了字段就单独补一次调用。二是失败重试。大模型接口偶尔超时、偶尔限流是家常便饭。管道代码里每个AI调用环节都必须加重试而且要区分“可重试错误”和“不可重试错误”。超时、限流可重试参数错误、内容触发了安全策略就别重试了记录下来让人排查。Spring框架自带的重试工具可以直接用。三是质量兜底。AI清洗不是万能的分类结果里会有少量明显错误。所以我在管道设计里一定会加一道人工复核队列置信度低的记录、模型自己判定为“不确定”的记录自动进复核清单。人只处理机器处理不了的少数几条成本低效果也稳。6. 常见问题与排查技巧实录6.1 向量化重复数据导致回答错乱症状同样的问题问两次答案不一致而且有时候答案里混着明显无关的内容。排查路径先看向量库里有没有重复块。最直接的办法统计一下同一份sourceDocId下的切块数量如果远超预期说明管道重复执行了但幂等没做好。另一个排查点检索时有没有把数据源过滤带上。向量库的相似度搜索支持按元数据过滤很多时候答案是拿A项目的资料回答B项目的问题就是因为没按项目ID过滤。6.2 切块边界切坏了语义症状模型回答时老觉得“话说到一半就停了”引用的资料断断续续。原因基本出在切块策略上。固定按token数切块最容易把一句话从中间断开。我做过的有效调整有两种第一是用段落感知的分割器先按段落、标题切再对超长段落内部做二次拆分第二是加大overlap值让相邻块有足够的重复内容这样检索时能互补多少补回来一点。6.3 模型输出解析失败症状调用.entity()的时候抛异常提示JSON解析失败。原因多半是模型输出的字段值不在你定义的枚举里尤其是“五选一”这种受限字段模型偶尔会自由发挥给你整出个第六种。解决办法分两层。Prompt里把候选值和输出模板写得更死板比如明确说“只输出以下五种之一不要加任何解释”代码里则要容忍未知值枚举解析用兜底默认值现在的AI接口偶尔抽风是常态代码一定要把异常和脏数据当成常规输入处理。6.4 Spring AI版本升级导致API变动这个问题最折磨人。Spring AI还处于快速迭代期版本升级经常带来方法签名变动、配置项改动。同一个功能这个版本用这个方法升级之后的方法名和参数全变了。我现在的策略是能锁版本就锁版本别追新升级之前先看迁移说明别靠猜。项目里把Spring AI的版本号单独提出来管理不跟Spring Boot的大版本走。6.5 常见问题速查表现象可能原因排查与解决答案内容与资料无关向量检索命中错误分块检查切块策略检索时带元数据过滤同样的数据反复被模型引用向量库重复数据给原始数据建唯一索引加载阶段保证幂等清洗打标结果格式乱模型输出不稳定用Spring AI输出解析器Prompt固定输出模板管道跑批越来越慢全量重跑无断点续跑增加处理状态字段支持失败记录跳过模型接口调用频繁超时并发过高或限流引入重试批次大小调小加指数退避新文档迟迟无法被问答命中管道未触发转换检查状态机流转确认向量化任务有调度调度7. 我的一些实操体会Spring AI ELT这套组合最值钱的地方其实不是某个具体技术点而是把“数据生命周期”的思维带进了AI应用开发。做过一段时间的AI功能之后你就会发现模型调用代码是全世界最相似的代码真正拉开项目质量差距的是数据管道是否可靠、是否可重跑、是否能容忍脏数据。我个人比较推荐的做法是先把不依赖AI的数据管道跑起来也就是抽取、加载、基础切块、向量化这条链路先通了能用现有数据完成问答再逐步引入AI辅助清洗和自动打标。一步一步走每个环节都能验证出问题也好定位。一上来就搞“AI全自动管道”的同学排查问题的时候通常很痛苦因为不知道哪一层出了问题。另外向量库的选择不用太纠结。先选一个社区活跃、Spring AI适配良好的跑通为主。真到了数据量几十万、并发要求高的阶段再根据性能瓶颈去评估替换也不迟。数据管道本身要留好抽象边界存储实现替换不会影响上下游的ETL逻辑。最后分享一个小经验在给AI管道做监控的时候别只盯着接口调用成功率这类基础设施指标一定要看数据质量指标。比如每次跑批之后处理了多少条、处理失败多少条、模型自报“不确定”的又有多少条。这些指标才是管道的真实健康度。数据不好上面做的所有AI功能都会跟着不好这个因果关系越早想明白越好。
