(rocketmq-a2a) branch main updated: Optimize the example (#13)

[jinrongtong]

This is an automated email from the ASF dual-hosted git repository.

jinrongtong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-a2a.git


The following commit(s) were added to refs/heads/main by this push:
     new 5d2648b  Optimize the example (#13)
5d2648b is described below

commit 5d2648b5d94ae7b474c3c99dc69e6959d83c895a
Author: Drizzle <464473...@qq.com>
AuthorDate: Thu Dec 25 10:07:41 2025 +0800

    Optimize the example (#13)
    
    * update
    
    Change-Id: Icb48afcf44893ad3c07fc0e243e333096d8e3ec0
    
    * optimize the code
    
    Change-Id: Iaa158a2bd2bc55475e7c55e7ae3392b4e2f28fa0
    
    * optimize the code
    
    Change-Id: I9fba77a503e387b5d9feab13cd144021b938ec1f
    
    ---------
    
    Co-authored-by: drizzle.zk <drizzle...@alibaba-inc.com>
---
 .../Common/src/main/java/common/QWModel.java          |  1 +
 .../src/main/java/org/example/common/TaskInfo.java    | 19 +++++++++++++++++++
 .../main/java/org/example/service/AgentService.java   | 16 +++++++++++-----
 3 files changed, 31 insertions(+), 5 deletions(-)

diff --git 
a/example/rocketmq-multiagent-base-adk/Common/src/main/java/common/QWModel.java 
b/example/rocketmq-multiagent-base-adk/Common/src/main/java/common/QWModel.java
index 3446ca6..5b05d4c 100644
--- 
a/example/rocketmq-multiagent-base-adk/Common/src/main/java/common/QWModel.java
+++ 
b/example/rocketmq-multiagent-base-adk/Common/src/main/java/common/QWModel.java
@@ -66,6 +66,7 @@ public class QWModel extends BaseLlm {
             .apiKey(API_KEY)
             .model(MODEL_NAME)
             .messages(messages)
+            .enableThinking(false)
             .resultFormat(GenerationParam.ResultFormat.MESSAGE)
             .build();
         return gen.call(param);
diff --git 
a/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/common/TaskInfo.java
 
b/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/common/TaskInfo.java
index ccd09ff..fe800bb 100644
--- 
a/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/common/TaskInfo.java
+++ 
b/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/common/TaskInfo.java
@@ -23,6 +23,7 @@ public class TaskInfo {
     private String taskDesc;
     private String userId;
     private String sessionId;
+    private String lastOutput = "";
     private Sinks.Many<String> sink;
 
     public TaskInfo(String taskId, String taskDesc, String sessionId, String 
userId, Sinks.Many<String> sink) {
@@ -33,6 +34,16 @@ public class TaskInfo {
         this.sink = sink;
     }
 
+    public TaskInfo(String taskId, String taskDesc, String userId, String 
sessionId, String lastOutput,
+        Many<String> sink) {
+        this.taskId = taskId;
+        this.taskDesc = taskDesc;
+        this.userId = userId;
+        this.sessionId = sessionId;
+        this.lastOutput = lastOutput;
+        this.sink = sink;
+    }
+
     public TaskInfo() {
     }
 
@@ -75,4 +86,12 @@ public class TaskInfo {
     public void setSink(Many<String> sink) {
         this.sink = sink;
     }
+
+    public String getLastOutput() {
+        return lastOutput;
+    }
+
+    public void setLastOutput(String lastOutput) {
+        this.lastOutput = lastOutput;
+    }
 }
diff --git 
a/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/service/AgentService.java
 
b/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/service/AgentService.java
index 512e17e..8d8c915 100644
--- 
a/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/service/AgentService.java
+++ 
b/example/rocketmq-multiagent-base-adk/SupervisorAgent-Web/src/main/java/org/example/service/AgentService.java
@@ -293,9 +293,13 @@ public class AgentService {
                     TaskState state = task.getStatus().state();
                     String msg = 
extractTextFromMessage(artifacts.get(artifacts.size() - 1));
                     log.info("receive msg: {}", msg);
-                    boolean result = emitMessage(sink, msg, false);
-                    if (!result) {
-                        throw new RuntimeException("client close stream");
+                    String lastOutput = taskInfo.getLastOutput();
+                    if (!lastOutput.equals(msg)) {
+                        boolean result = emitMessage(sink, msg, false);
+                        if (!result) {
+                            throw new RuntimeException("client close stream");
+                        }
+                        taskInfo.setLastOutput(msg);
                     }
                     if (state == TaskState.COMPLETED) {
                         StringBuilder stringBuilder = new StringBuilder();
@@ -361,13 +365,15 @@ public class AgentService {
         Flowable<Event> events = runner.runAsync(userId, sessionId, userMsg);
         events.blockingForEach(eventSub -> {
             boolean equals = lastQuestion.equals(eventSub.stringifyContent());
+            TaskInfo taskInfo = taskMap.get(taskId);
+            Many<String> sink = taskInfo.getSink();
             if (equals) {
+                sink.tryEmitComplete();
+                completeTask(taskInfo);
                 return;
             }
             lastQuestion = eventSub.stringifyContent();
             String content = lastQuestion;
-            TaskInfo taskInfo = taskMap.get(taskId);
-            Many<String> sink = taskInfo.getSink();
             if (!StringUtils.isEmpty(content)) {
                 if (content.startsWith("{")) {
                     try {