findByConditions(@Param("provider") String provider, @Param("keyword") String keyword,
+ @Param("isActive") Boolean isActive, @Param("maxTokens") Integer maxTokens,
+ @Param("modelType") String modelType);
+
+ @Insert("""
+ INSERT INTO model_config (provider, base_url, api_key, model_name, temperature, is_active, max_tokens,
+ model_type, completions_path, embeddings_path, created_time, updated_time, is_deleted,
+ proxy_enabled, proxy_host, proxy_port, proxy_username, proxy_password, chat_api_protocol)
+ VALUES (#{provider}, #{baseUrl}, #{apiKey}, #{modelName}, #{temperature}, #{isActive}, #{maxTokens},
+ #{modelType}, #{completionsPath}, #{embeddingsPath}, NOW(), NOW(), 0,
+ #{proxyEnabled}, #{proxyHost}, #{proxyPort}, #{proxyUsername}, #{proxyPassword}, #{chatApiProtocol})
+ """)
+ @Options(useGeneratedKeys = true, keyProperty = "id", keyColumn = "id")
+ int insert(ModelConfig modelConfig);
+
+ @Update("""
+
+ """)
+ int updateById(ModelConfig modelConfig);
+
+ @Update("""
+ UPDATE model_config SET is_deleted = 1 WHERE id = #{id}
+ """)
+ int deleteById(Integer id);
+
+}
diff --git a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/DynamicModelFactory.java b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/DynamicModelFactory.java
index d57a1458d..139f054f7 100644
--- a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/DynamicModelFactory.java
+++ b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/DynamicModelFactory.java
@@ -16,6 +16,9 @@
package com.alibaba.cloud.ai.dataagent.service.aimodelconfig;
import com.alibaba.cloud.ai.dataagent.dto.ModelConfigDTO;
+import com.alibaba.cloud.ai.dataagent.enums.ChatApiProtocol;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApiChatModel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hc.client5.http.auth.AuthScope;
@@ -49,15 +52,51 @@
public class DynamicModelFactory {
/**
- * 统一使用 OpenAiChatModel，通过 baseUrl 实现多厂商兼容
+ * 根据配置的接口协议创建 ChatModel。 RESPONSES 协议走自研 ResponsesApiChatModel； CHAT_COMPLETIONS（默认）走
+ * OpenAiChatModel。 两者均实现 ChatModel 接口，对上层完全透明。
*/
public ChatModel createChatModel(ModelConfigDTO config) {
- log.info("Creating NEW ChatModel instance. Provider: {}, Model: {}, BaseUrl: {}", config.getProvider(),
- config.getModelName(), config.getBaseUrl());
+ log.info("Creating NEW ChatModel instance. Provider: {}, Model: {}, BaseUrl: {}, Protocol: {}",
+ config.getProvider(), config.getModelName(), config.getBaseUrl(), config.getChatApiProtocol());
// 1. 验证参数
checkBasic(config);
+ // 按接口协议分支：RESPONSES 走自研适配层，其余保持现状走 OpenAiChatModel
+ if (ChatApiProtocol.RESPONSES.name().equalsIgnoreCase(config.getChatApiProtocol())) {
+ return createResponsesApiChatModel(config);
+ }
+
+ // 默认：Chat Completions 协议（现有逻辑，零变更）
+ return createCompletionsChatModel(config);
+ }
+
+ /**
+ * 创建基于 Responses API 的 ChatModel。 复用现有的 proxy RestClient/WebClient 构建体系，代理能力天然继承。
+ * completionsPath 在 RESPONSES 协议下复用为自定义 responses 路径（默认 /v1/responses）。
+ */
+ private ChatModel createResponsesApiChatModel(ModelConfigDTO config) {
+ String apiKey = StringUtils.hasText(config.getApiKey()) ? config.getApiKey() : "";
+ // completionsPath 复用为 Responses API 路径，默认 /v1/responses
+ String responsesPath = StringUtils.hasText(config.getCompletionsPath()) ? config.getCompletionsPath()
+ : "/v1/responses";
+
+ ResponsesApi responsesApi = new ResponsesApi(config.getBaseUrl(), apiKey, responsesPath,
+ getProxiedRestClientBuilder(config), getProxiedWebClientBuilder(config));
+
+ OpenAiChatOptions chatOptions = OpenAiChatOptions.builder()
+ .model(config.getModelName())
+ .temperature(config.getTemperature())
+ .maxTokens(config.getMaxTokens())
+ .build();
+
+ return new ResponsesApiChatModel(responsesApi, chatOptions);
+ }
+
+ /**
+ * 创建基于 Chat Completions 协议的 ChatModel（原有逻辑）
+ */
+ private ChatModel createCompletionsChatModel(ModelConfigDTO config) {
// 2. 构建 OpenAiApi (核心通讯对象)
String apiKey = StringUtils.hasText(config.getApiKey()) ? config.getApiKey() : "";
OpenAiApi.Builder apiBuilder = OpenAiApi.builder()
diff --git a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/ModelConfigDataServiceImpl.java b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/ModelConfigDataServiceImpl.java
index bac38186e..244c827ef 100644
--- a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/ModelConfigDataServiceImpl.java
+++ b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/ModelConfigDataServiceImpl.java
@@ -15,6 +15,7 @@
*/
package com.alibaba.cloud.ai.dataagent.service.aimodelconfig;
+import com.alibaba.cloud.ai.dataagent.enums.ChatApiProtocol;
import com.alibaba.cloud.ai.dataagent.enums.ModelType;
import com.alibaba.cloud.ai.dataagent.converter.ModelConfigConverter;
import com.alibaba.cloud.ai.dataagent.dto.ModelConfigDTO;
@@ -24,6 +25,7 @@
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
import java.time.LocalDateTime;
import java.util.List;
@@ -81,6 +83,12 @@ private void clean(ModelConfigDTO dto) {
if (dto.getEmbeddingsPath() != null) {
dto.setEmbeddingsPath(dto.getEmbeddingsPath().trim());
}
+ // 校验并归一化接口协议：前端下拉框虽限制了取值，但 API 直调可能传任意字符串，
+ // 非法值若不在保存入口拦截，会在 DynamicModelFactory 静默落入默认协议分支导致配置悄悄失效。
+ // fromCode 忽略大小写，归一为枚举标准值后存库，保证库内口径统一
+ if (StringUtils.hasText(dto.getChatApiProtocol())) {
+ dto.setChatApiProtocol(ChatApiProtocol.fromCode(dto.getChatApiProtocol().trim()).getCode());
+ }
}
/**
@@ -124,6 +132,7 @@ private static void mergeDtoToEntity(ModelConfigDTO dto, ModelConfig oldEntity)
oldEntity.setProxyPort(dto.getProxyPort());
oldEntity.setProxyUsername(dto.getProxyUsername());
oldEntity.setProxyPassword(dto.getProxyPassword());
+ oldEntity.setChatApiProtocol(dto.getChatApiProtocol());
// 只有当前端传来的 Key 不包含 "****" 时，才说明用户真的改了 Key，否则保持原样
if (dto.getApiKey() != null && !dto.getApiKey().contains("****")) {
diff --git a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApi.java b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApi.java
new file mode 100644
index 000000000..85fa531bc
--- /dev/null
+++ b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApi.java
@@ -0,0 +1,309 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.ai.retry.RetryUtils;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.util.StringUtils;
+import org.springframework.web.client.RestClient;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * OpenAI Responses API 低层 HTTP 客户端。 负责构建请求、解析非流式响应与流式 SSE 事件。 协议 DTO 以内嵌 record/class
+ * 形式定义，覆盖 DataAgent 实际用到的字段子集。
+ *
+ * 不传 store、previous_response_id、tools 等字段，DataAgent 不使用这些特性。
+ */
+@Slf4j
+public class ResponsesApi {
+
+ private final RestClient restClient;
+
+ private final WebClient webClient;
+
+ private final String responsesPath;
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+ public ResponsesApi(String baseUrl, String apiKey, String responsesPath, RestClient.Builder restClientBuilder,
+ WebClient.Builder webClientBuilder) {
+ this.responsesPath = StringUtils.hasText(responsesPath) ? responsesPath : "/v1/responses";
+
+ // 构建同步 RestClient（用于阻塞调用）。
+ // 必须挂载 DEFAULT_RESPONSE_ERROR_HANDLER：它把 4xx 映射为 NonTransientAiException、
+ // 5xx 映射为 TransientAiException，而上层 DEFAULT_RETRY_TEMPLATE 只对 TransientAiException
+ // 重试——
+ // 若不挂载，HTTP 错误抛出的是 RestClientResponseException，重试模板会直接放行，重试形同虚设
+ this.restClient = restClientBuilder.baseUrl(baseUrl)
+ .defaultHeader("Authorization", "Bearer " + apiKey)
+ .defaultHeader("Content-Type", "application/json")
+ .defaultStatusHandler(RetryUtils.DEFAULT_RESPONSE_ERROR_HANDLER)
+ .build();
+
+ // 构建异步 WebClient（用于 SSE 流式调用）
+ this.webClient = webClientBuilder.baseUrl(baseUrl)
+ .defaultHeader("Authorization", "Bearer " + apiKey)
+ .defaultHeader("Content-Type", "application/json")
+ .build();
+ }
+
+ // ======================== 同步调用 ========================
+
+ /**
+ * 非流式调用 Responses API，返回完整响应。
+ * @param request 请求体（stream 字段会被忽略/覆盖为 false）
+ * @return 完整的 Responses 响应
+ */
+ public ResponsesResponse call(ResponsesRequest request) {
+ // 确保非流式
+ ResponsesRequest syncRequest = request.withStream(false);
+
+ return restClient.post().uri(this.responsesPath).body(syncRequest).retrieve().body(ResponsesResponse.class);
+ }
+
+ // ======================== 流式调用 ========================
+
+ /**
+ * 流式调用 Responses API，返回 SSE 事件流。 解析 Responses API 特有的事件类型（response.output_text.delta、
+ * response.completed 等）。 对于未知事件类型一律忽略，保证向前兼容。
+ * @param request 请求体（stream 字段会被覆盖为 true）
+ * @return SSE 事件流，每个事件已解析为 StreamEvent
+ */
+ public Flux stream(ResponsesRequest request) {
+ // 确保流式
+ ResponsesRequest streamRequest = request.withStream(true);
+
+ return webClient.post()
+ .uri(this.responsesPath)
+ .bodyValue(streamRequest)
+ .accept(MediaType.TEXT_EVENT_STREAM)
+ .retrieve()
+ .bodyToFlux(new ParameterizedTypeReference>() {
+ })
+ .filter(sse -> StringUtils.hasText(sse.data()))
+ .mapNotNull(this::parseStreamEvent);
+ }
+
+ /**
+ * 解析单个 SSE 事件的 data 负载。 按事件 type 分发：delta / completed / incomplete / failed / error。
+ * 未知事件类型静默忽略（返回 null 被 mapNotNull 过滤掉）。 包私有可见性用于单元测试直接覆盖各事件分支。
+ */
+ StreamEvent parseStreamEvent(ServerSentEvent sse) {
+ String data = sse.data();
+ if (!StringUtils.hasText(data)) {
+ return null;
+ }
+ try {
+ Map eventMap = OBJECT_MAPPER.readValue(data, Map.class);
+ String type = (String) eventMap.get("type");
+ if (type == null) {
+ return null;
+ }
+
+ return switch (type) {
+ // 文本增量：产出一个 delta 文本块
+ case "response.output_text.delta" -> {
+ String delta = (String) eventMap.get("delta");
+ yield new StreamEvent(StreamEvent.Type.DELTA, delta, null, null);
+ }
+ // 响应完成：携带完整响应（含 usage）
+ case "response.completed" -> {
+ ResponsesResponse response = OBJECT_MAPPER.convertValue(eventMap.get("response"),
+ ResponsesResponse.class);
+ yield new StreamEvent(StreamEvent.Type.COMPLETED, null, response, null);
+ }
+ // 响应被截断（如达到 max_output_tokens）
+ case "response.incomplete" -> {
+ ResponsesResponse response = OBJECT_MAPPER.convertValue(eventMap.get("response"),
+ ResponsesResponse.class);
+ yield new StreamEvent(StreamEvent.Type.INCOMPLETE, null, response, null);
+ }
+ // 响应失败
+ case "response.failed" -> {
+ ResponsesResponse response = OBJECT_MAPPER.convertValue(eventMap.get("response"),
+ ResponsesResponse.class);
+ String errorMsg = extractErrorMessage(response);
+ yield new StreamEvent(StreamEvent.Type.ERROR, null, response, errorMsg);
+ }
+ // 全局错误事件
+ case "error" -> {
+ String errorMsg = extractErrorFromMap(eventMap);
+ yield new StreamEvent(StreamEvent.Type.ERROR, null, null, errorMsg);
+ }
+ // 其余事件（created、in_progress、output_item.added、output_text.done 等）一律忽略
+ default -> null;
+ };
+ }
+ catch (JsonProcessingException | IllegalArgumentException e) {
+ // readValue 抛 JsonProcessingException；convertValue 结构不匹配时抛
+ // IllegalArgumentException。
+ // 两者都按"单条事件损坏"处理：记录日志后跳过，不让一条畸形事件中断整个 SSE 流
+ log.warn("解析 Responses API SSE 事件失败，跳过: {}", abbreviate(data), e);
+ return null;
+ }
+ }
+
+ /** 日志输出时保留的 SSE 负载最大长度 */
+ private static final int LOG_PAYLOAD_MAX_LENGTH = 500;
+
+ /**
+ * 截断过长的 SSE 负载用于日志输出。 事件体可能携带完整模型输出，全量打印会造成日志膨胀并把对话内容写进日志， 截断后保留的前缀足够定位事件类型与结构问题。
+ */
+ private static String abbreviate(String data) {
+ if (data == null || data.length() <= LOG_PAYLOAD_MAX_LENGTH) {
+ return data;
+ }
+ return data.substring(0, LOG_PAYLOAD_MAX_LENGTH) + "...(已截断，总长度 " + data.length() + ")";
+ }
+
+ /**
+ * 从 ResponsesResponse 的 error 字段提取错误消息
+ */
+ private String extractErrorMessage(ResponsesResponse response) {
+ if (response != null && response.error() != null) {
+ return response.error().message();
+ }
+ return "Responses API 调用失败（未知原因）";
+ }
+
+ /**
+ * 从原始事件 map 的 error 字段提取错误消息
+ */
+ @SuppressWarnings("unchecked")
+ private String extractErrorFromMap(Map eventMap) {
+ Object error = eventMap.get("error");
+ if (error instanceof Map errorMap) {
+ Object message = errorMap.get("message");
+ if (message instanceof String msg) {
+ return msg;
+ }
+ }
+ return "Responses API 返回错误事件";
+ }
+
+ // ======================== 协议 DTO ========================
+
+ /**
+ * Responses API 请求体。 只包含 DataAgent 实际需要的字段，不传 tools/store/previous_response_id 等。
+ */
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public record ResponsesRequest(String model, String instructions, List input, Double temperature,
+ @JsonProperty("max_output_tokens") Integer maxOutputTokens, Boolean stream) {
+
+ /**
+ * 返回仅覆盖 stream 标志的副本。 call/stream 入口统一经此方法强制流式开关， 避免在多处手工重建
+ * record——后续给请求体加字段时只需改这一处，防止字段漏拷
+ */
+ public ResponsesRequest withStream(boolean stream) {
+ return new ResponsesRequest(model, instructions, input, temperature, maxOutputTokens, stream);
+ }
+ }
+
+ /**
+ * 输入消息项：对应 Responses API 的 input 数组元素。 DataAgent 只使用 user 角色的文本消息。
+ */
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public record InputItem(String role, List content) {
+
+ /**
+ * 构造 user 消息
+ */
+ public static InputItem userMessage(String text) {
+ return new InputItem("user", List.of(new ContentPart("input_text", text)));
+ }
+ }
+
+ /**
+ * 内容块：对应 input/output 中的 content 数组元素
+ */
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record ContentPart(String type, String text) {
+ }
+
+ /**
+ * Responses API 完整响应体
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record ResponsesResponse(String id, String model, String status, List output,
+ ResponsesUsage usage, @JsonProperty("incomplete_details") IncompleteDetails incompleteDetails,
+ ResponsesError error) {
+ }
+
+ /**
+ * 输出项：对应 output 数组元素。 type 为 "message" 时包含 content 数组；DataAgent 只关注 output_text
+ * 类型的文本内容。
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record OutputItem(String type, String role, List content) {
+ }
+
+ /**
+ * token 用量统计
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record ResponsesUsage(@JsonProperty("input_tokens") Integer inputTokens,
+ @JsonProperty("output_tokens") Integer outputTokens, @JsonProperty("total_tokens") Integer totalTokens) {
+ }
+
+ /**
+ * 截断详情（status=incomplete 时附带）
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record IncompleteDetails(String reason) {
+ }
+
+ /**
+ * 错误信息
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record ResponsesError(String type, String message, String code) {
+ }
+
+ /**
+ * 流式 SSE 事件的统一封装
+ */
+ public record StreamEvent(Type type, String delta, ResponsesResponse response, String errorMessage) {
+
+ public enum Type {
+
+ /** 文本增量 */
+ DELTA,
+ /** 响应完成（携带 usage） */
+ COMPLETED,
+ /** 响应被截断 */
+ INCOMPLETE,
+ /** 错误 */
+ ERROR
+
+ }
+ }
+
+}
diff --git a/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModel.java b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModel.java
new file mode 100644
index 000000000..1a6471711
--- /dev/null
+++ b/data-agent-management/src/main/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModel.java
@@ -0,0 +1,296 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses;
+
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ContentPart;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.InputItem;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.OutputItem;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesRequest;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesResponse;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesUsage;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.ai.chat.messages.AssistantMessage;
+import org.springframework.ai.chat.messages.Message;
+import org.springframework.ai.chat.messages.MessageType;
+import org.springframework.ai.chat.metadata.ChatGenerationMetadata;
+import org.springframework.ai.chat.metadata.ChatResponseMetadata;
+import org.springframework.ai.chat.metadata.DefaultUsage;
+import org.springframework.ai.chat.model.ChatModel;
+import org.springframework.ai.chat.model.ChatResponse;
+import org.springframework.ai.chat.model.Generation;
+import org.springframework.ai.chat.prompt.ChatOptions;
+import org.springframework.ai.chat.prompt.Prompt;
+import org.springframework.ai.openai.OpenAiChatOptions;
+import org.springframework.ai.retry.RetryUtils;
+import org.springframework.util.StringUtils;
+import reactor.core.publisher.Flux;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 基于 OpenAI Responses API 的 ChatModel 实现。 将 Spring AI 的 Prompt 转换为 Responses API 请求， 将
+ * Responses API 的响应转换回 Spring AI 的 ChatResponse。
+ *
+ * 设计要点：实现 ChatModel 接口后，对上层 AiModelRegistry / LlmService / 全部 16 个图节点完全透明，无需任何改动。
+ */
+@Slf4j
+public class ResponsesApiChatModel implements ChatModel {
+
+ private final ResponsesApi responsesApi;
+
+ private final OpenAiChatOptions defaultOptions;
+
+ public ResponsesApiChatModel(ResponsesApi responsesApi, OpenAiChatOptions defaultOptions) {
+ this.responsesApi = responsesApi;
+ this.defaultOptions = defaultOptions;
+ }
+
+ // ======================== 非流式调用 ========================
+
+ @Override
+ public ChatResponse call(Prompt prompt) {
+ ResponsesRequest request = buildRequest(prompt, false);
+ // 与 OpenAiChatModel 的容错行为对齐：同步调用使用统一重试模板，
+ // 网络抖动或服务端瞬时错误时自动重试，避免两种协议容错能力不一致
+ ResponsesResponse response = RetryUtils.DEFAULT_RETRY_TEMPLATE.execute(ctx -> responsesApi.call(request));
+ return convertToChatResponse(response);
+ }
+
+ // ======================== 流式调用 ========================
+
+ @Override
+ public Flux stream(Prompt prompt) {
+ ResponsesRequest request = buildRequest(prompt, true);
+
+ // 标记本次流是否收到过文本增量事件：个别兼容实现可能不推送 output_text.delta，
+ // 文本只出现在 completed 事件的完整 output 中，此时终包需降级携带全文，否则文本整体丢失。
+ // 每次订阅独立创建（defer），避免同一 Flux 被多次订阅时状态串扰
+ return Flux.defer(() -> {
+ AtomicBoolean deltaReceived = new AtomicBoolean(false);
+
+ // 将 SSE 事件转为 ChatResponse 流，与 OpenAiChatModel 的流式行为对齐：
+ // - delta 事件 → 包含文本块的 ChatResponse（逐 token 推送）
+ // - completed 事件 → 终包：空文本 + usage 元数据（Langfuse token 统计依赖此包）
+ // - incomplete 事件 → 终包：finishReason=LENGTH
+ // - error 事件 → Flux.error 向上传播，由节点重试机制接管
+ return responsesApi.stream(request).concatMap(event -> switch (event.type()) {
+ case DELTA -> {
+ // 文本增量：构建只含文本的 ChatResponse；delta 为 null 时按空串处理，避免 NPE
+ deltaReceived.set(true);
+ String delta = event.delta() != null ? event.delta() : "";
+ Generation generation = new Generation(new AssistantMessage(delta));
+ yield Flux.just(new ChatResponse(List.of(generation)));
+ }
+ case COMPLETED -> {
+ // 完成事件：构建携带 usage 元数据的终包
+ yield Flux.just(buildCompletedResponse(event.response(), "STOP", deltaReceived.get()));
+ }
+ case INCOMPLETE -> {
+ // 截断事件：finishReason 为 LENGTH
+ yield Flux.just(buildCompletedResponse(event.response(), "LENGTH", deltaReceived.get()));
+ }
+ case ERROR -> {
+ // 错误事件：转为异常向上传播
+ yield Flux.error(new RuntimeException("Responses API 错误: " + event.errorMessage()));
+ }
+ });
+ });
+ }
+
+ @Override
+ public ChatOptions getDefaultOptions() {
+ return this.defaultOptions;
+ }
+
+ // ======================== 请求构建 ========================
+
+ /**
+ * 将 Spring AI Prompt 转换为 Responses API 请求。 映射规则：SystemMessage → instructions
+ * 字段，UserMessage → input 数组。
+ */
+ private ResponsesRequest buildRequest(Prompt prompt, boolean stream) {
+ String instructions = null;
+ List inputItems = new ArrayList<>();
+
+ // 从 Prompt 中提取消息，分类映射到 Responses API 字段
+ for (Message message : prompt.getInstructions()) {
+ if (message.getMessageType() == MessageType.SYSTEM) {
+ // SystemMessage 映射为 instructions（Responses API 推荐的系统提示词方式）。
+ // 多条系统消息按出现顺序拼接：instructions 只有一个字段，直接赋值覆盖会静默丢失前文
+ instructions = mergeInstructions(instructions, message.getText());
+ }
+ else if (message.getMessageType() == MessageType.USER) {
+ // UserMessage 映射为 input 数组项
+ inputItems.add(InputItem.userMessage(message.getText()));
+ }
+ // AssistantMessage 和其他类型在 DataAgent 当前链路中不会出现，忽略
+ }
+
+ // 合并运行时 options 与默认 options，运行时优先。
+ // 使用 ChatOptions 接口取值而非 instanceof 具体实现类：
+ // ChatClient 可能传入 DefaultChatOptions 等任意实现，按接口读取保证运行时参数不被静默忽略
+ String model = defaultOptions.getModel();
+ Double temperature = defaultOptions.getTemperature();
+ Integer maxTokens = defaultOptions.getMaxTokens();
+
+ ChatOptions runtimeOptions = prompt.getOptions();
+ if (runtimeOptions != null) {
+ if (runtimeOptions.getModel() != null) {
+ model = runtimeOptions.getModel();
+ }
+ if (runtimeOptions.getTemperature() != null) {
+ temperature = runtimeOptions.getTemperature();
+ }
+ if (runtimeOptions.getMaxTokens() != null) {
+ maxTokens = runtimeOptions.getMaxTokens();
+ }
+ }
+
+ return new ResponsesRequest(model, instructions, inputItems.isEmpty() ? null : inputItems, temperature,
+ maxTokens, stream);
+ }
+
+ /**
+ * 合并多条 SystemMessage 为单个 instructions。 Responses API 只有一个 instructions 字段，
+ * 上游若注入多条系统消息，这里用空行拼接保证语义完整、顺序不变。
+ */
+ private String mergeInstructions(String existing, String text) {
+ if (!StringUtils.hasText(text)) {
+ return existing;
+ }
+ return existing == null ? text : existing + "\n\n" + text;
+ }
+
+ // ======================== 响应转换 ========================
+
+ /**
+ * 将 Responses API 完整响应转换为 ChatResponse（非流式场景）。 从 output 数组中提取 type=message 的文本内容，拼接为
+ * AssistantMessage。
+ */
+ private ChatResponse convertToChatResponse(ResponsesResponse response) {
+ String text = extractOutputText(response);
+ String finishReason = mapFinishReason(response);
+
+ Generation generation = new Generation(new AssistantMessage(text),
+ ChatGenerationMetadata.builder().finishReason(finishReason).build());
+
+ ChatResponseMetadata metadata = buildResponseMetadata(response);
+ return new ChatResponse(List.of(generation), metadata);
+ }
+
+ /**
+ * 构建流式完成/截断事件的终包 ChatResponse。 携带 usage 元数据，与 OpenAiChatModel streamUsage(true)
+ * 的末包行为对齐。
+ *
+ * 关键约束：正常路径下终包文本必须为空串。completed/incomplete 事件的 response 中携带完整 output 全文， 而上层 FluxUtil
+ * 会对流中每个 chunk 的文本做 append 聚合——若此处再返回全文， 聚合结果会出现"全部 delta + 整段全文"的重复，破坏 SQL/JSON 解析。
+ * 例外：若整个流从未收到 delta 事件（个别兼容实现不推送增量），说明文本只存在于终包的完整 output 中， 此时降级在终包携带全文，否则文本会整体丢失。
+ * @param deltaReceived 本次流是否已收到过文本增量事件，决定终包是否需要降级携带全文
+ */
+ private ChatResponse buildCompletedResponse(ResponsesResponse response, String finishReason,
+ boolean deltaReceived) {
+ // 正常路径终包为空文本（见上方约束）；从未收到 delta 且完整 output 中有文本时降级携带全文
+ String text = "";
+ if (!deltaReceived) {
+ String fullText = extractOutputText(response);
+ if (StringUtils.hasText(fullText)) {
+ log.warn("Responses API 流式响应未收到任何文本增量事件，终包降级携带完整文本（长度 {}）", fullText.length());
+ text = fullText;
+ }
+ }
+
+ Generation generation = new Generation(new AssistantMessage(text),
+ ChatGenerationMetadata.builder().finishReason(finishReason).build());
+
+ ChatResponseMetadata metadata = buildResponseMetadata(response);
+ return new ChatResponse(List.of(generation), metadata);
+ }
+
+ /**
+ * 从 ResponsesResponse 的 output 数组中提取所有文本内容并拼接。 只处理 type=message 且 content 中
+ * type=output_text 的内容块。
+ */
+ private String extractOutputText(ResponsesResponse response) {
+ if (response == null || response.output() == null) {
+ return "";
+ }
+
+ StringBuilder sb = new StringBuilder();
+ for (OutputItem outputItem : response.output()) {
+ // 只处理 message 类型的输出项
+ if ("message".equals(outputItem.type()) && outputItem.content() != null) {
+ for (ContentPart part : outputItem.content()) {
+ // 只提取 output_text 类型的文本内容
+ if ("output_text".equals(part.type()) && part.text() != null) {
+ sb.append(part.text());
+ }
+ }
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * 根据 Responses API 的 status 映射 finishReason。 completed → STOP；incomplete +
+ * max_output_tokens → LENGTH；failed → ERROR
+ */
+ private String mapFinishReason(ResponsesResponse response) {
+ if (response == null || response.status() == null) {
+ return "STOP";
+ }
+
+ return switch (response.status()) {
+ case "completed" -> "STOP";
+ case "incomplete" -> {
+ // 检查截断原因
+ if (response.incompleteDetails() != null
+ && "max_output_tokens".equals(response.incompleteDetails().reason())) {
+ yield "LENGTH";
+ }
+ yield "STOP";
+ }
+ case "failed" -> "ERROR";
+ default -> "STOP";
+ };
+ }
+
+ /**
+ * 构建包含 usage 统计的响应元数据。 将 Responses API 的 input_tokens/output_tokens 映射为 Spring AI 的
+ * promptTokens/completionTokens。
+ */
+ private ChatResponseMetadata buildResponseMetadata(ResponsesResponse response) {
+ ChatResponseMetadata.Builder builder = ChatResponseMetadata.builder();
+ if (response != null) {
+ if (response.id() != null) {
+ builder.id(response.id());
+ }
+ if (response.model() != null) {
+ builder.model(response.model());
+ }
+ // usage 映射：input_tokens → promptTokens，output_tokens → completionTokens
+ if (response.usage() != null) {
+ ResponsesUsage u = response.usage();
+ builder.usage(new DefaultUsage(u.inputTokens() != null ? u.inputTokens() : 0,
+ u.outputTokens() != null ? u.outputTokens() : 0,
+ u.totalTokens() != null ? u.totalTokens() : 0));
+ }
+ }
+ return builder.build();
+ }
+
+}
diff --git a/data-agent-management/src/main/resources/sql/h2/schema-h2.sql b/data-agent-management/src/main/resources/sql/h2/schema-h2.sql
index 5eadd0907..611c30435 100644
--- a/data-agent-management/src/main/resources/sql/h2/schema-h2.sql
+++ b/data-agent-management/src/main/resources/sql/h2/schema-h2.sql
@@ -265,5 +265,6 @@ CREATE TABLE IF NOT EXISTS `model_config` (
`proxy_port` int(11) DEFAULT NULL COMMENT '代理端口',
`proxy_username` varchar(255) DEFAULT NULL COMMENT '代理用户名（可选）',
`proxy_password` varchar(255) DEFAULT NULL COMMENT '代理密码（可选）',
+ `chat_api_protocol` varchar(32) DEFAULT 'CHAT_COMPLETIONS' COMMENT 'Chat模型接口协议：CHAT_COMPLETIONS（默认Chat Completions格式）、RESPONSES（OpenAI Responses API格式）。仅model_type=CHAT时生效',
PRIMARY KEY (`id`)
) ENGINE=InnoDB;
diff --git a/data-agent-management/src/main/resources/sql/schema.sql b/data-agent-management/src/main/resources/sql/schema.sql
index 160bcedf0..7478e01e8 100644
--- a/data-agent-management/src/main/resources/sql/schema.sql
+++ b/data-agent-management/src/main/resources/sql/schema.sql
@@ -266,5 +266,6 @@ CREATE TABLE IF NOT EXISTS `model_config` (
`proxy_port` int(11) DEFAULT NULL COMMENT '代理端口',
`proxy_username` varchar(255) DEFAULT NULL COMMENT '代理用户名（可选）',
`proxy_password` varchar(255) DEFAULT NULL COMMENT '代理密码（可选）',
+ `chat_api_protocol` varchar(32) DEFAULT 'CHAT_COMPLETIONS' COMMENT 'Chat模型接口协议：CHAT_COMPLETIONS（默认Chat Completions格式）、RESPONSES（OpenAI Responses API格式）。仅model_type=CHAT时生效',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/converter/ModelConfigConverterTest.java b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/converter/ModelConfigConverterTest.java
index 2b0907e57..aa4259de1 100644
--- a/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/converter/ModelConfigConverterTest.java
+++ b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/converter/ModelConfigConverterTest.java
@@ -131,4 +131,36 @@ void toEntity_embeddingModelType_mapsCorrectly() {
assertEquals(ModelType.EMBEDDING, entity.getModelType());
}
+ // ======================== chatApiProtocol 默认值兜底 ========================
+
+ private ModelConfigDTO buildChatDto(String chatApiProtocol) {
+ return ModelConfigDTO.builder()
+ .provider("openai")
+ .baseUrl("https://api.example.com")
+ .apiKey("sk-test")
+ .modelName("test-model")
+ .modelType("CHAT")
+ .chatApiProtocol(chatApiProtocol)
+ .build();
+ }
+
+ @Test
+ void toEntity_backfillsDefaultProtocolWhenNull() {
+ // chatApiProtocol 为空时必须回填默认协议：INSERT 显式写入 NULL 会绕过数据库列默认值
+ ModelConfig entity = ModelConfigConverter.toEntity(buildChatDto(null));
+ assertEquals("CHAT_COMPLETIONS", entity.getChatApiProtocol());
+ }
+
+ @Test
+ void toEntity_backfillsDefaultProtocolWhenBlank() {
+ ModelConfig entity = ModelConfigConverter.toEntity(buildChatDto(" "));
+ assertEquals("CHAT_COMPLETIONS", entity.getChatApiProtocol());
+ }
+
+ @Test
+ void toEntity_keepsExplicitProtocol() {
+ ModelConfig entity = ModelConfigConverter.toEntity(buildChatDto("RESPONSES"));
+ assertEquals("RESPONSES", entity.getChatApiProtocol());
+ }
+
}
diff --git a/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/enums/ChatApiProtocolTest.java b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/enums/ChatApiProtocolTest.java
new file mode 100644
index 000000000..5a092ef90
--- /dev/null
+++ b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/enums/ChatApiProtocolTest.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.cloud.ai.dataagent.enums;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * ChatApiProtocol 协议解析单元测试。 重点回归：非法值必须抛异常拦截在保存入口， 不能静默落入默认协议分支导致用户配置悄悄失效。
+ */
+class ChatApiProtocolTest {
+
+ @Test
+ void fromCode_resolvesStandardValues() {
+ assertEquals(ChatApiProtocol.CHAT_COMPLETIONS, ChatApiProtocol.fromCode("CHAT_COMPLETIONS"));
+ assertEquals(ChatApiProtocol.RESPONSES, ChatApiProtocol.fromCode("RESPONSES"));
+ }
+
+ @Test
+ void fromCode_ignoresCase() {
+ // 与 DynamicModelFactory 的 equalsIgnoreCase 容忍口径一致
+ assertEquals(ChatApiProtocol.RESPONSES, ChatApiProtocol.fromCode("responses"));
+ assertEquals(ChatApiProtocol.CHAT_COMPLETIONS, ChatApiProtocol.fromCode("chat_completions"));
+ }
+
+ @Test
+ void fromCode_rejectsUnknownValue() {
+ // 拼写错误（如漏掉 S）必须报错，而不是静默回退默认协议
+ assertThrows(IllegalArgumentException.class, () -> ChatApiProtocol.fromCode("RESPONSE"));
+ assertThrows(IllegalArgumentException.class, () -> ChatApiProtocol.fromCode(""));
+ assertThrows(IllegalArgumentException.class, () -> ChatApiProtocol.fromCode(null));
+ }
+
+}
diff --git a/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModelTest.java b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModelTest.java
new file mode 100644
index 000000000..00d8e3490
--- /dev/null
+++ b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiChatModelTest.java
@@ -0,0 +1,266 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses;
+
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ContentPart;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.IncompleteDetails;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.OutputItem;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesRequest;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesResponse;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.ResponsesUsage;
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.StreamEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.ai.chat.messages.SystemMessage;
+import org.springframework.ai.chat.messages.UserMessage;
+import org.springframework.ai.chat.model.ChatResponse;
+import org.springframework.ai.chat.prompt.ChatOptions;
+import org.springframework.ai.chat.prompt.Prompt;
+import org.springframework.ai.openai.OpenAiChatOptions;
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * ResponsesApiChatModel 协议转换单元测试。 覆盖 Prompt → Responses 请求映射、非流式响应转换、
+ * 流式事件转换（重点回归：终包必须为空文本，避免上层聚合时全文重复）。
+ */
+class ResponsesApiChatModelTest {
+
+ private ResponsesApi responsesApi;
+
+ private ResponsesApiChatModel chatModel;
+
+ @BeforeEach
+ void setUp() {
+ responsesApi = mock(ResponsesApi.class);
+ OpenAiChatOptions defaultOptions = OpenAiChatOptions.builder()
+ .model("default-model")
+ .temperature(0.1)
+ .maxTokens(1000)
+ .build();
+ chatModel = new ResponsesApiChatModel(responsesApi, defaultOptions);
+ }
+
+ /**
+ * 构造一个携带文本输出和 usage 的完整响应
+ */
+ private ResponsesResponse completedResponse(String text) {
+ OutputItem item = new OutputItem("message", "assistant", List.of(new ContentPart("output_text", text)));
+ return new ResponsesResponse("resp_1", "gpt-test", "completed", List.of(item), new ResponsesUsage(10, 5, 15),
+ null, null);
+ }
+
+ // ======================== 请求映射 ========================
+
+ @Test
+ void call_mapsSystemToInstructionsAndUserToInput() {
+ when(responsesApi.call(any())).thenReturn(completedResponse("ok"));
+
+ Prompt prompt = new Prompt(List.of(new SystemMessage("you are helpful"), new UserMessage("hi")));
+ chatModel.call(prompt);
+
+ ArgumentCaptor captor = ArgumentCaptor.forClass(ResponsesRequest.class);
+ verify(responsesApi).call(captor.capture());
+ ResponsesRequest request = captor.getValue();
+
+ assertEquals("you are helpful", request.instructions());
+ assertEquals(1, request.input().size());
+ assertEquals("user", request.input().get(0).role());
+ assertEquals("hi", request.input().get(0).content().get(0).text());
+ // 未指定运行时 options 时使用默认配置
+ assertEquals("default-model", request.model());
+ assertEquals(0.1, request.temperature());
+ assertEquals(1000, request.maxOutputTokens());
+ }
+
+ @Test
+ void call_runtimeOptionsOverrideDefaults_viaChatOptionsInterface() {
+ when(responsesApi.call(any())).thenReturn(completedResponse("ok"));
+
+ // 使用通用 ChatOptions 实现（非 OpenAiChatOptions），验证按接口取值不会被静默忽略
+ ChatOptions runtimeOptions = ChatOptions.builder()
+ .model("runtime-model")
+ .temperature(0.9)
+ .maxTokens(123)
+ .build();
+ Prompt prompt = new Prompt(List.of(new UserMessage("hi")), runtimeOptions);
+ chatModel.call(prompt);
+
+ ArgumentCaptor captor = ArgumentCaptor.forClass(ResponsesRequest.class);
+ verify(responsesApi).call(captor.capture());
+ ResponsesRequest request = captor.getValue();
+
+ assertEquals("runtime-model", request.model());
+ assertEquals(0.9, request.temperature());
+ assertEquals(123, request.maxOutputTokens());
+ }
+
+ @Test
+ void call_multipleSystemMessages_concatenatedIntoInstructions() {
+ when(responsesApi.call(any())).thenReturn(completedResponse("ok"));
+
+ // 多条系统消息必须按顺序拼接为单个 instructions，直接覆盖会静默丢失前文
+ Prompt prompt = new Prompt(
+ List.of(new SystemMessage("first rule"), new SystemMessage("second rule"), new UserMessage("hi")));
+ chatModel.call(prompt);
+
+ ArgumentCaptor captor = ArgumentCaptor.forClass(ResponsesRequest.class);
+ verify(responsesApi).call(captor.capture());
+ assertEquals("first rule\n\nsecond rule", captor.getValue().instructions());
+ }
+
+ // ======================== 非流式响应转换 ========================
+
+ @Test
+ void call_extractsTextAndUsage() {
+ when(responsesApi.call(any())).thenReturn(completedResponse("hello world"));
+
+ ChatResponse response = chatModel.call(new Prompt(List.of(new UserMessage("hi"))));
+
+ assertEquals("hello world", response.getResult().getOutput().getText());
+ assertEquals("STOP", response.getResult().getMetadata().getFinishReason());
+ assertEquals(10, response.getMetadata().getUsage().getPromptTokens());
+ assertEquals(5, response.getMetadata().getUsage().getCompletionTokens());
+ }
+
+ @Test
+ void call_incompleteWithMaxTokens_mapsToLengthFinishReason() {
+ OutputItem item = new OutputItem("message", "assistant", List.of(new ContentPart("output_text", "partial")));
+ ResponsesResponse incomplete = new ResponsesResponse("resp_2", "gpt-test", "incomplete", List.of(item), null,
+ new IncompleteDetails("max_output_tokens"), null);
+ when(responsesApi.call(any())).thenReturn(incomplete);
+
+ ChatResponse response = chatModel.call(new Prompt(List.of(new UserMessage("hi"))));
+
+ assertEquals("partial", response.getResult().getOutput().getText());
+ assertEquals("LENGTH", response.getResult().getMetadata().getFinishReason());
+ }
+
+ // ======================== 流式响应转换 ========================
+
+ @Test
+ void stream_terminalChunkMustBeEmptyText_toAvoidDuplication() {
+ // completed 事件的 response 中携带完整全文；上层 FluxUtil 会对每个 chunk 的文本做 append 聚合，
+ // 因此终包文本必须为空串，否则聚合结果为"全部 delta + 整段全文"的重复（P0 回归用例）
+ when(responsesApi.stream(any())).thenReturn(Flux.just(new StreamEvent(StreamEvent.Type.DELTA, "Hello", null, null),
+ new StreamEvent(StreamEvent.Type.DELTA, " world", null, null),
+ new StreamEvent(StreamEvent.Type.COMPLETED, null, completedResponse("Hello world"), null)));
+
+ List chunks = chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ assertNotNull(chunks);
+ assertEquals(3, chunks.size());
+ assertEquals("Hello", chunks.get(0).getResult().getOutput().getText());
+ assertEquals(" world", chunks.get(1).getResult().getOutput().getText());
+ // 终包：空文本 + finishReason + usage
+ assertEquals("", chunks.get(2).getResult().getOutput().getText());
+ assertEquals("STOP", chunks.get(2).getResult().getMetadata().getFinishReason());
+ assertEquals(10, chunks.get(2).getMetadata().getUsage().getPromptTokens());
+
+ // 模拟上层聚合行为：全部 chunk 文本拼接后应恰好等于一遍全文
+ StringBuilder aggregated = new StringBuilder();
+ chunks.forEach(c -> aggregated.append(c.getResult().getOutput().getText()));
+ assertEquals("Hello world", aggregated.toString());
+ }
+
+ @Test
+ void stream_incompleteEvent_mapsToLengthTerminal() {
+ ResponsesResponse incomplete = new ResponsesResponse("resp_3", "gpt-test", "incomplete", null, null,
+ new IncompleteDetails("max_output_tokens"), null);
+ when(responsesApi.stream(any()))
+ .thenReturn(Flux.just(new StreamEvent(StreamEvent.Type.DELTA, "part", null, null),
+ new StreamEvent(StreamEvent.Type.INCOMPLETE, null, incomplete, null)));
+
+ List chunks = chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ assertNotNull(chunks);
+ assertEquals(2, chunks.size());
+ assertEquals("", chunks.get(1).getResult().getOutput().getText());
+ assertEquals("LENGTH", chunks.get(1).getResult().getMetadata().getFinishReason());
+ }
+
+ @Test
+ void stream_nullDelta_treatedAsEmptyText() {
+ // delta 字段缺失时按空串处理，不应抛 NPE
+ when(responsesApi.stream(any()))
+ .thenReturn(Flux.just(new StreamEvent(StreamEvent.Type.DELTA, null, null, null)));
+
+ List chunks = chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ assertNotNull(chunks);
+ assertEquals("", chunks.get(0).getResult().getOutput().getText());
+ }
+
+ @Test
+ void stream_noDeltaReceived_terminalCarriesFullText() {
+ // 个别兼容实现不推送 output_text.delta，文本只出现在 completed 事件的完整 output 中，
+ // 此时终包必须降级携带全文，否则文本整体丢失
+ when(responsesApi.stream(any())).thenReturn(
+ Flux.just(new StreamEvent(StreamEvent.Type.COMPLETED, null, completedResponse("full text only"), null)));
+
+ List chunks = chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ assertNotNull(chunks);
+ assertEquals(1, chunks.size());
+ assertEquals("full text only", chunks.get(0).getResult().getOutput().getText());
+ assertEquals("STOP", chunks.get(0).getResult().getMetadata().getFinishReason());
+ }
+
+ @Test
+ void stream_withDelta_terminalStaysEmptyEvenIfOutputHasText() {
+ // 已收到 delta 时终包必须保持空文本（防聚合重复），不能因 output 中有全文而降级
+ when(responsesApi.stream(any()))
+ .thenReturn(Flux.just(new StreamEvent(StreamEvent.Type.DELTA, "Hello world", null, null),
+ new StreamEvent(StreamEvent.Type.COMPLETED, null, completedResponse("Hello world"), null)));
+
+ List chunks = chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ assertNotNull(chunks);
+ assertEquals(2, chunks.size());
+ assertEquals("", chunks.get(1).getResult().getOutput().getText());
+ }
+
+ @Test
+ void stream_errorEvent_propagatesAsFluxError() {
+ when(responsesApi.stream(any()))
+ .thenReturn(Flux.just(new StreamEvent(StreamEvent.Type.ERROR, null, null, "boom")));
+
+ Flux flux = chatModel.stream(new Prompt(List.of(new UserMessage("hi"))));
+
+ RuntimeException ex = assertThrows(RuntimeException.class, flux::blockLast);
+ assertTrue(ex.getMessage().contains("boom"));
+ }
+
+ @Test
+ void stream_requestHasStreamFlagTrue() {
+ when(responsesApi.stream(any())).thenReturn(Flux.empty());
+
+ chatModel.stream(new Prompt(List.of(new UserMessage("hi")))).collectList().block();
+
+ ArgumentCaptor captor = ArgumentCaptor.forClass(ResponsesRequest.class);
+ verify(responsesApi).stream(captor.capture());
+ assertTrue(captor.getValue().stream());
+ }
+
+}
diff --git a/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiTest.java b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiTest.java
new file mode 100644
index 000000000..f0b643944
--- /dev/null
+++ b/data-agent-management/src/test/java/com/alibaba/cloud/ai/dataagent/service/aimodelconfig/responses/ResponsesApiTest.java
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses;
+
+import com.alibaba.cloud.ai.dataagent.service.aimodelconfig.responses.ResponsesApi.StreamEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.web.client.RestClient;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * ResponsesApi SSE 事件解析单元测试。 覆盖 delta / completed / incomplete / failed / error
+ * 各事件分支，以及畸形事件的容错行为。
+ */
+class ResponsesApiTest {
+
+ private ResponsesApi responsesApi;
+
+ @BeforeEach
+ void setUp() {
+ // 仅测试解析逻辑，不发起真实 HTTP 请求，baseUrl 等参数使用占位值
+ responsesApi = new ResponsesApi("http://localhost", "test-key", null, RestClient.builder(),
+ WebClient.builder());
+ }
+
+ private StreamEvent parse(String data) {
+ return responsesApi.parseStreamEvent(ServerSentEvent.builder(data).build());
+ }
+
+ @Test
+ void parseDeltaEvent_returnsDeltaText() {
+ StreamEvent event = parse("{\"type\":\"response.output_text.delta\",\"delta\":\"Hello\"}");
+ assertNotNull(event);
+ assertEquals(StreamEvent.Type.DELTA, event.type());
+ assertEquals("Hello", event.delta());
+ }
+
+ @Test
+ void parseCompletedEvent_returnsResponseWithUsage() {
+ String data = """
+ {"type":"response.completed","response":{"id":"resp_1","model":"gpt-test","status":"completed",
+ "output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"hi"}]}],
+ "usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}
+ """;
+ StreamEvent event = parse(data);
+ assertNotNull(event);
+ assertEquals(StreamEvent.Type.COMPLETED, event.type());
+ assertEquals("resp_1", event.response().id());
+ assertEquals(10, event.response().usage().inputTokens());
+ assertEquals(5, event.response().usage().outputTokens());
+ }
+
+ @Test
+ void parseIncompleteEvent_returnsIncompleteType() {
+ String data = """
+ {"type":"response.incomplete","response":{"id":"resp_2","status":"incomplete",
+ "incomplete_details":{"reason":"max_output_tokens"}}}
+ """;
+ StreamEvent event = parse(data);
+ assertNotNull(event);
+ assertEquals(StreamEvent.Type.INCOMPLETE, event.type());
+ assertEquals("max_output_tokens", event.response().incompleteDetails().reason());
+ }
+
+ @Test
+ void parseFailedEvent_returnsErrorWithMessage() {
+ String data = """
+ {"type":"response.failed","response":{"id":"resp_3","status":"failed",
+ "error":{"type":"server_error","message":"boom","code":"500"}}}
+ """;
+ StreamEvent event = parse(data);
+ assertNotNull(event);
+ assertEquals(StreamEvent.Type.ERROR, event.type());
+ assertEquals("boom", event.errorMessage());
+ }
+
+ @Test
+ void parseErrorEvent_returnsErrorWithMessage() {
+ StreamEvent event = parse("{\"type\":\"error\",\"error\":{\"message\":\"rate limited\"}}");
+ assertNotNull(event);
+ assertEquals(StreamEvent.Type.ERROR, event.type());
+ assertEquals("rate limited", event.errorMessage());
+ }
+
+ @Test
+ void parseUnknownEventType_returnsNull() {
+ // 未知事件类型（如 response.created）应静默忽略，保证向前兼容
+ assertNull(parse("{\"type\":\"response.created\",\"response\":{\"id\":\"resp_4\"}}"));
+ }
+
+ @Test
+ void parseMalformedJson_returnsNull() {
+ // 畸形 JSON 不应抛异常中断流，应按单条事件损坏跳过
+ assertNull(parse("{not valid json"));
+ }
+
+ @Test
+ void parseCompletedEventWithWrongStructure_returnsNull() {
+ // response 字段结构不匹配时 convertValue 抛 IllegalArgumentException，
+ // 必须被捕获并跳过，不能让一条畸形事件中断整个 SSE 流（回归用例）
+ assertNull(parse("{\"type\":\"response.completed\",\"response\":\"not-an-object\"}"));
+ }
+
+ @Test
+ void parseEventWithoutType_returnsNull() {
+ assertNull(parse("{\"delta\":\"orphan\"}"));
+ }
+
+}