流式响应接口

// src/chat-room.ts
import { DurableObject } from 'cloudflare:workers'
 
type Env = { AI: Ai }
 
interface ChatMessage {
 
  role: 'user' | 'assistant'
 
  content: string
 
}
 
export class ChatRoom extends DurableObject<Env> {
 
  async fetch(request: Request): Promise<Response> {
 
    // 1. 必须是 WebSocket upgrade 请求
 
    if (request.headers.get('Upgrade') !== 'websocket') {
 
      return new Response('Expected WebSocket', { status: 426 })
 
    }
 
    // 2. 创建一对 WebSocket
 
    const pair = new WebSocketPair()
 
    const [client, server] = Object.values(pair)
 
    // 3. Hibernatable 模式接收服务端一侧
 
    this.ctx.acceptWebSocket(server)
 
    // 4. 返回客户端一侧给上游
 
    return new Response(null, { status: 101, webSocket: client })
 
  }
 
  // Hibernatable 版本的消息处理器（注意不是 addEventListener）
 
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
 
    if (typeof message !== 'string') return
 
    const userMsg: ChatMessage = { role: 'user', content: message }
 
    // 取出历史消息
 
    const history = (await this.ctx.storage.get<ChatMessage[]>('history')) ?? []
 
    history.push(userMsg)
 
    // 调 LLM
 
    const result = await this.env.AI.run('@cf/meta/llama-3.1-8b-instruct', {
 
      messages: history,
 
    })
 
    // Workers AI 文本生成返回 { response: string }，类型声明比较宽泛所以要断言一下
 
    const assistantMsg: ChatMessage = {
 
      role: 'assistant',
 
      content: (result as { response: string }).response,
 
    }
 
    history.push(assistantMsg)
 
    // 保存历史
 
    await this.ctx.storage.put('history', history)
 
    // 把回答推给客户端
 
    ws.send(JSON.stringify(assistantMsg))
 
  }
 
  async webSocketClose(ws: WebSocket, code: number) {
 
    // 连接关闭时的清理逻辑
 
    console.log(`WebSocket closed with code ${code}`)
 
  }
 
}

// src/index.ts
import { Hono } from 'hono'
 
import { ChatRoom } from './chat-room'
 
type Bindings = {
 
  CHAT_ROOM: DurableObjectNamespace<ChatRoom>
 
  AI: Ai
 
}
 
const app = new Hono<{ Bindings: Bindings }>()
 
app.get('/chat/:userId', async (c) => {
 
  // 每个用户一个独立的 ChatRoom
 
  const userId = c.req.param('userId')
 
  const id = c.env.CHAT_ROOM.idFromName(userId)
 
  const stub = c.env.CHAT_ROOM.get(id)
 
  // 把 upgrade 请求直接转给 DO，DO 自己处理 101 响应
 
  return stub.fetch(c.req.raw)
 
})
 
export { ChatRoom }
 
export default app

// client.ts
const ws = new WebSocket('wss://your-worker.workers.dev/chat/alice')
 
ws.onmessage = (e) => {
 
  const msg = JSON.parse(e.data)
 
  console.log('AI:', msg.content)
 
}
 
ws.send('你好，介绍一下 Cloudflare Workers')

// src/meeting-room.ts
import { DurableObject } from 'cloudflare:workers'
 
type Env = { AI: Ai }
 
export class MeetingRoom extends DurableObject<Env> {
 
  async fetch(request: Request): Promise<Response> {
 
    const url = new URL(request.url)
 
    const username = url.searchParams.get('user') ?? 'anon'
 
    const pair = new WebSocketPair()
 
    const [client, server] = Object.values(pair)
 
    // serializeAttachment 把数据附加到这个连接上
 
    // DO 休眠再唤醒后，用 deserializeAttachment 能拿回来
 
    server.serializeAttachment({ username })
 
    this.ctx.acceptWebSocket(server)
 
    return new Response(null, { status: 101, webSocket: client })
 
  }
 
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
 
    if (typeof message !== 'string') return
 
    const { username } = ws.deserializeAttachment() as { username: string }
 
    const payload = JSON.stringify({ from: username, text: message })
 
    // 广播给房间里所有其他连接
 
    for (const client of this.ctx.getWebSockets()) {
 
      if (client === ws) continue
 
      try {
 
        client.send(payload)
 
      } catch {
 
        // 客户端掉线时可能报错，忽略即可
 
      }
 
    }
 
  }
 
}

// src/chat-room.ts
export class ChatRoom extends DurableObject<Env> {
 
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
 
    // ... 处理消息 ...
 
    // 每次有消息，就把「闹钟」推到 30 分钟后
 
    await this.ctx.storage.setAlarm(Date.now() + 30 * 60 * 1000)
 
  }
 
  async alarm() {
 
    // 30 分钟没消息了，归档历史并重置
 
    const history = await this.ctx.storage.get('history')
 
    if (history) {
 
      await archiveToR2(history)  // 保存到 R2
 
      await this.ctx.storage.delete('history')
 
    }
 
  }
 
}