feat: add streaming output for studio#565
Conversation
|
|
|
|
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
LearningGp
left a comment
LearningGp
left a comment
There was a problem hiding this comment.
The current implementation feels more like sending discrete messages (each chunk is a separate message) rather than streaming, which isn't quite what we expected.
Consider reusing StudioWebSocketClient to handle the streaming implementation.
Also, we need to filter out the isLast flag in the events to avoid duplicates.
Let us know if you'd like to discuss this further.
Thanks a lot for the detailed feedback, this is very helpful!
@LearningGp I‘m looking forward to any feedback. |
LGTM overall. Not sure if studio-side modifications are required, but feel free to proceed as designed. Let's focus on the final outcome. |
There was a problem hiding this comment.
Pull request overview
Adds a Studio-side bridge for AgentScope’s StreamableAgent.stream(...) API so that selected Event emissions can be forwarded to AgentScope Studio in real time via WebSocket, while persisting the final message via the existing HTTP client.
Changes:
- Added
StudioStreamingBridgeto forward aFlux<Event>to Studio WebSocket and persist a finalMsgviaStudioClient. - Extended
StudioWebSocketClientwithsendStreamEvent(...)/sendStreamCompleted()to emit new Socket.IO events. - Added a
StudioManager.streamToStudio(...)entry point and a new unit test class.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| agentscope-extensions/agentscope-extensions-studio/src/main/java/io/agentscope/core/studio/StudioWebSocketClient.java | Adds WebSocket emit methods for streaming events/completion (payload construction & text extraction). |
| agentscope-extensions/agentscope-extensions-studio/src/main/java/io/agentscope/core/studio/StudioStreamingBridge.java | New bridge that consumes Flux<Event> and forwards selected events to Studio, persisting a final message. |
| agentscope-extensions/agentscope-extensions-studio/src/main/java/io/agentscope/core/studio/StudioStreamingRunner.java | New runner wrapper around StreamableAgent.stream(...) + bridge forwarding. |
| agentscope-extensions/agentscope-extensions-studio/src/main/java/io/agentscope/core/studio/StudioManager.java | Adds streamToStudio(...) convenience method for streaming integration. |
| agentscope-extensions/agentscope-extensions-studio/src/test/java/io/agentscope/core/studio/StudioStreamingBridgeTest.java | Adds initial unit tests for streaming forwarding and completion/persistence behavior. |
| /** | ||
| * Sends a stream event to Studio in real-time. | ||
| */ | ||
| public void sendStreamEvent(Event event) { | ||
| if (socket == null || !socket.connected()) { | ||
| logger.warn("Attempted to send stream event while WebSocket is not connected"); | ||
| return; |
There was a problem hiding this comment.
sendStreamEvent/sendStreamCompleted Javadocs don’t follow the conventions used in this module (missing @param/@return details and behavioral notes such as what happens when the socket is disconnected). Please expand the Javadoc to include parameter description(s) and clarify whether these calls are best-effort or guaranteed delivery.
| private String extractTextFromMsg(Msg msg) { | ||
| if (msg == null || msg.getContent() == null || msg.getContent().isEmpty()) { | ||
| return null; | ||
| } | ||
| Object first = msg.getContent().get(0); | ||
| if (first instanceof TextBlock tb && tb.getText() != null) { | ||
| return tb.getText(); | ||
| } |
There was a problem hiding this comment.
extractTextFromMsg only reads the first content block and only supports TextBlock. In actual streaming events, reasoning commonly uses ThinkingBlock and tool results use ToolResultBlock (with text nested in its output blocks), so Studio will often receive stream events with no text field. Update extraction to handle the relevant block types (and consider concatenating multiple text-bearing blocks) so streamed reasoning/tool output is visible.
| if (type == EventType.AGENT_RESULT && !isLast) { | ||
| webSocketClient.sendStreamEvent(event); | ||
| return; | ||
| } | ||
|
|
||
| if (!isLast | ||
| && (type == EventType.REASONING | ||
| || type == EventType.TOOL_RESULT | ||
| || type == EventType.SUMMARY)) { | ||
| webSocketClient.sendStreamEvent(event); | ||
| } |
There was a problem hiding this comment.
handleEvent currently forwards REASONING / TOOL_RESULT / SUMMARY only when isLast == false. However, the core streaming implementation emits the final reasoning/tool messages with isLast == true (and may emit no chunks at all depending on StreamOptions), so these outputs can be dropped entirely. Consider forwarding these event types regardless of isLast, and reserve the special terminal-handling logic for the final AGENT_RESULT that ends the stream.
| public class StudioStreamingBridge { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(StudioStreamingBridge.class); | ||
|
|
||
| private final StudioClient studioClient; | ||
| private final StudioWebSocketClient webSocketClient; | ||
|
|
||
| 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"); | ||
| } |
There was a problem hiding this comment.
This class introduces a new public API but lacks class-level Javadoc and constructor Javadoc, while other public Studio integration classes document purpose/usage and public constructors. Please add class-level Javadoc (what is bridged, which events are forwarded, and how completion is signaled) and document the public constructor parameters.
| public class StudioStreamingRunner { | ||
|
|
||
| private final StudioStreamingBridge streamingBridge; | ||
|
|
||
| public StudioStreamingRunner(StudioStreamingBridge streamingBridge) { | ||
| this.streamingBridge = streamingBridge; | ||
| } | ||
|
|
||
| public Mono<Void> runWithStreaming(StreamableAgent agent, Msg input, StreamOptions options) { | ||
| StreamOptions effectiveOptions = | ||
| options != null ? options : StreamOptions.defaults(); | ||
|
|
||
| Flux<Event> eventFlux = agent.stream(input, effectiveOptions); | ||
| return streamingBridge.forwardToStudio(eventFlux); | ||
| } |
There was a problem hiding this comment.
StudioStreamingRunner is currently unused in the codebase (no references found), which adds public API surface without providing functionality. Either wire it into the integration entry points (e.g., StudioManager) or remove it to avoid maintaining an unused abstraction.
| public class StudioStreamingRunner { | ||
|
|
||
| private final StudioStreamingBridge streamingBridge; | ||
|
|
||
| public StudioStreamingRunner(StudioStreamingBridge streamingBridge) { | ||
| this.streamingBridge = streamingBridge; | ||
| } | ||
|
|
||
| public Mono<Void> runWithStreaming(StreamableAgent agent, Msg input, StreamOptions options) { | ||
| StreamOptions effectiveOptions = |
There was a problem hiding this comment.
This public class/constructor/method are missing Javadoc, which is inconsistent with the rest of the Studio integration module’s public APIs. Please add class-level Javadoc and document the constructor and runWithStreaming(...) parameters/behavior.
| /** | ||
| * Stream agent events to Studio. | ||
| */ | ||
| public static Mono<Void> streamToStudio(StreamableAgent agent, Msg input, StreamOptions options) { | ||
| if (!isInitialized() || client == null || wsClient == null) { | ||
| return Mono.error(new IllegalStateException("Studio is not initialized")); | ||
| } | ||
| StreamOptions effective = options != null ? options : StreamOptions.defaults(); | ||
| Flux<Event> events = agent.stream(input, effective); | ||
| StudioStreamingBridge bridge = new StudioStreamingBridge(client, wsClient); | ||
| return bridge.forwardToStudio(events); | ||
| } |
There was a problem hiding this comment.
streamToStudio(...) has minimal Javadoc and is missing standard tags/behavioral notes (params, return, error conditions) that other public methods in StudioManager include. Please expand the Javadoc to describe parameters (agent/input/options), what events are forwarded, and what the returned Mono represents.
| import static org.mockito.ArgumentMatchers.any; | ||
| import static org.mockito.Mockito.*; | ||
|
|
There was a problem hiding this comment.
Avoid wildcard imports (e.g., import static org.mockito.Mockito.*;). Please import only the specific Mockito members used in this test to keep imports explicit and consistent with typical project style checks.
| Event reasoning1 = new Event(EventType.REASONING, msg("thinking 1"), false); | ||
| Event reasoning2 = new Event(EventType.REASONING, msg("thinking 2"), false); | ||
|
|
||
| Event tool1 = new Event(EventType.TOOL_RESULT, msg("tool out 1"), false); | ||
|
|
||
| Event answerChunk1 = new Event(EventType.AGENT_RESULT, msg("hello "), false); | ||
| Event answerChunk2 = new Event(EventType.AGENT_RESULT, msg("world"), false); | ||
|
|
||
| Msg finalMsg = msg("hello world"); | ||
| Event answerFinal = new Event(EventType.AGENT_RESULT, finalMsg, true); | ||
|
|
There was a problem hiding this comment.
Current tests only use Msg with a single TextBlock, but real streaming events often carry ThinkingBlock for reasoning and ToolResultBlock for tool output (and the final reasoning/tool emissions are typically isLast == true). Add test cases that reflect these real event shapes so the streaming-to-Studio path is validated end-to-end for reasoning/tool output visibility.
|
@LearningGp I have submitted the code, could you please take another look when you are convenient❤️ |
|
Based on E2E validation, I've found that the current implementation doesn't achieve streaming output. While there are a few bugs to address, the primary blocker is that the Studio Server doesn't support streaming yet. Consequently, we’ll need to hold off on merging this PR for the time being. That said, would you be interested in contributing to the Studio project to help implement the necessary server-side support? |
Thanks for the clarification! I’d be happy to contribute to the Studio project and help implement the server-side streaming support. Please let me know what I need to do next.❤️ |
AgentScopeJavaBot
added
enhancement
AgentScope-Java Version
1.0.8
[The version of AgentScope-Java you are working on, e.g. 1.0.7, check your pom.xml dependency version or run
mvn dependency:tree | grep agentscope-parent:pom(only mac/linux)]Description
[Please describe the background, purpose, changes made, and how to test this PR]
Feat #416
Checklist
Please check the following items before code is ready to be reviewed.
mvn spotless:applymvn test)