Flink异步IO调用大模型实战：架构设计与性能优化指南

如果你正在构建一个实时数据处理系统，比如实时推荐、欺诈检测或智能客服，你可能会面临一个经典困境：流处理引擎（如 Flink）擅长处理海量、高速的结构化数据，但面对文本理解、情感分析、图像识别等复杂任务时，却显得力不从心。而另一边，大模型（LLM）在这些认知任务上表现出色，但其推理延迟高、资源消耗大，难以直接嵌入到低延迟的流处理管道中。

那么，一个自然的想法是：能否让 Flink 调用大模型，将流处理的实时性与大模型的智能性结合起来？这个想法听起来很美，但实际效果如何？是“1+1>2”的架构创新，还是“牛头不对马嘴”的技术缝合？

本文将通过一个完整的实战项目，带你深入探索 Flink 调用大模型的真实效果。我们将从架构设计、代码实现、性能瓶颈到最佳实践，逐一拆解。读完本文，你将能清晰地判断：在你的业务场景下，Flink + 大模型是否值得投入，以及如何规避其中的“深坑”。

1. 这篇文章真正要解决的问题

Flink 调用大模型，核心要解决的是“实时流”与“慢推理”之间的矛盾。这不是一个简单的 API 调用问题，而是一个涉及系统架构、资源管理、容错性和成本控制的复杂工程挑战。

很多开发者容易陷入两个误区：

1. 过度乐观：认为只需在 Flink 的MapFunction里发个 HTTP 请求调用大模型 API 就万事大吉，忽略了延迟激增、背压、API 限流和成本爆炸等问题。
2. 过度悲观：认为两者根本不适合结合，从而放弃探索更高效的实时智能应用可能性。

本文要解决的，正是介于这两者之间的务实路径。我们将探讨：

• 什么场景下值得尝试这种组合？（例如：对延迟有一定容忍度的实时内容审核、异步的个性化摘要生成）
• 如何设计架构来平衡实时性与大模型开销？（例如：异步调用、批处理窗口、旁路输出）
• 在代码层面如何实现稳定、高效的调用？（包括重试、降级、监控）
• 实际运行时会遇到哪些性能瓶颈？如何量化评估“效果”？（不仅是功能效果，更是系统效果）
• 有哪些现成的模式和最佳实践可以借鉴？

如果你正在评估或设计一个需要实时智能决策的系统，这篇文章将为你提供从理论到实践的完整路线图。

2. 基础概念与核心原理

在深入实战之前，我们需要统一几个关键概念，并理解其结合的内在逻辑。

2.1 Flink：流处理引擎的核心能力

Apache Flink 是一个分布式、高性能、高可用的流处理框架。它的核心优势在于：

• 有状态计算：能够在处理无界数据流时维护状态（如计数器、聚合值、窗口内容），这是实现复杂事件处理的基础。
• 精确一次（Exactly-Once）语义：确保数据即使在发生故障时也不会丢失或重复处理，对于金融、计费等场景至关重要。
• 低延迟与高吞吐：通过内存计算、流水线执行和优化算子链，实现毫秒级延迟和每秒百万级事件的处理能力。
• 丰富的API：提供了DataStream API（更灵活、更底层）和Table API / SQL（声明式，更易用）来构建流式应用。

2.2 大模型（LLM）的推理特点

这里的大模型主要指用于自然语言处理（NLP）或视觉任务的大型预训练模型（如 GPT、LLaMA、ChatGLM 等）。其推理过程的特点是：

• 计算密集：需要强大的 GPU 或 NPU 进行张量计算。
• 延迟较高：一次生成式推理通常在几百毫秒到数秒之间，远高于传统数据库查询或规则计算。
• 非确定性（一定程度）：相同输入可能产生不同输出（取决于温度参数）。
• 常通过 API 服务化：大多数团队通过部署模型服务（如使用 vLLM、TGI、或直接调用 OpenAI、DeepSeek 等云端 API）来提供推理能力。

2.3 结合点与核心挑战

将 Flink 与大模型结合的典型模式是：Flink 处理实时流，将需要“智能处理”的数据（如一条用户评论、一张图片）发送给大模型服务，然后将模型返回的结果（如情感标签、摘要）与原始流继续向下游处理或输出。

核心挑战由此产生：

1. 同步调用阻塞流：如果在 Flink 算子内同步调用大模型 API，整个算子的处理线程会被阻塞，等待数秒。这会导致严重的背压（Backpressure），上游数据无法及时处理，最终可能拖垮整个作业。
2. 资源管理困难：大模型服务是独立资源池。Flink 作业的并发度（Parallelism）变化，如何动态匹配模型服务的承载能力？如何避免对模型服务的洪峰请求？
3. 容错与一致性：如果模型服务调用失败，Flink 作业该如何处理？重试可能导致重复消费和状态不一致。如何保证“精确一次”语义在涉及外部系统时依然有效？
4. 成本与效率：大模型 API 调用通常按 token 计费。流式数据可能产生大量、细小且频繁的请求，导致 API 调用成本高昂且效率低下。

理解了这些挑战，我们才能设计出合理的解决方案。

3. 环境准备与前置条件

我们将构建一个模拟场景：一个实时新闻流处理系统，Flink 消费新闻标题流，调用大模型 API 为每条新闻生成一个简短的分类标签（如“科技”、“体育”、“财经”）。

环境清单：

• Flink 环境：本地单机模式（便于演示）。建议使用 Flink 1.17+ 版本。
• Java 开发环境：JDK 8 或 11（推荐 11），Maven 3.6+。
• 大模型服务：为了普适性和可复现性，我们使用OpenAI 兼容的 API作为示例。你可以替换为任何提供 HTTP API 的模型服务，如本地部署的 LLaMA、通义千问、DeepSeek 等。
  • 你需要一个可用的 API 端点（Endpoint）和 API Key。
  • 本地备选方案：可以使用 Ollama 在本地运行一个轻量级模型（如llama3.2:1b），其也提供类似 OpenAI 的 API 接口 (http://localhost:11434/v1/chat/completions)。
• 网络：确保运行 Flink 作业的机器可以访问你的大模型服务地址。

项目初始化：创建一个标准的 Flink Maven 项目。

<!-- pom.xml 关键依赖 --> <dependencies> <!-- Flink 核心依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.17.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>1.17.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>1.17.2</version> </dependency> <!-- HTTP 客户端，用于调用大模型 API --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-http</artifactId> <version>1.17.2</version> </dependency> <!-- 或使用更通用的异步 HTTP 客户端，如 AsyncHttpClient --> <dependency> <groupId>org.asynchttpclient</groupId> <artifactId>async-http-client</artifactId> <version>2.12.3</version> </dependency> <!-- JSON 解析 --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.2</version> </dependency> </dependencies>

4. 架构设计：如何优雅地调用？

直接同步调用是“灾难”的起点。我们需要更高级的模式。以下是三种渐进式的架构方案：

4.1 方案一：异步 I/O（Async I/O）—— 官方推荐

这是 Flink 为访问外部系统（如数据库、HTTP 服务）而设计的原生模式。它允许单个算子实例并发处理多个请求，通过回调函数非阻塞地接收结果，极大提升吞吐量。

工作原理：

1. Flink 接收到一条数据。
2. 发出一个异步请求（如 HTTP 请求）到外部服务，并立即释放该算子的线程去处理下一条数据。
3. 当外部服务返回结果时，由回调函数将结果与原始数据关联，并发送到下游。

优点：高效利用资源，避免线程阻塞，是处理高延迟外部调用的标准答案。缺点：需要外部客户端支持异步模式（如AsyncHttpClient），且对开发者的异步编程能力有一定要求。

4.2 方案二：批量请求窗口（Batch Request Window）

针对大模型 API 调用成本高的问题，我们可以将短时间内到达的多条数据攒成一个微批次（Micro-batch），然后一次性发送给大模型 API（如果 API 支持批量处理）。

工作原理：

1. 使用 Flink 的window操作（如滚动窗口、滑动窗口）将流数据分组。
2. 在窗口触发时，将窗口内所有数据拼接成一个批量请求体。
3. 调用大模型 API 的批量处理接口。
4. 将批量返回的结果拆解，分别对应到原始数据上。

优点：显著减少 API 调用次数，降低成本，提高整体吞吐量。缺点：引入了窗口延迟（需要等待窗口关闭），牺牲了部分实时性。且需要大模型服务支持批量推理。

4.3 方案三：旁路输出与异步处理（Side Output & Async Processing）

这是一种更解耦的架构。主数据流正常处理，将需要调用大模型的数据通过“旁路输出”（Side Output）发送到一个独立的、专门处理慢任务的流中。这个慢任务流可以采用更宽松的延迟策略（如更大的检查点间隔、更低的并行度），甚至使用不同的计算框架（如 Spark）来处理。

工作原理：

1. 主 Flink 作业识别出需要智能处理的数据。
2. 使用OutputTag将这类数据输出到侧输出流。
3. 侧输出流连接一个专门负责调用大模型的算子（或另一个独立的 Flink 作业）。
4. 处理完成后，结果可以写回 Kafka 等消息队列，供主流程或其他系统消费。

优点：实现关注点分离，避免慢任务阻塞核心实时链路。容错性更好，慢任务流的故障不影响主流程。缺点：架构更复杂，需要维护多个作业，数据一致性需要额外设计。

对于大多数场景，方案一（异步 I/O）是平衡复杂度和效果的优选。接下来，我们将基于此方案进行代码实现。

5. 核心流程拆解与代码实现

我们将实现一个基于异步 I/O的 Flink 作业，调用 OpenAI 兼容 API 为新闻标题分类。

5.1 定义数据流与 POJO

首先，定义输入数据（新闻事件）和输出数据（带分类的新闻事件）。

// 文件路径：src/main/java/com/example/flinkllm/NewsEvent.java import java.time.Instant; public class NewsEvent { private String id; // 新闻ID private String title; // 新闻标题 private Instant timestamp; // 事件时间 // 省略构造函数、Getter/Setter、toString 方法 }

// 文件路径：src/main/java/com/example/flinkllm/ClassifiedNewsEvent.java public class ClassifiedNewsEvent { private String id; private String title; private String category; // 大模型返回的分类标签 private Instant timestamp; // 省略构造函数、Getter/Setter、toString 方法 }

5.2 实现异步 I/O 函数

这是最核心的部分。我们需要继承RichAsyncFunction，并实现asyncInvoke方法。

// 文件路径：src/main/java/com/example/flinkllm/LLMAsyncFunction.java import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.async.ResultFuture; import org.apache.flink.streaming.api.functions.async.RichAsyncFunction; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import org.asynchttpclient.*; import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; public class LLMAsyncFunction extends RichAsyncFunction<NewsEvent, ClassifiedNewsEvent> { private transient AsyncHttpClient asyncHttpClient; private transient ObjectMapper objectMapper; private final String apiUrl; private final String apiKey; private final String modelName; public LLMAsyncFunction(String apiUrl, String apiKey, String modelName) { this.apiUrl = apiUrl; this.apiKey = apiKey; this.modelName = modelName; } @Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 初始化异步 HTTP 客户端 this.asyncHttpClient = Dsl.asyncHttpClient(); this.objectMapper = new ObjectMapper(); } @Override public void close() throws Exception { super.close(); if (asyncHttpClient != null) { asyncHttpClient.close(); } } @Override public void asyncInvoke(NewsEvent input, ResultFuture<ClassifiedNewsEvent> resultFuture) throws Exception { // 1. 构建请求体 (OpenAI 兼容格式) ObjectNode requestBody = objectMapper.createObjectNode(); requestBody.put("model", modelName); requestBody.putArray("messages").addObject() .put("role", "user") .put("content", "请将以下新闻标题分类为‘科技’、‘体育’、‘财经’、‘娱乐’、‘其他’中的一个。标题：" + input.getTitle()); requestBody.put("temperature", 0.1); // 低随机性，保证分类稳定 requestBody.put("max_tokens", 10); String requestBodyStr = objectMapper.writeValueAsString(requestBody); // 2. 构建异步 HTTP 请求 BoundRequestBuilder requestBuilder = asyncHttpClient.preparePost(apiUrl) .addHeader("Content-Type", "application/json") .addHeader("Authorization", "Bearer " + apiKey) .setBody(requestBodyStr) .setRequestTimeout(10000); // 设置10秒超时 // 3. 发起异步请求 CompletableFuture<Response> future = requestBuilder.execute() .toCompletableFuture() .exceptionally(e -> { // 异常处理：记录日志，返回一个标记失败的Response或null System.err.println("API调用失败: " + e.getMessage()); return null; }); // 4. 处理异步结果 future.thenAccept(response -> { try { if (response != null && response.getStatusCode() == 200) { String responseBody = response.getResponseBody(); ObjectNode responseJson = (ObjectNode) objectMapper.readTree(responseBody); // 解析大模型返回的文本内容 String category = responseJson .path("choices").get(0) .path("message").path("content").asText() .trim() .replaceAll("^[\"']|[\"']$", ""); // 去除可能的引号 // 构建输出结果 ClassifiedNewsEvent output = new ClassifiedNewsEvent(); output.setId(input.getId()); output.setTitle(input.getTitle()); output.setCategory(category); output.setTimestamp(input.getTimestamp()); // 将单个结果传递给下游 resultFuture.complete(Collections.singleton(output)); } else { // 处理HTTP错误或空响应 String errorMsg = (response == null) ? "No response" : "Status: " + response.getStatusCode(); System.err.println("API请求失败: " + errorMsg); // 可以选择降级处理，例如赋予一个默认分类 ClassifiedNewsEvent output = new ClassifiedNewsEvent(); output.setId(input.getId()); output.setTitle(input.getTitle()); output.setCategory("未知"); output.setTimestamp(input.getTimestamp()); resultFuture.complete(Collections.singleton(output)); } } catch (Exception e) { System.err.println("解析响应失败: " + e.getMessage()); resultFuture.completeExceptionally(e); } }); } // 超时处理函数 @Override public void timeout(NewsEvent input, ResultFuture<ClassifiedNewsEvent> resultFuture) throws Exception { System.err.println("请求超时 for: " + input.getTitle()); // 超时降级处理 ClassifiedNewsEvent output = new ClassifiedNewsEvent(); output.setId(input.getId()); output.setTitle(input.getTitle()); output.setCategory("超时"); output.setTimestamp(input.getTimestamp()); resultFuture.complete(Collections.singleton(output)); } }

关键点解析：

1. 异步客户端：使用AsyncHttpClient发起非阻塞请求。
2. 超时控制：在asyncInvoke中通过setRequestTimeout和在 Flink 配置中通过AsyncWaitOperator的超时参数共同控制。
3. 容错与降级：在 HTTP 失败、解析失败或超时（timeout方法）时，我们都提供了降级策略（返回“未知”或“超时”分类），避免作业因单次调用失败而崩溃。这是生产环境必须的。
4. 资源管理：在open和close生命周期方法中初始化和关闭 HTTP 客户端。

5.3 构建主 Flink 作业流

现在，我们将所有部分组合起来。

// 文件路径：src/main/java/com/example/flinkllm/NewsClassificationJob.java import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import org.apache.flink.streaming.api.datastream.AsyncDataStream; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import com.fasterxml.jackson.databind.ObjectMapper; import java.time.Duration; import java.time.Instant; import java.util.concurrent.TimeUnit; public class NewsClassificationJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); // 设置并行度 // 1. 定义 Kafka Source (模拟数据源) KafkaSource<String> kafkaSource = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("news-titles") .setGroupId("flink-llm-demo") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> kafkaStream = env.fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)), "Kafka Source"); // 2. 解析 JSON 字符串为 NewsEvent 对象 ObjectMapper mapper = new ObjectMapper(); SingleOutputStreamOperator<NewsEvent> newsStream = kafkaStream .map(value -> { try { // 假设Kafka消息是JSON格式：{"id":"1", "title":"某公司发布AI芯片", "timestamp":"2023-...} return mapper.readValue(value, NewsEvent.class); } catch (Exception e) { System.err.println("解析JSON失败: " + value); return null; // 过滤掉解析失败的数据 } }) .filter(event -> event != null); // 3. 应用异步 I/O 函数调用大模型 String apiUrl = "https://api.openai.com/v1/chat/completions"; // 替换为你的API地址 String apiKey = "your-api-key-here"; // 替换为你的API Key String modelName = "gpt-3.5-turbo"; // 替换为你的模型名 LLMAsyncFunction asyncFunction = new LLMAsyncFunction(apiUrl, apiKey, modelName); // 使用 unorderedWait 模式，允许结果乱序到达，以获得更高的吞吐量。 // 参数：输入流，异步函数，超时时间，时间单位，容量（最多允许多少个异步请求同时挂起） DataStream<ClassifiedNewsEvent> classifiedStream = AsyncDataStream .unorderedWait(newsStream, asyncFunction, 15, TimeUnit.SECONDS, 100); // 4. 打印结果到控制台 (生产环境应输出到Kafka、数据库等) classifiedStream.print(); // 5. 执行作业 env.execute("Flink LLM News Classification"); } }

6. 运行结果与效果验证

运行步骤：

1. 启动 Kafka，并创建news-titles主题。
2. 向 Kafka 发送测试数据：

   # 使用 kafka-console-producer ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic news-titles

   输入 JSON 消息：

   {"id":"1","title":"OpenAI发布新一代语言模型","timestamp":"2023-10-27T10:00:00Z"} {"id":"2","title":"世界杯决赛阿根廷对阵法国","timestamp":"2023-10-27T10:00:01Z"} {"id":"3","title":"美联储宣布维持利率不变","timestamp":"2023-10-27T10:00:02Z"}

3. 运行 Flink 作业。在 IDE 中直接运行NewsClassificationJob的 main 方法，或在打包后使用flink run命令提交。
4. 观察控制台输出。你应该能看到类似以下的输出：

   10> ClassifiedNewsEvent{id='1', title='OpenAI发布新一代语言模型', category='科技', timestamp=2023-10-27T10:00:00Z} 11> ClassifiedNewsEvent{id='2', title='世界杯决赛阿根廷对阵法国', category='体育', timestamp=2023-10-27T10:00:01Z} 12> ClassifiedNewsEvent{id='3', title='美联储宣布维持利率不变', category='财经', timestamp=2023-10-27T10:00:02Z}

效果验证要点：

• 功能正确性：大模型是否正确地对新闻标题进行了分类。
• 系统吞吐量：观察作业的处理速率。你可以使用 Flink Web UI（默认端口 8081）的 Metrics 选项卡，查看numRecordsOutPerSecond等指标。注意：这个速率将严重受限于大模型 API 的响应延迟和 QPS 限制。
• 延迟分析：在LLMAsyncFunction中添加日志，记录请求发出和收到响应的时间差，可以直观看到大模型调用引入的延迟。
• 资源占用：观察 TaskManager 的 CPU 和内存使用情况。异步 I/O 虽然不阻塞线程，但大量并发请求会占用网络和内存资源。

7. 常见问题与排查思路

在实际运行中，你几乎一定会遇到以下问题。下表提供了系统的排查思路：

8. 最佳实践与工程建议

要让 Flink 调用大模型在生产环境中稳定运行，仅靠基础代码是不够的。以下是从实战中总结出的关键建议：

8.1 性能优化

• 提示词工程：精心设计发送给大模型的提示词（Prompt），使其尽可能简短、明确，直接输出结构化或限定格式的结果（如 JSON），减少不必要的文本生成，能显著降低延迟和成本。
• 模型选择：在效果可接受的范围内，选择更小、更快的模型。例如，对于分类任务，gpt-3.5-turbo通常比gpt-4快一个数量级，成本也更低。
• 本地化部署：如果对延迟和隐私要求极高，考虑在 Kubernetes 集群中部署开源模型（如 LLaMA、ChatGLM），并使用高性能推理框架（如 vLLM、TGI），将网络延迟降至最低。
• 异步客户端调优：配置AsyncHttpClient的连接池大小、超时时间、重试策略，以匹配你的流量模式。

8.2 稳定性与容错

• 完善的降级策略：如代码所示，对网络超时、API 错误、解析失败等情况必须有降级方案（返回默认值、将数据导入死信队列等）。绝不能因为外部服务不稳定导致 Flink 作业失败。
• 熔断机制：当连续失败或超时次数超过阈值时，应暂时“熔断”对大模型的调用，直接走降级逻辑，并定期尝试恢复。可以使用 Resilience4j 等库在asyncInvoke方法中实现。
• 监控与告警：对以下指标进行监控：
  • Flink 作业：异步 I/O 算子的吞吐量、延迟、背压状态、numRecordsIn/Out。
  • 大模型服务：API 调用成功率、平均响应时间、错误码分布。
  • 业务指标：分类准确率（可通过抽样人工评估）。

8.3 架构演进

• 引入消息队列解耦：对于核心链路，可以采用方案三（旁路输出）。主流程将需要处理的数据写入一个 Kafka Topic，由另一个独立的、弹性更强的消费者服务（可以是另一个 Flink 作业，也可以是其他服务）来消费并调用大模型，再将结果写回。这样彻底隔离了风险。
• 向量化与缓存：对于重复或相似的问题（例如，“今天天气怎么样？”），可以将大模型的回答进行向量化并存入向量数据库（如 Milvus、Weaviate）。当新问题到来时，先进行向量相似度搜索，如果找到高度相似的缓存结果，则直接返回，避免重复调用大模型。这尤其适用于客服、问答场景。
• 批处理优先：对于实时性要求不高的任务（如每日报告生成、用户行为分析），完全可以采用 Flink Batch 或 Spark 进行离线处理，成本更低，控制更灵活。

9. 总结与后续学习方向

回到最初的问题：Flink 调用大模型，效果如何？

答案是：效果取决于架构设计和场景匹配度。它是一个强大的模式，但绝非“即插即用”。

• 效果好的场景：对延迟有一定容忍度（秒级）、调用量可控、且有明确降级方案的近实时智能处理。例如：实时评论情感分析（正面/负面/中性）、新闻自动打标、低代码平台的自然语言生成 SQL 等。
• 效果差或需慎用的场景：要求毫秒级响应的交易风控、高频的实时推荐、或预算有限且调用量巨大的场景。在这些场景下，传统的规则引擎、小模型或离线预处理可能是更优解。

本文为你铺平了从零到一实践的道路。你学会了使用 Flink 异步 I/O 来协调流处理与慢服务，实现了基本的容错降级，并了解了性能瓶颈与优化方向。

如果你想继续深入，建议从以下几个方向探索：

1. 深入 Flink 异步 I/O：研究其底层原理，如何与 Checkpoint 机制协同工作，保证状态一致性。
2. 探索 Flink ML Pipeline：虽然目前对深度学习集成还不成熟，但可以关注社区动态，看是否有更原生的集成方式出现。
3. 学习大模型服务部署：掌握如何使用 vLLM、TensorRT-LLM 等工具在 GPU 集群上高效部署和运维开源大模型，摆脱对商用 API 的依赖。
4. 设计混合智能系统：思考如何将规则引擎、传统机器学习模型、向量检索与大模型结合，在成本、速度和效果间取得最佳平衡。

技术组合的魅力在于解决单一技术无法解决的复杂问题。Flink 与大模型的结合，正是流处理智能化演进中的一个重要探索。希望本文能成为你探索路上的实用指南。