Refactor streaming chunk model and visualizer

Author: enyst

README.md

@@ -145,17 +145,17 @@ llm = registry.get("default")
 
 You can receive incremental deltas from the Responses API by supplying a token
 callback when constructing a conversation. Each callback receives an
-``LLMStreamEvent`` describing the delta.
+``LLMStreamChunk`` describing the delta.
 
 ```python
 from pathlib import Path
-from openhands.sdk import Conversation, LLMStreamEvent
+from openhands.sdk import Conversation, LLMStreamChunk
 
 log_dir = Path("logs/stream")
 log_dir.mkdir(parents=True, exist_ok=True)
 
-def on_token(event: LLMStreamEvent) -> None:
-    print(event.text or event.arguments or "", end="", flush=True)
+def on_token(event: LLMStreamChunk) -> None:
+    print(event.text_delta or event.arguments_delta or "", end="", flush=True)
 
 conversation = Conversation(agent=agent, token_callbacks=[on_token])
 conversation.send_message("Summarize the benefits of token streaming.")

examples/01_standalone_sdk/24_responses_streaming.py

@@ -15,23 +15,28 @@
 
 from pydantic import SecretStr
 
-from openhands.sdk import Conversation, LLMStreamEvent, get_logger
+from openhands.sdk import Conversation, LLMStreamChunk, get_logger
 from openhands.sdk.llm import LLM
 from openhands.tools.preset.default import get_default_agent
 
 
+PRINT_STREAM_TO_STDOUT = False
+
+
 logger = get_logger(__name__)
 LOG_DIR = Path("logs/stream")
 
 
-def _serialize_event(event: LLMStreamEvent) -> dict[str, Any]:
+def _serialize_event(event: LLMStreamChunk) -> dict[str, Any]:
     record = {
         "type": event.type,
-        "text": event.text,
-        "arguments": event.arguments,
+        "part_kind": event.part_kind,
+        "text": event.text_delta,
+        "arguments": event.arguments_delta,
         "output_index": event.output_index,
         "content_index": event.content_index,
         "item_id": event.item_id,
+        "response_id": event.response_id,
         "is_final": event.is_final,
     }
     return record
@@ -58,15 +63,15 @@ def main() -> None:
     LOG_DIR.mkdir(parents=True, exist_ok=True)
     log_path = LOG_DIR / f"responses_stream_{timestamp}.jsonl"
 
-    def on_token(event: LLMStreamEvent) -> None:
+    def on_token(event: LLMStreamChunk) -> None:
         record = _serialize_event(event)
         with log_path.open("a", encoding="utf-8") as fp:
             fp.write(json.dumps(record) + "\n")
 
-        stream_chunk = event.text or event.arguments
-        if stream_chunk:
-            print(stream_chunk, end="", flush=True)
-        if event.is_final:
+        delta = event.text_delta or event.arguments_delta
+        if delta and PRINT_STREAM_TO_STDOUT:
+            print(delta, end="", flush=True)
+        if event.is_final and event.part_kind == "status" and PRINT_STREAM_TO_STDOUT:
             print("\n--- stream complete ---")
 
     conversation = Conversation(
@@ -83,7 +88,7 @@ def on_token(event: LLMStreamEvent) -> None:
     conversation.run()
 
     cleanup_prompt = (
-        "Thank you. Please delete streaming_story.md now that I've read it, "
+        "Thank you. Please delete the streaming story file now that I've read it, "
         "then confirm the deletion."
     )
     conversation.send_message(cleanup_prompt)

openhands/sdk/__init__.py

@@ -20,7 +20,7 @@
     LLM,
     ImageContent,
     LLMRegistry,
-    LLMStreamEvent,
+    LLMStreamChunk,
     Message,
     RedactedThinkingBlock,
     RegistryEvent,
@@ -60,7 +60,7 @@
 __all__ = [
     "LLM",
     "LLMRegistry",
-    "LLMStreamEvent",
+    "LLMStreamChunk",
     "TokenCallbackType",
     "ConversationStats",
     "RegistryEvent",

openhands/sdk/conversation/impl/local_conversation.py

@@ -20,7 +20,7 @@
     StreamingDeltaEvent,
     UserRejectObservation,
 )
-from openhands.sdk.llm import LLM, LLMStreamEvent, Message, TextContent
+from openhands.sdk.llm import LLM, LLMStreamChunk, Message, TextContent
 from openhands.sdk.llm.llm_registry import LLMRegistry
 from openhands.sdk.logger import get_logger
 from openhands.sdk.security.confirmation_policy import (
@@ -129,15 +129,15 @@ def _composed(event):
             _compose_token_callbacks(token_callbacks) if token_callbacks else None
         )
 
-        def _handle_stream_event(stream_event: LLMStreamEvent) -> None:
+        def _handle_stream_event(stream_chunk: LLMStreamChunk) -> None:
             try:
                 self._on_event(
-                    StreamingDeltaEvent(source="agent", stream_event=stream_event)
+                    StreamingDeltaEvent(source="agent", stream_chunk=stream_chunk)
                 )
             except Exception:
                 logger.exception("stream_event_processing_error", exc_info=True)
             if user_token_callback:
-                user_token_callback(stream_event)
+                user_token_callback(stream_chunk)
 
         self._on_token = _handle_stream_event