Convex Agent Component 与 Cloudflare Agents SDK：关页后继续生成并恢复聊天

npm install agents @cloudflare/ai-chat ai workers-ai-provider

// src/agents/chat-agent.ts
import {
  AIChatAgent,
  type OnChatMessageOptions,
} from "@cloudflare/ai-chat";
import {
  convertToModelMessages,
  streamText,
  type GenerateTextOnFinishCallback,
  type ToolSet,
} from "ai";
import { createWorkersAI } from "workers-ai-provider";

export class ChatAgent extends AIChatAgent<Env> {
  // resumable streaming 只处理浏览器断线；chatRecovery 还处理
  // Durable Object eviction、部署或资源限制导致的 turn 中断。
  override chatRecovery = {
    maxAttempts: 6,
    noProgressTimeoutMs: 5 * 60_000,
    maxRecoveryWork: 100,
    terminalMessage: "回答生成过程中断且无法恢复，请重试。",
  };

  override async onChatMessage(
    _onFinish: GenerateTextOnFinishCallback<ToolSet>,
    options?: OnChatMessageOptions,
  ) {
    const workersai = createWorkersAI({ binding: this.env.AI });
    const result = streamText({
      model: workersai("@cf/zai-org/glm-4.7-flash"),
      messages: await convertToModelMessages(this.messages),
      abortSignal: options?.abortSignal,
    });

    return result.toUIMessageStreamResponse();
  }
}

{
  "$schema": "node_modules/wrangler/config-schema.json",
  "name": "chat-app",
  "main": "src/server.ts",
  "compatibility_date": "2026-08-14",
  "compatibility_flags": ["nodejs_compat"],
  "ai": { "binding": "AI" },
  "durable_objects": {
    "bindings": [{ "name": "ChatAgent", "class_name": "ChatAgent" }]
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["ChatAgent"] }
  ]
}

// src/routes/chat.$conversationId.tsx
import { useAgentChat } from "@cloudflare/ai-chat/react";
import { useAgent } from "agents/react";
import type { ChatAgent } from "../agents/chat-agent";

export function Chat({ conversationId }: { conversationId: string }) {
  const agent = useAgent<ChatAgent>({
    agent: "ChatAgent",
    name: conversationId,
  });

  const chat = useAgentChat({
    agent,
    resume: true,
    cancelOnClientAbort: false,
  });

  return (
    <main>
      {chat.messages.map((message) => (
        <article key={message.id}>
          {message.parts.map((part, index) =>
            part.type === "text" ? <span key={index}>{part.text}</span> : null,
          )}
        </article>
      ))}

      <form
        onSubmit={(event) => {
          event.preventDefault();
          const form = event.currentTarget;
          const data = new FormData(form);
          const text = String(data.get("message") ?? "").trim();
          if (!text) return;
          chat.sendMessage({ text });
          form.reset();
        }}
      >
        <input name="message" />
        <button type="submit" disabled={chat.status !== "ready"}>
          发送
        </button>
      </form>
    </main>
  );
}

npm install @cloudflare/think

// src/agents/chat-agent.ts
import { Think } from "@cloudflare/think";
import { callable } from "agents";
import { createWorkersAI } from "workers-ai-provider";

type SubmitInput = {
  messageId: string;
  text: string;
};

export class ChatAgent extends Think<Env> {
  getModel() {
    return createWorkersAI({ binding: this.env.AI })(
      "@cf/moonshotai/kimi-k2.6",
    );
  }

  @callable()
  async submitUserMessage(input: SubmitInput) {
    const result = await this.submitMessages(
      [
        {
          id: input.messageId,
          role: "user",
          parts: [{ type: "text", text: input.text }],
        },
      ],
      { idempotencyKey: input.messageId },
    );

    return {
      submissionId: result.submissionId,
      status: result.status,
      accepted: result.accepted,
    };
  }
}

// vite.config.ts
import { cloudflare } from "@cloudflare/vite-plugin";
import { tanstackStart } from "@tanstack/react-start/plugin/vite";
import react from "@vitejs/plugin-react";
import agents from "agents/vite";
import { defineConfig } from "vite";

export default defineConfig({
  plugins: [
    agents(),
    cloudflare({ viteEnvironment: { name: "ssr" } }),
    tanstackStart(),
    react(),
  ],
});

import { useMutation } from "@tanstack/react-query";
import { useAgent } from "agents/react";
import type { ChatAgent } from "../agents/chat-agent";

function useSubmitTurn(conversationId: string) {
  const agent = useAgent<ChatAgent>({
    agent: "ChatAgent",
    name: conversationId,
  });

  return useMutation({
    mutationFn: (input: { messageId: string; text: string }) =>
      agent.stub.submitUserMessage(input),
    retry: 2,
    // 只让同一会话的提交串行；它不是幂等保证。
    scope: { id: conversationId },
  });
}

function Composer({ conversationId }: { conversationId: string }) {
  const submitTurn = useSubmitTurn(conversationId);

  async function submitText(text: string) {
    // ID 在 mutationFn 外生成；网络重试必须复用同一个值。
    const input = { messageId: crypto.randomUUID(), text };
    const ack = await submitTurn.mutateAsync(input);

    // Promise resolve 后才显示“服务端已接收，可以关闭页面”。
    return ack;
  }

  // 表单调用 submitText(text)，并根据 submitTurn.status 渲染 ACK 状态。
  return null;
}

// src/server.ts
import startHandler from "@tanstack/react-start/server-entry";
import { routeAgentRequest } from "agents";

export { ChatAgent } from "./agents/chat-agent";

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const url = new URL(request.url);

    if (url.pathname.startsWith("/agents/chat-agent/")) {
      const session = await requireSession(request, env);
      const encodedId = url.pathname.split("/")[3];
      if (!encodedId) return new Response("Not found", { status: 404 });
      const conversationId = decodeURIComponent(encodedId);
      await assertConversationOwner(env, session.userId, conversationId);

      const response = await routeAgentRequest(request, env, {
        props: { userId: session.userId },
      });
      if (response) return response;
    }

    return startHandler.fetch(request, env, ctx);
  },
} satisfies ExportedHandler<Env>;

// convex/chat.ts
import {
  listUIMessages,
  saveMessage,
  syncStreams,
  vStreamArgs,
} from "@convex-dev/agent";
import { paginationOptsValidator } from "convex/server";
import { v } from "convex/values";
import { components, internal } from "./_generated/api";
import { internalAction, mutation, query } from "./_generated/server";
import { chatAgent } from "./agent";

export const sendMessage = mutation({
  args: { threadId: v.string(), prompt: v.string() },
  handler: async (ctx, { threadId, prompt }) => {
    await authorizeThreadAccess(ctx, threadId);

    const { messageId } = await saveMessage(ctx, components.agent, {
      threadId,
      prompt,
      skipEmbeddings: true,
    });

    const scheduledId = await ctx.scheduler.runAfter(
      0,
      internal.chat.generateResponse,
      { threadId, promptMessageId: messageId },
    );

    return { messageId, scheduledId };
  },
});

export const generateResponse = internalAction({
  args: { threadId: v.string(), promptMessageId: v.string() },
  handler: async (ctx, { threadId, promptMessageId }) => {
    const result = await chatAgent.streamText(
      ctx,
      { threadId },
      { promptMessageId },
      {
        saveStreamDeltas: {
          chunking: "word",
          throttleMs: 250,
        },
      },
    );

    // 后台 action 必须消费完整 stream，不能启动后直接 return。
    await result.consumeStream();
  },
});

export const listThreadMessages = query({
  args: {
    threadId: v.string(),
    paginationOpts: paginationOptsValidator,
    streamArgs: vStreamArgs,
  },
  handler: async (ctx, args) => {
    await authorizeThreadAccess(ctx, args.threadId);
    const page = await listUIMessages(ctx, components.agent, args);
    const streams = await syncStreams(ctx, components.agent, args);
    return { ...page, streams };
  },
});

import {
  optimisticallySendMessage,
  useUIMessages,
} from "@convex-dev/agent/react";
import { useMutation } from "convex/react";
import { api } from "../../convex/_generated/api";

function ConvexChat({ threadId }: { threadId: string }) {
  const { results: messages } = useUIMessages(
    api.chat.listThreadMessages,
    { threadId },
    { initialNumItems: 20, stream: true },
  );

  const sendMessage = useMutation(api.chat.sendMessage).withOptimisticUpdate(
    optimisticallySendMessage(api.chat.listThreadMessages),
  );

  async function submit(prompt: string) {
    await sendMessage({ threadId, prompt });
    // 只有 Promise resolve 后，才显示“服务端已接收”。
  }

  // 根据 messages 渲染 transcript，表单调用 submit(prompt)。
  return null;
}

await step.runAction(
  internal.chat.generateResponse,
  { threadId, promptMessageId },
  { retry: true },
);