From af03d4eba99a1cb14ce300811fd8c12952065435 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Fri, 17 Jul 2026 13:00:37 -0700 Subject: [PATCH 1/5] feat(ai-sdk): add streamText/streamObject streaming sample and bump to AI SDK v7 Adds streamingAgent and streamObjectAgent that publish deltas to a WorkflowStream topic, a subscribing client consumer, and an offline test. Migrates workflow imports to the @temporalio/ai-sdk/workflow subpath and bumps ai/@ai-sdk deps to v7. Co-Authored-By: Claude Opus 4.8 (1M context) --- ai-sdk/README.md | 13 +++ ai-sdk/package.json | 11 +- ai-sdk/src/client.ts | 51 +++++++- ai-sdk/src/mocha/streaming.test.ts | 179 +++++++++++++++++++++++++++++ ai-sdk/src/workflows.ts | 94 +++++++++++++-- ai-sdk/tsconfig.json | 6 + 6 files changed, 341 insertions(+), 13 deletions(-) create mode 100644 ai-sdk/src/mocha/streaming.test.ts diff --git a/ai-sdk/README.md b/ai-sdk/README.md index 7ce2bf8ab..768dc0192 100644 --- a/ai-sdk/README.md +++ b/ai-sdk/README.md @@ -15,3 +15,16 @@ This project demonstrates some uses of the AI SDK inside Temporal. 1. `npm run workflow tools` 1. `npm run workflow mcp` 1. `npm run workflow middleware` +1. `npm run workflow stream` +1. `npm run workflow stream-object` + +### Streaming + +The `stream` and `stream-object` samples show how to stream model output out of a +Workflow. `streamText` (and `streamObject` for structured output) run the model call in +an activity that publishes each delta onto a +[Workflow Stream](https://docs.temporal.io/develop/typescript/workflows/workflow-streams) +topic. The Workflow hosts the stream (`new WorkflowStream()`) and durably reassembles the +final result, while external consumers subscribe by Workflow id to render the tokens live +as they arrive — see `src/client.ts`. Both stream functions flow through the same model +path, so no extra wiring is needed beyond choosing a `streamingTopic`. diff --git a/ai-sdk/package.json b/ai-sdk/package.json index 3270f5316..b6def7caa 100644 --- a/ai-sdk/package.json +++ b/ai-sdk/package.json @@ -23,17 +23,18 @@ ] }, "dependencies": { - "@ai-sdk/openai": "^3.0.0", - "@ai-sdk/provider": "^3.0.0", - "@ai-sdk/mcp": "^1.0.0", - "@modelcontextprotocol/sdk": "^1.10.2", + "@ai-sdk/openai": "^4.0.0", + "@ai-sdk/provider": "^4.0.0", + "@ai-sdk/mcp": "^2.0.0", + "@modelcontextprotocol/sdk": "^1.25.2", "@temporalio/activity": "^1.20.3", "@temporalio/ai-sdk": "^1.20.3", "@temporalio/client": "^1.20.3", "@temporalio/envconfig": "^1.20.3", "@temporalio/worker": "^1.20.3", "@temporalio/workflow": "^1.20.3", - "ai": "^6.0.0", + "@temporalio/workflow-streams": "^1.20.3", + "ai": "^7.0.0", "nanoid": "3.x", "zod": "^3.25.76" }, diff --git a/ai-sdk/src/client.ts b/ai-sdk/src/client.ts index 9671725c9..a7227905a 100644 --- a/ai-sdk/src/client.ts +++ b/ai-sdk/src/client.ts @@ -1,8 +1,33 @@ import { Connection, Client } from '@temporalio/client'; import { loadClientConnectConfig } from '@temporalio/envconfig'; -import { haikuAgent, mcpAgent, middlewareAgent, toolsAgent } from './workflows'; +import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; +import { + haikuAgent, + mcpAgent, + middlewareAgent, + streamingAgent, + streamObjectAgent, + toolsAgent, + STREAM_TOPIC, + OBJECT_STREAM_TOPIC, +} from './workflows'; import { nanoid } from 'nanoid'; +// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-consumer +// Subscribe to a Workflow's stream topic and render each text delta live as it +// is published by the streaming activity. Each item's payload is the +// JSON-encoded AI SDK stream part; `resultType: true` decodes it to raw bytes. +async function renderStream(client: Client, workflowId: string, topic: string): Promise { + const streamClient = WorkflowStreamClient.create(client, workflowId); + for await (const item of streamClient.subscribe(topic, 0, { resultType: true })) { + const part = JSON.parse(new TextDecoder().decode(item.data)); + if (part.type === 'text-delta') process.stdout.write(part.delta); + if (part.type === 'finish') break; + } + process.stdout.write('\n'); +} +// @@@SNIPEND + async function run() { const args = process.argv; const workflow = args[2] ?? 'haiku'; @@ -14,6 +39,30 @@ async function run() { let handle; switch (workflow) { + case 'stream': { + const streamHandle = await client.workflow.start(streamingAgent, { + taskQueue: 'ai-sdk', + args: ['Temporal'], + workflowId: 'workflow-' + nanoid(), + }); + console.log(`Started workflow ${streamHandle.workflowId}`); + await renderStream(client, streamHandle.workflowId, STREAM_TOPIC); + console.log(await streamHandle.result()); + await connection.close(); + return; + } + case 'stream-object': { + const streamHandle = await client.workflow.start(streamObjectAgent, { + taskQueue: 'ai-sdk', + args: ['Generate a lasagna recipe.'], + workflowId: 'workflow-' + nanoid(), + }); + console.log(`Started workflow ${streamHandle.workflowId}`); + await renderStream(client, streamHandle.workflowId, OBJECT_STREAM_TOPIC); + console.log(await streamHandle.result()); + await connection.close(); + return; + } case 'middleware': handle = await client.workflow.start(middlewareAgent, { taskQueue: 'ai-sdk', diff --git a/ai-sdk/src/mocha/streaming.test.ts b/ai-sdk/src/mocha/streaming.test.ts new file mode 100644 index 000000000..2621818f7 --- /dev/null +++ b/ai-sdk/src/mocha/streaming.test.ts @@ -0,0 +1,179 @@ +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { after, before, describe, it } from 'mocha'; +import { Worker } from '@temporalio/worker'; +import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; +import type { + LanguageModelV4, + LanguageModelV4CallOptions, + LanguageModelV4GenerateResult, + LanguageModelV4StreamPart, + LanguageModelV4StreamResult, + ProviderV4, +} from '@ai-sdk/provider'; +import assert from 'assert'; +import { streamingAgent, streamObjectAgent, STREAM_TOPIC, OBJECT_STREAM_TOPIC } from '../workflows'; +import * as activities from '../activities'; +import { AiSdkPlugin } from '@temporalio/ai-sdk'; + +// A deterministic, offline model that streams a fixed set of text deltas so +// these tests need no OPENAI_API_KEY and can assert on exact output. +class MockStreamModel implements LanguageModelV4 { + readonly specificationVersion = 'v4'; + readonly provider = 'mock'; + readonly modelId = 'mock-model'; + private readonly chunks: string[]; + + constructor(chunks: string[]) { + this.chunks = chunks; + } + + get supportedUrls(): Record { + return {}; + } + + doGenerate(_options: LanguageModelV4CallOptions): Promise { + throw new Error('generate not supported by mock'); + } + + doStream(_options: LanguageModelV4CallOptions): Promise { + const chunks = this.chunks; + const parts: LanguageModelV4StreamPart[] = [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 't1' }, + ...chunks.map((delta): LanguageModelV4StreamPart => ({ type: 'text-delta', id: 't1', delta })), + { type: 'text-end', id: 't1' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: undefined }, + usage: { + inputTokens: { total: 1, noCache: undefined, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: chunks.length, text: undefined, reasoning: undefined }, + }, + }, + ]; + return Promise.resolve({ + stream: new ReadableStream({ + start(controller) { + for (const part of parts) controller.enqueue(part); + controller.close(); + }, + }), + request: {}, + response: {}, + }); + } +} + +function mockProvider(chunks: string[]): ProviderV4 { + return { + specificationVersion: 'v4', + languageModel: () => new MockStreamModel(chunks), + embeddingModel: () => { + throw new Error('not implemented'); + }, + imageModel: () => { + throw new Error('not implemented'); + }, + }; +} + +// Collect the live deltas an external subscriber sees on a topic. +async function collectDeltas(client: any, workflowId: string, topic: string): Promise { + const deltas: string[] = []; + const streamClient = WorkflowStreamClient.create(client, workflowId); + for await (const item of streamClient.subscribe(topic, 0, { resultType: true })) { + const part = JSON.parse(new TextDecoder().decode(item.data)); + if (part.type === 'text-delta') deltas.push(part.delta); + if (part.type === 'finish') break; + } + return deltas; +} + +describe('streaming agents', function () { + this.timeout(30_000); + + let testEnv: TestWorkflowEnvironment; + + before(async () => { + testEnv = await TestWorkflowEnvironment.createLocal(); + }); + + after(async () => { + await testEnv?.teardown(); + }); + + it('streamingAgent publishes live deltas and returns the full text', async () => { + const { client, nativeConnection } = testEnv; + const taskQueue = 'test-stream-text'; + const chunks = ['Dur', 'able ', 'streams ', 'of ', 'thought']; + + const worker = await Worker.create({ + connection: nativeConnection, + plugins: [new AiSdkPlugin({ modelProvider: mockProvider(chunks) })], + taskQueue, + workflowsPath: require.resolve('../workflows'), + activities, + }); + + await worker.runUntil(async () => { + const handle = await client.workflow.start(streamingAgent, { + args: ['Temporal'], + workflowId: 'test-stream-text-' + Date.now(), + taskQueue, + }); + + const deltasPromise = collectDeltas(client, handle.workflowId, STREAM_TOPIC); + const result = await handle.result(); + const deltas = await deltasPromise; + + // The workflow durably reassembles the full text from the replayed stream. + assert.strictEqual(result, chunks.join('')); + // The external subscriber saw the response arrive incrementally. + assert.ok(deltas.length > 1, `expected multiple live deltas, got ${deltas.length}`); + assert.strictEqual(deltas.join(''), chunks.join('')); + }); + }); + + it('streamObjectAgent publishes JSON deltas and returns the parsed object', async () => { + const { client, nativeConnection } = testEnv; + const taskQueue = 'test-stream-object'; + // A valid JSON document matching streamObjectAgent's schema, split so the + // subscriber observes it building up incrementally. + const object = { + recipe: { + name: 'Classic Lasagna', + ingredients: [{ name: 'pasta', amount: '1 box' }], + steps: ['Layer', 'Bake'], + }, + }; + const json = JSON.stringify(object); + const chunks = json.match(/.{1,12}/gs) ?? [json]; + + const worker = await Worker.create({ + connection: nativeConnection, + plugins: [new AiSdkPlugin({ modelProvider: mockProvider(chunks) })], + taskQueue, + workflowsPath: require.resolve('../workflows'), + activities, + }); + + await worker.runUntil(async () => { + const handle = await client.workflow.start(streamObjectAgent, { + args: ['Generate a lasagna recipe.'], + workflowId: 'test-stream-object-' + Date.now(), + taskQueue, + }); + + const deltasPromise = collectDeltas(client, handle.workflowId, OBJECT_STREAM_TOPIC); + const result = await handle.result(); + const deltas = await deltasPromise; + + // The workflow durably resolves the final, validated object. + assert.strictEqual(result, object.recipe.name); + // The external subscriber saw the JSON build up incrementally, and the + // concatenated deltas reconstruct the full document. + assert.ok(deltas.length > 1, `expected multiple live deltas, got ${deltas.length}`); + assert.strictEqual(deltas.join(''), json); + }); + }); +}); diff --git a/ai-sdk/src/workflows.ts b/ai-sdk/src/workflows.ts index 943c8cd0c..23cf65414 100644 --- a/ai-sdk/src/workflows.ts +++ b/ai-sdk/src/workflows.ts @@ -1,10 +1,10 @@ -import '@temporalio/ai-sdk/lib/load-polyfills'; -import { generateText, stepCountIs, tool, wrapLanguageModel } from 'ai'; -import { TemporalMCPClient, temporalProvider } from '@temporalio/ai-sdk'; +import { generateText, streamText, streamObject, stepCountIs, tool, wrapLanguageModel } from 'ai'; +import type { LanguageModelMiddleware } from 'ai'; +import { TemporalMCPClient, TemporalProvider, temporalProvider } from '@temporalio/ai-sdk/workflow'; +import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import type * as activities from './activities'; -import { proxyActivities } from '@temporalio/workflow'; +import { proxyActivities, sleep } from '@temporalio/workflow'; import z from 'zod'; -import { LanguageModelV3Middleware } from '@ai-sdk/provider'; const { getWeather } = proxyActivities({ startToCloseTimeout: '1 minute', @@ -45,8 +45,8 @@ export async function toolsAgent(question: string): Promise { // @@@SNIPSTART typescript-vercel-ai-sdk-middleware-agent export async function middlewareAgent(prompt: string): Promise { const cache = new Map(); - const middleware: LanguageModelV3Middleware = { - specificationVersion: 'v3', + const middleware: LanguageModelMiddleware = { + specificationVersion: 'v4', wrapGenerate: async ({ doGenerate, params }) => { const cacheKey = JSON.stringify(params); if (cache.has(cacheKey)) { @@ -88,3 +88,83 @@ export async function mcpAgent(prompt: string): Promise { return result.text; } // @@@SNIPEND + +// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-topic +// The topic that streamed model deltas are published to. External consumers +// subscribe to this topic by workflow id to receive live tokens as they are +// generated. See `client.ts` for the consumer side. +export const STREAM_TOPIC = 'text-stream'; +// @@@SNIPEND + +// A provider whose language-model calls stream their deltas onto STREAM_TOPIC. +// Setting `streamingTopic` enables `doStream`; without it, streaming calls +// throw. Use a distinct topic per concurrent streaming call. +const streamingProvider = new TemporalProvider({ + languageModel: { streamingTopic: STREAM_TOPIC }, +}); + +// @@@SNIPSTART typescript-vercel-ai-sdk-streaming-agent +export async function streamingAgent(prompt: string): Promise { + // Host the WorkflowStream as the first statement of the workflow so its + // publish-signal handler is registered before the streaming activity starts + // publishing deltas to it. + new WorkflowStream(); + + const result = streamText({ + model: streamingProvider.languageModel('gpt-4o-mini'), + prompt, + system: 'You only respond in haikus.', + }); + + // The model call runs in an activity that publishes each delta to + // STREAM_TOPIC for external subscribers. Inside the workflow the deltas are + // replayed after the activity completes, so this loop durably reassembles + // the full text. + let text = ''; + for await (const delta of result.textStream) { + text += delta; + } + + // Briefly hold the run open so a subscriber's final poll can deliver the + // last deltas before the workflow exits and its in-memory stream log is gone. + await sleep('500 milliseconds'); + return text; +} +// @@@SNIPEND + +// A provider for streaming structured output onto its own topic. `streamObject` +// flows through the same `doStream` path as `streamText`, so no extra wiring is +// needed — only a distinct topic so the two streams stay separable. +export const OBJECT_STREAM_TOPIC = 'object-stream'; +const objectStreamingProvider = new TemporalProvider({ + languageModel: { streamingTopic: OBJECT_STREAM_TOPIC }, +}); + +// @@@SNIPSTART typescript-vercel-ai-sdk-stream-object-agent +export async function streamObjectAgent(prompt: string): Promise { + new WorkflowStream(); + + const result = streamObject({ + model: objectStreamingProvider.languageModel('gpt-4o-mini'), + schema: z.object({ + recipe: z.object({ + name: z.string(), + ingredients: z.array(z.object({ name: z.string(), amount: z.string() })), + steps: z.array(z.string()), + }), + }), + prompt, + }); + + // External subscribers see the object build up incrementally via the partial + // JSON deltas published to OBJECT_STREAM_TOPIC. The workflow drains the + // partial stream and durably resolves the final, validated object. + for await (const _partial of result.partialObjectStream) { + // Draining drives the stream; the consumer renders the live partials. + } + const object = await result.object; + + await sleep('500 milliseconds'); + return object.recipe.name; +} +// @@@SNIPEND diff --git a/ai-sdk/tsconfig.json b/ai-sdk/tsconfig.json index 488f2c62a..ec1db90a8 100644 --- a/ai-sdk/tsconfig.json +++ b/ai-sdk/tsconfig.json @@ -3,6 +3,12 @@ "version": "5.6.3", "compilerOptions": { "lib": ["es2021"], + // The Vercel AI SDK v7 (`ai`) is ESM-only. Use classic CommonJS module + // resolution (matching the @temporalio/ai-sdk build) so it can be required + // from this CommonJS sample and the `@temporalio/ai-sdk/workflow` subpath + // resolves via `typesVersions`. + "module": "commonjs", + "moduleResolution": "node", "declaration": true, "declarationMap": true, "sourceMap": true, From f662d20a7b3c24399fa3de1187640117fad67752 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 23 Jul 2026 11:57:12 -0700 Subject: [PATCH 2/5] chore(ai-sdk): exclude from shared tsconfig copy The ai-sdk sample needs custom module settings (module: commonjs, moduleResolution: node) so the ESM-only AI SDK v7 works from this CommonJS sample. Add it to TSCONFIG_EXCLUDE so copy-shared-files doesn't clobber ai-sdk/tsconfig.json on push. Co-Authored-By: Claude Opus 4.8 (1M context) --- .scripts/copy-shared-files.mjs | 1 + 1 file changed, 1 insertion(+) diff --git a/.scripts/copy-shared-files.mjs b/.scripts/copy-shared-files.mjs index 00d0a3e5d..2c284e583 100644 --- a/.scripts/copy-shared-files.mjs +++ b/.scripts/copy-shared-files.mjs @@ -15,6 +15,7 @@ const HAS_CHILD_SAMPLES = [ // Some samples have different config files from those in .shared/ // that we don't want to overwrite const TSCONFIG_EXCLUDE = [ + 'ai-sdk', 'nextjs-ecommerce-oneclick', 'monorepo-folders', 'fetch-esm', From 15143c9b4125dd838fb268da529f830fd3c08582 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 23 Jul 2026 12:53:41 -0700 Subject: [PATCH 3/5] fix(ai-sdk): load @ai-sdk/openai lazily to fix Node 22 CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The static top-level `import { openai } from '@ai-sdk/openai'` compiles to require() under the sample's CommonJS build. On Node 22's require(ESM) implementation this throws "Unexpected module status 0" at module-load time, aborting the entire mocha run — even though the workflow test is skipped when OPENAI_API_KEY is unset (as in CI). Node 20 and 24 are unaffected. Load @ai-sdk/openai via a native dynamic import() (wrapped in `new Function` so TypeScript doesn't down-level it to require()), inside the `before` hook so it only runs when the suite isn't skipped. Verified against Node 22.23.1: old code reproduces the failure, new code passes. Co-Authored-By: Claude Opus 4.8 (1M context) --- ai-sdk/src/mocha/workflows.test.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/ai-sdk/src/mocha/workflows.test.ts b/ai-sdk/src/mocha/workflows.test.ts index 1963327bd..71488d2b7 100644 --- a/ai-sdk/src/mocha/workflows.test.ts +++ b/ai-sdk/src/mocha/workflows.test.ts @@ -1,12 +1,21 @@ import { TestWorkflowEnvironment } from '@temporalio/testing'; -import { before, describe, it } from 'mocha'; +import { after, before, describe, it } from 'mocha'; import { Worker } from '@temporalio/worker'; import { haikuAgent } from '../workflows'; import * as activities from '../activities'; import { AiSdkPlugin } from '@temporalio/ai-sdk'; -import { openai } from '@ai-sdk/openai'; import assert from 'assert'; +// `@ai-sdk/openai` is ESM-only. Under this sample's CommonJS build a static +// `import` is emitted as `require()`, which throws on Node 22's require(ESM) +// implementation ("Unexpected module status 0") at module-load time — aborting +// the whole mocha run. Load it lazily via a native dynamic `import()` (wrapped +// in `new Function` so TypeScript doesn't down-level it back to `require`), +// inside `before` so it only runs when this suite isn't skipped. +const importESM = new Function('specifier', 'return import(specifier)') as ( + specifier: string, +) => Promise; + const hasOpenAIKey = Boolean(process.env.OPENAI_API_KEY); const describeWorkflow = hasOpenAIKey ? describe : describe.skip; @@ -16,9 +25,11 @@ describeWorkflow( this.timeout(30_000); let testEnv: TestWorkflowEnvironment; + let openai: (typeof import('@ai-sdk/openai'))['openai']; before(async () => { testEnv = await TestWorkflowEnvironment.createLocal(); + ({ openai } = await importESM('@ai-sdk/openai')); }); after(async () => { From b9b42cc59048c7b17088f7cee43b36a8e2acacbe Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 23 Jul 2026 13:34:47 -0700 Subject: [PATCH 4/5] fix(ai-sdk): enable require(ESM) so tests pass on Node < 22.12 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sample's CommonJS mocha/ts-node tests must require() the ESM-only `ai` package (via workflows.ts). That needs require(ESM), which is only enabled by default on Node >= 22.12 / 20.19 / 24. The Windows CI runner uses Node 22.11.0, where require(ESM) is still flagged — so requiring workflows.ts throws ERR_REQUIRE_ESM, mocha falls back to import()-ing the .ts test file, and Node's ESM loader rejects it ("Unknown file extension .ts"). Pass `--node-option experimental-require-module` via mocha (cross-platform, unlike inline NODE_OPTIONS) to enable require(ESM) on older Node. The flag is a no-op/accepted on 20.19, 22.12+, and 24. Verified locally on Node 20.19.4, 22.11.0, 22.23.1, and 24: all pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- ai-sdk/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ai-sdk/package.json b/ai-sdk/package.json index 7bfe070b0..e02252cf9 100644 --- a/ai-sdk/package.json +++ b/ai-sdk/package.json @@ -11,7 +11,7 @@ "start": "ts-node src/worker.ts", "start.watch": "nodemon src/worker.ts", "workflow": "ts-node src/client.ts", - "test": "mocha --exit --require ts-node/register --require source-map-support/register src/mocha/*.test.ts" + "test": "mocha --exit --node-option experimental-require-module --require ts-node/register --require source-map-support/register src/mocha/*.test.ts" }, "nodemonConfig": { "execMap": { From 4165e76e5f3f14cad723473f75b6bc1e41520f94 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Thu, 23 Jul 2026 16:52:41 -0700 Subject: [PATCH 5/5] refactor(ai-sdk): remove deprecated streamObject, fix streaming teardown Address PR #497 review comments: - Remove the streamObject sample; streamObject is @deprecated in ai@7.0.36 ("Use streamText with an output setting instead"). Drops streamObjectAgent, its provider/topic, the stream-object CLI case, test, and README entry. - Add the missing SNIPEND for middlewareAgent so its snippet no longer swallows the mcp-agent below it. - Replace the arbitrary 500ms linger-sleep in streamingAgent with a consumer-ack signal: the workflow waits for the subscriber to confirm it received the final delta (with a timeout fallback) instead of racing its in-memory stream log being discarded on completion. Co-Authored-By: Claude Opus 4.8 (1M context) --- ai-sdk/README.md | 8 ++-- ai-sdk/src/client.ts | 18 ++------- ai-sdk/src/mocha/streaming.test.ts | 47 ++-------------------- ai-sdk/src/workflows.ts | 63 ++++++++++-------------------- 4 files changed, 31 insertions(+), 105 deletions(-) diff --git a/ai-sdk/README.md b/ai-sdk/README.md index 768dc0192..381d1d77e 100644 --- a/ai-sdk/README.md +++ b/ai-sdk/README.md @@ -16,15 +16,13 @@ This project demonstrates some uses of the AI SDK inside Temporal. 1. `npm run workflow mcp` 1. `npm run workflow middleware` 1. `npm run workflow stream` -1. `npm run workflow stream-object` ### Streaming -The `stream` and `stream-object` samples show how to stream model output out of a -Workflow. `streamText` (and `streamObject` for structured output) run the model call in +The `stream` sample shows how to stream model output out of a +Workflow. `streamText` runs the model call in an activity that publishes each delta onto a [Workflow Stream](https://docs.temporal.io/develop/typescript/workflows/workflow-streams) topic. The Workflow hosts the stream (`new WorkflowStream()`) and durably reassembles the final result, while external consumers subscribe by Workflow id to render the tokens live -as they arrive — see `src/client.ts`. Both stream functions flow through the same model -path, so no extra wiring is needed beyond choosing a `streamingTopic`. +as they arrive — see `src/client.ts`. diff --git a/ai-sdk/src/client.ts b/ai-sdk/src/client.ts index a7227905a..929d377d9 100644 --- a/ai-sdk/src/client.ts +++ b/ai-sdk/src/client.ts @@ -6,10 +6,9 @@ import { mcpAgent, middlewareAgent, streamingAgent, - streamObjectAgent, toolsAgent, STREAM_TOPIC, - OBJECT_STREAM_TOPIC, + consumerDoneSignal, } from './workflows'; import { nanoid } from 'nanoid'; @@ -25,6 +24,9 @@ async function renderStream(client: Client, workflowId: string, topic: string): if (part.type === 'finish') break; } process.stdout.write('\n'); + // Acknowledge receipt so the Workflow can complete without racing this final + // poll against its in-memory stream log being discarded. + await client.workflow.getHandle(workflowId).signal(consumerDoneSignal); } // @@@SNIPEND @@ -51,18 +53,6 @@ async function run() { await connection.close(); return; } - case 'stream-object': { - const streamHandle = await client.workflow.start(streamObjectAgent, { - taskQueue: 'ai-sdk', - args: ['Generate a lasagna recipe.'], - workflowId: 'workflow-' + nanoid(), - }); - console.log(`Started workflow ${streamHandle.workflowId}`); - await renderStream(client, streamHandle.workflowId, OBJECT_STREAM_TOPIC); - console.log(await streamHandle.result()); - await connection.close(); - return; - } case 'middleware': handle = await client.workflow.start(middlewareAgent, { taskQueue: 'ai-sdk', diff --git a/ai-sdk/src/mocha/streaming.test.ts b/ai-sdk/src/mocha/streaming.test.ts index 2621818f7..17e2292e7 100644 --- a/ai-sdk/src/mocha/streaming.test.ts +++ b/ai-sdk/src/mocha/streaming.test.ts @@ -11,7 +11,7 @@ import type { ProviderV4, } from '@ai-sdk/provider'; import assert from 'assert'; -import { streamingAgent, streamObjectAgent, STREAM_TOPIC, OBJECT_STREAM_TOPIC } from '../workflows'; +import { streamingAgent, STREAM_TOPIC, consumerDoneSignal } from '../workflows'; import * as activities from '../activities'; import { AiSdkPlugin } from '@temporalio/ai-sdk'; @@ -86,6 +86,8 @@ async function collectDeltas(client: any, workflowId: string, topic: string): Pr if (part.type === 'text-delta') deltas.push(part.delta); if (part.type === 'finish') break; } + // Mirror the real consumer: acknowledge receipt so the workflow can complete. + await client.workflow.getHandle(workflowId).signal(consumerDoneSignal); return deltas; } @@ -133,47 +135,4 @@ describe('streaming agents', function () { assert.strictEqual(deltas.join(''), chunks.join('')); }); }); - - it('streamObjectAgent publishes JSON deltas and returns the parsed object', async () => { - const { client, nativeConnection } = testEnv; - const taskQueue = 'test-stream-object'; - // A valid JSON document matching streamObjectAgent's schema, split so the - // subscriber observes it building up incrementally. - const object = { - recipe: { - name: 'Classic Lasagna', - ingredients: [{ name: 'pasta', amount: '1 box' }], - steps: ['Layer', 'Bake'], - }, - }; - const json = JSON.stringify(object); - const chunks = json.match(/.{1,12}/gs) ?? [json]; - - const worker = await Worker.create({ - connection: nativeConnection, - plugins: [new AiSdkPlugin({ modelProvider: mockProvider(chunks) })], - taskQueue, - workflowsPath: require.resolve('../workflows'), - activities, - }); - - await worker.runUntil(async () => { - const handle = await client.workflow.start(streamObjectAgent, { - args: ['Generate a lasagna recipe.'], - workflowId: 'test-stream-object-' + Date.now(), - taskQueue, - }); - - const deltasPromise = collectDeltas(client, handle.workflowId, OBJECT_STREAM_TOPIC); - const result = await handle.result(); - const deltas = await deltasPromise; - - // The workflow durably resolves the final, validated object. - assert.strictEqual(result, object.recipe.name); - // The external subscriber saw the JSON build up incrementally, and the - // concatenated deltas reconstruct the full document. - assert.ok(deltas.length > 1, `expected multiple live deltas, got ${deltas.length}`); - assert.strictEqual(deltas.join(''), json); - }); - }); }); diff --git a/ai-sdk/src/workflows.ts b/ai-sdk/src/workflows.ts index 23cf65414..2a70e8245 100644 --- a/ai-sdk/src/workflows.ts +++ b/ai-sdk/src/workflows.ts @@ -1,9 +1,9 @@ -import { generateText, streamText, streamObject, stepCountIs, tool, wrapLanguageModel } from 'ai'; +import { generateText, streamText, stepCountIs, tool, wrapLanguageModel } from 'ai'; import type { LanguageModelMiddleware } from 'ai'; import { TemporalMCPClient, TemporalProvider, temporalProvider } from '@temporalio/ai-sdk/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import type * as activities from './activities'; -import { proxyActivities, sleep } from '@temporalio/workflow'; +import { proxyActivities, condition, defineSignal, setHandler } from '@temporalio/workflow'; import z from 'zod'; const { getWeather } = proxyActivities({ @@ -73,6 +73,7 @@ export async function middlewareAgent(prompt: string): Promise { }); return result.text; } +// @@@SNIPEND // @@@SNIPSTART typescript-vercel-ai-sdk-mcp-agent export async function mcpAgent(prompt: string): Promise { @@ -96,6 +97,12 @@ export async function mcpAgent(prompt: string): Promise { export const STREAM_TOPIC = 'text-stream'; // @@@SNIPEND +// @@@SNIPSTART typescript-vercel-ai-sdk-consumer-done-signal +// A subscriber sends this signal once it has received the stream's final delta, +// so the workflow knows it is safe to complete. See `client.ts` for the sender. +export const consumerDoneSignal = defineSignal('consumer-done'); +// @@@SNIPEND + // A provider whose language-model calls stream their deltas onto STREAM_TOPIC. // Setting `streamingTopic` enables `doStream`; without it, streaming calls // throw. Use a distinct topic per concurrent streaming call. @@ -110,6 +117,12 @@ export async function streamingAgent(prompt: string): Promise { // publishing deltas to it. new WorkflowStream(); + // A subscriber flips this once it has consumed the final delta (see below). + let consumerDone = false; + setHandler(consumerDoneSignal, () => { + consumerDone = true; + }); + const result = streamText({ model: streamingProvider.languageModel('gpt-4o-mini'), prompt, @@ -125,46 +138,12 @@ export async function streamingAgent(prompt: string): Promise { text += delta; } - // Briefly hold the run open so a subscriber's final poll can deliver the - // last deltas before the workflow exits and its in-memory stream log is gone. - await sleep('500 milliseconds'); + // The workflow's stream log lives in memory and is discarded once the run + // completes, which can race a subscriber's final poll. Rather than guess at a + // fixed delay, wait for the subscriber to signal that it received the last + // delta. The timeout is a fallback for when nothing is subscribed, so the run + // can't hang forever. + await condition(() => consumerDone, '10 seconds'); return text; } // @@@SNIPEND - -// A provider for streaming structured output onto its own topic. `streamObject` -// flows through the same `doStream` path as `streamText`, so no extra wiring is -// needed — only a distinct topic so the two streams stay separable. -export const OBJECT_STREAM_TOPIC = 'object-stream'; -const objectStreamingProvider = new TemporalProvider({ - languageModel: { streamingTopic: OBJECT_STREAM_TOPIC }, -}); - -// @@@SNIPSTART typescript-vercel-ai-sdk-stream-object-agent -export async function streamObjectAgent(prompt: string): Promise { - new WorkflowStream(); - - const result = streamObject({ - model: objectStreamingProvider.languageModel('gpt-4o-mini'), - schema: z.object({ - recipe: z.object({ - name: z.string(), - ingredients: z.array(z.object({ name: z.string(), amount: z.string() })), - steps: z.array(z.string()), - }), - }), - prompt, - }); - - // External subscribers see the object build up incrementally via the partial - // JSON deltas published to OBJECT_STREAM_TOPIC. The workflow drains the - // partial stream and durably resolves the final, validated object. - for await (const _partial of result.partialObjectStream) { - // Draining drives the stream; the consumer renders the live partials. - } - const object = await result.object; - - await sleep('500 milliseconds'); - return object.recipe.name; -} -// @@@SNIPEND