-
Notifications
You must be signed in to change notification settings - Fork 932
feat: add streaming output for studio #565
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
lokidundun
wants to merge
8
commits into
agentscope-ai:main
Choose a base branch
from
lokidundun:addStreamingOutput
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
8 commits
Select commit
Hold shift + click to select a range
0cd98ec
feat: add streaming output for studio
lokidundun 1e8ac60
fix the code format
lokidundun 1d69a1f
feature: add stream output
lokidundun 7b9c0c6
fix ci
lokidundun b2a61bd
fomate the code style
lokidundun 1faf197
add some tests
lokidundun b8df4df
Merge branch 'main' into addStreamingOutput
lokidundun 05dac8f
formate code style
lokidundun File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
186 changes: 186 additions & 0 deletions
186
...cope-extensions-studio/src/main/java/io/agentscope/core/studio/StudioStreamingBridge.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,186 @@ | ||
| /* | ||
| * Copyright 2024-2026 the original author or authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.agentscope.core.studio; | ||
|
|
||
| import io.agentscope.core.agent.Event; | ||
| import io.agentscope.core.agent.EventType; | ||
| import io.agentscope.core.message.Msg; | ||
| import java.util.Objects; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import reactor.core.publisher.Flux; | ||
| import reactor.core.publisher.Mono; | ||
|
|
||
| /** | ||
| * Bridges agent streaming events to AgentScope Studio. | ||
| * | ||
| * <p>This component subscribes to a {@link reactor.core.publisher.Flux} of | ||
| * {@link io.agentscope.core.agent.Event} instances produced by a | ||
| * {@code StreamableAgent} and forwards them to Studio using both the HTTP | ||
| * {@link StudioClient} and the WebSocket-based {@link StudioWebSocketClient}. | ||
| * | ||
| * <p>The bridge forwards REASONING, TOOL_RESULT, and SUMMARY events directly to the | ||
| * Studio frontend via {@link StudioWebSocketClient#sendStreamEvent(Event)} so that | ||
| * intermediate and final reasoning/tool outputs are visible in real time. Non-terminal | ||
| * {@link EventType#AGENT_RESULT} events are also streamed as incremental chunks. | ||
| * | ||
| * <p>The final {@link EventType#AGENT_RESULT} event with {@code isLast == true} is treated as | ||
| * a control signal: its {@link Msg} payload is cached for persistence and it triggers | ||
| * {@link StudioWebSocketClient#sendStreamCompleted()} but is not forwarded again as a | ||
| * streaming chunk. After the stream completes, the selected final {@link Msg} (terminal | ||
| * AGENT_RESULT if present, otherwise the last non-terminal AGENT_RESULT) is persisted once | ||
| * via {@link StudioClient#pushMessage(Msg)}. | ||
| */ | ||
| public class StudioStreamingBridge { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(StudioStreamingBridge.class); | ||
|
|
||
| private final StudioClient studioClient; | ||
| private final StudioWebSocketClient webSocketClient; | ||
|
|
||
| /** | ||
| * Creates a new streaming bridge for forwarding agent events to Studio. | ||
| * | ||
| * @param studioClient HTTP client used to persist the final agent message to Studio | ||
| * @param webSocketClient WebSocket client used to stream incremental events and | ||
| * completion signals to the Studio frontend | ||
| */ | ||
| public StudioStreamingBridge(StudioClient studioClient, StudioWebSocketClient webSocketClient) { | ||
| this.studioClient = Objects.requireNonNull(studioClient, "studioClient must not be null"); | ||
| this.webSocketClient = | ||
| Objects.requireNonNull(webSocketClient, "webSocketClient must not be null"); | ||
| } | ||
|
Comment on lines
+47
to
+65
|
||
|
|
||
| /** | ||
| * Forwards a stream of agent events to Studio for real-time visualization and | ||
| * final message persistence. | ||
| * | ||
| * <ol> | ||
| * <li>All REASONING, TOOL_RESULT, and SUMMARY events are forwarded to the frontend via | ||
| * {@link StudioWebSocketClient#sendStreamEvent(Event)} regardless of the {@code isLast} | ||
| * flag so that final reasoning/tool outputs remain visible even when emitted as | ||
| * terminal events.</li> | ||
| * <li>Non-terminal AGENT_RESULT events ({@code isLast == false}) are also forwarded as | ||
| * streaming chunks.</li> | ||
| * <li>The terminal AGENT_RESULT event with {@code isLast == true} is used to: | ||
| * <ul> | ||
| * <li>cache the final {@link Msg} for persistence, and</li> | ||
| * <li>trigger {@link StudioWebSocketClient#sendStreamCompleted()}.</li> | ||
| * </ul> | ||
| * It is not sent again as a streaming chunk.</li> | ||
| * <li>If a final {@link Msg} exists, it is persisted once via | ||
| * {@link StudioClient#pushMessage(Msg)} after the stream completes.</li> | ||
| * </ol> | ||
| * | ||
| * <p>Note: {@code isLast} is treated purely as a control flag for AGENT_RESULT and does not | ||
| * generate additional visible content.</p> | ||
| */ | ||
| public Mono<Void> forwardToStudio(Flux<Event> eventFlux) { | ||
| return Mono.defer( | ||
| () -> { | ||
| FinalMsgHolder holder = new FinalMsgHolder(); | ||
|
|
||
| return eventFlux | ||
| .doOnNext(event -> handleEvent(event, holder)) | ||
| .doOnError( | ||
| ex -> | ||
| logger.error( | ||
| "Error occurred during streaming to Studio", | ||
| ex)) | ||
| .doFinally( | ||
| signalType -> { | ||
| if (!holder.completedSignalSent) { | ||
| try { | ||
| webSocketClient.sendStreamCompleted(); | ||
| } catch (Exception e) { | ||
| logger.error( | ||
| "Failed to send streamCompleted in" | ||
| + " doFinally", | ||
| e); | ||
| } | ||
| holder.completedSignalSent = true; | ||
| } | ||
| }) | ||
| .then( | ||
| Mono.defer( | ||
| () -> { | ||
| Msg finalMsg = holder.getEffectiveFinalMsg(); | ||
| if (finalMsg == null) { | ||
| logger.debug( | ||
| "No final Msg determined from stream;" | ||
| + " skip pushMessage"); | ||
| return Mono.empty(); | ||
| } | ||
| return studioClient.pushMessage(finalMsg); | ||
| })); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Handles individual events from the stream. | ||
| */ | ||
| private void handleEvent(Event event, FinalMsgHolder holder) { | ||
| if (event == null) { | ||
| return; | ||
| } | ||
|
|
||
| EventType type = event.getType(); | ||
| Msg msg = event.getMessage(); | ||
| boolean isLast = event.isLast(); | ||
|
|
||
| if (type == EventType.AGENT_RESULT && msg != null) { | ||
| holder.lastAgentResultMsg = msg; | ||
| } | ||
|
|
||
| if (type == EventType.AGENT_RESULT && isLast) { | ||
| if (msg != null) { | ||
| holder.terminalAgentResultMsg = msg; | ||
| } | ||
| if (!holder.completedSignalSent) { | ||
| webSocketClient.sendStreamCompleted(); | ||
| holder.completedSignalSent = true; | ||
| } | ||
| return; | ||
| } | ||
|
|
||
| if (type == EventType.AGENT_RESULT && !isLast) { | ||
| webSocketClient.sendStreamEvent(event); | ||
| return; | ||
| } | ||
|
|
||
| if (type == EventType.REASONING | ||
| || type == EventType.TOOL_RESULT | ||
| || type == EventType.SUMMARY) { | ||
| webSocketClient.sendStreamEvent(event); | ||
| } | ||
|
Comment on lines
+159
to
+168
|
||
| } | ||
|
|
||
| private static class FinalMsgHolder { | ||
| private Msg terminalAgentResultMsg; | ||
| private Msg lastAgentResultMsg; | ||
| private boolean completedSignalSent; | ||
|
|
||
| Msg getEffectiveFinalMsg() { | ||
| if (terminalAgentResultMsg != null) { | ||
| return terminalAgentResultMsg; | ||
| } | ||
| if (lastAgentResultMsg != null) { | ||
| return lastAgentResultMsg; | ||
| } | ||
| return null; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,7 +15,13 @@ | |
| */ | ||
| package io.agentscope.core.studio; | ||
|
|
||
| import io.agentscope.core.agent.Event; | ||
| import io.agentscope.core.agent.EventType; | ||
| import io.agentscope.core.message.ContentBlock; | ||
| import io.agentscope.core.message.Msg; | ||
| import io.agentscope.core.message.TextBlock; | ||
| import io.agentscope.core.message.ThinkingBlock; | ||
| import io.agentscope.core.message.ToolResultBlock; | ||
| import io.agentscope.core.util.JsonUtils; | ||
| import io.socket.client.IO; | ||
| import io.socket.client.Socket; | ||
|
|
@@ -395,4 +401,127 @@ public Map<String, Object> getStructuredInput() { | |
| return structuredInput; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Sends a stream event to Studio in real-time. | ||
| * | ||
| * <p>This method delivers best-effort, non-guaranteed notifications over the | ||
| * existing Socket.IO connection. If the underlying WebSocket is {@code null} | ||
| * or not connected, the method logs a warning and returns without throwing | ||
| * an exception and without attempting any reconnection. | ||
| * | ||
| * @param event the agent event to forward to Studio; if {@code null}, the | ||
| * call is ignored and no payload is emitted | ||
| */ | ||
| public void sendStreamEvent(Event event) { | ||
| if (socket == null || !socket.connected()) { | ||
| logger.warn("Attempted to send stream event while WebSocket is not connected"); | ||
| return; | ||
|
Comment on lines
+405
to
+419
|
||
| } | ||
| if (event == null) { | ||
| return; | ||
| } | ||
|
|
||
| try { | ||
| JSONObject payload = new JSONObject(); | ||
| EventType type = event.getType(); | ||
| payload.put("eventType", type != null ? type.name() : "UNKNOWN"); | ||
| payload.put("isLast", event.isLast()); | ||
|
|
||
| Msg msg = event.getMessage(); | ||
| if (msg != null) { | ||
| String text = extractTextFromMsg(msg); | ||
| if (text != null) { | ||
| payload.put("text", text); | ||
| } | ||
| if (msg.getRole() != null) { | ||
| payload.put("role", msg.getRole().name()); | ||
| } | ||
| if (msg.getName() != null) { | ||
| payload.put("name", msg.getName()); | ||
| } | ||
| if (msg.getMetadata() != null) { | ||
| String metadataJson = JsonUtils.getJsonCodec().toJson(msg.getMetadata()); | ||
| payload.put("metadata", new JSONObject(metadataJson)); | ||
| } | ||
| } | ||
|
|
||
| socket.emit("studioStreamEvent", payload); | ||
| } catch (Exception e) { | ||
| logger.error("Failed to send stream event to Studio", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Sends a stream completed event to Studio. | ||
| * | ||
| * <p>This method delivers a best-effort completion signal over the | ||
| * existing Socket.IO connection. If the underlying WebSocket is {@code null} | ||
| * or not connected, the method logs a warning and returns without throwing | ||
| * an exception and without attempting any reconnection. | ||
| */ | ||
| public void sendStreamCompleted() { | ||
| if (socket == null || !socket.connected()) { | ||
| logger.warn("Attempted to send streamCompleted while WebSocket is not connected"); | ||
| return; | ||
| } | ||
| try { | ||
| JSONObject payload = new JSONObject(); | ||
| payload.put("status", "completed"); | ||
| socket.emit("studioStreamCompleted", payload); | ||
| } catch (Exception e) { | ||
| logger.error("Failed to send streamCompleted event to Studio", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Extracts text content from a Msg for streaming to Studio. | ||
| */ | ||
| private String extractTextFromMsg(Msg msg) { | ||
| if (msg == null || msg.getContent() == null || msg.getContent().isEmpty()) { | ||
| return null; | ||
| } | ||
|
|
||
| StringBuilder builder = new StringBuilder(); | ||
| for (Object block : msg.getContent()) { | ||
| if (!(block instanceof ContentBlock)) { | ||
| continue; | ||
| } | ||
| if (block instanceof TextBlock) { | ||
| TextBlock tb = (TextBlock) block; | ||
| if (tb.getText() != null && !tb.getText().isEmpty()) { | ||
| appendWithNewline(builder, tb.getText()); | ||
| } | ||
| } else if (block instanceof ThinkingBlock) { | ||
| ThinkingBlock tb = (ThinkingBlock) block; | ||
| if (tb.getThinking() != null && !tb.getThinking().isEmpty()) { | ||
| appendWithNewline(builder, tb.getThinking()); | ||
| } | ||
| } else if (block instanceof ToolResultBlock) { | ||
| ToolResultBlock trb = (ToolResultBlock) block; | ||
| for (ContentBlock outputBlock : trb.getOutput()) { | ||
| if (outputBlock instanceof TextBlock) { | ||
| TextBlock otb = (TextBlock) outputBlock; | ||
| if (otb.getText() != null && !otb.getText().isEmpty()) { | ||
| appendWithNewline(builder, otb.getText()); | ||
| } | ||
| } else if (outputBlock instanceof ThinkingBlock) { | ||
| ThinkingBlock otb = (ThinkingBlock) outputBlock; | ||
| if (otb.getThinking() != null && !otb.getThinking().isEmpty()) { | ||
| appendWithNewline(builder, otb.getThinking()); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
Comment on lines
+480
to
+516
|
||
|
|
||
| return builder.length() > 0 ? builder.toString() : null; | ||
| } | ||
|
|
||
| private void appendWithNewline(StringBuilder builder, String text) { | ||
| if (builder.length() > 0) { | ||
| builder.append('\n'); | ||
| } | ||
| builder.append(text); | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
streamToStudio(...)has minimal Javadoc and is missing standard tags/behavioral notes (params, return, error conditions) that other public methods inStudioManagerinclude. Please expand the Javadoc to describe parameters (agent/input/options), what events are forwarded, and what the returned Mono represents.