Slight tweaks and improvements

Author: ericallam

README.md

@@ -1,4 +1,4 @@
-## Trigger.dev Realtime Demo
+## Trigger.dev Realtime streams Demo
 
 This demo shows off a variety of features of [Trigger.dev Realtime](https://trigger.dev/docs/realtime)
 
@@ -16,7 +16,23 @@ npm run dev
 npm run dev:trigger # Run this in a separate terminal
 ```
 
-Open [http://localhost:3000](http://localhost:3000) with your browser to see the result.
+Open [http://localhost:3000/weather](http://localhost:3000/weather) with your browser to see the result.
+
+## Realtime + ai SDK + streams
+
+<video
+  src="https://content.trigger.dev/realtime-streams-demo.mp4"
+  preload="auto"
+  controls={true}
+  loop
+  muted
+  autoPlay={true}
+  width="100%"
+  height="100%"
+/>
+
+- View the Trigger.dev task code in the [src/trigger/ai.ts](src/trigger/ai.ts) file.
+- View the frontend code in the [src/app/weather/page.tsx](src/app/weather/page.tsx) file and the [src/components/GetStreamingWeatherButton.tsx](src/components/GetStreamingWeatherButton.tsx) file.
 
 ## Learn More
 

src/app/page.tsx

@@ -1,7 +1,7 @@
 import RunButton from "@/components/RunButton";
 import BatchRunButton from "@/components/BatchRunButton";
 import TriggerButton from "@/components/TriggerButton";
-import TriggerButtonWithStreaming from "@/components/TriggerButtonWithStreaming";
+import GetStreamingWeatherButton from "@/components/GetStreamingWeatherButton";
 import { ImageUploadDropzone } from "@/components/ImageUploadButton";
 import { auth } from "@trigger.dev/sdk/v3";
 
@@ -36,7 +36,7 @@ export default async function Home() {
         <RunButton />
         <BatchRunButton />
         <TriggerButton accessToken={publicAccessToken} />
-        <TriggerButtonWithStreaming accessToken={publicAccessToken} />
+        <GetStreamingWeatherButton accessToken={publicAccessToken} />
       </div>
     </main>
   );

src/app/weather/page.tsx

@@ -1,4 +1,4 @@
-import TriggerButtonWithStreaming from "@/components/TriggerButtonWithStreaming";
+import GetStreamingWeatherButton from "@/components/GetStreamingWeatherButton";
 import { auth } from "@trigger.dev/sdk/v3";
 
 export default async function WeatherPage() {
@@ -13,7 +13,7 @@ export default async function WeatherPage() {
   return (
     <main className="grid grid-rows-[1fr_auto] min-h-screen items-center justify-center w-full bg-gray-900">
       <div className="flex flex-col space-y-8">
-        <TriggerButtonWithStreaming accessToken={publicAccessToken} />
+        <GetStreamingWeatherButton accessToken={publicAccessToken} />
       </div>
     </main>
   );

src/components/TriggerButtonWithStreaming.tsx renamed to src/components/GetStreamingWeatherButton.tsx

@@ -1,7 +1,7 @@
 "use client";
 
 import { Button } from "@/components/ui/button";
-import type { STREAMS, openaiStreaming } from "@/trigger/ai";
+import type { STREAMS, aiWeather } from "@/trigger/ai";
 import { useRealtimeTaskTriggerWithStreams } from "@trigger.dev/react-hooks";
 import { useCallback, useState } from "react";
 import {
@@ -14,17 +14,17 @@ import {
 } from "./ui/dialog";
 import { Card, CardContent, CardFooter } from "./ui/card";
 
-export default function TriggerButton({
+export default function GetStreamingWeather({
   accessToken,
 }: {
   accessToken: string;
 }) {
   const [isOpen, setIsOpen] = useState(false);
 
   const { submit, isLoading, run, streams } = useRealtimeTaskTriggerWithStreams<
-    typeof openaiStreaming,
+    typeof aiWeather,
     STREAMS
-  >("openai-streaming", {
+  >("ai-weather", {
     accessToken,
     baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
     experimental_throttleInMs: 100,

src/trigger/ai.ts

@@ -1,87 +1,12 @@
 import { openai } from "@ai-sdk/openai";
-import {
-  logger,
-  metadata,
-  runs,
-  schemaTask,
-  task,
-  toolTask,
-  wait,
-} from "@trigger.dev/sdk/v3";
-import { streamText, type TextStreamPart, tool } from "ai";
-import { setTimeout } from "node:timers/promises";
+import { logger, metadata, schemaTask, toolTask } from "@trigger.dev/sdk/v3";
+import { streamText, type TextStreamPart } from "ai";
 import { z } from "zod";
 
 export type STREAMS = {
   openai: TextStreamPart<{ getWeather: typeof weatherTask.tool }>;
 };
 
-export const openaiConsumer = schemaTask({
-  id: "openai-consumer",
-  schema: z.object({
-    model: z.string().default("gpt-3.5-turbo"),
-    prompt: z.string().default("Hello, how are you?"),
-  }),
-  run: async ({ model, prompt }) => {
-    const handle = await openaiStreaming.trigger({ model, prompt });
-
-    let openaiCompletion = "";
-
-    for await (const part of runs
-      .subscribeToRun(handle)
-      .withStreams<STREAMS>()) {
-      switch (part.type) {
-        case "run": {
-          logger.info("Received run chunk", { run: part.run });
-          break;
-        }
-        case "openai": {
-          logger.info("Received OpenAI chunk", {
-            chunk: part.chunk,
-            run: part.run,
-          });
-
-          switch (part.chunk.type) {
-            case "text-delta": {
-              openaiCompletion += part.chunk.textDelta;
-              break;
-            }
-            case "tool-call": {
-              switch (part.chunk.toolName) {
-                case "getWeather": {
-                  console.log("Calling getWeather tool with args", {
-                    args: part.chunk.args,
-                  });
-                }
-              }
-              break;
-            }
-            case "tool-result": {
-              switch (part.chunk.toolName) {
-                case "getWeather": {
-                  console.log("Received getWeather tool result", {
-                    result: part.chunk.result,
-                  });
-                }
-              }
-              break;
-            }
-          }
-        }
-      }
-    }
-
-    return { openaiCompletion };
-  },
-});
-
-export const waitUntilExamples = task({
-  id: "wait-until-examples",
-  run: async () => {
-    await setTimeout(30_000);
-  },
-});
-
 export const weatherTask = toolTask({
   id: "weather",
   description: "Get the weather for a location",
@@ -97,11 +22,11 @@ export const weatherTask = toolTask({
   },
 });
 
-export const openaiStreaming = schemaTask({
-  id: "openai-streaming",
-  description: "Stream data from OpenAI to get the weather",
+export const aiWeather = schemaTask({
+  id: "ai-weather",
+  description: "Send the fullStream from the ai SDK to the metadata system",
   schema: z.object({
-    model: z.string().default("chatgpt-4o-latest"),
+    model: z.string().default("gpt-4o-mini"),
     prompt: z.string().default("Hello, how are you?"),
   }),
   run: async ({ model, prompt }) => {
@@ -131,33 +56,3 @@ export const openaiStreaming = schemaTask({
     return { text };
   },
 });
-
-export const openaiO1Model = schemaTask({
-  id: "openai-o1-model",
-  description: "Stream data from OpenAI to get the weather",
-  schema: z.object({
-    model: z.string().default("o1-preview"),
-    prompt: z.string().default("Hello, how are you?"),
-  }),
-  run: async ({ model, prompt }) => {
-    logger.info("Running OpenAI model", { model, prompt });
-
-    const result = await streamText({
-      model: openai(model),
-      prompt,
-      experimental_continueSteps: true,
-    });
-
-    const stream = await metadata.stream("openai", result.textStream);
-
-    let text = "";
-
-    for await (const chunk of stream) {
-      logger.log("Received chunk", { chunk });
-
-      text += chunk;
-    }
-
-    return { text };
-  },
-});