Streaming restaurant sample

samples/agent/adk/restaurant_finder/agent.py

@@ -27,6 +27,7 @@
     Part,
     TextPart,
 )
+from google.adk.agents import run_config
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.artifacts import InMemoryArtifactService
 from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
@@ -45,7 +46,11 @@
 from a2ui.core.parser.parser import parse_response, ResponsePart
 from a2ui.basic_catalog.provider import BasicCatalog
 from a2ui.core.schema.common_modifiers import remove_strict_validation
-from a2ui.a2a import create_a2ui_part, get_a2ui_agent_extension, parse_response_to_parts
+from a2ui.a2a import (
+    get_a2ui_agent_extension,
+    parse_response_to_parts,
+    stream_response_to_parts,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -63,6 +68,7 @@ def __init__(self, base_url: str):
 
     self._schema_managers: Dict[str, A2uiSchemaManager] = {}
     self._ui_runners: Dict[str, Runner] = {}
+    self._parsers = {}
 
     for version in [VERSION_0_8, VERSION_0_9]:
       schema_manager = self._build_schema_manager(version)
@@ -229,45 +235,46 @@ async def stream(
       current_message = types.Content(
           role="user", parts=[types.Part.from_text(text=current_query_text)]
       )
-      final_response_content = None
 
-      async for event in runner.run_async(
-          user_id=self._user_id,
-          session_id=session.id,
-          new_message=current_message,
-      ):
-        logger.info(f"Event from runner: {event}")
-        if event.is_final_response():
-          if event.content and event.content.parts and event.content.parts[0].text:
-            final_response_content = "\n".join(
-                [p.text for p in event.content.parts if p.text]
-            )
-          break  # Got the final response, stop consuming events
-        else:
-          logger.info(f"Intermediate event: {event}")
-          # Yield intermediate updates on every attempt
+      full_content_list = []
+
+      async def token_stream():
+        async for event in runner.run_async(
+            user_id=self._user_id,
+            session_id=session.id,
+            run_config=run_config.RunConfig(
+                streaming_mode=run_config.StreamingMode.SSE
+            ),
+            new_message=current_message,
+        ):
+          if event.content and event.content.parts:
+            for p in event.content.parts:
+              if p.text:
+                full_content_list.append(p.text)
+                yield p.text
+
+      if selected_catalog:
+        from a2ui.core.parser.streaming import A2uiStreamParser
+
+        if session_id not in self._parsers:
+          self._parsers[session_id] = A2uiStreamParser(catalog=selected_catalog)
+
+        async for part in stream_response_to_parts(
+            self._parsers[session_id],
+            token_stream(),
+        ):
+          yield {
+              "is_task_complete": False,
+              "parts": [part],
+          }
+      else:
+        async for token in token_stream():
           yield {
               "is_task_complete": False,
-              "updates": self.get_processing_message(),
+              "updates": token,
           }
 
-      if final_response_content is None:
-        logger.warning(
-            "--- RestaurantAgent.stream: Received no final response content from"
-            f" runner (Attempt {attempt}). ---"
-        )
-        if attempt <= max_retries:
-          current_query_text = (
-              "I received no response. Please try again."
-              f"Please retry the original request: '{query}'"
-          )
-          continue  # Go to next retry
-        else:
-          # Retries exhausted on no-response
-          final_response_content = (
-              "I'm sorry, I encountered an error and couldn't process your request."
-          )
-          # Fall through to send this as a text-only error
+      final_response_content = "".join(full_content_list)
 
       is_valid = False
       error_message = ""

samples/agent/adk/restaurant_finder/agent_executor.py

@@ -124,10 +124,14 @@ async def execute(
     async for item in self._agent.stream(query, task.context_id, active_ui_version):
       is_task_complete = item["is_task_complete"]
       if not is_task_complete:
-        await updater.update_status(
-            TaskState.working,
-            new_agent_text_message(item["updates"], task.context_id, task.id),
-        )
+        message = None
+        if "parts" in item:
+          message = new_agent_parts_message(item["parts"], task.context_id, task.id)
+        elif "updates" in item:
+          message = new_agent_text_message(item["updates"], task.context_id, task.id)
+
+        if message:
+          await updater.update_status(TaskState.working, message)
         continue
 
       final_state = (

samples/client/angular/projects/restaurant/src/app/app.css

@@ -225,3 +225,51 @@ form {
     rotate: 360deg;
   }
 }
+
+.streaming-indicator {
+  display: flex;
+  flex-direction: column;
+  gap: 8px;
+  padding: 16px;
+  background: light-dark(var(--n-95), var(--n-20));
+  border-radius: 8px;
+  margin-bottom: 16px;
+  animation: fadeIn 0.5s ease-out;
+
+  & span {
+    font-size: 14px;
+    color: light-dark(var(--n-40), var(--n-70));
+    font-weight: 500;
+  }
+}
+
+.progress-bar {
+  width: 100%;
+  height: 4px;
+  background-color: light-dark(var(--n-90), var(--n-30));
+  border-radius: 2px;
+  overflow: hidden;
+  position: relative;
+
+  &::after {
+    content: "";
+    position: absolute;
+    left: 0;
+    top: 0;
+    height: 100%;
+    width: 30%;
+    background-color: var(--p-60);
+    border-radius: 2px;
+    animation: loading-bar 2s infinite ease-in-out;
+  }
+}
+
+@keyframes loading-bar {
+  0% {
+    left: -30%;
+  }
+
+  100% {
+    left: 100%;
+  }
+}

samples/client/angular/projects/restaurant/src/app/app.html

@@ -14,41 +14,43 @@
  limitations under the License.
  -->
 
-@if (client.isLoading()) {
+@if (!hasData()) {
+<form (submit)="handleSubmit($event)">
+  <img class="hero-img" src="hero-img" src="hero.png" alt="Image of the restaurant">
+
+  <h1 class="app-title">Restaurant Finder</h1>
+  <div>
+    <input required value="Top 5 Chinese restaurants in New York." autocomplete="off" id="body" name="body" type="text"
+      [disabled]="client.isLoading()" />
+    <button type="submit" [disabled]="client.isLoading()">
+      <span class="g-icon filled-heavy">send</span>
+    </button>
+  </div>
+</form>
+} @else {
+<div class="surfaces">
+  @let surfaces = processor.getSurfaces();
+
+  @if (client.isLoading() && surfaces.size === 0) {
   <div class="pending">
     <div class="spinner"></div>
     <div class="loading-text">{{loadingTextLines[loadingTextIndex()]}}</div>
   </div>
-} @else if (!hasData()) {
-  <form (submit)="handleSubmit($event)">
-    <img class="hero-img" src="hero-img" src="hero.png" alt="Image of the restaurant">
-
-    <h1 class="app-title">Restaurant Finder</h1>
-    <div>
-      <input
-        required
-        value="Top 5 Chinese restaurants in New York."
-        autocomplete="off"
-        id="body"
-        name="body"
-        type="text"
-        [disabled]="client.isLoading()"
-      />
-      <button type="submit" [disabled]="client.isLoading()">
-        <span class="g-icon filled-heavy">send</span>
-      </button>
-    </div>
-  </form>
-} @else {
-  <div class="surfaces">
-    @let surfaces = processor.getSurfaces();
-
-    @for (entry of surfaces; track entry[0]) {
-      <a2ui-surface [surfaceId]="entry[0]" [surface]="entry[1]"/>
-    }
+  } @else {
+  @if (client.isLoading()) {
+  <div class="streaming-indicator">
+    <div class="progress-bar"></div>
+    <span>Finding the best spots for you...</span>
   </div>
+  }
+
+  @for (entry of surfaces; track entry[0]) {
+  <a2ui-surface [surfaceId]="entry[0]" [surface]="entry[1]" />
+  }
+  }
+</div>
 }
 
 <button (click)="toggleTheme(toggleButton)" class="theme-toggle" #toggleButton>
   <span class="g-icon filled-heavy"></span>
-</button>
+</button>

samples/client/angular/projects/restaurant/src/app/app.ts

@@ -54,8 +54,8 @@ export class App {
     if (body) {
       this.startLoadingAnimation();
       const message = body as Types.A2UIClientEventMessage;
-      await this.client.makeRequest(message);
       this.hasData.set(true);
+      await this.client.makeRequest(message);
       this.stopLoadingAnimation();
     }
   }