feat(chat): expose typed chat.stream, add deepResearch subtask example, per-chat model persistence, debug panel
- Export chat.stream (typed RealtimeDefinedStream<UIMessageChunk>) for writing custom data to the chat stream - Add deepResearch subtask using data-* chunks to stream progress back to parent chat via target: root - Use AI SDK data-research-progress chunk protocol with id-based updates for live progress - Add ResearchProgress component and generic data-* fallback renderer in frontend - Persist model per chat in DB (schema + onChatStart), model selector only on new chats - Add collapsible debug panel showing run ID (with dashboard link), chat ID, model, status, session info - Document chat.stream API, data-* chunks, and subtask streaming pattern in docs
This commit is contained in:
@@ -14,7 +14,7 @@ import {
|
||||
type TaskSchema,
|
||||
type TaskWithSchema,
|
||||
} from "@trigger.dev/core/v3";
|
||||
import type { ModelMessage, UIMessage } from "ai";
|
||||
import type { ModelMessage, UIMessage, UIMessageChunk } from "ai";
|
||||
import type { StreamWriteResult } from "@trigger.dev/core/v3";
|
||||
import { convertToModelMessages, dynamicTool, generateId as generateMessageId, jsonSchema, JSONSchema7, Schema, Tool, ToolCallOptions, zodSchema } from "ai";
|
||||
import { type Attributes, trace } from "@opentelemetry/api";
|
||||
@@ -175,6 +175,29 @@ export const CHAT_STREAM_KEY = _CHAT_STREAM_KEY;
|
||||
// Re-export input stream IDs for advanced usage
|
||||
export { CHAT_MESSAGES_STREAM_ID, CHAT_STOP_STREAM_ID };
|
||||
|
||||
/**
|
||||
* Typed chat output stream. Provides `.writer()`, `.pipe()`, `.append()`,
|
||||
* and `.read()` methods pre-bound to the chat stream key and typed to `UIMessageChunk`.
|
||||
*
|
||||
* Use from within a `chat.task` run to write custom chunks:
|
||||
* ```ts
|
||||
* const { waitUntilComplete } = chat.stream.writer({
|
||||
* execute: ({ write }) => {
|
||||
* write({ type: "text-start", id: "status-1" });
|
||||
* write({ type: "text-delta", id: "status-1", delta: "Processing..." });
|
||||
* write({ type: "text-end", id: "status-1" });
|
||||
* },
|
||||
* });
|
||||
* await waitUntilComplete();
|
||||
* ```
|
||||
*
|
||||
* Use from a subtask to stream back to the parent chat:
|
||||
* ```ts
|
||||
* chat.stream.pipe(myStream, { target: "root" });
|
||||
* ```
|
||||
*/
|
||||
const chatStream = streams.define<UIMessageChunk>({ id: _CHAT_STREAM_KEY });
|
||||
|
||||
/**
|
||||
* The wire payload shape sent by `TriggerChatTransport`.
|
||||
* Uses `metadata` to match the AI SDK's `ChatRequestOptions` field name.
|
||||
@@ -1452,6 +1475,8 @@ export const chat = {
|
||||
isStopped,
|
||||
/** Clean up aborted parts from a UIMessage. See {@link cleanupAbortedParts}. */
|
||||
cleanupAbortedParts,
|
||||
/** Typed chat output stream for writing custom chunks or piping from subtasks. */
|
||||
stream: chatStream,
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@@ -70,6 +70,46 @@ function ToolInvocation({ part }: { part: any }) {
|
||||
);
|
||||
}
|
||||
|
||||
function ResearchProgress({ part }: { part: any }) {
|
||||
const data = part.data as {
|
||||
status: "fetching" | "done";
|
||||
query: string;
|
||||
current: number;
|
||||
total: number;
|
||||
currentUrl?: string;
|
||||
completedUrls: string[];
|
||||
};
|
||||
|
||||
const isDone = data.status === "done";
|
||||
|
||||
return (
|
||||
<div className="my-2 rounded border border-blue-200 bg-blue-50 px-3 py-2 text-xs">
|
||||
<div className="flex items-center gap-2 font-medium text-blue-700">
|
||||
{isDone ? (
|
||||
<span className="text-green-600">✓</span>
|
||||
) : (
|
||||
<span className="inline-block h-3 w-3 animate-spin rounded-full border-2 border-blue-300 border-t-blue-600" />
|
||||
)}
|
||||
<span>
|
||||
{isDone
|
||||
? `Research complete — ${data.total} sources fetched`
|
||||
: `Researching "${data.query}" (${data.current}/${data.total})`}
|
||||
</span>
|
||||
</div>
|
||||
{data.currentUrl && !isDone && (
|
||||
<div className="mt-1 truncate text-blue-500">Fetching {data.currentUrl}</div>
|
||||
)}
|
||||
{data.completedUrls.length > 0 && (
|
||||
<div className="mt-1 space-y-0.5 text-blue-400">
|
||||
{data.completedUrls.map((url, i) => (
|
||||
<div key={i} className="truncate">✓ {url}</div>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
function DebugPanel({
|
||||
chatId,
|
||||
model,
|
||||
@@ -321,10 +361,25 @@ export function Chat({
|
||||
);
|
||||
}
|
||||
|
||||
if (part.type === "data-research-progress") {
|
||||
return <ResearchProgress key={i} part={part} />;
|
||||
}
|
||||
|
||||
if (part.type.startsWith("tool-") || part.type === "dynamic-tool") {
|
||||
return <ToolInvocation key={i} part={part} />;
|
||||
}
|
||||
|
||||
if (part.type.startsWith("data-")) {
|
||||
return (
|
||||
<div key={i} className="my-1 rounded border border-gray-200 bg-gray-50 p-2 text-xs text-gray-500">
|
||||
<span className="font-medium">{part.type}</span>
|
||||
<pre className="mt-1 overflow-x-auto whitespace-pre-wrap">
|
||||
{JSON.stringify((part as any).data, null, 2)}
|
||||
</pre>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
return null;
|
||||
})}
|
||||
</div>
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { chat } from "@trigger.dev/sdk/ai";
|
||||
import { streamText, tool, stepCountIs } from "ai";
|
||||
import { chat, ai } from "@trigger.dev/sdk/ai";
|
||||
import { schemaTask } from "@trigger.dev/sdk";
|
||||
import { streamText, tool, stepCountIs, generateId } from "ai";
|
||||
import type { LanguageModel } from "ai";
|
||||
import { openai } from "@ai-sdk/openai";
|
||||
import { anthropic } from "@ai-sdk/anthropic";
|
||||
@@ -135,6 +136,105 @@ const userContext = chat.local<{
|
||||
messageCount: number;
|
||||
}>();
|
||||
|
||||
// --------------------------------------------------------------------------
|
||||
// Subtask: deep research — fetches multiple URLs and streams progress
|
||||
// back to the parent chat via chat.stream using data-* chunks
|
||||
// --------------------------------------------------------------------------
|
||||
export const deepResearch = schemaTask({
|
||||
id: "deep-research",
|
||||
description:
|
||||
"Research a topic by fetching multiple URLs and synthesizing the results. " +
|
||||
"Streams progress updates to the chat as it works.",
|
||||
schema: z.object({
|
||||
query: z.string().describe("The research query or topic"),
|
||||
urls: z.array(z.string().url()).describe("URLs to fetch and analyze"),
|
||||
}),
|
||||
run: async ({ query, urls }) => {
|
||||
const partId = generateId();
|
||||
const results: { url: string; status: number; snippet: string }[] = [];
|
||||
|
||||
// Stream progress using data-research-progress chunks.
|
||||
// Using the same id means each write updates the same part in the message.
|
||||
function streamProgress(progress: {
|
||||
status: "fetching" | "done";
|
||||
query: string;
|
||||
current: number;
|
||||
total: number;
|
||||
currentUrl?: string;
|
||||
completedUrls: string[];
|
||||
}) {
|
||||
return chat.stream.writer({
|
||||
target: "root",
|
||||
execute: ({ write }) => {
|
||||
write({
|
||||
type: "data-research-progress" as any,
|
||||
id: partId,
|
||||
data: progress,
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
for (let i = 0; i < urls.length; i++) {
|
||||
const url = urls[i]!;
|
||||
|
||||
// Update progress — fetching
|
||||
const { waitUntilComplete } = streamProgress({
|
||||
status: "fetching",
|
||||
query,
|
||||
current: i + 1,
|
||||
total: urls.length,
|
||||
currentUrl: url,
|
||||
completedUrls: results.map((r) => r.url),
|
||||
});
|
||||
await waitUntilComplete();
|
||||
|
||||
try {
|
||||
const response = await fetch(url);
|
||||
let text = await response.text();
|
||||
const contentType = response.headers.get("content-type") ?? "";
|
||||
|
||||
if (contentType.includes("html")) {
|
||||
text = text
|
||||
.replace(/<script[\s\S]*?<\/script>/gi, "")
|
||||
.replace(/<style[\s\S]*?<\/style>/gi, "")
|
||||
.replace(/<[^>]+>/g, " ")
|
||||
.replace(/ /g, " ")
|
||||
.replace(/&/g, "&")
|
||||
.replace(/</g, "<")
|
||||
.replace(/>/g, ">")
|
||||
.replace(/\s+/g, " ")
|
||||
.trim();
|
||||
}
|
||||
|
||||
results.push({
|
||||
url,
|
||||
status: response.status,
|
||||
snippet: text.slice(0, 500),
|
||||
});
|
||||
} catch (err) {
|
||||
results.push({
|
||||
url,
|
||||
status: 0,
|
||||
snippet: `Error: ${err instanceof Error ? err.message : String(err)}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Final progress update — done
|
||||
const { waitUntilComplete: waitForDone } = streamProgress({
|
||||
status: "done",
|
||||
query,
|
||||
current: urls.length,
|
||||
total: urls.length,
|
||||
completedUrls: results.map((r) => r.url),
|
||||
});
|
||||
await waitForDone();
|
||||
|
||||
return { query, results };
|
||||
},
|
||||
});
|
||||
|
||||
export const aiChat = chat.task({
|
||||
id: "ai-chat",
|
||||
clientDataSchema: z.object({ model: z.string().optional(), userId: z.string() }),
|
||||
@@ -228,7 +328,11 @@ export const aiChat = chat.task({
|
||||
model: getModel(modelId),
|
||||
system: `You are a helpful assistant for ${userContext.name} (${userContext.plan} plan). Be concise and friendly.`,
|
||||
messages,
|
||||
tools: { inspectEnvironment, webFetch },
|
||||
tools: {
|
||||
inspectEnvironment,
|
||||
webFetch,
|
||||
deepResearch: ai.tool(deepResearch),
|
||||
},
|
||||
stopWhen: stepCountIs(10),
|
||||
abortSignal: stopSignal,
|
||||
providerOptions: {
|
||||
|
||||
Reference in New Issue
Block a user