trigger-chat-agent-advanced

chat.agent：进阶与实战

chat.agent 构建于 Sessions 之上：一对持久、绑定任务、双向的 I/O 通道，通过稳定的 externalId（例如 chatId）关联，其生命周期长于任何单次运行。本技能涵盖日常 Agent 之下及其周边的各个层面：原始 sessions API、服务端 AgentChat、持久子 Agent、操作 / 后台注入、快速启动、压缩与恢复，以及自定义传输的线路协议。

两个 chat 命名空间很容易混淆：Agent 定义从 @trigger.dev/sdk/ai 导入 chat；Head Start / Node-listener 服务端条目从 @trigger.dev/sdk/chat-server 导入 chat。

设置

推荐用法：通过服务端代码（任务、webhook 或脚本）使用 AgentChat 驱动 Agent。

import { AgentChat } from "@trigger.dev/sdk/chat";
import type { myAgent } from "./trigger/my-agent";

const chat = new AgentChat<typeof myAgent>({ agent: "my-chat", clientData: { userId: "user_123" } });
const stream = await chat.sendMessage("Review PR #42");
const text = await stream.text();
await chat.close();

首次调用 sendMessage() 会触发运行，之后则通过输入流复用该运行。ChatStream 提供 text()、result()（{ text, toolCalls, toolResults }）、messages()（UIMessage 快照）以及原始 .stream。其他方法：steer(text)、stop()、sendRaw(uiMessages)、sendAction(action)、preload()、reconnect()。

核心模式

1. 用于非聊天双向 I/O 的原始 Sessions

当聊天抽象不适用时，直接使用 sessions：例如 Agent 收件箱、审批流程、服务端到服务端的流水线。sessions.start 对 (env, externalId) 幂等；externalId 不能以 session_ 开头。

import { sessions } from "@trigger.dev/sdk";

const { id, publicAccessToken } = await sessions.start({
  type: "chat.agent",
  externalId: chatId,
  taskIdentifier: "my-chat",
  triggerConfig: { tags: [`chat:${chatId}`], basePayload: { chatId, trigger: "preload" } },
});

const session = sessions.open(chatId); // no network call; methods are lazy
await session.out.append({ kind: "message", text: "hello" });
const next = await session.in.once<MyEvent>({ timeoutMs: 30_000 });

sessions.open(id).in 还提供 send、on(handler)、peek、wait（挂起运行，仅限在 task.run() 内使用）和 waitWithIdleTimeout。.out 提供 append、pipe、writer、read、writeControl 和 trimTo。使用 sessions.list({ type, tag, status, ... }) 列出（for await），使用 sessions.update 修改，使用 sessions.close 结束（终态操作，幂等）。

2. 作为流式工具的持久子 Agent

在 AI SDK 的 tool() 中使用 AgentChat，可委派给持久子 Agent；其响应会作为初步工具结果流式传回。为该工具提供 toModelOutput，让模型看到简洁摘要。

import { tool } from "ai";
import { AgentChat } from "@trigger.dev/sdk/chat";
import { z } from "zod";

const researchTool = tool({
  description: "Delegate research to a specialist agent.",
  inputSchema: z.object({ topic: z.string() }),
  execute: async function* ({ topic }, { abortSignal }) {
    const chat = new AgentChat({ agent: "research-agent" });
    const stream = await chat.sendMessage(topic, { abortSignal });
    yield* stream.messages(); // UIMessage snapshots become preliminary tool results
    await chat.close();
  },
  toModelOutput: ({ output: message }) => {
    const lastText = message?.parts?.findLast((p: { type: string }) => p.type === "text") as
      | { text?: string }
      | undefined;
    return { type: "text", value: lastText?.text ?? "Done." };
  },
});

对于通过 execute: ai.toolExecute(task) 暴露的子任务，使用 chat.stream.writer({ target: "root" }) 将进度流式传输到 Agent 的运行中。target 接受 "self" | "parent" | "root" | <runId>。在子任务内，使用 ai.toolCallId() 和 ai.chatContextOrThrow<typeof myChat>()（{ chatId, turn, continuation, clientData }）读取上下文。

import { chat, ai } from "@trigger.dev/sdk/ai";

const { waitUntilComplete } = chat.stream.writer({
  target: "root",
  execute: ({ write }) =>
    write({ type: "data-research-status", id: partId, data: { query, status: "in-progress" } }),
});
await waitUntilComplete();

3. 后台注入：延后与注入

chat.defer(promise) 会在流式传输期间并行执行工作（所有延后的 promise 都会在 onTurnComplete 前等待完成，超时为 5 秒）。chat.inject(messages) 会将 ModelMessage[] 排入队列，并在下一轮开始时或 prepareStep 边界处排空。

有两个通道，由角色决定。role: "system" 消息会进入模型指令，与系统提示词一样受信任，且仅应用于下一轮。任何其他角色都会加入对话，天然不受信任，因此可在其中放置可核实的事实，并将指令放在系统通道中。指令通道只有通过受管理的 streamText（或展开 chat.toStreamTextOptions()）才能到达模型，因为 SDK 只能在此处设置指令。

export const myChat = chat.agent({
  id: "my-chat",
  onTurnComplete: async ({ messages }) => {
    chat.defer(
      (async () => {
        const analysis = await analyzeConversation(messages);
        chat.inject([{ role: "system", content: `[Analysis]\n\n${analysis}` }]);
      })()
    );
  },
  registry,
  run: async ({ messages, signal, streamText }) =>
    streamText({ messages, abortSignal: signal, stopWhen: stepCountIs(15) }),
});

4. 压缩（基于阈值）

compaction.shouldCompact 决定何时压缩，summarize 则生成用于替换模型消息的摘要。默认情况下会保留 UI 消息（可通过 compactUIMessages 自定义）。执行内部循环压缩的 prepareStep 依托受管理的 streamText，该方法会组合你传入的 prepareStep。如果展开 chat.toStreamTextOptions() 后再传入自己的 prepareStep，就会替换原有配置，从而关闭压缩。

compaction: {
  shouldCompact: ({ totalTokens }) => (totalTokens ?? 0) > 80_000,
  summarize: async ({ messages }) =>
    (await generateText({
      model: anthropic("claude-haiku-4-5"),
      messages: [...messages, { role: "user", content: "Summarize concisely." }],
    })).text,
},

5. 操作：编辑状态，并可选择作答

actionSchema 负责验证；onAction 通过 chat.history（slice、replace、rollbackTo、 remove、getPendingToolCalls、extractNewToolResults）进行编辑。操作只会触发 hydrateMessages 和 onAction。对于只编辑的操作，不返回任何内容：不会调用模型，轮次计数器也不会递增。编辑后返回 chat.turn() 即可作答：该轮会基于编辑后的历史运行，并包含一轮所需的全部内容（Agent 的系统提示词和工具、 引导、压缩、注入的指令、onTurnStart 和 onTurnComplete、持久化），run() 会收到它，并带有 trigger: "action-turn"。onAction 没有 streamText 参数，从中返回 StreamTextResult、字符串或 UIMessage 会抛出异常。

持久化方式因模型而异。使用转录存储（chat.agent 上的 storage，默认使用平台快照）时，在操作改变对话后， 运行时会向存储传递一个带有 reason: "action" 的变更集：撤销对应一次 truncateAfter；重新生成对应一次 truncateAfter，并在该轮完成时为新答案执行 put；编辑对应为已编辑的 id 执行 put。使用已弃用的 hydrateMessages 时，你的存储是事实来源，运行时不会写入，因此必须自行同步每次变更：重新生成对应删除和插入； 调用 chat.turn() 后的答案则像任何一轮的答案一样，通过 onTurnComplete 到达。

export const myChat = chat.agent({
  id: "my-chat",
  actionSchema: z.discriminatedUnion("type", [
    z.object({ type: z.literal("undo") }),
    z.object({ type: z.literal("regenerate") }),
    z.object({ type: z.literal("rollback"), targetMessageId: z.string() }),
  ]),
  onAction: async ({ action }) => {
    if (action.type === "undo") chat.history.slice(0, -2); // edit only
    if (action.type === "rollback") chat.history.rollbackTo(action.targetMessageId);
    if (action.type === "regenerate") {
      chat.history.slice(0, -1);
      return chat.turn(); // answer the edited history
    }
  },
  run: async ({ messages, signal, streamText }) =>
    streamText({ model: anthropic("claude-sonnet-4-5"), messages, abortSignal: signal }),
});

从浏览器通过 useChat 发送，这样一轮生成的答案会像其他轮次一样渲染： sendMessage(undefined, { body: { action: { type: "regenerate" } } })、regenerate({ body: { action } })， 或从 @trigger.dev/sdk/chat/react 导入 useChatActions({ sendMessage })。重新生成时，请使用 useChat（Vercel AI SDK）中的 regenerate()，它会在流式传输新答案前移除最后一条助手消息；使用 sendMessage 携带重新生成操作会追加消息而不移除原消息，导致两个答案都可见。transport.sendAction(chatId, action) 会返回一个原始流，调用方必须自行读取并应用。在服务端，使用 agentChat.sendAction({ type: "rollback", targetMessageId: "msg-3" })。

6. 快速启动：Head Start

chat.headStart（来自 @trigger.dev/sdk/chat-server，而非 /ai）会返回一个 Web Fetch 处理程序， 由你自己的预热进程处理第 1 轮，然后在第 2 轮及之后交由 Agent 处理。此处传入的工具必须仅包含 模式（仅导入 ai 和 zod 的模块）；耗费资源的执行逻辑应留在任务中。

import { chat } from "@trigger.dev/sdk/chat-server";
import { anthropic } from "@ai-sdk/anthropic";
import { headStartTools } from "@/lib/chat-tools/schemas";

export const chatHandler = chat.headStart({
  agentId: "my-chat",
  // `streamText` from the run argument owns `messages`, `prompt`, `stopWhen`
  // and `abortSignal`: the handover needs `stopWhen: stepCountIs(1)` so the agent,
  // not this handler, runs step 2 onward. Passing any of them is a type error.
  run: async ({ streamText }) =>
    streamText({
      model: anthropic("claude-sonnet-4-6"),
      system: "You are helpful.",
      tools: headStartTools,
    }),
});
// Next.js: export const POST = chatHandler;  Transport: headStart: "/api/chat"

仅支持 Node 的框架可使用 chat.toNodeListener(handler) 包装 Web Fetch 处理程序。两端使用相同的 模型，以避免第 1 轮与第 2 轮及之后的语气发生变化。

7. chat.local：在 onBoot 中初始化，而非 onChatStart

chat.local<T>({ id }) 是模块级、浅代理、按运行范围划分的状态。在 onBoot 中初始化 （每个新的 worker 都会触发，包括续接运行），切勿在 onChatStart 中初始化。

const userContext = chat.local<{ name: string; plan: "free" | "pro" }>({ id: "userContext" });

export const myChat = chat.agent({
  id: "my-chat",
  onBoot: async ({ clientData }) => userContext.init({ name: "Alice", plan: "pro" }),
  run: async ({ messages, signal }) => streamText({ /* ... */ }),
});

8. 待处理消息（流式传输期间的用户输入）

一轮正在流式传输时发送的消息不应取消该流。在 Agent 上配置 pendingMessages（shouldInject、prepare、onReceived、onInjected），让托管的 streamText 的 prepareStep 在下一个边界处将其并入。注入的引导消息属于钩子所见对话的一部分，因此会出现在 onTurnComplete 的 uiMessages 和 newUIMessages 中，从那里进行持久化的应用无需额外操作即可存储它。在前端， usePendingMessages 会返回 pending、steer(text)、queue(text) 和 promoteToSteering(id)；通过 transport.sendPendingMessage(chatId, uiMessage, metadata?) 发送。

9. 恢复与版本升级

onRecoveryBoot 仅当末尾存在部分助手消息时触发（部署中断、崩溃、OOM 重试）。调用 chat.requestUpgrade() 时不会触发，因为这是正常退出，不会留下部分消息。chat.requestUpgrade()（在 onTurnStart / onValidateMessages 中调用以跳过 run()，或在 run() / chat.defer() 中调用以在轮次结束后退出）会将 Session 的 currentRunId 轮换为最新部署上的一次运行，无需客户端重新连接。请将其与 clientData 上的契约版本搭配使用。

const SUPPORTED_VERSIONS = new Set(["v2", "v3"]);
onTurnStart: async ({ clientData }) => {
  if (clientData?.protocolVersion && !SUPPORTED_VERSIONS.has(clientData.protocolVersion)) {
    chat.requestUpgrade();
  }
},

为增强 OOM 恢复能力，请在 Agent 上设置 oomMachine（以及 machine），以便重试时使用更大的预设。

10. 使用 mockChatAgent 进行离线测试

@trigger.dev/sdk/ai/test 会在内存中运行真实的轮次循环。请在 Agent 模块之前导入它，以便安装资源目录。使用 sendMessage、sendRegenerate、sendAction、sendStop、sendHeadStart、sendHandover 驱动测试；使用 seedSnapshot / seedSessionOutTail / seedSessionOutPartial / seedSessionInTail 初始化状态；针对 turn.chunks 和 harness.allChunks 进行断言。

import { mockChatAgent } from "@trigger.dev/sdk/ai/test"; // BEFORE the agent module
import { myChatAgent } from "./my-chat.js";

const harness = mockChatAgent(myChatAgent, { chatId: "test-1", clientData: { model } });
try {
  const turn = await harness.sendMessage({ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] });
  // assert against turn.chunks
} finally {
  await harness.close();
}

选项包括 mode（"preload" | "submit-message" | "handover-prepare" | "continuation"）、preload、continuation、previousRunId、snapshot、taskContext 和 setupLocals。将 taskContext.ctx.attempt.number > 1 设为满足条件，以模拟 OOM 重试尝试。runInMockTaskContext 可离线驱动非聊天任务。

11. 自定义传输：线路协议

端点：POST /api/v1/sessions（创建）、GET /realtime/v1/sessions/{id}/out（SSE）、POST /realtime/v1/sessions/{id}/in/append、POST /api/v1/sessions/{id}/close。ChatInputChunk 的格式为 { kind: "message"; payload: ChatTaskWirePayload } | { kind: "stop"; message? }。ChatTaskWirePayload 包含 chatId、trigger（submit-message | regenerate-message | preload | close | action | handover-prepare）、message?、metadata?、action?、continuation?、previousRunId? 等。控制记录采用标头形式：trigger-control: turn-complete（可选包含 public-access-token、session-in-event-id）以及 trigger-control: upgrade-required。TS 辅助函数 SSEStreamSubscription 和 controlSubtype(headers)（文档见 docs/ai-chat/client-protocol.mdx）会为你处理批量解码和控制记录筛选。

常见错误

• 严重：通过重新 POST POST /api/v1/sessions 来发送后续消息。

  // Wrong - a cached re-POST silently drops basePayload.message; basePayload is trigger config, not a channel
  await fetch("/api/v1/sessions", { method: "POST", body: JSON.stringify({ ...createBody }) });
  // Correct - append to the session's input channel
  await fetch(`/realtime/v1/sessions/${id}/in/append`, { method: "POST", body: JSON.stringify({ kind: "message", payload }) });
  

• 为 .in / .out 使用了错误的令牌。 使用创建响应正文中的 publicAccessToken（作用于会话）。x-trigger-jwt 响应标头作用于运行，无法用于订阅。

• 在 onChatStart 中初始化 chat.local。 续接运行会跳过此步骤，因此 run() 会因 chat.local can only be modified after initialization 而崩溃。请在 onBoot 中初始化。

• 使用 chat.defer 写入消息历史。 流传输期间刷新会读到 []。请在模型开始流式输出前，内联 await 这项写入；chat.defer 应留给分析、审计和缓存预热。

• 为 HITL 工具提供 execute。 streamText 会立即调用它。不要设置 execute；由前端通过 addToolOutput + sendAutomaticallyWhen 提供答案。

• 只在 streamText 上声明子 Agent / 重型工具。 还要在 chat.agent({ tools }) 上声明（或在自定义 Agent 中传给 convertToModelMessages(uiMessages, { tools })），这样每轮都会重新应用 toModelOutput。

• 将执行繁重操作的工具导入 Head Start 路由模块。 这是构建时的导入链问题；运行时剥离辅助函数无法解决。请将架构定义放在仅依赖 ai + zod 的模块中。

• 在流中返回数兆字节的工具输出。 单条超过约 1 MiB 的 tool-output-available 记录会抛出 ChatChunkTooLargeError。请将内容持久化到存储中，先写入记录，然后只发送 ID。

• 在活动发送路径上设置 X-Peek-Settled: 1。 这会与新一轮的首个数据块发生竞争，并提前关闭流。仅在重新加载时重连的路径中使用。

文档术语说明：部分文档中的 Agent 端示例仍使用旧版 trigger:turn-complete 数据块类型。这是 Agent 发出的术语。自定义读取器必须 根据 trigger-control 标头进行筛选，而不是根据 chunk.type。

由 MCP 驱动的 Agent 聊天（list_agents、start_agent_chat、send_agent_message、 close_agent_chat）是供 Claude Code / Cursor 使用的 MCP 服务工具，并非可导入的 SDK 函数。请参阅 /mcp-tools#agent-chat-tools。

参考资料

• trigger-authoring-chat-agent 技能——介绍日常使用的 chat.agent({...}) 定义、生命周期钩子，以及 useTriggerChatTransport 的顺畅使用方式。使用此技能前，请先从这里开始。
• trigger-realtime 技能——介绍实时钩子，以及超出聊天传输范围的前端流式传输。
• trigger-tasks 技能——介绍基础 task() 语义、ctx 和标准生命周期钩子。

参考文档与此技能一同发布在同一软件包中。请在本地阅读（无需联网），并使用与你安装版本匹配的文档。上方的 sources: frontmatter 列出了此技能引用的所有文档，均位于 @trigger.dev/sdk/docs/ai-chat/ 下（包括 patterns/）。对于 HITL、会话和子 Agent，请从 sessions.mdx、server-chat.mdx、client-protocol.mdx、patterns/human-in-the-loop.mdx、patterns/sub-agents.mdx 开始阅读。

对于 trigger.config.ts 以及聊天 Agent 任务可能需要的构建扩展（Prisma、Playwright、Python 等），请阅读 @trigger.dev/sdk/docs/config/ 下随附的配置文档（各扩展的设置说明位于 config/extensions/）。

版本

此技能包含在 @trigger.dev/sdk 中，并直接从 node_modules 读取，因此始终与你安装的 SDK 版本一致（请参阅旁边的 package.json）。这些 API 的完整文档随附在 @trigger.dev/sdk/docs/ 下。