diff --git a/README.md b/README.md index 10e46d36..129e566d 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ Conversation conversation = new Conversation(Protocol.HTTP.getValue()); Conversation conversation = new Conversation(Protocol.WEBSOCKET.getValue()); ``` -The conversation interface supports both stream and non-stream queries. These queries all accept `ConversationParam` as input, and returns `ConversationResult` as output . Each model has a unique input structure and output structure which derives from the two data classes mentioned above. Please use these sub classes when you are using the coordinating model. +The conversation interface supports both stream and non-stream queries. These queries all accept `ConversationParam` as input, and returns `ConversationResult` as output . Each model has a unique input structure and output structure which derives from the two data classes mentioned above. Please use these sub classes when you are using the coordinating model. Here shows the usages of each method, with the examples of `qwen-turbo` model. diff --git a/pom.xml b/pom.xml index 55decafe..534a5f30 100644 --- a/pom.xml +++ b/pom.xml @@ -40,7 +40,7 @@ DashScope Java SDK com.alibaba dashscope-sdk-java - 2.22.20 + 2.22.21 8 diff --git a/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGeneration.java b/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGeneration.java index fb07c79b..877611a8 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGeneration.java +++ b/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGeneration.java @@ -1,16 +1,26 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.codegeneration; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; import io.reactivex.Flowable; import lombok.extern.slf4j.Slf4j; @Slf4j -public class CodeGeneration { +public final class CodeGeneration { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; diff --git a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletion.java b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletion.java index f6168da4..3a2a6d2d 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletion.java +++ b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletion.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.completion; import com.alibaba.dashscope.common.DashScopeResult; diff --git a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletionStreamOptions.java b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletionStreamOptions.java index 774f6f3c..101a8877 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletionStreamOptions.java +++ b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletionStreamOptions.java @@ -1,12 +1,16 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.completion; -import com.google.gson.annotations.SerializedName; import lombok.Data; +import lombok.EqualsAndHashCode; import lombok.experimental.SuperBuilder; +/** + * @deprecated Use {@link com.alibaba.dashscope.common.StreamOptions} instead for a shared + * implementation across all service classes. + */ +@Deprecated @Data @SuperBuilder -public class ChatCompletionStreamOptions { - @SerializedName("include_usage") - private Boolean includeUsage; -} +@EqualsAndHashCode(callSuper = true) +public class ChatCompletionStreamOptions extends com.alibaba.dashscope.common.StreamOptions {} diff --git a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletions.java b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletions.java index d0013df5..30aabfa9 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletions.java +++ b/src/main/java/com/alibaba/dashscope/aigc/completion/ChatCompletions.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.completion; import com.alibaba.dashscope.api.GeneralApi; diff --git a/src/main/java/com/alibaba/dashscope/aigc/generation/Generation.java b/src/main/java/com/alibaba/dashscope/aigc/generation/Generation.java index c329b66a..aca4b7e4 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/generation/Generation.java +++ b/src/main/java/com/alibaba/dashscope/aigc/generation/Generation.java @@ -18,8 +18,8 @@ import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import com.alibaba.dashscope.tools.ToolCallBase; -import com.alibaba.dashscope.tools.ToolCallFunction; import com.alibaba.dashscope.utils.ParamUtils; +import com.alibaba.dashscope.utils.StreamingMerger; import com.alibaba.dashscope.utils.StringUtils; import io.reactivex.Flowable; import java.util.ArrayList; @@ -33,9 +33,6 @@ public final class Generation { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; - private final ThreadLocal> accumulatedDataMap = - ThreadLocal.withInitial(HashMap::new); - public static class Models { /** @deprecated use QWEN_TURBO instead */ @Deprecated public static final String QWEN_V1 = "qwen-v1"; @@ -64,6 +61,21 @@ private ApiServiceOption defaultApiServiceOption() { .build(); } + /** Creates a copy of the shared serviceOption for thread-safe per-call usage. */ + private ApiServiceOption copyServiceOption() { + return ApiServiceOption.builder() + .protocol(serviceOption.getProtocol()) + .httpMethod(serviceOption.getHttpMethod()) + .streamingMode(serviceOption.getStreamingMode()) + .outputMode(serviceOption.getOutputMode()) + .taskGroup(serviceOption.getTaskGroup()) + .task(serviceOption.getTask()) + .function(serviceOption.getFunction()) + .baseHttpUrl(serviceOption.getBaseHttpUrl()) + .baseWebSocketUrl(serviceOption.getBaseWebSocketUrl()) + .build(); + } + public Generation() { serviceOption = defaultApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); @@ -78,7 +90,7 @@ public Generation(String protocol) { public Generation(String protocol, String baseUrl) { serviceOption = defaultApiServiceOption(); serviceOption.setProtocol(Protocol.of(protocol)); - if (protocol.equals(Protocol.HTTP.getValue())) { + if (Protocol.HTTP.getValue().equals(protocol)) { serviceOption.setBaseHttpUrl(baseUrl); } else { serviceOption.setBaseWebSocketUrl(baseUrl); @@ -89,7 +101,7 @@ public Generation(String protocol, String baseUrl) { public Generation(String protocol, String baseUrl, ConnectionOptions connectionOptions) { serviceOption = defaultApiServiceOption(); serviceOption.setProtocol(Protocol.of(protocol)); - if (protocol.equals(Protocol.HTTP.getValue())) { + if (Protocol.HTTP.getValue().equals(protocol)) { serviceOption.setBaseHttpUrl(baseUrl); } else { serviceOption.setBaseWebSocketUrl(baseUrl); @@ -108,9 +120,10 @@ public Generation(String protocol, String baseUrl, ConnectionOptions connectionO public GenerationResult call(HalfDuplexServiceParam param) throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); - return GenerationResult.fromDashScopeResult(syncApi.call(param)); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); + return GenerationResult.fromDashScopeResult(syncApi.call(param, callOption)); } /** @@ -125,8 +138,9 @@ public GenerationResult call(HalfDuplexServiceParam param) public void call(HalfDuplexServiceParam param, ResultCallback callback) throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); syncApi.call( param, new ResultCallback() { @@ -144,7 +158,8 @@ public void onComplete() { public void onError(Exception e) { callback.onError(e); } - }); + }, + callOption); } /** @@ -168,31 +183,26 @@ public Flowable streamCall(HalfDuplexServiceParam param) String userAgentSuffix = StringUtils.format("incremental_to_full/%d", flagValue); param.putHeader("user-agent", userAgentSuffix); - serviceOption.setIsSSE(true); - serviceOption.setStreamingMode(StreamingMode.OUT); - return syncApi - .streamCall(param) - .map(GenerationResult::fromDashScopeResult) - .flatMap( - result -> { - GenerationResult merged = mergeSingleResponse(result, toMergeResponse, param); - if (merged == null) { - return Flowable.empty(); - } - return Flowable.just(merged); - }) - .doOnComplete( - () -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }) - .doOnError( - throwable -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(true); + callOption.setStreamingMode(StreamingMode.OUT); + return Flowable.defer( + () -> { + Map accumulatedData = new HashMap<>(); + return syncApi + .streamCall(param, callOption) + .map(GenerationResult::fromDashScopeResult) + .flatMap( + result -> { + GenerationResult merged = + mergeSingleResponse(result, toMergeResponse, param, accumulatedData); + if (merged == null) { + return Flowable.empty(); + } + return Flowable.just(merged); + }) + .doFinally(accumulatedData::clear); + }); } public void streamCall(HalfDuplexServiceParam param, ResultCallback callback) @@ -207,15 +217,18 @@ public void streamCall(HalfDuplexServiceParam param, ResultCallback accumulatedData = new HashMap<>(); syncApi.streamCall( param, new ResultCallback() { @Override public void onEvent(DashScopeResult msg) { GenerationResult result = GenerationResult.fromDashScopeResult(msg); - GenerationResult mergedResult = mergeSingleResponse(result, toMergeResponse, param); + GenerationResult mergedResult = + mergeSingleResponse(result, toMergeResponse, param, accumulatedData); if (mergedResult != null) { callback.onEvent(mergedResult); } @@ -223,20 +236,17 @@ public void onEvent(DashScopeResult msg) { @Override public void onComplete() { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onComplete(); } @Override public void onError(Exception e) { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onError(e); } - }); + }, + callOption); } /** @@ -270,16 +280,18 @@ private boolean modifyIncrementalOutput(HalfDuplexServiceParam param) { * @param result The GenerationResult to merge * @param toMergeResponse Whether to perform merging (based on original incrementalOutput setting) * @param param The HalfDuplexServiceParam to get n parameter + * @param accumulatedData The per-stream accumulated data * @return The merged GenerationResult, or null if should be filtered out */ private GenerationResult mergeSingleResponse( - GenerationResult result, boolean toMergeResponse, HalfDuplexServiceParam param) { + GenerationResult result, + boolean toMergeResponse, + HalfDuplexServiceParam param, + Map accumulatedData) { if (!toMergeResponse || result == null || result.getOutput() == null) { return result; } - Map accumulatedData = accumulatedDataMap.get(); - // Get n parameter Integer n = null; if (param instanceof GenerationParam) { @@ -344,10 +356,10 @@ private GenerationResult mergeSingleResponse( choice.getMessage().setReasoningContent(accumulated.reasoningContent.toString()); } - // Handle tool_calls accumulation + // Handle tool_calls accumulation (delegate to shared utility) List currentToolCalls = choice.getMessage().getToolCalls(); if (currentToolCalls != null && !currentToolCalls.isEmpty()) { - mergeToolCalls(currentToolCalls, accumulated.toolCalls); + StreamingMerger.mergeToolCalls(currentToolCalls, accumulated.toolCalls); } // Always set accumulated tool_calls if we have any if (!accumulated.toolCalls.isEmpty()) { @@ -526,110 +538,6 @@ private GenerationResult mergeSingleResponse( return result; } - /** Merges tool calls from current response with accumulated tool calls. */ - private void mergeToolCalls( - List currentToolCalls, List accumulatedToolCalls) { - for (ToolCallBase currentCall : currentToolCalls) { - if (currentCall == null || currentCall.getIndex() == null) { - continue; - } - - int index = currentCall.getIndex(); - - // Find existing accumulated call with same index - ToolCallBase existingCall = null; - for (ToolCallBase accCall : accumulatedToolCalls) { - if (accCall != null && accCall.getIndex() != null && accCall.getIndex().equals(index)) { - existingCall = accCall; - break; - } - } - - if (existingCall instanceof ToolCallFunction && currentCall instanceof ToolCallFunction) { - // Merge function calls - ToolCallFunction existingFunctionCall = (ToolCallFunction) existingCall; - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - - if (currentFunctionCall.getFunction() != null) { - // Ensure existing function call has a function object - if (existingFunctionCall.getFunction() == null) { - existingFunctionCall.setFunction(existingFunctionCall.new CallFunction()); - } - - // Accumulate arguments if present - if (currentFunctionCall.getFunction().getArguments() != null) { - String existingArguments = existingFunctionCall.getFunction().getArguments(); - if (existingArguments == null) { - existingArguments = ""; - } - String currentArguments = currentFunctionCall.getFunction().getArguments(); - existingFunctionCall.getFunction().setArguments(existingArguments + currentArguments); - } - - // Accumulate function name if present - if (currentFunctionCall.getFunction().getName() != null) { - String existingName = existingFunctionCall.getFunction().getName(); - if (existingName == null) { - existingName = ""; - } - String currentName = currentFunctionCall.getFunction().getName(); - existingFunctionCall.getFunction().setName(existingName + currentName); - } - - // Update function output if present - if (currentFunctionCall.getFunction().getOutput() != null) { - existingFunctionCall - .getFunction() - .setOutput(currentFunctionCall.getFunction().getOutput()); - } - } - - // Update other fields with latest non-empty values - if (currentFunctionCall.getIndex() != null) { - existingFunctionCall.setIndex(currentFunctionCall.getIndex()); - } - if (currentFunctionCall.getId() != null && !currentFunctionCall.getId().isEmpty()) { - existingFunctionCall.setId(currentFunctionCall.getId()); - } - if (currentFunctionCall.getType() != null) { - existingFunctionCall.setType(currentFunctionCall.getType()); - } - } else { - // Add new tool call (create a copy) - if (currentCall instanceof ToolCallFunction) { - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - ToolCallFunction newFunctionCall = new ToolCallFunction(); - newFunctionCall.setIndex(currentFunctionCall.getIndex()); - newFunctionCall.setId(currentFunctionCall.getId()); - newFunctionCall.setType(currentFunctionCall.getType()); - - if (currentFunctionCall.getFunction() != null) { - ToolCallFunction.CallFunction newCallFunction = newFunctionCall.new CallFunction(); - newCallFunction.setName(currentFunctionCall.getFunction().getName()); - newCallFunction.setArguments(currentFunctionCall.getFunction().getArguments()); - newCallFunction.setOutput(currentFunctionCall.getFunction().getOutput()); - newFunctionCall.setFunction(newCallFunction); - } - - accumulatedToolCalls.add(newFunctionCall); - } else { - // For other types of tool calls, add directly (assuming they are immutable or don't need - // merging) - accumulatedToolCalls.add(currentCall); - } - } - } - } - - /** - * Clears accumulated data for the current thread. Should be called when streaming is complete or - * encounters error. - */ - private void clearAccumulatedData() { - accumulatedDataMap.get().clear(); - accumulatedDataMap.remove(); - } - /** Inner class to store accumulated data for response merging. */ private static class AccumulatedData { StringBuilder content = new StringBuilder(); diff --git a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationOutput.java b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationOutput.java index d8eae904..4f4f2b6c 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationOutput.java +++ b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationOutput.java @@ -20,6 +20,10 @@ public class Choice { private Message message; private GenerationLogprobs logprobs; + + /** Search information for this specific choice, returned when search_options is configured. */ + @SerializedName("search_info") + private SearchInfo searchInfo; } // output text private String text; diff --git a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationParam.java b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationParam.java index 17269c5a..120751fb 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationParam.java +++ b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationParam.java @@ -5,6 +5,7 @@ import static com.alibaba.dashscope.utils.ApiKeywords.HISTORY; import static com.alibaba.dashscope.utils.ApiKeywords.MAX_TOKENS; import static com.alibaba.dashscope.utils.ApiKeywords.MESSAGES; +import static com.alibaba.dashscope.utils.ApiKeywords.PRESENCE_PENALTY; import static com.alibaba.dashscope.utils.ApiKeywords.PROMPT; import static com.alibaba.dashscope.utils.ApiKeywords.REPETITION_PENALTY; import static com.alibaba.dashscope.utils.ApiKeywords.STOP; @@ -13,6 +14,7 @@ import com.alibaba.dashscope.common.ResponseFormat; import com.alibaba.dashscope.common.Role; import com.alibaba.dashscope.common.SearchOptions; +import com.alibaba.dashscope.common.StreamOptions; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.tools.ToolBase; import com.alibaba.dashscope.utils.ApiKeywords; @@ -36,8 +38,8 @@ @SuperBuilder public class GenerationParam extends GenerationParamBase { public static class ResultFormat { - public static String TEXT = "text"; - public static String MESSAGE = "message"; + public static final String TEXT = "text"; + public static final String MESSAGE = "message"; } private List messages; @@ -96,13 +98,25 @@ public static class ResultFormat { * apple * */ - @Builder.Default private Boolean incrementalOutput; + private Boolean incrementalOutput; /** Maximum tokens to generate. */ private Integer maxTokens; /** repetition penalty */ private Float repetitionPenalty; + /** + * Presence penalty: reduces the model's likelihood to repeat the exact tokens that have already + * appeared in the output. A value of 0 indicates no penalty. Default value: 0.0 + */ + private Float presencePenalty; + + /** + * Frequency penalty: reduces the model's likelihood to repeat tokens based on their frequency in + * the output so far. A value of 0 indicates no penalty. Default value: 0.0 + */ + private Float frequencyPenalty; + /** stopString and token are mutually exclusive. */ @Singular("stopString") private List stopStrings; @@ -143,6 +157,9 @@ public static class ResultFormat { /** 翻译参数 */ private TranslationOptions translationOptions; + /** Streaming options controlling what extra info is included in stream chunks. */ + private StreamOptions streamOptions; + @Override public JsonObject getInput() { JsonObject jsonObject = new JsonObject(); @@ -207,6 +224,14 @@ public Map getParameters() { if (repetitionPenalty != null) { params.put(REPETITION_PENALTY, repetitionPenalty); } + + if (presencePenalty != null) { + params.put(PRESENCE_PENALTY, presencePenalty); + } + + if (frequencyPenalty != null) { + params.put("frequency_penalty", frequencyPenalty); + } if (maxTokens != null) { params.put(MAX_TOKENS, maxTokens); } @@ -262,6 +287,10 @@ public Map getParameters() { params.put("translation_options", translationOptions); } + if (streamOptions != null) { + params.put("stream_options", streamOptions); + } + params.putAll(parameters); return params; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationUsage.java b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationUsage.java index 4a338b60..608793bb 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationUsage.java +++ b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationUsage.java @@ -1,6 +1,7 @@ // Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.generation; +import com.alibaba.dashscope.common.Plugins; import com.google.gson.annotations.SerializedName; import lombok.Data; import lombok.experimental.SuperBuilder; @@ -35,23 +36,6 @@ public static class CacheCreation { private CacheCreation cacheCreation; } - @Data - @SuperBuilder - public static class Plugins { - - @Data - public static class Search { - @SerializedName("count") - private Integer count; - - @SerializedName("strategy") - private String strategy; - } - - @SerializedName("search") - private Search search; - } - @SerializedName("input_tokens") private Integer inputTokens; diff --git a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGeneration.java b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGeneration.java index 4864ed8b..81e5847d 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGeneration.java +++ b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGeneration.java @@ -3,18 +3,27 @@ import com.alibaba.dashscope.api.AsynchronousApi; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Role; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.exception.UploadFileException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; import com.alibaba.dashscope.task.AsyncTaskListParam; -import com.alibaba.dashscope.tools.ToolCallBase; -import com.alibaba.dashscope.tools.ToolCallFunction; import com.alibaba.dashscope.utils.OSSUploadCertificate; import com.alibaba.dashscope.utils.ParamUtils; import com.alibaba.dashscope.utils.PreprocessMessageInput; +import com.alibaba.dashscope.utils.StreamingMerger; import com.alibaba.dashscope.utils.StringUtils; import io.reactivex.Flowable; import java.util.ArrayList; @@ -31,9 +40,6 @@ public final class ImageGeneration { private final ApiServiceOption serviceOption; private final String baseUrl; - private final ThreadLocal> accumulatedDataMap = - ThreadLocal.withInitial(HashMap::new); - public static class Models { public static final String WanX2_6_T2I = "wan2.6-t2i"; public static final String WanX2_6_IMAGE = "wan2.6-image"; @@ -50,6 +56,21 @@ private ApiServiceOption defaultSyncApiServiceOption() { .build(); } + /** Creates a copy of the shared serviceOption for thread-safe per-call usage. */ + private ApiServiceOption copyServiceOption() { + return ApiServiceOption.builder() + .protocol(serviceOption.getProtocol()) + .httpMethod(serviceOption.getHttpMethod()) + .streamingMode(serviceOption.getStreamingMode()) + .outputMode(serviceOption.getOutputMode()) + .taskGroup(serviceOption.getTaskGroup()) + .task(serviceOption.getTask()) + .function(serviceOption.getFunction()) + .baseHttpUrl(serviceOption.getBaseHttpUrl()) + .baseWebSocketUrl(serviceOption.getBaseWebSocketUrl()) + .build(); + } + public ImageGeneration() { serviceOption = defaultSyncApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); @@ -103,11 +124,12 @@ public ImageGeneration(String protocol, String baseUrl, ConnectionOptions connec */ public ImageGenerationResult call(ImageGenerationParam param) throws ApiException, NoApiKeyException, UploadFileException { - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); - serviceOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); + callOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); preprocessInput(param); - return ImageGenerationResult.fromDashScopeResult(syncApi.call(param)); + return ImageGenerationResult.fromDashScopeResult(syncApi.call(param, callOption)); } /** @@ -122,9 +144,10 @@ public ImageGenerationResult call(ImageGenerationParam param) */ public void call(ImageGenerationParam param, ResultCallback callback) throws ApiException, NoApiKeyException, UploadFileException { - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); - serviceOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); + callOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); preprocessInput(param); syncApi.call( param, @@ -143,7 +166,8 @@ public void onComplete() { public void onError(Exception e) { callback.onError(e); } - }); + }, + callOption); } /** @@ -157,9 +181,10 @@ public void onError(Exception e) { public ImageGenerationResult asyncCall(ImageGenerationParam param) throws ApiException, NoApiKeyException, UploadFileException { preprocessInput(param); - serviceOption.setTask(Task.IMAGE_GENERATION.getValue()); - serviceOption.setIsAsyncTask(true); - return ImageGenerationResult.fromDashScopeResult(asyncApi.asyncCall(param, serviceOption)); + ApiServiceOption callOption = copyServiceOption(); + callOption.setTask(Task.IMAGE_GENERATION.getValue()); + callOption.setIsAsyncTask(true); + return ImageGenerationResult.fromDashScopeResult(asyncApi.asyncCall(param, callOption)); } public ImageGenerationListResult list(AsyncTaskListParam param) @@ -234,26 +259,20 @@ public Flowable streamCall(ImageGenerationParam param) String userAgentSuffix = StringUtils.format("incremental_to_full/%d", flagValue); param.putHeader("user-agent", userAgentSuffix); - serviceOption.setIsSSE(true); - serviceOption.setStreamingMode(StreamingMode.OUT); - serviceOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(true); + callOption.setStreamingMode(StreamingMode.OUT); + callOption.setTask(Task.MULTIMODAL_GENERATION.getValue()); preprocessInput(param); - return syncApi - .streamCall(param) - .map(ImageGenerationResult::fromDashScopeResult) - .map(result -> mergeSingleResponse(result, toMergeResponse)) - .doOnComplete( - () -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }) - .doOnError( - throwable -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }); + return Flowable.defer( + () -> { + Map accumulatedData = new HashMap<>(); + return syncApi + .streamCall(param, callOption) + .map(ImageGenerationResult::fromDashScopeResult) + .map(result -> mergeSingleResponse(result, toMergeResponse, accumulatedData)) + .doFinally(accumulatedData::clear); + }); } /** @@ -278,36 +297,38 @@ public void streamCall(ImageGenerationParam param, ResultCallback accumulatedData = new HashMap<>(); syncApi.streamCall( param, new ResultCallback() { @Override public void onEvent(DashScopeResult msg) { ImageGenerationResult result = ImageGenerationResult.fromDashScopeResult(msg); - ImageGenerationResult mergedResult = mergeSingleResponse(result, toMergeResponse); - callback.onEvent(mergedResult); + ImageGenerationResult mergedResult = + mergeSingleResponse(result, toMergeResponse, accumulatedData); + if (mergedResult != null) { + callback.onEvent(mergedResult); + } } @Override public void onComplete() { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onComplete(); } @Override public void onError(Exception e) { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onError(e); } - }); + }, + callOption); } private void preprocessInput(ImageGenerationParam param) @@ -358,16 +379,17 @@ private boolean modifyIncrementalOutput(ImageGenerationParam param) { * * @param result The ImageGenerationResult to merge * @param toMergeResponse Whether to perform merging (based on original incrementalOutput setting) + * @param accumulatedData The per-stream accumulated data * @return The merged ImageGenerationResult */ private ImageGenerationResult mergeSingleResponse( - ImageGenerationResult result, boolean toMergeResponse) { + ImageGenerationResult result, + boolean toMergeResponse, + Map accumulatedData) { if (!toMergeResponse || result == null || result.getOutput() == null) { return result; } - Map accumulatedData = accumulatedDataMap.get(); - // Handle choices format: output.choices[].message.content if (result.getOutput().getChoices() != null) { List choices = result.getOutput().getChoices(); @@ -382,11 +404,11 @@ private ImageGenerationResult mergeSingleResponse( // Handle content accumulation (text content in content list) List> currentContent = choice.getMessage().getContent(); if (currentContent != null && !currentContent.isEmpty()) { - mergeTextContent(currentContent, accumulated); + StreamingMerger.mergeTextContent(currentContent, accumulated.content); } // Always set the accumulated content if we have any if (!accumulated.content.isEmpty()) { - choice.getMessage().setContent(accumulated.content); + choice.getMessage().setContent(StreamingMerger.copyContent(accumulated.content)); } } } @@ -395,147 +417,6 @@ private ImageGenerationResult mergeSingleResponse( return result; } - /** - * Merges text content from current response with accumulated content. For MultiModal, content is - * a List> where text content is in maps with "text" key. - */ - private void mergeTextContent( - List> currentContent, AccumulatedData accumulated) { - for (Map contentItem : currentContent) { - if (contentItem.containsKey("text")) { - String textValue = (String) contentItem.get("text"); - if (textValue != null && !textValue.isEmpty()) { - // Find or create text content item in accumulated content - Map accumulatedTextItem = null; - for (Map accItem : accumulated.content) { - if (accItem.containsKey("text")) { - accumulatedTextItem = accItem; - break; - } - } - - if (accumulatedTextItem == null) { - // Create new text content item - accumulatedTextItem = new HashMap<>(); - accumulatedTextItem.put("text", textValue); - accumulated.content.add(accumulatedTextItem); - } else { - // Append to existing text content - String existingText = (String) accumulatedTextItem.get("text"); - if (existingText == null) { - existingText = ""; - } - accumulatedTextItem.put("text", existingText + textValue); - } - } - } - } - } - - /** Merges tool calls from current response with accumulated tool calls. */ - private void mergeToolCalls( - List currentToolCalls, List accumulatedToolCalls) { - for (ToolCallBase currentCall : currentToolCalls) { - if (currentCall == null || currentCall.getIndex() == null) { - continue; - } - - int index = currentCall.getIndex(); - - // Find existing accumulated call with same index - ToolCallBase existingCall = null; - for (ToolCallBase accCall : accumulatedToolCalls) { - if (accCall != null && accCall.getIndex() != null && accCall.getIndex().equals(index)) { - existingCall = accCall; - break; - } - } - - if (existingCall instanceof ToolCallFunction && currentCall instanceof ToolCallFunction) { - // Merge function calls - ToolCallFunction existingFunctionCall = (ToolCallFunction) existingCall; - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - - if (currentFunctionCall.getFunction() != null) { - // Ensure existing function call has a function object - if (existingFunctionCall.getFunction() == null) { - existingFunctionCall.setFunction(existingFunctionCall.new CallFunction()); - } - - // Accumulate arguments if present - if (currentFunctionCall.getFunction().getArguments() != null) { - String existingArguments = existingFunctionCall.getFunction().getArguments(); - if (existingArguments == null) { - existingArguments = ""; - } - String currentArguments = currentFunctionCall.getFunction().getArguments(); - existingFunctionCall.getFunction().setArguments(existingArguments + currentArguments); - } - - // Accumulate function name if present - if (currentFunctionCall.getFunction().getName() != null) { - String existingName = existingFunctionCall.getFunction().getName(); - if (existingName == null) { - existingName = ""; - } - String currentName = currentFunctionCall.getFunction().getName(); - existingFunctionCall.getFunction().setName(existingName + currentName); - } - - // Update function output if present - if (currentFunctionCall.getFunction().getOutput() != null) { - existingFunctionCall - .getFunction() - .setOutput(currentFunctionCall.getFunction().getOutput()); - } - } - - // Update other fields with latest non-empty values - if (currentFunctionCall.getIndex() != null) { - existingFunctionCall.setIndex(currentFunctionCall.getIndex()); - } - if (currentFunctionCall.getId() != null && !currentFunctionCall.getId().isEmpty()) { - existingFunctionCall.setId(currentFunctionCall.getId()); - } - if (currentFunctionCall.getType() != null) { - existingFunctionCall.setType(currentFunctionCall.getType()); - } - } else { - // Add new tool call (create a copy) - if (currentCall instanceof ToolCallFunction) { - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - ToolCallFunction newFunctionCall = new ToolCallFunction(); - newFunctionCall.setIndex(currentFunctionCall.getIndex()); - newFunctionCall.setId(currentFunctionCall.getId()); - newFunctionCall.setType(currentFunctionCall.getType()); - - if (currentFunctionCall.getFunction() != null) { - ToolCallFunction.CallFunction newCallFunction = newFunctionCall.new CallFunction(); - newCallFunction.setName(currentFunctionCall.getFunction().getName()); - newCallFunction.setArguments(currentFunctionCall.getFunction().getArguments()); - newCallFunction.setOutput(currentFunctionCall.getFunction().getOutput()); - newFunctionCall.setFunction(newCallFunction); - } - - accumulatedToolCalls.add(newFunctionCall); - } else { - // For other types of tool calls, add directly (assuming they are immutable or don't need - // merging) - accumulatedToolCalls.add(currentCall); - } - } - } - } - - /** - * Clears accumulated data for the current thread. Should be called when streaming is complete or - * encounters error. - */ - private void clearAccumulatedData() { - accumulatedDataMap.get().clear(); - accumulatedDataMap.remove(); - } - /** Inner class to store accumulated data for response merging. */ private static class AccumulatedData { List> content = new ArrayList<>(); diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java index 93b80594..e7d3fa80 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java @@ -2,17 +2,28 @@ package com.alibaba.dashscope.aigc.multimodalconversation; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.MultiModalMessage; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Role; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.exception.UploadFileException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; import com.alibaba.dashscope.tools.ToolCallBase; -import com.alibaba.dashscope.tools.ToolCallFunction; import com.alibaba.dashscope.utils.OSSUploadCertificate; import com.alibaba.dashscope.utils.ParamUtils; import com.alibaba.dashscope.utils.PreprocessMessageInput; +import com.alibaba.dashscope.utils.StreamingMerger; import com.alibaba.dashscope.utils.StringUtils; import io.reactivex.Flowable; import java.util.ArrayList; @@ -27,9 +38,6 @@ public final class MultiModalConversation { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; - private final ThreadLocal> accumulatedDataMap = - ThreadLocal.withInitial(HashMap::new); - public static class Models { public static final String QWEN_VL_CHAT_V1 = "qwen-vl-chat-v1"; public static final String QWEN_VL_PLUS = "qwen-vl-plus"; @@ -47,6 +55,21 @@ private ApiServiceOption defaultApiServiceOption() { .build(); } + /** Creates a copy of the shared serviceOption for thread-safe per-call usage. */ + private ApiServiceOption copyServiceOption() { + return ApiServiceOption.builder() + .protocol(serviceOption.getProtocol()) + .httpMethod(serviceOption.getHttpMethod()) + .streamingMode(serviceOption.getStreamingMode()) + .outputMode(serviceOption.getOutputMode()) + .taskGroup(serviceOption.getTaskGroup()) + .task(serviceOption.getTask()) + .function(serviceOption.getFunction()) + .baseHttpUrl(serviceOption.getBaseHttpUrl()) + .baseWebSocketUrl(serviceOption.getBaseWebSocketUrl()) + .build(); + } + public MultiModalConversation() { serviceOption = defaultApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); @@ -90,10 +113,11 @@ public MultiModalConversation( */ public MultiModalConversationResult call(MultiModalConversationParam param) throws ApiException, NoApiKeyException, UploadFileException { - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); preprocessInput(param); - return MultiModalConversationResult.fromDashScopeResult(syncApi.call(param)); + return MultiModalConversationResult.fromDashScopeResult(syncApi.call(param, callOption)); } /** @@ -109,8 +133,9 @@ public MultiModalConversationResult call(MultiModalConversationParam param) public void call( MultiModalConversationParam param, ResultCallback callback) throws ApiException, NoApiKeyException, UploadFileException { - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); preprocessInput(param); syncApi.call( param, @@ -129,7 +154,8 @@ public void onComplete() { public void onError(Exception e) { callback.onError(e); } - }); + }, + callOption); } /** @@ -151,25 +177,19 @@ public Flowable streamCall(MultiModalConversationP String userAgentSuffix = StringUtils.format("incremental_to_full/%d", flagValue); param.putHeader("user-agent", userAgentSuffix); - serviceOption.setIsSSE(true); - serviceOption.setStreamingMode(StreamingMode.OUT); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(true); + callOption.setStreamingMode(StreamingMode.OUT); preprocessInput(param); - return syncApi - .streamCall(param) - .map(MultiModalConversationResult::fromDashScopeResult) - .map(result -> mergeSingleResponse(result, toMergeResponse)) - .doOnComplete( - () -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }) - .doOnError( - throwable -> { - if (toMergeResponse) { - clearAccumulatedData(); - } - }); + return Flowable.defer( + () -> { + Map accumulatedData = new HashMap<>(); + return syncApi + .streamCall(param, callOption) + .map(MultiModalConversationResult::fromDashScopeResult) + .map(result -> mergeSingleResponse(result, toMergeResponse, accumulatedData)) + .doFinally(accumulatedData::clear); + }); } /** @@ -195,9 +215,11 @@ public void streamCall( String userAgentSuffix = StringUtils.format("incremental_to_full/%d", flagValue); param.putHeader("user-agent", userAgentSuffix); - serviceOption.setIsSSE(true); - serviceOption.setStreamingMode(StreamingMode.OUT); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(true); + callOption.setStreamingMode(StreamingMode.OUT); preprocessInput(param); + Map accumulatedData = new HashMap<>(); syncApi.streamCall( param, new ResultCallback() { @@ -206,26 +228,25 @@ public void onEvent(DashScopeResult msg) { MultiModalConversationResult result = MultiModalConversationResult.fromDashScopeResult(msg); MultiModalConversationResult mergedResult = - mergeSingleResponse(result, toMergeResponse); - callback.onEvent(mergedResult); + mergeSingleResponse(result, toMergeResponse, accumulatedData); + if (mergedResult != null) { + callback.onEvent(mergedResult); + } } @Override public void onComplete() { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onComplete(); } @Override public void onError(Exception e) { - if (toMergeResponse) { - clearAccumulatedData(); - } + accumulatedData.clear(); callback.onError(e); } - }); + }, + callOption); } private void preprocessInput(MultiModalConversationParam param) @@ -291,16 +312,17 @@ private boolean modifyIncrementalOutput(MultiModalConversationParam param) { * * @param result The MultiModalConversationResult to merge * @param toMergeResponse Whether to perform merging (based on original incrementalOutput setting) + * @param accumulatedData The per-stream accumulated data * @return The merged MultiModalConversationResult */ private MultiModalConversationResult mergeSingleResponse( - MultiModalConversationResult result, boolean toMergeResponse) { + MultiModalConversationResult result, + boolean toMergeResponse, + Map accumulatedData) { if (!toMergeResponse || result == null || result.getOutput() == null) { return result; } - Map accumulatedData = accumulatedDataMap.get(); - // Handle choices format: output.choices[].message.content if (result.getOutput().getChoices() != null) { List choices = result.getOutput().getChoices(); @@ -315,11 +337,11 @@ private MultiModalConversationResult mergeSingleResponse( // Handle content accumulation (text content in content list) List> currentContent = choice.getMessage().getContent(); if (currentContent != null && !currentContent.isEmpty()) { - mergeTextContent(currentContent, accumulated); + StreamingMerger.mergeTextContent(currentContent, accumulated.content); } // Always set the accumulated content if we have any if (!accumulated.content.isEmpty()) { - choice.getMessage().setContent(accumulated.content); + choice.getMessage().setContent(StreamingMerger.copyContent(accumulated.content)); } // Handle reasoning_content accumulation @@ -332,10 +354,10 @@ private MultiModalConversationResult mergeSingleResponse( choice.getMessage().setReasoningContent(accumulated.reasoningContent.toString()); } - // Handle tool_calls accumulation + // Handle tool_calls accumulation (delegate to shared utility) List currentToolCalls = choice.getMessage().getToolCalls(); if (currentToolCalls != null && !currentToolCalls.isEmpty()) { - mergeToolCalls(currentToolCalls, accumulated.toolCalls); + StreamingMerger.mergeToolCalls(currentToolCalls, accumulated.toolCalls); } // Always set accumulated tool_calls if we have any if (!accumulated.toolCalls.isEmpty()) { @@ -348,147 +370,6 @@ private MultiModalConversationResult mergeSingleResponse( return result; } - /** - * Merges text content from current response with accumulated content. For MultiModal, content is - * a List> where text content is in maps with "text" key. - */ - private void mergeTextContent( - List> currentContent, AccumulatedData accumulated) { - for (Map contentItem : currentContent) { - if (contentItem.containsKey("text")) { - String textValue = (String) contentItem.get("text"); - if (textValue != null && !textValue.isEmpty()) { - // Find or create text content item in accumulated content - Map accumulatedTextItem = null; - for (Map accItem : accumulated.content) { - if (accItem.containsKey("text")) { - accumulatedTextItem = accItem; - break; - } - } - - if (accumulatedTextItem == null) { - // Create new text content item - accumulatedTextItem = new HashMap<>(); - accumulatedTextItem.put("text", textValue); - accumulated.content.add(accumulatedTextItem); - } else { - // Append to existing text content - String existingText = (String) accumulatedTextItem.get("text"); - if (existingText == null) { - existingText = ""; - } - accumulatedTextItem.put("text", existingText + textValue); - } - } - } - } - } - - /** Merges tool calls from current response with accumulated tool calls. */ - private void mergeToolCalls( - List currentToolCalls, List accumulatedToolCalls) { - for (ToolCallBase currentCall : currentToolCalls) { - if (currentCall == null || currentCall.getIndex() == null) { - continue; - } - - int index = currentCall.getIndex(); - - // Find existing accumulated call with same index - ToolCallBase existingCall = null; - for (ToolCallBase accCall : accumulatedToolCalls) { - if (accCall != null && accCall.getIndex() != null && accCall.getIndex().equals(index)) { - existingCall = accCall; - break; - } - } - - if (existingCall instanceof ToolCallFunction && currentCall instanceof ToolCallFunction) { - // Merge function calls - ToolCallFunction existingFunctionCall = (ToolCallFunction) existingCall; - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - - if (currentFunctionCall.getFunction() != null) { - // Ensure existing function call has a function object - if (existingFunctionCall.getFunction() == null) { - existingFunctionCall.setFunction(existingFunctionCall.new CallFunction()); - } - - // Accumulate arguments if present - if (currentFunctionCall.getFunction().getArguments() != null) { - String existingArguments = existingFunctionCall.getFunction().getArguments(); - if (existingArguments == null) { - existingArguments = ""; - } - String currentArguments = currentFunctionCall.getFunction().getArguments(); - existingFunctionCall.getFunction().setArguments(existingArguments + currentArguments); - } - - // Accumulate function name if present - if (currentFunctionCall.getFunction().getName() != null) { - String existingName = existingFunctionCall.getFunction().getName(); - if (existingName == null) { - existingName = ""; - } - String currentName = currentFunctionCall.getFunction().getName(); - existingFunctionCall.getFunction().setName(existingName + currentName); - } - - // Update function output if present - if (currentFunctionCall.getFunction().getOutput() != null) { - existingFunctionCall - .getFunction() - .setOutput(currentFunctionCall.getFunction().getOutput()); - } - } - - // Update other fields with latest non-empty values - if (currentFunctionCall.getIndex() != null) { - existingFunctionCall.setIndex(currentFunctionCall.getIndex()); - } - if (currentFunctionCall.getId() != null && !currentFunctionCall.getId().isEmpty()) { - existingFunctionCall.setId(currentFunctionCall.getId()); - } - if (currentFunctionCall.getType() != null) { - existingFunctionCall.setType(currentFunctionCall.getType()); - } - } else { - // Add new tool call (create a copy) - if (currentCall instanceof ToolCallFunction) { - ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; - ToolCallFunction newFunctionCall = new ToolCallFunction(); - newFunctionCall.setIndex(currentFunctionCall.getIndex()); - newFunctionCall.setId(currentFunctionCall.getId()); - newFunctionCall.setType(currentFunctionCall.getType()); - - if (currentFunctionCall.getFunction() != null) { - ToolCallFunction.CallFunction newCallFunction = newFunctionCall.new CallFunction(); - newCallFunction.setName(currentFunctionCall.getFunction().getName()); - newCallFunction.setArguments(currentFunctionCall.getFunction().getArguments()); - newCallFunction.setOutput(currentFunctionCall.getFunction().getOutput()); - newFunctionCall.setFunction(newCallFunction); - } - - accumulatedToolCalls.add(newFunctionCall); - } else { - // For other types of tool calls, add directly (assuming they are immutable or don't need - // merging) - accumulatedToolCalls.add(currentCall); - } - } - } - } - - /** - * Clears accumulated data for the current thread. Should be called when streaming is complete or - * encounters error. - */ - private void clearAccumulatedData() { - accumulatedDataMap.get().clear(); - accumulatedDataMap.remove(); - } - /** Inner class to store accumulated data for response merging. */ private static class AccumulatedData { List> content = new ArrayList<>(); diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationOutput.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationOutput.java index f2a00c7f..c8c420cb 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationOutput.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationOutput.java @@ -14,7 +14,13 @@ public static class Choice { @SerializedName("finish_reason") private String finishReason; + private Integer index; + private MultiModalMessage message; + + /** Search information for this specific choice, returned when search_options is configured. */ + @SerializedName("search_info") + private SearchInfo searchInfo; } private List choices; diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationParam.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationParam.java index 202bfa53..304859c1 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationParam.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationParam.java @@ -3,6 +3,7 @@ import com.alibaba.dashscope.base.HalfDuplexServiceParam; import com.alibaba.dashscope.common.ResponseFormat; import com.alibaba.dashscope.common.SearchOptions; +import com.alibaba.dashscope.common.StreamOptions; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.tools.ToolBase; import com.alibaba.dashscope.utils.ApiKeywords; @@ -55,6 +56,12 @@ public class MultiModalConversationParam extends HalfDuplexServiceParam { */ private Float presencePenalty; + /** + * Frequency penalty: reduces the model's likelihood to repeat tokens based on their frequency in + * the output so far. A value of 0 indicates no penalty. Default value: 0.0 + */ + private Float frequencyPenalty; + /* Whether to enable web search(quark). Currently works best only on the first round of conversation. Default to False */ @@ -162,6 +169,9 @@ public class MultiModalConversationParam extends HalfDuplexServiceParam { /** thinking budget */ private Integer thinkingBudget; + /** Streaming options controlling what extra info is included in stream chunks. */ + private StreamOptions streamOptions; + @Override public JsonObject getHttpBody() { JsonObject requestObject = new JsonObject(); @@ -239,6 +249,10 @@ public Map getParameters() { params.put(ApiKeywords.PRESENCE_PENALTY, presencePenalty); } + if (frequencyPenalty != null) { + params.put("frequency_penalty", frequencyPenalty); + } + // Apply different logic based on model version if (ParamUtils.isQwenVersionThreeOrHigher(getModel())) { if (incrementalOutput != null) { @@ -318,6 +332,10 @@ public Map getParameters() { params.put("thinking_budget", thinkingBudget); } + if (streamOptions != null) { + params.put("stream_options", streamOptions); + } + params.putAll(parameters); return params; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationUsage.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationUsage.java index 2ec3934e..99df869b 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationUsage.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationUsage.java @@ -1,27 +1,13 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.aigc.multimodalconversation; +import com.alibaba.dashscope.common.Plugins; import com.google.gson.annotations.SerializedName; import lombok.Data; @Data public class MultiModalConversationUsage { - @Data - public static class Plugins { - - @Data - public static class Search { - @SerializedName("count") - private Integer count; - - @SerializedName("strategy") - private String strategy; - } - - @SerializedName("search") - private Search search; - } - @SerializedName("input_tokens") private Integer inputTokens; diff --git a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesis.java b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesis.java index 8b3aca28..6269cf38 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesis.java +++ b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesis.java @@ -80,23 +80,7 @@ public static class AudioSetting { /** * Create ApiServiceOption * - * @return ApiServiceOption - */ - private ApiServiceOption getApiServiceOption() { - return ApiServiceOption.builder() - .protocol(Protocol.HTTP) - .httpMethod(HttpMethod.POST) - .streamingMode(StreamingMode.NONE) - .taskGroup(taskGroup) - .task(task) - .function(function) - .isAsyncTask(true) - .build(); - } - - /** - * Create ApiServiceOption - * + * @param task The task name for the API service. * @return ApiServiceOption */ private ApiServiceOption getApiServiceOption(String task) { @@ -115,7 +99,7 @@ private ApiServiceOption getApiServiceOption(String task) { public VideoSynthesis() { // only support http asyncApi = new AsynchronousApi<>(); - createServiceOptions = getApiServiceOption(); + createServiceOptions = getApiServiceOption(task); this.baseUrl = null; } @@ -127,7 +111,7 @@ public VideoSynthesis() { public VideoSynthesis(String baseUrl) { // only support http asyncApi = new AsynchronousApi<>(); - createServiceOptions = getApiServiceOption(); + createServiceOptions = getApiServiceOption(task); this.baseUrl = baseUrl; } @@ -138,6 +122,7 @@ public VideoSynthesis(String baseUrl) { */ public VideoSynthesis(String baseUrl, String task) { // only support http + this.task = task; asyncApi = new AsynchronousApi<>(); createServiceOptions = getApiServiceOption(task); this.baseUrl = baseUrl; diff --git a/src/main/java/com/alibaba/dashscope/api/SynchronizeHalfDuplexApi.java b/src/main/java/com/alibaba/dashscope/api/SynchronizeHalfDuplexApi.java index d321caf3..0de0845a 100644 --- a/src/main/java/com/alibaba/dashscope/api/SynchronizeHalfDuplexApi.java +++ b/src/main/java/com/alibaba/dashscope/api/SynchronizeHalfDuplexApi.java @@ -58,6 +58,21 @@ public DashScopeResult call(ParamT param) throws ApiException, NoApiKeyException return client.send(req); } + /** + * Call the server to get the whole result with custom service option. + * + * @param param The input param, should be the subclass of `ConversationParam`. + * @param serviceOption The custom service option for this call. + * @return The output structure, should be the subclass of `ConversationResult`. + * @throws NoApiKeyException Can not find api key + * @throws ApiException The request failed, possibly due to a network or data error. + */ + public DashScopeResult call(ParamT param, ServiceOption serviceOption) + throws ApiException, NoApiKeyException { + HalfDuplexRequest req = new HalfDuplexRequest(param, serviceOption); + return client.send(req); + } + /** * Call the server to get the result in the callback function. * @@ -72,6 +87,22 @@ public void call(ParamT param, ResultCallback callback) client.send(req, callback); } + /** + * Call the server to get the result in the callback function with custom service option. + * + * @param param The input param, should be the subclass of `Param`. + * @param callback The callback to receive response, should be the subclass of `Result`. + * @param serviceOption The custom service option for this call. + * @throws NoApiKeyException Can not find api key + * @throws ApiException The request failed, possibly due to a network or data error. + */ + public void call( + ParamT param, ResultCallback callback, ServiceOption serviceOption) + throws ApiException, NoApiKeyException { + HalfDuplexRequest req = new HalfDuplexRequest(param, serviceOption); + client.send(req, callback); + } + /** * Call the server to get the result by stream. * @@ -85,6 +116,21 @@ public Flowable streamCall(ParamT param) throws ApiException, N return client.streamOut(req); } + /** + * Call the server to get the result by stream with custom service option. + * + * @param param The input param, should be the subclass of `Param`. + * @param serviceOption The custom service option for this call. + * @return A `Flowable` of the output structure, which is the subclass of `Result`. + * @throws NoApiKeyException Can not find api key + * @throws ApiException The request failed, possibly due to a network or data error. + */ + public Flowable streamCall(ParamT param, ServiceOption serviceOption) + throws ApiException, NoApiKeyException { + HalfDuplexRequest req = new HalfDuplexRequest(param, serviceOption); + return client.streamOut(req); + } + /** * Call the server to get the result by stream. * @@ -99,6 +145,22 @@ public void streamCall(ParamT param, ResultCallback callback) client.streamOut(req, callback); } + /** + * Call the server to get the result by stream with custom service option. + * + * @param param The input param, should be the subclass of `Param`. + * @param callback The result callback. + * @param serviceOption The custom service option for this call. + * @throws NoApiKeyException Can not find api key + * @throws ApiException The request failed, possibly due to a network or data error. + */ + public void streamCall( + ParamT param, ResultCallback callback, ServiceOption serviceOption) + throws ApiException, NoApiKeyException { + HalfDuplexRequest req = new HalfDuplexRequest(param, serviceOption); + client.streamOut(req, callback); + } + public boolean close(int code, String reason) { if (client != null) { return client.close(code, reason); diff --git a/src/main/java/com/alibaba/dashscope/app/Application.java b/src/main/java/com/alibaba/dashscope/app/Application.java index b869814b..f788fe9b 100644 --- a/src/main/java/com/alibaba/dashscope/app/Application.java +++ b/src/main/java/com/alibaba/dashscope/app/Application.java @@ -12,13 +12,11 @@ import io.reactivex.Flowable; /** - * Title Ap completion calls.
- * Description App completion calls.
- * Created at 2024-02-23 15:38 + * Application completion calls. * * @since jdk8 */ -public class Application { +public final class Application { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; @@ -42,7 +40,7 @@ public Application(String baseUrl) { } /** - * app completion call for http request + * App completion call for http request * * @param param app completion params * @return app completion result @@ -61,7 +59,7 @@ public ApplicationResult call(ApplicationParam param) } /** - * app completion call for http request by sse stream + * App completion call for http request by sse stream * * @param param app completion params * @return flowable stream of app completion result @@ -88,8 +86,5 @@ private void setRequestOption(ApiServiceOption serviceOption, String resourceId) serviceOption.setTaskGroup("apps"); serviceOption.setTask(resourceId); serviceOption.setFunction("completion"); - // serviceOption.setResource("apps"); - // serviceOption.setResourceId(resourceId); - // serviceOption.setAction("completion"); } } diff --git a/src/main/java/com/alibaba/dashscope/common/Plugins.java b/src/main/java/com/alibaba/dashscope/common/Plugins.java new file mode 100644 index 00000000..4f3ebe6c --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/common/Plugins.java @@ -0,0 +1,23 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.common; + +import com.google.gson.annotations.SerializedName; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class Plugins { + + @Data + public static class Search { + @SerializedName("count") + private Integer count; + + @SerializedName("strategy") + private String strategy; + } + + @SerializedName("search") + private Search search; +} diff --git a/src/main/java/com/alibaba/dashscope/common/ResponseFormat.java b/src/main/java/com/alibaba/dashscope/common/ResponseFormat.java index f7820bd2..26e212b4 100644 --- a/src/main/java/com/alibaba/dashscope/common/ResponseFormat.java +++ b/src/main/java/com/alibaba/dashscope/common/ResponseFormat.java @@ -2,6 +2,7 @@ package com.alibaba.dashscope.common; import com.google.gson.annotations.SerializedName; +import lombok.Builder; import lombok.Data; import lombok.experimental.SuperBuilder; @@ -12,10 +13,41 @@ public class ResponseFormat { public static final String JSON_OBJECT = "json_object"; + public static final String JSON_SCHEMA = "json_schema"; + @SerializedName("type") private Object type; + /** + * JSON schema definition when type is JSON_SCHEMA. Used to enforce structured output from the + * model. + */ + @SerializedName("json_schema") + private JsonSchema jsonSchema; + + @SuperBuilder + @Data + public static class JsonSchema { + /** A friendly name for the JSON schema. */ + private String name; + + /** The JSON schema definition as a string or object. */ + private Object schema; + + /** Whether to strictly validate the output against the schema. Default: true. */ + @Builder.Default + @SerializedName("strict") + private Boolean strict = true; + } + public static ResponseFormat from(Object type) { return ResponseFormat.builder().type(type).build(); } + + public static ResponseFormat fromJsonSchema(String name, Object schema) { + return ResponseFormat.builder() + .type(JSON_SCHEMA) + .jsonSchema(JsonSchema.builder().name(name).schema(schema).build()) + .build(); + } } diff --git a/src/main/java/com/alibaba/dashscope/common/StreamOptions.java b/src/main/java/com/alibaba/dashscope/common/StreamOptions.java new file mode 100644 index 00000000..b3910fe2 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/common/StreamOptions.java @@ -0,0 +1,13 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.common; + +import com.google.gson.annotations.SerializedName; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class StreamOptions { + @SerializedName("include_usage") + private Boolean includeUsage; +} diff --git a/src/main/java/com/alibaba/dashscope/embeddings/BatchTextEmbedding.java b/src/main/java/com/alibaba/dashscope/embeddings/BatchTextEmbedding.java index 76e7c5ed..542d06cf 100644 --- a/src/main/java/com/alibaba/dashscope/embeddings/BatchTextEmbedding.java +++ b/src/main/java/com/alibaba/dashscope/embeddings/BatchTextEmbedding.java @@ -16,12 +16,12 @@ import com.alibaba.dashscope.task.AsyncTaskListParam; import com.alibaba.dashscope.task.AsyncTaskListResult; -public class BatchTextEmbedding { +public final class BatchTextEmbedding { private final AsynchronousApi asyncApi; private final ApiServiceOption serviceOption; private final String baseUrl; - public final class Models { + public static class Models { public static final String TEXT_EMBEDDING_ASYNC_V1 = "text-embedding-async-v1"; public static final String TEXT_EMBEDDING_ASYNC_V2 = "text-embedding-async-v2"; } diff --git a/src/main/java/com/alibaba/dashscope/embeddings/MultiModalEmbedding.java b/src/main/java/com/alibaba/dashscope/embeddings/MultiModalEmbedding.java index 3a535451..61ca2baf 100644 --- a/src/main/java/com/alibaba/dashscope/embeddings/MultiModalEmbedding.java +++ b/src/main/java/com/alibaba/dashscope/embeddings/MultiModalEmbedding.java @@ -1,5 +1,4 @@ // Copyright (c) Alibaba, Inc. and its affiliates. - package com.alibaba.dashscope.embeddings; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; @@ -18,7 +17,7 @@ import com.alibaba.dashscope.protocol.StreamingMode; import com.alibaba.dashscope.utils.PreprocessMessageInput; -public class MultiModalEmbedding { +public final class MultiModalEmbedding { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; @@ -28,7 +27,7 @@ public static class Models { public static final String MULTIMODAL_EMBEDDING_V1 = "multimodal-embedding-v1"; } - private ApiServiceOption defaulApiServiceOption() { + private ApiServiceOption defaultApiServiceOption() { return ApiServiceOption.builder() .protocol(Protocol.HTTP) .httpMethod(HttpMethod.POST) @@ -41,12 +40,12 @@ private ApiServiceOption defaulApiServiceOption() { } public MultiModalEmbedding() { - serviceOption = defaulApiServiceOption(); + serviceOption = defaultApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } public MultiModalEmbedding(String baseUrl) { - serviceOption = defaulApiServiceOption(); + serviceOption = defaultApiServiceOption(); serviceOption.setBaseHttpUrl(baseUrl); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } @@ -54,8 +53,9 @@ public MultiModalEmbedding(String baseUrl) { /** * Call the server to get the result in the callback function. * - * @param param The input param of class `GenerationParam`. - * @param callback The callback to receive response, the template class is `GenerationResult`. + * @param param The input param of class `MultiModalEmbeddingParam`. + * @param callback The callback to receive response, the template class is + * `MultiModalEmbeddingResult`. * @throws NoApiKeyException Can not find api key * @throws ApiException The request failed, possibly due to a network or data error. * @throws UploadFileException File upload failed. @@ -87,8 +87,8 @@ public void onError(Exception e) { /** * Call the server to get the whole result, only http protocol * - * @param param The input param of class `ConversationParam`. - * @return The output structure of `QWenConversationResult`. + * @param param The input param of class `MultiModalEmbeddingParam`. + * @return The output structure of `MultiModalEmbeddingResult`. * @throws NoApiKeyException Can not find api key * @throws ApiException The request failed, possibly due to a network or data error. * @throws UploadFileException File upload failed. diff --git a/src/main/java/com/alibaba/dashscope/embeddings/TextEmbedding.java b/src/main/java/com/alibaba/dashscope/embeddings/TextEmbedding.java index d5fd6767..09273071 100644 --- a/src/main/java/com/alibaba/dashscope/embeddings/TextEmbedding.java +++ b/src/main/java/com/alibaba/dashscope/embeddings/TextEmbedding.java @@ -1,5 +1,4 @@ // Copyright (c) Alibaba, Inc. and its affiliates. - package com.alibaba.dashscope.embeddings; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; @@ -16,18 +15,18 @@ import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; -public class TextEmbedding { +public final class TextEmbedding { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; - public final class Models { + public static class Models { public static final String TEXT_EMBEDDING_V1 = "text-embedding-v1"; public static final String TEXT_EMBEDDING_V2 = "text-embedding-v2"; public static final String TEXT_EMBEDDING_V3 = "text-embedding-v3"; public static final String TEXT_EMBEDDING_V4 = "text-embedding-v4"; } - private ApiServiceOption defaulApiServiceOption() { + private ApiServiceOption defaultApiServiceOption() { return ApiServiceOption.builder() .protocol(Protocol.HTTP) .httpMethod(HttpMethod.POST) @@ -40,12 +39,12 @@ private ApiServiceOption defaulApiServiceOption() { } public TextEmbedding() { - serviceOption = defaulApiServiceOption(); + serviceOption = defaultApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } public TextEmbedding(String baseUrl) { - serviceOption = defaulApiServiceOption(); + serviceOption = defaultApiServiceOption(); serviceOption.setBaseHttpUrl(baseUrl); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } @@ -53,8 +52,8 @@ public TextEmbedding(String baseUrl) { /** * Call the server to get the result in the callback function. * - * @param param The input param of class `GenerationParam`. - * @param callback The callback to receive response, the template class is `GenerationResult`. + * @param param The input param of class `TextEmbeddingParam`. + * @param callback The callback to receive response, the template class is `TextEmbeddingResult`. * @throws NoApiKeyException Can not find api key * @throws ApiException The request failed, possibly due to a network or data error. */ @@ -83,8 +82,8 @@ public void onError(Exception e) { /** * Call the server to get the whole result, only http protocol * - * @param param The input param of class `ConversationParam`. - * @return The output structure of `QWenConversationResult`. + * @param param The input param of class `TextEmbeddingParam`. + * @return The output structure of `TextEmbeddingResult`. * @throws NoApiKeyException Can not find api key * @throws ApiException The request failed, possibly due to a network or data error. */ diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java index 64c463b3..5f6ae37a 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java @@ -1,8 +1,14 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal; import com.alibaba.dashscope.Version; import com.alibaba.dashscope.api.SynchronizeFullDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogApiKeyWords.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogApiKeyWords.java index 8b1cadc0..61280bfa 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogApiKeyWords.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogApiKeyWords.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal; /** author songsong.shao date 2025/4/25 */ diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogCallback.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogCallback.java index 269e756d..3eed5d9a 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogCallback.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialogCallback.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal; import com.google.gson.JsonObject; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java index fa75ebf0..98643378 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal; /** author songsong.shao date 2025/4/24 */ diff --git a/src/main/java/com/alibaba/dashscope/multimodal/State.java b/src/main/java/com/alibaba/dashscope/multimodal/State.java index b701cfca..f56d821d 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/State.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/State.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal; import lombok.Getter; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWu.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWu.java index fc865c05..27cb762c 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWu.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWu.java @@ -1,12 +1,17 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal.tingwu; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; import com.alibaba.dashscope.base.HalfDuplexServiceParam; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.ResultCallback; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; /** The tingwu client. */ public final class TingWu { @@ -24,6 +29,22 @@ private ApiServiceOption defaultApiServiceOption() { .build(); } + /** Creates a copy of the shared serviceOption for thread-safe per-call usage. */ + private ApiServiceOption copyServiceOption() { + return ApiServiceOption.builder() + .protocol(serviceOption.getProtocol()) + .httpMethod(serviceOption.getHttpMethod()) + .streamingMode(serviceOption.getStreamingMode()) + .outputMode(serviceOption.getOutputMode()) + .taskGroup(serviceOption.getTaskGroup()) + .task(serviceOption.getTask()) + .function(serviceOption.getFunction()) + .isService(false) + .baseHttpUrl(serviceOption.getBaseHttpUrl()) + .baseWebSocketUrl(serviceOption.getBaseWebSocketUrl()) + .build(); + } + public TingWu() { serviceOption = defaultApiServiceOption(); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); @@ -31,13 +52,14 @@ public TingWu() { public TingWu(String protocol) { serviceOption = defaultApiServiceOption(); + serviceOption.setProtocol(Protocol.of(protocol)); syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } public TingWu(String protocol, String baseUrl) { serviceOption = defaultApiServiceOption(); serviceOption.setProtocol(Protocol.of(protocol)); - if (protocol.equals(Protocol.HTTP.getValue())) { + if (Protocol.HTTP.getValue().equals(protocol)) { serviceOption.setBaseHttpUrl(baseUrl); } else { serviceOption.setBaseWebSocketUrl(baseUrl); @@ -48,7 +70,7 @@ public TingWu(String protocol, String baseUrl) { public TingWu(String protocol, String baseUrl, ConnectionOptions connectionOptions) { serviceOption = defaultApiServiceOption(); serviceOption.setProtocol(Protocol.of(protocol)); - if (protocol.equals(Protocol.HTTP.getValue())) { + if (Protocol.HTTP.getValue().equals(protocol)) { serviceOption.setBaseHttpUrl(baseUrl); } else { serviceOption.setBaseWebSocketUrl(baseUrl); @@ -60,7 +82,26 @@ public TingWu(String protocol, String baseUrl, ConnectionOptions connectionOptio public DashScopeResult call(HalfDuplexServiceParam param) throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); - serviceOption.setIsSSE(false); - return syncApi.call(param); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + return syncApi.call(param, callOption); + } + + /** + * Call the server to get the result in the callback function. + * + * @param param The input param. + * @param callback The callback to receive response. + * @throws NoApiKeyException Can not find api key. + * @throws ApiException The request failed, possibly due to a network or data error. + * @throws InputRequiredException Missing inputs. + */ + public void call(HalfDuplexServiceParam param, ResultCallback callback) + throws ApiException, NoApiKeyException, InputRequiredException { + java.util.Objects.requireNonNull(callback, "callback must not be null"); + param.validate(); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + syncApi.call(param, callback, callOption); } } diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuParam.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuParam.java index 546c898d..f5472ad1 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuParam.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuParam.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal.tingwu; import com.alibaba.dashscope.base.HalfDuplexServiceParam; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java index 552edc24..1fb79461 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java @@ -3,7 +3,12 @@ package com.alibaba.dashscope.multimodal.tingwu; import com.alibaba.dashscope.api.SynchronizeFullDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeCallback.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeCallback.java index a7344b0e..44174e4b 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeCallback.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeCallback.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal.tingwu; import com.google.gson.JsonObject; diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeParam.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeParam.java index ce8791f3..4c8606b4 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeParam.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtimeParam.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.multimodal.tingwu; import static com.alibaba.dashscope.multimodal.MultiModalDialogApiKeyWords.CONST_NAME_DIRECTIVE; diff --git a/src/main/java/com/alibaba/dashscope/nlp/understanding/Understanding.java b/src/main/java/com/alibaba/dashscope/nlp/understanding/Understanding.java index a4b5123b..5cd4720b 100644 --- a/src/main/java/com/alibaba/dashscope/nlp/understanding/Understanding.java +++ b/src/main/java/com/alibaba/dashscope/nlp/understanding/Understanding.java @@ -1,15 +1,25 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.nlp.understanding; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.Function; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.Task; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; import lombok.extern.slf4j.Slf4j; @Slf4j -public class Understanding { +public final class Understanding { private final SynchronizeHalfDuplexApi syncApi; private final ApiServiceOption serviceOption; @@ -21,13 +31,12 @@ private ApiServiceOption defaultApiServiceOption() { return ApiServiceOption.builder() .protocol(Protocol.HTTP) .httpMethod(HttpMethod.POST) - .streamingMode(StreamingMode.OUT) + .streamingMode(StreamingMode.NONE) .outputMode(OutputMode.ACCUMULATE) .taskGroup(TaskGroup.NLP.getValue()) .task(Task.NLU.getValue()) .function(Function.UNDERSTANDING.getValue()) .isSSE(false) - .streamingMode(StreamingMode.NONE) .build(); } @@ -53,6 +62,17 @@ public Understanding(String protocol, String baseUrl) { syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); } + public Understanding(String protocol, String baseUrl, ConnectionOptions connectionOptions) { + serviceOption = defaultApiServiceOption(); + serviceOption.setProtocol(Protocol.of(protocol)); + if (Protocol.HTTP.getValue().equals(protocol)) { + serviceOption.setBaseHttpUrl(baseUrl); + } else { + serviceOption.setBaseWebSocketUrl(baseUrl); + } + syncApi = new SynchronizeHalfDuplexApi<>(connectionOptions, serviceOption); + } + /** * Call the server to get the whole result, only http protocol * diff --git a/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingParam.java b/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingParam.java index 21df75b5..5c72bfcc 100644 --- a/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingParam.java +++ b/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingParam.java @@ -1,3 +1,4 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. package com.alibaba.dashscope.nlp.understanding; import com.alibaba.dashscope.base.HalfDuplexServiceParam; diff --git a/src/main/java/com/alibaba/dashscope/rerank/TextReRank.java b/src/main/java/com/alibaba/dashscope/rerank/TextReRank.java index 5e56a470..25771283 100644 --- a/src/main/java/com/alibaba/dashscope/rerank/TextReRank.java +++ b/src/main/java/com/alibaba/dashscope/rerank/TextReRank.java @@ -2,11 +2,18 @@ package com.alibaba.dashscope.rerank; import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; -import com.alibaba.dashscope.common.*; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.common.TaskGroup; import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; -import com.alibaba.dashscope.protocol.*; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; import lombok.extern.slf4j.Slf4j; @Slf4j @@ -64,6 +71,21 @@ public TextReRank(String protocol, String baseUrl, ConnectionOptions connectionO syncApi = new SynchronizeHalfDuplexApi<>(connectionOptions, serviceOption); } + /** Creates a copy of the shared serviceOption for thread-safe per-call usage. */ + private ApiServiceOption copyServiceOption() { + return ApiServiceOption.builder() + .protocol(serviceOption.getProtocol()) + .httpMethod(serviceOption.getHttpMethod()) + .streamingMode(serviceOption.getStreamingMode()) + .outputMode(serviceOption.getOutputMode()) + .taskGroup(serviceOption.getTaskGroup()) + .task(serviceOption.getTask()) + .function(serviceOption.getFunction()) + .baseHttpUrl(serviceOption.getBaseHttpUrl()) + .baseWebSocketUrl(serviceOption.getBaseWebSocketUrl()) + .build(); + } + /** * Call the server to get the whole result. * @@ -75,8 +97,45 @@ public TextReRank(String protocol, String baseUrl, ConnectionOptions connectionO public TextReRankResult call(TextReRankParam param) throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); - serviceOption.setIsSSE(false); - serviceOption.setStreamingMode(StreamingMode.NONE); - return TextReRankResult.fromDashScopeResult(syncApi.call(param)); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); + return TextReRankResult.fromDashScopeResult(syncApi.call(param, callOption)); + } + + /** + * Call the server to get the result in the callback function. + * + * @param param The input param of class `TextReRankParam`. + * @param callback The callback to receive response, the template class is `TextReRankResult`. + * @throws NoApiKeyException Can not find api key + * @throws ApiException The request failed, possibly due to a network or data error. + */ + public void call(TextReRankParam param, ResultCallback callback) + throws ApiException, NoApiKeyException, InputRequiredException { + java.util.Objects.requireNonNull(callback, "callback must not be null"); + param.validate(); + ApiServiceOption callOption = copyServiceOption(); + callOption.setIsSSE(false); + callOption.setStreamingMode(StreamingMode.NONE); + syncApi.call( + param, + new ResultCallback() { + @Override + public void onEvent(DashScopeResult message) { + callback.onEvent(TextReRankResult.fromDashScopeResult(message)); + } + + @Override + public void onComplete() { + callback.onComplete(); + } + + @Override + public void onError(Exception e) { + callback.onError(e); + } + }, + callOption); } } diff --git a/src/main/java/com/alibaba/dashscope/utils/StreamingMerger.java b/src/main/java/com/alibaba/dashscope/utils/StreamingMerger.java new file mode 100644 index 00000000..9e73e85c --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/utils/StreamingMerger.java @@ -0,0 +1,160 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.utils; + +import com.alibaba.dashscope.tools.ToolCallBase; +import com.alibaba.dashscope.tools.ToolCallFunction; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public final class StreamingMerger { + + private StreamingMerger() {} + + public static void mergeTextContent( + List> currentContent, List> accumulatedContent) { + if (currentContent == null || accumulatedContent == null) { + return; + } + + for (Map contentItem : currentContent) { + if (contentItem == null || !contentItem.containsKey("text")) { + continue; + } + + Object textObj = contentItem.get("text"); + if (!(textObj instanceof String)) { + continue; + } + String textValue = (String) textObj; + if (textValue == null || textValue.isEmpty()) { + continue; + } + + Map accumulatedTextItem = null; + for (Map accumulatedItem : accumulatedContent) { + if (accumulatedItem != null && accumulatedItem.containsKey("text")) { + accumulatedTextItem = accumulatedItem; + break; + } + } + + if (accumulatedTextItem == null) { + accumulatedTextItem = new HashMap<>(); + accumulatedTextItem.put("text", textValue); + accumulatedContent.add(accumulatedTextItem); + } else { + String existingText = (String) accumulatedTextItem.get("text"); + if (existingText == null) { + existingText = ""; + } + accumulatedTextItem.put("text", existingText + textValue); + } + } + } + + public static List> copyContent( + List> accumulatedContent) { + List> copiedContent = new ArrayList<>(); + if (accumulatedContent == null) { + return copiedContent; + } + + for (Map contentItem : accumulatedContent) { + if (contentItem == null) { + copiedContent.add(null); + } else { + copiedContent.add(new HashMap<>(contentItem)); + } + } + return copiedContent; + } + + public static void mergeToolCalls( + List currentToolCalls, List accumulatedToolCalls) { + if (currentToolCalls == null || accumulatedToolCalls == null) { + return; + } + for (ToolCallBase currentCall : currentToolCalls) { + if (currentCall == null || currentCall.getIndex() == null) { + continue; + } + + int index = currentCall.getIndex(); + + ToolCallBase existingCall = null; + for (ToolCallBase accCall : accumulatedToolCalls) { + if (accCall != null && accCall.getIndex() != null && accCall.getIndex().equals(index)) { + existingCall = accCall; + break; + } + } + + if (existingCall instanceof ToolCallFunction && currentCall instanceof ToolCallFunction) { + ToolCallFunction existingFunctionCall = (ToolCallFunction) existingCall; + ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; + + if (currentFunctionCall.getFunction() != null) { + if (existingFunctionCall.getFunction() == null) { + existingFunctionCall.setFunction(existingFunctionCall.new CallFunction()); + } + + if (currentFunctionCall.getFunction().getArguments() != null) { + String existingArguments = existingFunctionCall.getFunction().getArguments(); + if (existingArguments == null) { + existingArguments = ""; + } + String currentArguments = currentFunctionCall.getFunction().getArguments(); + existingFunctionCall.getFunction().setArguments(existingArguments + currentArguments); + } + + if (currentFunctionCall.getFunction().getName() != null) { + String existingName = existingFunctionCall.getFunction().getName(); + if (existingName == null) { + existingName = ""; + } + String currentName = currentFunctionCall.getFunction().getName(); + existingFunctionCall.getFunction().setName(existingName + currentName); + } + + if (currentFunctionCall.getFunction().getOutput() != null) { + existingFunctionCall + .getFunction() + .setOutput(currentFunctionCall.getFunction().getOutput()); + } + } + + if (currentFunctionCall.getIndex() != null) { + existingFunctionCall.setIndex(currentFunctionCall.getIndex()); + } + if (currentFunctionCall.getId() != null && !currentFunctionCall.getId().isEmpty()) { + existingFunctionCall.setId(currentFunctionCall.getId()); + } + if (currentFunctionCall.getType() != null) { + existingFunctionCall.setType(currentFunctionCall.getType()); + } + } else { + if (currentCall instanceof ToolCallFunction) { + ToolCallFunction currentFunctionCall = (ToolCallFunction) currentCall; + ToolCallFunction newFunctionCall = new ToolCallFunction(); + newFunctionCall.setIndex(currentFunctionCall.getIndex()); + newFunctionCall.setId(currentFunctionCall.getId()); + newFunctionCall.setType(currentFunctionCall.getType()); + + if (currentFunctionCall.getFunction() != null) { + ToolCallFunction.CallFunction newCallFunction = newFunctionCall.new CallFunction(); + newCallFunction.setName(currentFunctionCall.getFunction().getName()); + newCallFunction.setArguments(currentFunctionCall.getFunction().getArguments()); + newCallFunction.setOutput(currentFunctionCall.getFunction().getOutput()); + newFunctionCall.setFunction(newCallFunction); + } + + accumulatedToolCalls.add(newFunctionCall); + } else { + accumulatedToolCalls.add(currentCall); + } + } + } + } +} diff --git a/src/test/java/com/alibaba/dashscope/TestFullDuplexErrorHandling.java b/src/test/java/com/alibaba/dashscope/TestFullDuplexErrorHandling.java index ec95488d..1c9adb38 100644 --- a/src/test/java/com/alibaba/dashscope/TestFullDuplexErrorHandling.java +++ b/src/test/java/com/alibaba/dashscope/TestFullDuplexErrorHandling.java @@ -13,7 +13,6 @@ import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; -import com.alibaba.dashscope.utils.Constants; import com.alibaba.dashscope.utils.JsonUtils; import io.reactivex.BackpressureStrategy; import io.reactivex.Flowable; @@ -42,15 +41,18 @@ @Slf4j @SetEnvironmentVariable(key = "DASHSCOPE_API_KEY", value = "1234") public class TestFullDuplexErrorHandling { - private ApiServiceOption serviceOption = - ApiServiceOption.builder() - .protocol(Protocol.WEBSOCKET) - .streamingMode(StreamingMode.DUPLEX) - .outputMode(OutputMode.ACCUMULATE) - .taskGroup("group") - .task("task") - .function("function") - .build(); + private ApiServiceOption createServiceOption(String baseWebSocketUrl) { + return ApiServiceOption.builder() + .protocol(Protocol.WEBSOCKET) + .streamingMode(StreamingMode.DUPLEX) + .outputMode(OutputMode.ACCUMULATE) + .taskGroup("group") + .task("task") + .function("function") + .baseWebSocketUrl(baseWebSocketUrl) + .build(); + } + MockWebServer server; WebSocketRecorder serverListener; int msgIndex = 0; @@ -167,7 +169,8 @@ public void testAddressInvalid() throws ApiException, NoApiKeyException, Interru serverListener = new WebSocketRecorder("server"); server.enqueue(new MockResponse().withWebSocketUpgrade(serverListener)); int port = this.server.getPort() + 1; // address is invalid - Constants.baseWebsocketApiUrl = String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port); + ApiServiceOption serviceOption = + createServiceOption(String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port)); FullDuplexTestParam param = getStreamMixParam(); ApiException thrown = assertThrows( @@ -192,7 +195,8 @@ public void testErrorOnSendingStreamData() serverListener = new WebSocketRecorder("server"); server.enqueue(new MockResponse().withWebSocketUpgrade(serverListener)); int port = this.server.getPort(); - Constants.baseWebsocketApiUrl = String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port); + ApiServiceOption serviceOption = + createServiceOption(String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port)); FullDuplexTestParam param = getStreamErrorData(); Exception thrown = assertThrows( @@ -208,7 +212,19 @@ public void testErrorOnSendingStreamData() }); log.error(thrown.getMessage()); thrown.printStackTrace(); - assertTrue(thrown.getMessage().contains("emitter on error")); + assertTrue(hasMessageInCauseChain(thrown, "emitter on error")); + } + + private boolean hasMessageInCauseChain(Throwable throwable, String expectedMessage) { + Throwable currentThrowable = throwable; + while (currentThrowable != null) { + String message = currentThrowable.getMessage(); + if (message != null && message.contains(expectedMessage)) { + return true; + } + currentThrowable = currentThrowable.getCause(); + } + return false; } private FullDuplexTestParam getStreamDataWithNullStreamData() { @@ -230,7 +246,8 @@ public void testSendingStreamDataNullFlowable() serverListener = new WebSocketRecorder("server"); server.enqueue(new MockResponse().withWebSocketUpgrade(serverListener)); int port = this.server.getPort(); - Constants.baseWebsocketApiUrl = String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port); + ApiServiceOption serviceOption = + createServiceOption(String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port)); FullDuplexTestParam param = getStreamDataWithNullStreamData(); Exception thrown = assertThrows( @@ -257,7 +274,8 @@ public void testErrorOnReceiveData() serverListener = new WebSocketRecorder("server"); server.enqueue(new MockResponse().withWebSocketUpgrade(serverListener)); int port = this.server.getPort(); - Constants.baseWebsocketApiUrl = String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port); + ApiServiceOption serviceOption = + createServiceOption(String.format("ws://127.0.0.1:%s/api-ws/v1/inference/", port)); FullDuplexTestParam param = getStreamTextParam(); SynchronizeFullDuplexApi api = new SynchronizeFullDuplexApi<>(serviceOption); diff --git a/src/test/resources/mismatch.jsonl b/src/test/resources/mismatch.jsonl index ba8bf4b6..ef0cc609 100644 --- a/src/test/resources/mismatch.jsonl +++ b/src/test/resources/mismatch.jsonl @@ -1,7 +1,7 @@ -{"HumanAnswer": "２个读音　　一个是ｘｉ　一个是ｑｉａｎ", -"Python": [24918, 18947, 57553, 78685, 22441, 22441, 108075, 144767, 144155, 22441, 108075, 145529, 144155, 144127, 144270], +{"HumanAnswer": "２个读音　　一个是ｘｉ　一个是ｑｉａｎ", +"Python": [24918, 18947, 57553, 78685, 22441, 22441, 108075, 144767, 144155, 22441, 108075, 145529, 144155, 144127, 144270], "Java": [24918, 18947, 57553, 78685, 43429, 108075, 144767, 144155, 22441, 108075, 145529, 144155, 144127, 144270]} -{"HumanAnswer": "炸腐竹·油温要适中·炸后浸温水 +{"HumanAnswer": "炸腐竹·油温要适中·炸后浸温水 　　要腐竹炸得刚好而不焦黑，秘诀是当生油烧至六成熟时转慢火，然后放下腐竹，迅速地翻动，让腐竹全身炸透，呈现白泡，并即投入温水中浸透，再切段。炸时如油温过热，会将腐竹炸焦，变深黄色或黑色，如油温不够热，则不会起白泡。 ", "Python": [100893, 100298, 102045, 13935, 99318, 99416, 30534, 99340, 15946, 13935, 100893, 33447, 101595, 99416, 52510, 2529, 22441, 22441, 30534, 100298, 102045, 100893, 49828, 109586, 105145, 100479, 56652, 3837, 114928, 20412, 39165, 21287, 99318, 100228, 56137, 99566, 102226, 13343, 46670, 99843, 79599, 3837, 101889, 106162, 100298, 102045, 3837, 104015, 29490, 100712, 27733, 3837, 99258, 100298, 102045, 105418, 100893, 99844, 3837, 104401, 99243, 100454, 90395, 91676, 100709, 99416, 107629, 101595, 99844, 3837, 87256, 99322, 37474, 1773, 100893, 13343, 29524, 99318, 99416, 38182, 99259, 3837, 36993, 44063, 100298, 102045, 100893, 100479, 3837, 74040, 99194, 106140, 57191, 104723, 3837, 29524, 99318, 99416, 104222, 99259, 95053, 99670, 71618, 99243, 100454, 1773, 220], "Java": [100893, 100298, 102045, 13935, 99318, 99416, 30534, 99340, 15946, 13935, 100893, 33447, 101595, 99416, 52510, 2529, 43429, 30534, 100298, 102045, 100893, 49828, 109586, 105145, 100479, 56652, 3837, 114928, 20412, 39165, 21287, 99318, 100228, 56137, 99566, 102226, 13343, 46670, 99843, 79599, 3837, 101889, 106162, 100298, 102045, 3837, 104015, 29490, 100712, 27733, 3837, 99258, 100298, 102045, 105418, 100893, 99844, 3837, 104401, 99243, 100454, 90395, 91676, 100709, 99416, 107629, 101595, 99844, 3837, 87256, 99322, 37474, 1773, 100893, 13343, 29524, 99318, 99416, 38182, 99259, 3837, 36993, 44063, 100298, 102045, 100893, 100479, 3837, 74040, 99194, 106140, 57191, 104723, 3837, 29524, 99318, 99416, 104222, 99259, 95053, 99670, 71618, 99243, 100454, 1773, 220]} {"HumanAnswer": "　　你好，癫痫病频频发作容易造成癫痫患者的脑发育障碍、组织结构毁坏、代谢异常，必然也影响其所有心理发育的领域，对于脑器质性损害伴发的癫痫，往往伴有智力损害及行为问题。一般癫痫发生愈早，程度愈重，对心理、精神行为的影响也愈大。 ", "Python": [22441, 22441, 108386, 3837, 113282, 111972, 107253, 100047, 101090, 103440, 108723, 99931, 106115, 102544, 5373, 99877, 100166, 100882, 100250, 5373, 103225, 70633, 3837, 102309, 74763, 99564, 41146, 55338, 100438, 106115, 9370, 100650, 3837, 100002, 99931, 31548, 99178, 33071, 105252, 99595, 28291, 9370, 103440, 3837, 101207, 115563, 107941, 105252, 81217, 101070, 86119, 1773, 100141, 103440, 99726, 100939, 99391, 3837, 100069, 100939, 29258, 3837, 32664, 100438, 5373, 100150, 101070, 104126, 74763, 100939, 26288, 37265], "Java": [43429, 108386, 3837, 113282, 111972, 107253, 100047, 101090, 103440, 108723, 99931, 106115, 102544, 5373, 99877, 100166, 100882, 100250, 5373, 103225, 70633, 3837, 102309, 74763, 99564, 41146, 55338, 100438, 106115, 9370, 100650, 3837, 100002, 99931, 31548, 99178, 33071, 105252, 99595, 28291, 9370, 103440, 3837, 101207, 115563, 107941, 105252, 81217, 101070, 86119, 1773, 100141, 103440, 99726, 100939, 99391, 3837, 100069, 100939, 29258, 3837, 32664, 100438, 5373, 100150, 101070, 104126, 74763, 100939, 26288, 37265]} @@ -10,7 +10,7 @@ ", "Python": [85336, 116350, 104160, 39907, 99604, 99999, 97480, 36993, 99518, 102478, 3837, 102716, 104160, 18830, 319, 22441, 16, 5373, 99950, 100553, 24339, 9909, 99393, 99451, 24339, 7552, 85336, 116350, 220, 22441, 22441, 17, 5373, 29767, 101953, 24339, 85336, 116350, 220, 22441, 18, 5373, 105608, 24339, 85336, 116350, 220, 22441, 22441, 19, 5373, 99759, 64952, 32320, 99368, 24339, 85336, 116350, 220, 22441, 20, 5373, 47815, 76837, 99322, 29767, 24339, 85336, 116350, 220, 100700, 99559, 30440, 118393, 100849, 105157, 104316, 31139, 319], "Java": [85336, 116350, 104160, 39907, 99604, 99999, 97480, 36993, 99518, 102478, 3837, 102716, 104160, 18830, 319, 22441, 16, 5373, 99950, 100553, 24339, 9909, 99393, 99451, 24339, 7552, 85336, 116350, 220, 43429, 17, 5373, 29767, 101953, 24339, 85336, 116350, 220, 22441, 18, 5373, 105608, 24339, 85336, 116350, 220, 43429, 19, 5373, 99759, 64952, 32320, 99368, 24339, 85336, 116350, 220, 22441, 20, 5373, 47815, 76837, 99322, 29767, 24339, 85336, 116350, 220, 100700, 99559, 30440, 118393, 100849, 105157, 104316, 31139, 319]} {"HumanAnswer": "　　你好,过敏性鼻炎喷嚏、鼻痒、流涕和鼻堵是最常见的四大症状。目前临床上主要是应用内窥镜下等离子低温消融术，选择性地进行筛前神经、翼管神经阻断术，降低鼻腔副交感神经的兴奋性，消除过敏区域神经对外界各种刺激，如气候、灰尘、花粉等的敏感性，过敏性鼻炎就可以自然而愈，不易复发。 ", "Python": [22441, 22441, 108386, 11, 107270, 33071, 102228, 100439, 102125, 121875, 5373, 102228, 108681, 5373, 88653, 119392, 33108, 102228, 101476, 104890, 102716, 104978, 101368, 1773, 100004, 99635, 101717, 104312, 99892, 31843, 112808, 100811, 16872, 49567, 108510, 109878, 32320, 99368, 99216, 3837, 50404, 33071, 29490, 71817, 101737, 24562, 102398, 5373, 101401, 35551, 102398, 100318, 63789, 99216, 3837, 101134, 102228, 102504, 99310, 38109, 98650, 102398, 9370, 105487, 33071, 3837, 105109, 107270, 101065, 102398, 102330, 97120, 100646, 104313, 3837, 29524, 102730, 5373, 114267, 5373, 99232, 99742, 49567, 9370, 105521, 33071, 3837, 107270, 33071, 102228, 100439, 102003, 117536, 100939, 3837, 106921, 111574, 8997], "Java": [43429, 108386, 11, 107270, 33071, 102228, 100439, 102125, 121875, 5373, 102228, 108681, 5373, 88653, 119392, 33108, 102228, 101476, 104890, 102716, 104978, 101368, 1773, 100004, 99635, 101717, 104312, 99892, 31843, 112808, 100811, 16872, 49567, 108510, 109878, 32320, 99368, 99216, 3837, 50404, 33071, 29490, 71817, 101737, 24562, 102398, 5373, 101401, 35551, 102398, 100318, 63789, 99216, 3837, 101134, 102228, 102504, 99310, 38109, 98650, 102398, 9370, 105487, 33071, 3837, 105109, 107270, 101065, 102398, 102330, 97120, 100646, 104313, 3837, 29524, 102730, 5373, 114267, 5373, 99232, 99742, 49567, 9370, 105521, 33071, 3837, 107270, 33071, 102228, 100439, 102003, 117536, 100939, 3837, 106921, 111574, 8997]} -{"HumanAnswer": "　　新生婴儿应在出生后一个月内，到父亲或母亲户口所在地的户口登记机关（辖区派出所）办理出生入户登记。 +{"HumanAnswer": "　　新生婴儿应在出生后一个月内，到父亲或母亲户口所在地的户口登记机关（辖区派出所）办理出生入户登记。 　　办理时需提供以下材料： 　　1、小孩的出生医学证明 　　2、父母的户口簿、身份证和结婚证。", "Python": [22441, 22441, 102900, 102833, 110391, 102246, 33447, 105033, 31843, 3837, 26939, 102037, 57191, 102079, 109200, 108176, 9370, 109200, 104141, 100776, 9909, 103022, 106644, 7552, 103990, 102246, 111981, 104141, 1773, 2529, 22441, 22441, 103990, 13343, 58362, 99553, 87752, 100643, 5122, 319, 22441, 22441, 16, 5373, 104902, 9370, 102246, 104316, 104022, 319, 22441, 22441, 17, 5373, 103953, 9370, 109200, 113649, 5373, 105765, 33108, 104388, 33477, 1773], "Java": [43429, 102900, 102833, 110391, 102246, 33447, 105033, 31843, 3837, 26939, 102037, 57191, 102079, 109200, 108176, 9370, 109200, 104141, 100776, 9909, 103022, 106644, 7552, 103990, 102246, 111981, 104141, 1773, 2529, 43429, 103990, 13343, 58362, 99553, 87752, 100643, 5122, 319, 43429, 16, 5373, 104902, 9370, 102246, 104316, 104022, 319, 43429, 17, 5373, 103953, 9370, 109200, 113649, 5373, 105765, 33108, 104388, 33477, 1773]} @@ -59,7 +59,7 @@ {"HumanAnswer": "1．及时治疗：主要是改善脑血液循环。可以在医生的指导下使用扩血管药物和银杏叶制剂等。　　2．早期预防：对脑供血不足的防治重点在脑血管方面，尤其是血脂和低密度脂蛋白增高要对症处理。　　3．合理饮食：平时多吃新鲜蔬菜(如洋葱、西红柿等)、水果、鱼、黑木耳、少量醋、干红葡萄酒等，可以起抗氧化作用，延缓脑动脉硬化的发生。　　4．适当的户外活动：如快走、慢跑、散步等，每次30-40分钟，每周至少5天，或者打打太极拳、垂钓、登山等。　　5．保持良好的心态和健康用脑：平时看看电视、报纸；做些手工劳作或家务事；也可以参加一些文体活动，如唱歌、跳舞、书法、打球等，调制性情，增强脑的思维活动；避免情绪激动和过度疲劳。", "Python": [16, 58883, 100667, 101899, 5122, 104312, 104009, 99931, 116417, 1773, 104964, 103998, 9370, 109767, 37029, 99834, 102890, 104459, 33108, 99660, 108202, 100027, 115031, 49567, 1773, 22441, 22441, 17, 58883, 105184, 104332, 5122, 32664, 99931, 83744, 99389, 102004, 9370, 102681, 99887, 18493, 99931, 102890, 99522, 3837, 104033, 118112, 33108, 99285, 106651, 100553, 102835, 117015, 30534, 32664, 99769, 54542, 1773, 22441, 22441, 18, 58883, 100745, 104579, 5122, 104325, 108851, 104838, 104451, 7, 29524, 117956, 5373, 117700, 49567, 8, 5373, 104618, 5373, 100655, 5373, 56652, 116690, 5373, 108521, 106960, 5373, 99251, 99425, 107358, 49567, 3837, 73670, 71618, 118283, 100154, 3837, 99771, 99982, 99931, 109826, 99927, 105302, 99726, 1773, 22441, 22441, 19, 58883, 109776, 105813, 99600, 5122, 29524, 99234, 99314, 5373, 99843, 99803, 5373, 111261, 49567, 3837, 102173, 18, 15, 12, 19, 15, 83031, 3837, 106854, 104102, 20, 35727, 3837, 100631, 75437, 75437, 116674, 5373, 102566, 103429, 5373, 112357, 49567, 1773, 22441, 22441, 20, 58883, 100662, 104205, 106479, 33108, 99722, 11622, 99931, 5122, 104325, 101997, 100234, 5373, 107574, 24968, 99190, 97084, 107171, 99697, 19403, 57191, 114791, 29826, 24968, 104047, 101061, 101883, 111048, 99600, 3837, 29524, 109283, 5373, 112084, 5373, 102702, 5373, 117248, 49567, 3837, 47872, 43316, 33071, 39374, 3837, 101138, 99931, 9370, 102141, 99600, 24968, 101153, 104405, 105694, 33108, 105831, 108351, 1773], "Java": [16, 58883, 100667, 101899, 5122, 104312, 104009, 99931, 116417, 1773, 104964, 103998, 9370, 109767, 37029, 99834, 102890, 104459, 33108, 99660, 108202, 100027, 115031, 49567, 1773, 43429, 17, 58883, 105184, 104332, 5122, 32664, 99931, 83744, 99389, 102004, 9370, 102681, 99887, 18493, 99931, 102890, 99522, 3837, 104033, 118112, 33108, 99285, 106651, 100553, 102835, 117015, 30534, 32664, 99769, 54542, 1773, 43429, 18, 58883, 100745, 104579, 5122, 104325, 108851, 104838, 104451, 7, 29524, 117956, 5373, 117700, 49567, 8, 5373, 104618, 5373, 100655, 5373, 56652, 116690, 5373, 108521, 106960, 5373, 99251, 99425, 107358, 49567, 3837, 73670, 71618, 118283, 100154, 3837, 99771, 99982, 99931, 109826, 99927, 105302, 99726, 1773, 43429, 19, 58883, 109776, 105813, 99600, 5122, 29524, 99234, 99314, 5373, 99843, 99803, 5373, 111261, 49567, 3837, 102173, 18, 15, 12, 19, 15, 83031, 3837, 106854, 104102, 20, 35727, 3837, 100631, 75437, 75437, 116674, 5373, 102566, 103429, 5373, 112357, 49567, 1773, 43429, 20, 58883, 100662, 104205, 106479, 33108, 99722, 11622, 99931, 5122, 104325, 101997, 100234, 5373, 107574, 24968, 99190, 97084, 107171, 99697, 19403, 57191, 114791, 29826, 24968, 104047, 101061, 101883, 111048, 99600, 3837, 29524, 109283, 5373, 112084, 5373, 102702, 5373, 117248, 49567, 3837, 47872, 43316, 33071, 39374, 3837, 101138, 99931, 9370, 102141, 99600, 24968, 101153, 104405, 105694, 33108, 105831, 108351, 1773]} {"HumanAnswer": "您好：肠道检查包括： (1)粪便检查： -　　(2)直肠指诊： +　　(2)直肠指诊： 　　(3)乙状结肠镜检查： 　　(4)钡灌肠X线检查： 　　(5)纤维结肠镜检查 @@ -93,6 +93,6 @@ {"HumanAnswer": "不合法  属于盗用   协议也是无效的   不受保护", "Python": [16530, 101431, 4102, 4102, 100409, 100591, 11622, 9238, 4102, 101136, 100000, 107128, 9370, 9238, 4102, 107524, 100153], "Java": [16530, 101431, 9238, 100409, 100591, 11622, 45393, 101136, 100000, 107128, 9370, 45393, 107524, 100153]} {"HumanAnswer": "您好  一般是可以的，具体的咨询社保部门", "Python": [111308, 4102, 4102, 100141, 105903, 9370, 3837, 108247, 100703, 106732, 99667], "Java": [111308, 9238, 100141, 105903, 9370, 3837, 108247, 100703, 106732, 99667]} {"HumanAnswer": "针对你的问题答复如下：    《民法通则》第六十六条 没有代理权、超越代理权或者代理权终止后的行为，只有经过被代理人的追认，被代理人才承担民事责任。未经追认的行为，由行为人承担民事责任。本人知道他人以本人名义实施民事行为而不作否认表示的，视为同意。以上答复，如果满意，敬请采纳，并给予评价。", "Python": [101092, 103929, 86119, 110463, 104506, 5122, 45393, 4102, 26940, 69721, 24339, 31935, 46448, 25067, 102651, 117602, 4102, 80443, 101259, 40981, 5373, 104840, 101259, 40981, 100631, 101259, 40981, 107660, 33447, 104796, 3837, 101043, 101897, 99250, 101259, 103947, 99626, 28951, 3837, 99250, 101259, 101050, 102016, 108669, 99531, 1773, 102887, 99626, 28951, 104796, 3837, 67071, 101070, 17340, 102016, 108669, 99531, 1773, 102043, 99392, 104287, 23031, 102043, 107765, 100355, 108669, 101070, 105145, 19403, 109433, 51463, 9370, 3837, 103097, 101185, 1773, 70589, 110463, 3837, 62244, 100545, 3837, 114131, 115250, 90395, 102062, 103964, 1773], "Java": [101092, 103929, 86119, 110463, 104506, 5122, 19070, 26940, 69721, 24339, 31935, 46448, 25067, 102651, 117602, 4102, 80443, 101259, 40981, 5373, 104840, 101259, 40981, 100631, 101259, 40981, 107660, 33447, 104796, 3837, 101043, 101897, 99250, 101259, 103947, 99626, 28951, 3837, 99250, 101259, 101050, 102016, 108669, 99531, 1773, 102887, 99626, 28951, 104796, 3837, 67071, 101070, 17340, 102016, 108669, 99531, 1773, 102043, 99392, 104287, 23031, 102043, 107765, 100355, 108669, 101070, 105145, 19403, 109433, 51463, 9370, 3837, 103097, 101185, 1773, 70589, 110463, 3837, 62244, 100545, 3837, 114131, 115250, 90395, 102062, 103964, 1773]} -{"HumanAnswer": "你在网上的棋牌室花了人民币多少钱啊  你说的跑路是什么意思啊", -"Python": [107740, 31139, 101913, 101386, 99478, 108225, 104142, 105957, 103924, 4102, 4102, 105665, 9370, 99803, 45995, 102021, 100313, 103924], +{"HumanAnswer": "你在网上的棋牌室花了人民币多少钱啊  你说的跑路是什么意思啊", +"Python": [107740, 31139, 101913, 101386, 99478, 108225, 104142, 105957, 103924, 4102, 4102, 105665, 9370, 99803, 45995, 102021, 100313, 103924], "Java": [107740, 31139, 101913, 101386, 99478, 108225, 104142, 105957, 103924, 9238, 105665, 9370, 99803, 45995, 102021, 100313, 103924]} diff --git a/src/test/resources/stream_runs.json b/src/test/resources/stream_runs.json index 98ac44fc..d1128fa7 100644 --- a/src/test/resources/stream_runs.json +++ b/src/test/resources/stream_runs.json @@ -55,7 +55,7 @@ { "event": "done", "data": "[DONE]" - } + } ], "create_stream_thread_and_run_response": [ {"event": "thread.created", @@ -120,7 +120,7 @@ { "event": "done", "data": "[DONE]" - } + } ], "create_stream_require_action_function_1": [ { @@ -182,7 +182,7 @@ { "event": "done", "data": "[DONE]" - } + } ], "create_stream_require_action_function_2": [ {