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管道特有的大头，决定了下游检索质量。服务（Serve）：Spring 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 List<RawDocument> 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 List<DocumentChunk> transform(RawDocument doc, int chunkSize, int overlap) { var splitter = new TokenTextSplitter(chunkSize, overlap, null, null, null); List<org.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到800，overlap取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) { List<RawDocument> docs = extractor.extract(offset, BATCH_SIZE); for (RawDocument doc : docs) { List<DocumentChunk> chunks = transformer.transform(doc, 600, 80); List<org.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) { List<org.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) {} Map<String, 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功能都会跟着不好，这个因果关系越早想明白越好。
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管道特有的大头，决定了下游检索质量。服务（Serve）：Spring 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 List<RawDocument> 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 List<DocumentChunk> transform(RawDocument doc, int chunkSize, int overlap) { var splitter = new TokenTextSplitter(chunkSize, overlap, null, null, null); List<org.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到800，overlap取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) { List<RawDocument> docs = extractor.extract(offset, BATCH_SIZE); for (RawDocument doc : docs) { List<DocumentChunk> chunks = transformer.transform(doc, 600, 80); List<org.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) { List<org.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) {} Map<String, 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功能都会跟着不好，这个因果关系越早想明白越好。