在实际数据处理和实时计算场景中，Flink 作为流处理引擎，与大模型（LLM）的结合正成为一种探索方向。开发者希望利用 Flink 强大的实时数据流处理能力，来调用大模型进行文本分析、内容生成、智能决策等任务。当面临模型选型时，一个常见的问题是：在 Flink 框架下调用，GLM 和 DeepSeek 哪个更“厉害”？这里的“厉害”通常指向几个维度：API 调用的便捷性、推理速度、成本、模型能力（如代码生成、文本理解）以及对中文的支持度。本文将从工程实践的角度，探讨在 Flink 项目中集成大模型 API 的方案，并对比 GLM 与 DeepSeek 在关键指标上的差异，帮助你根据项目需求做出技术选型。

需要明确的是，Flink 本身并不直接提供大模型推理能力，其角色是作为数据流的编排和调度者。我们的目标是在 Flink 的算子（例如ProcessFunction或通过Async I/O）中，异步调用外部的大模型 API 服务，将模型推理无缝嵌入到实时数据处理管道中。因此，选型的核心在于评估不同大模型 API 的服务质量、接口稳定性、成本以及它们与 Flink 异步编程模型的契合度。

1. 理解 Flink 调用大模型的核心架构与挑战

在 Flink 流处理作业中直接进行同步 HTTP 调用是危险的，因为网络延迟和模型推理耗时可能长达数秒，会严重阻塞数据处理管道，导致背压甚至作业失败。因此，异步调用是必须遵循的核心原则。

1.1 为什么必须使用 Async I/O？

Flink 的 Async I/O 功能允许单个算子并发处理多个请求并异步等待结果，从而在等待外部服务响应时，不会阻塞算子的计算资源。这对于调用延迟高的大模型 API 至关重要。其工作流程可以概括为：

1. 数据流中的每条记录触发一个异步请求。
2. 请求被分发到线程池，由线程池管理并发请求。
3. 算子继续处理后续数据，不等待当前请求返回。
4. 异步请求完成后，结果被收集并发送到下游。

1.2 通用集成架构

一个典型的 Flink 作业调用大模型 API 的架构如下：

Kafka Source -> Map/ProcessFunction (数据预处理) -> Async I/O (调用大模型 API) -> Sink (结果写入 Kafka/DB)

在Async I/O算子中，我们会封装一个AsyncFunction，其内部使用 HTTP 客户端（如 Apache HttpClient、OkHttp 或异步客户端如 AsyncHttpClient）向大模型的 API 端点发起请求。

1.3 主要技术挑战

• 容错与重试：网络波动或 API 服务暂时不可用。需要在AsyncFunction中实现指数退避等重试机制。
• 速率限制：所有大模型 API 都有 QPS（每秒查询率）或 RPM（每分钟请求数）限制。需要在 Flink 侧实现限流，例如使用 Guava 的RateLimiter。
• 结果解析与错误处理：需要健壮地解析 API 返回的 JSON，并处理各种错误码（如429代表限流，503代表服务过载）。
• 状态管理：某些场景下可能需要关联请求与响应，或者累计某些指标，会用到 Flink 的状态编程。

2. 环境准备与项目依赖配置

在开始编写代码前，需要搭建一个基础的 Flink 开发环境，并引入必要的依赖。这里我们以 Java 项目为例，使用 Maven 进行依赖管理。

2.1 基础环境要求

• Java: JDK 8 或 11（推荐 11，与 Flink 1.17+ 兼容性更好）。
• Flink: 版本 1.16 或 1.17（本文示例基于 1.17.1）。
• 构建工具: Maven 3.6+ 或 Gradle。
• 集成开发环境（IDE）: IntelliJ IDEA 或 Eclipse。

2.2 Maven 核心依赖

创建一个新的 Maven 项目，在pom.xml中添加以下依赖：

<properties> <flink.version>1.17.1</flink.version> <scala.binary.version>2.12</scala.binary.version> </properties> <dependencies> <!-- Flink 核心依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <!-- Flink Async I/O 需要连接器依赖，通常已包含在 streaming-java 中 --> <!-- 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> <!-- 日志 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.36</version> <scope>runtime</scope> </dependency> </dependencies>

注意：Flink 核心依赖的scope设置为provided，是因为在提交到集群运行时，集群环境已经提供了这些 Jar 包。本地测试时，IDE 或mvn exec:java命令可以正确处理。

2.3 获取大模型 API 密钥

要调用 GLM 或 DeepSeek 的 API，你需要先注册相应的平台账号并获取 API Key。

• GLM (智谱AI): 访问智谱AI开放平台，注册后可在控制台创建 API Key。
• DeepSeek: 访问 DeepSeek 开放平台，完成注册和认证后获取 API Key。

请妥善保管你的 API Key，不要在代码中硬编码，建议通过环境变量或配置文件传入。

3. 实现 Flink AsyncFunction 调用大模型 API

我们将实现一个通用的AsyncFunction，它可以通过配置来适配不同的大模型 API。这里以文本补全（Chat Completion）任务为例。

3.1 定义数据流 POJO 和配置类

首先，定义输入输出数据的结构。

// 输入事件：包含需要模型处理的文本 public class InputEvent { private String id; // 用于关联请求和响应 private String text; // 待处理的原始文本 // 省略构造函数、getter、setter } // 输出事件：包含模型返回的结果 public class OutputEvent { private String id; private String originalText; private String modelResponse; private long timestamp; // 省略构造函数、getter、setter } // 大模型 API 配置 public class LLMConfig { private String apiKey; private String apiEndpoint; // 如 GLM 的 https://open.bigmodel.cn/api/paas/v4/chat/completions private String modelName; // 如 “glm-4”, “deepseek-chat” private int maxTokens; private double temperature; // 省略其他参数和 getter/setter }

3.2 实现通用的 AsyncLLMInvokeFunction

这是最核心的类，继承RichAsyncFunction。

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 org.asynchttpclient.*; import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; public class AsyncLLMInvokeFunction extends RichAsyncFunction<InputEvent, OutputEvent> { private transient AsyncHttpClient asyncHttpClient; private final LLMConfig llmConfig; private final RateLimiter rateLimiter; // 假设已引入Guava RateLimiter public AsyncLLMInvokeFunction(LLMConfig config) { this.llmConfig = config; this.rateLimiter = RateLimiter.create(10.0); // 初始限制 10 QPS } @Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 初始化异步 HTTP 客户端 DefaultAsyncHttpClientConfig.Builder clientBuilder = Dsl.config() .setConnectTimeout(5000) .setRequestTimeout(30000) // 大模型响应可能较慢，超时设长 .setMaxRequestRetry(1); // 重试策略可在外部实现 this.asyncHttpClient = Dsl.asyncHttpClient(clientBuilder.build()); } @Override public void close() throws Exception { super.close(); if (asyncHttpClient != null) { asyncHttpClient.close(); } } @Override public void asyncInvoke(InputEvent input, ResultFuture<OutputEvent> resultFuture) throws Exception { // 1. 限流 rateLimiter.acquire(); // 2. 构建请求 JSON String requestBody = buildRequestBody(input.getText()); // 3. 构建异步 HTTP 请求 BoundRequestBuilder requestBuilder = asyncHttpClient.preparePost(llmConfig.getApiEndpoint()) .addHeader("Content-Type", "application/json") .addHeader("Authorization", "Bearer " + llmConfig.getApiKey()) .setBody(requestBody); // 4. 执行异步请求，并将 Future 转换为 CompletableFuture CompletableFuture<Response> responseFuture = requestBuilder.execute() .toCompletableFuture() .exceptionally(ex -> { // 记录异常，返回一个自定义的错误响应或抛出 System.err.println("HTTP请求失败: " + ex.getMessage()); return null; // 实际应返回一个包含错误信息的Response包装对象 }); // 5. 处理响应，完成后调用 resultFuture.complete responseFuture.thenAccept(response -> { if (response != null && response.getStatusCode() == 200) { String responseBody = response.getResponseBody(); String modelOutput = parseModelResponse(responseBody); OutputEvent output = new OutputEvent(input.getId(), input.getText(), modelOutput, System.currentTimeMillis()); resultFuture.complete(Collections.singleton(output)); } else { // 处理错误，例如记录日志、重试或发送到侧输出流 System.err.println("API调用失败，状态码: " + (response != null ? response.getStatusCode() : "N/A")); resultFuture.completeExceptionally(new RuntimeException("LLM API call failed")); } }); } // 构建请求体（以GLM API v4格式为例） private String buildRequestBody(String prompt) { // 使用Jackson或简单字符串拼接构建JSON // 示例：{"model": "glm-4", "messages": [{"role": "user", "content": prompt}], "max_tokens": 500} return String.format( "{\"model\": \"%s\", \"messages\": [{\"role\": \"user\", \"content\": \"%s\"}], \"max_tokens\": %d}", llmConfig.getModelName(), prompt.replace("\"", "\\\""), // 简单转义 llmConfig.getMaxTokens() ); } // 解析响应体，提取模型生成的文本 private String parseModelResponse(String responseBody) { // 使用Jackson解析JSON // 示例解析：从 responseBody 的 JSON 中提取 choices[0].message.content // 这里为简化，直接返回原始响应或截取部分 try { com.fasterxml.jackson.databind.JsonNode root = objectMapper.readTree(responseBody); return root.path("choices").get(0).path("message").path("content").asText(); } catch (Exception e) { return "Error parsing response: " + e.getMessage(); } } }

3.3 在主程序中组装流处理作业

现在，在 Flink 主程序中创建数据流并使用这个异步函数。

import org.apache.flink.streaming.api.datastream.AsyncDataStream; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.util.concurrent.TimeUnit; public class FlinkLLMJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); // 根据并发请求数设置并行度 // 1. 定义数据源（这里用集合模拟，实际可能是 Kafka） DataStream<InputEvent> inputStream = env.fromElements( new InputEvent("1", "请用Java写一个快速排序函数"), new InputEvent("2", "解释一下什么是机器学习"), new InputEvent("3", "将'Hello, world'翻译成法语") ); // 2. 配置大模型参数 LLMConfig glmConfig = new LLMConfig(); glmConfig.setApiKey(System.getenv("GLM_API_KEY")); glmConfig.setApiEndpoint("https://open.bigmodel.cn/api/paas/v4/chat/completions"); glmConfig.setModelName("glm-4"); glmConfig.setMaxTokens(500); // 3. 应用异步 I/O 转换 DataStream<OutputEvent> outputStream = AsyncDataStream .unorderedWait( // 使用无序等待，效率更高，除非需要严格顺序 inputStream, new AsyncLLMInvokeFunction(glmConfig), 30000, // 超时时间 30秒 TimeUnit.MILLISECONDS, 100 // 最大并发请求数 ); // 4. 输出结果（这里打印，实际可写入 Kafka、JDBC Sink 等） outputStream.print(); // 5. 执行作业 env.execute("Flink LLM API Invocation Job"); } }

4. GLM 与 DeepSeek 在 Flink 集成中的关键对比

在 Flink 异步调用的架构下，评估 GLM（以智谱 GLM-4 为代表）和 DeepSeek（以 DeepSeek Chat 为代表）的“厉害”之处，需要聚焦于工程集成相关的指标。以下是从实际调用角度总结的对比：

关键判断：从纯技术集成角度看，两者在 Flink 中调用的复杂度几乎没有区别。选型的决定性因素往往在于业务需求侧重点（更看重代码生成还是中文创作）、实时性要求（毫秒级差异是否关键）以及成本预算。

4.1 如何根据场景选择？

• 选择 GLM 的场景：
  • 业务内容以中文内容创作、润色、摘要为主。
  • 需要处理非常本土化的语境和表达。
  • 团队对智谱的生态工具（如 ChatGLM 系列开源模型）有前期技术积累。
• 选择 DeepSeek 的场景：
  • 任务核心是代码生成、补全、审查或算法逻辑推理。
  • 对推理速度有极致要求，希望进一步降低流处理延迟。
  • 希望尝试在代码能力上表现更突出的模型。

5. 生产环境部署的注意事项与最佳实践

在本地测试通过后，将 Flink 调用大模型的作业部署到生产环境（如 YARN 或 Kubernetes 集群），还需要考虑更多因素。

5.1 配置管理

切勿将 API Key 等敏感信息硬编码在代码中。推荐做法：

• 使用 Flink Configuration：通过ExecutionConfig或ParameterTool从启动参数传入。
• 使用外部化配置：将配置存储在 Hadoop 分布式缓存、Kubernetes ConfigMap 或专门的配置中心（如 Apollo, Nacos），在open()方法中读取。
• 环境变量：在集群节点或容器中设置环境变量，通过System.getenv()获取。

// 在main方法中 ParameterTool parameters = ParameterTool.fromArgs(args); String apiKey = parameters.get("llm.api.key"); // 在AsyncFunction的open方法中 LLMConfig config = new LLMConfig(); config.setApiKey(getRuntimeContext().getExecutionConfig().getGlobalJobParameters().get("apiKey"));

5.2 性能与稳定性优化

1. 连接池与超时：确保AsyncHttpClient配置了合理的连接池大小、超时时间和重试策略。
2. 背压处理：如果大模型 API 响应变慢，会导致 Flink 作业产生背压。需要监控 Flink Web UI 中的背压指标。解决方案包括：
   • 增加Async I/O算子的并行度。
   • 在源端（如 Kafka）降低消费速率。
   • 在AsyncFunction中实现更严格的限流，避免压垮下游 API。
3. 容错与重试：实现一个带退避机制的重试策略，而不是简单失败。

// 简化的带指数退避的重试逻辑 private CompletableFuture<Response> executeWithRetry(BoundRequestBuilder request, int maxRetries) { CompletableFuture<Response> future = new CompletableFuture<>(); retryInternal(request, future, maxRetries, 1); return future; } private void retryInternal(BoundRequestBuilder request, CompletableFuture<Response> resultFuture, int maxRetries, int attempt) { request.execute().toCompletableFuture() .thenAccept(response -> { if (response.getStatusCode() == 200) { resultFuture.complete(response); } else if (response.getStatusCode() == 429 && attempt <= maxRetries) { // 限流，等待后重试 long waitTime = (long) (Math.pow(2, attempt) * 1000 + Math.random() * 1000); scheduler.schedule(() -> retryInternal(request, resultFuture, maxRetries, attempt + 1), waitTime, TimeUnit.MILLISECONDS); } else { resultFuture.completeExceptionally(new RuntimeException("Failed after retries")); } }) .exceptionally(ex -> { if (attempt <= maxRetries) { long waitTime = (long) (Math.pow(2, attempt) * 1000); scheduler.schedule(() -> retryInternal(request, resultFuture, maxRetries, attempt + 1), waitTime, TimeUnit.MILLISECONDS); } else { resultFuture.completeExceptionally(ex); } return null; }); }

5.3 监控与告警

• Flink Metrics：利用 Flink 内置的 Metrics 系统，暴露Async I/O算子的队列长度、平均等待时间、请求成功率等指标。
• 日志聚合：将作业日志集中收集到 ELK 或类似平台，重点关注 HTTP 请求的异常状态码和超时日志。
• API 侧监控：关注大模型服务商控制台提供的调用量、延迟、错误率仪表盘。

6. 常见问题排查清单

在开发和运行过程中，你可能会遇到以下问题。这里提供一个排查路径。

7. 扩展方向与进阶思考

在完成基础集成后，可以考虑以下方向来增强系统的能力和鲁棒性：

1. 动态模型路由：实现一个RouterAsyncFunction，根据输入内容的特征（如语言、任务类型）动态选择调用 GLM 或 DeepSeek，甚至其他模型，实现成本与效果的最优平衡。
2. 结果缓存：对于重复或相似的查询，可以在 Flink 状态中引入一个简单的缓存（如 Guava Cache），避免重复调用大模型，显著降低成本并提升速度。
3. 流批结合与模型微调：使用 Flink 实时处理用户反馈数据（如对模型生成结果的点赞/点踩），定期（批处理）用这些数据对开源小模型进行微调，再通过 Flink 将轻量级的微调模型部署为实时服务，形成闭环。
4. 复杂工作流编排：将一次大模型调用升级为多步链式调用（如先总结，再翻译，最后情感分析）。可以利用 Flink 的迭代操作或拆分成多个连续的Async I/O步骤来实现，但需仔细设计错误处理和状态一致性。

最终，在 Flink 中调用 GLM 还是 DeepSeek，并非一个非此即彼的问题。更成熟的架构应该具备可插拔性，允许你通过配置轻松切换或同时使用多个模型。工程上的重点始终是构建一个高吞吐、低延迟、具备容错和监控能力的异步调用框架，这将是你应对未来任何大模型 API 变更或新模型出现的最坚实保障。建议在实际项目中，针对你的核心业务场景，用同样的测试数据集对两个模型进行效果和性能的基准测试，让数据驱动决策。