Streaming

// response.txt
HTTP/1.1 200 OK
 
Content-Type: text/event-stream
 
Cache-Control: no-cache
 
Connection: keep-alive
 
event: thinking
 
data: {"node": "memory_retrieval"}
 
event: token
 
data: {"content": "我"}
 
event: token
 
data: {"content": "记得"}
 
event: token
 
data: {"content": "你上次"}
 
event: done
 
data: {"emotion": "happy", "memories_used": 3}

// pipeline.txt
LangGraph 节点执行
 
    ↓ yield { type: 'node_start', node: 'memory_retrieval' }
 
    ↓ yield { type: 'node_end', node: 'memory_retrieval' }
 
    ↓ yield { type: 'node_start', node: 'llm_generate' }
 
    ↓ yield { type: 'token', content: '我' }
 
    ↓ yield { type: 'token', content: '记得' }
 
    ↓ ...
 
    ↓ yield { type: 'done', metadata: { emotion, memories_used } }
 
Hono SSE 中间层
 
    ↓ event: thinking\ndata: {...}
 
    ↓ event: token\ndata: {...}
 
    ↓ event: done\ndata: {...}
 
React 前端
 
    → 显示 "正在思考..."
 
    → 逐字追加文字
 
    → 完成，显示情绪标签

// normal.ts
// 普通响应：等 LLM 生成完毕，一次性返回
 
app.post('/chat', async (c) => {
 
  const { message } = await c.req.json()
 
  const reply = await callLLM(message)  // 阻塞 1-3 秒
 
  return c.json({ reply })               // 全部完成后才返回
 
})

// streaming.ts
// 流式响应：边生成边返回
 
app.post('/chat/stream', async (c) => {
 
  const { message } = await c.req.json()
 
  // 立即返回响应头，body 持续写入
 
  return streamSSE(c, async (stream) => {
 
    const llmStream = await callLLM(message, { stream: true })
 
    for await (const chunk of llmStream) {
 
      await stream.writeSSE({           // 每个 token 立即推送
 
        event: 'token',
 
        data: JSON.stringify({ content: chunk.text })
 
      })
 
    }
 
  })
 
})

// minimal.ts
import { Hono } from 'hono'
 
import { streamSSE } from 'hono/streaming'
 
const app = new Hono()
 
app.get('/hello-stream', async (c) => {
 
  return streamSSE(c, async (stream) => {
 
    // stream 对象是你和前端之间的管道
 
    // 通过 writeSSE 方法往管道里写入 SSE 事件
 
    await stream.writeSSE({
 
      event: 'greeting',
 
      data: '你好'
 
    })
 
    // 模拟延迟
 
    await stream.sleep(1000)
 
    await stream.writeSSE({
 
      event: 'greeting',
 
      data: '世界'
 
    })
 
  })
 
  // 回调函数执行完毕后，流自动关闭
 
})
 
export default app

// events.ts
// 管线阶段事件：告诉前端当前执行到了哪个节点
 
interface ThinkingEvent {
 
  type: 'thinking'
 
  node: string       // 'safety_check' | 'memory_retrieval' | 'llm_generate' | ...
 
  status: 'start' | 'end'
 
}
 
// 文本生成事件：LLM 逐 token 输出
 
interface TokenEvent {
 
  type: 'token'
 
  content: string    // 一个或多个字符
 
}
 
// 完成事件：对话成功结束，携带元信息
 
interface DoneEvent {
 
  type: 'done'
 
  emotion: string
 
  memoriesUsed: number
 
  tokensConsumed: number
 
}
 
// 错误事件：出了问题，附带兜底回复
 
interface ErrorEvent {
 
  type: 'error'
 
  code: string       // 'safety_blocked' | 'llm_timeout' | 'internal_error'
 
  message?: string
 
  fallbackReply?: string  // 兜底话术，前端直接展示即可
 
}
 
type StreamEvent = ThinkingEvent | TokenEvent | DoneEvent | ErrorEvent

// llm-response.txt
data: {"choices": [{"delta": {"content": "我"}}]}
 
data: {"choices": [{"delta": {"content": "记得"}}]}
 
data: {"choices": [{"delta": {"content": "你"}}]}
 
data: [DONE]

// llm.ts
// AsyncGenerator 函数：用 async function* 声明，内部用 yield 逐个产出结果
 
// 调用方用 for await...of 消费
 
async function* callLLMStream(
 
  systemPrompt: string,
 
  userMessage: string,
 
  apiKey: string
 
): AsyncGenerator<string> {
 
  // 第一步：发起 HTTP 请求，开启流式模式
 
  const response = await fetch('https://api.deepseek.com/chat/completions', {
 
    method: 'POST',
 
    headers: {
 
      'Content-Type': 'application/json',
 
      'Authorization': `Bearer ${apiKey}`
 
    },
 
    body: JSON.stringify({
 
      model: 'deepseek-chat',
 
      messages: [
 
        { role: 'system', content: systemPrompt },
 
        { role: 'user', content: userMessage }
 
      ],
 
      stream: true  // 关键：开启流式返回
 
    })
 
  })
 
  // 第二步：拿到响应体的 ReadableStream，逐块读取
 
  const reader = response.body!.getReader()
 
  const decoder = new TextDecoder()
 
  let buffer = ''
 
  while (true) {
 
    // reader.read() 每次返回一块数据（Uint8Array）
 
    // done=true 表示流结束
 
    const { done, value } = await reader.read()
 
    if (done) break
 
    // 第三步：将二进制数据解码为文本，拼接到缓冲区
 
    // stream: true 告诉 decoder 后续还有数据，不要丢弃不完整的多字节字符
 
    buffer += decoder.decode(value, { stream: true })
 
    // 第四步：按行切分，解析 SSE 数据
 
    // LLM 返回的也是 SSE 格式，所以我们要在服务端解析它
 
    const lines = buffer.split('\n')
 
    // 最后一行可能不完整，保留到下次处理
 
    buffer = lines.pop() ?? ''
 
    for (const line of lines) {
 
      // 跳过空行和非 data 行
 
      if (!line.startsWith('data: ')) continue
 
      // [DONE] 是流结束标记
 
      if (line === 'data: [DONE]') return
 
      // 第五步：解析 JSON，提取文本内容
 
      const json = JSON.parse(line.slice(6))  // 去掉 'data: ' 前缀
 
      const content = json.choices?.[0]?.delta?.content
 
      if (content) {
 
        yield content  // 产出一个文本片段，调用方立即收到
 
      }
 
    }
 
  }
 
}

// usage.ts
// 消费方式：像遍历数组一样简单
 
for await (const text of callLLMStream(systemPrompt, message, apiKey)) {
 
  console.log(text)  // 依次输出："我"、"记得"、"你"、...
 
}

// chat.ts
import { Hono } from 'hono'
 
import { streamSSE } from 'hono/streaming'
 
type Bindings = {
 
  KV: KVNamespace
 
  DB: D1Database
 
  VECTORIZE: VectorizeIndex
 
  AI: Ai
 
  DEEPSEEK_API_KEY: string
 
}
 
const app = new Hono<{ Bindings: Bindings }>()
 
app.post('/chat/stream', async (c) => {
 
  const { message, sessionId } = await c.req.json()
 
  return streamSSE(c, async (stream) => {
 
    try {
 
      // ========== 阶段 1：安全检查 ==========
 
      await stream.writeSSE({
 
        event: 'thinking',
 
        data: JSON.stringify({ node: 'safety_check', status: 'start' })
 
      })
 
      const safetyResult = await runSafetyCheck(message)
 
      if (!safetyResult.safe) {
 
        // 安全拦截：不走后续管线，直接返回兜底话术
 
        await stream.writeSSE({
 
          event: 'error',
 
          data: JSON.stringify({
 
            code: 'safety_blocked',
 
            fallbackReply: '这个话题我们换一个方向聊聊吧~'
 
          })
 
        })
 
        return  // 直接结束，流自动关闭
 
      }
 
      await stream.writeSSE({
 
        event: 'thinking',
 
        data: JSON.stringify({ node: 'safety_check', status: 'end' })
 
      })
 
      // ========== 阶段 2：记忆检索 + 情绪读取（并行） ==========
 
      await stream.writeSSE({
 
        event: 'thinking',
 
        data: JSON.stringify({ node: 'memory_retrieval', status: 'start' })
 
      })
 
      // 这两个操作互不依赖，用 Promise.all 并行执行
 
      // 记忆检索约 50-150ms，情绪读取约 5-15ms
 
      // 并行后总耗时 = max(150, 15) = 150ms，而非 150+15=165ms
 
      const [memories, emotion] = await Promise.all([
 
        retrieveMemories(c.env, message, sessionId),
 
        readEmotion(c.env, sessionId)
 
      ])
 
      await stream.writeSSE({
 
        event: 'thinking',
 
        data: JSON.stringify({ node: 'memory_retrieval', status: 'end' })
 
      })
 
      // ========== 阶段 3：Prompt 组装 ==========
 
      // 把记忆、情绪、人设拼装成完整的 System Prompt
 
      // 这一步是纯 CPU 计算，耗时 < 1ms，不需要 thinking 事件
 
      const systemPrompt = assemblePrompt(memories, emotion)
 
      // ========== 阶段 4：LLM 流式生成 ==========
 
      await stream.writeSSE({
 
        event: 'thinking',
 
        data: JSON.stringify({ node: 'llm_generate', status: 'start' })
 
      })
 
      // callLLMStream 返回 AsyncGenerator，不会阻塞
 
      // 真正的等待发生在 for await 的第一次迭代（等待首个 token）
 
      const llmStream = callLLMStream(
 
        systemPrompt, message, c.env.DEEPSEEK_API_KEY
 
      )
 
      let fullReply = ''
 
      for await (const text of llmStream) {
 
        fullReply += text
 
        // 每个 token 立即推送给前端，不做缓冲
 
        await stream.writeSSE({
 
          event: 'token',
 
          data: JSON.stringify({ content: text })
 
        })
 
      }
 
      // ========== 阶段 5：发送完成事件 ==========
 
      await stream.writeSSE({
 
        event: 'done',
 
        data: JSON.stringify({
 
          emotion: emotion.current,
 
          memoriesUsed: memories.length,
 
          tokensConsumed: fullReply.length
 
        })
 
      })
 
      // ========== 阶段 6：异步后处理 ==========
 
      // waitUntil 告诉 Workers："HTTP 响应已经发完了，
 
      // 但请保持进程存活，直到这个 Promise 完成"
 
      // 这样情绪更新、记忆写入等操作不会因为流关闭而被中断
 
      c.executionCtx.waitUntil(
 
        postProcess(c.env, sessionId, message, fullReply, emotion)
 
      )
 
    } catch (err) {
 
      // 兜底：任何未预期的异常都通过 error 事件通知前端
 
      // 前端收到后展示 fallbackReply，不会白屏
 
      await stream.writeSSE({
 
        event: 'error',
 
        data: JSON.stringify({
 
          code: 'internal_error',
 
          fallbackReply: '抱歉，我刚才走神了，你再说一次好吗？'
 
        })
 
      })
 
    }
 
  })
 
})

// waituntil.ts
// 异步后处理函数：在响应发出后执行
 
async function postProcess(
 
  env: Bindings,
 
  sessionId: string,
 
  userMessage: string,
 
  aiReply: string,
 
  emotion: EmotionState
 
) {
 
  // 这三个操作互不依赖，并行执行
 
  await Promise.all([
 
    // 根据本轮对话更新情绪状态
 
    updateEmotion(env, sessionId, userMessage, aiReply),
 
    // 从对话中提取记忆片段，写入向量库
 
    extractAndWriteMemories(env, sessionId, userMessage, aiReply),
 
    // 更新亲密度分值
 
    updateIntimacy(env, sessionId, emotion)
 
  ])
 
}

// useChat.ts
function useStreamChat() {
 
  const [status, setStatus] = useState<'idle' | 'thinking' | 'generating' | 'done'>('idle')
 
  const [thinkingNode, setThinkingNode] = useState('')
 
  const [reply, setReply] = useState('')
 
  const [metadata, setMetadata] = useState<{ emotion: string } | null>(null)
 
  const sendMessage = useCallback(async (message: string) => {
 
    setStatus('thinking')
 
    setReply('')
 
    let hasDone = false
 
    let partialReply = ''
 
    // 第一步：发起 POST 请求
 
    const response = await fetch('/api/chat/stream', {
 
      method: 'POST',
 
      headers: { 'Content-Type': 'application/json' },
 
      body: JSON.stringify({ message, sessionId: getSessionId() })
 
    })
 
    // 第二步：获取 ReadableStream 的 reader
 
    // 和服务端的 callLLMStream 一样，用 reader 逐块读取
 
    const reader = response.body!.getReader()
 
    const decoder = new TextDecoder()
 
    let buffer = ''
 
    // 第三步：循环读取，解析 SSE 事件
 
    while (true) {
 
      const { done, value } = await reader.read()
 
      if (done) break
 
      buffer += decoder.decode(value, { stream: true })
 
      // 解析缓冲区中所有完整的 SSE 事件
 
      const { parsed, remaining } = parseSSEEvents(buffer)
 
      buffer = remaining
 
      // 根据事件类型更新 UI 状态
 
      for (const event of parsed) {
 
        switch (event.type) {
 
          case 'thinking':
 
            setThinkingNode(event.data.node)
 
            break
 
          case 'token':
 
            setStatus('generating')
 
            partialReply += event.data.content
 
            setReply(prev => prev + event.data.content)
 
            break
 
          case 'done':
 
            hasDone = true
 
            setStatus('done')
 
            setMetadata(event.data)
 
            break
 
          case 'error':
 
            hasDone = true
 
            setStatus('done')
 
            setReply(event.data.fallbackReply ?? '出错了，请重试')
 
            break
 
        }
 
      }
 
    }
 
    // 流结束后，不要读取闭包里可能过期的 React state
 
    // 用局部变量判断是否正常收到了 done / error 事件
 
    if (!hasDone) {
 
      if (partialReply.length > 0) {
 
        setReply(prev => prev + '\n\n[回复未完成，点击重试]')
 
      } else {
 
        setReply('网络不太稳定，请重新发送消息')
 
      }
 
      setStatus('done')
 
    }
 
  }, [])
 
  return { status, thinkingNode, reply, metadata, sendMessage }
 
}

// parser.ts
function parseSSEEvents(buffer: string) {
 
  const events: Array<{ type: string; data: any }> = []
 
  const parts = buffer.split('\n\n')
 
  const remaining = parts.pop() ?? ''
 
  for (const part of parts) {
 
    if (!part.trim()) continue
 
    let eventType = 'message'
 
    let data = ''
 
    for (const line of part.split('\n')) {
 
      if (line.startsWith('event: ')) eventType = line.slice(7)
 
      else if (line.startsWith('data: ')) data = line.slice(6)
 
    }
 
    if (data) {
 
      events.push({ type: eventType, data: JSON.parse(data) })
 
    }
 
  }
 
  return { parsed: events, remaining }
 
}

// ChatBubble.tsx
function AIBubble({ status, thinkingNode, reply, metadata }: AIBubbleProps) {
 
  return (
 
    <div className="flex gap-3 items-start">
 
      <Avatar emotion={metadata?.emotion} />
 
      <div className="flex-1 space-y-2">
 
        {/* 思考阶段：显示当前执行的管线节点 */}
 
        {status === 'thinking' && (
 
          <ThinkingIndicator node={thinkingNode} />
 
        )}
 
        {/* 生成阶段：逐字显示回复 + 光标闪烁 */}
 
        {(status === 'generating' || status === 'done') && (
 
          <div className="prose prose-sm">
 
            <MarkdownRenderer content={reply} />
 
            {status === 'generating' && <BlinkingCursor />}
 
          </div>
 
        )}
 
        {/* 完成后：显示情绪标签 */}
 
        {status === 'done' && metadata && (
 
          <EmotionTag emotion={metadata.emotion} />
 
        )}
 
      </div>
 
    </div>
 
  )
 
}

// ThinkingIndicator.tsx
const nodeLabels: Record<string, string> = {
 
  safety_check: '安全检查中',
 
  memory_retrieval: '回忆与你的过往',
 
  llm_generate: '组织语言中'
 
}
 
function ThinkingIndicator({ node }: { node: string }) {
 
  return (
 
    <div className="flex items-center gap-2 text-sm text-zinc-400">
 
      <LoadingDots />
 
      <span>{nodeLabels[node] ?? '思考中'}</span>
 
    </div>
 
  )
 
}

// timeout.ts
async function forwardWithTimeout(
 
  stream: { writeSSE: Function; close: Function },
 
  llmIterator: AsyncGenerator<string>,
 
  gapMs: number = 15000  // 两个 token 之间最多等 15 秒
 
) {
 
  let timer: ReturnType<typeof setTimeout>
 
  const resetTimer = () => {
 
    clearTimeout(timer)
 
    timer = setTimeout(async () => {
 
      await stream.writeSSE({
 
        event: 'error',
 
        data: JSON.stringify({
 
          code: 'llm_timeout',
 
          fallbackReply: '我刚才想说的太多了，脑子有点转不过来...你能再说一次吗？'
 
        })
 
      })
 
      stream.close()
 
    }, gapMs)
 
  }
 
  resetTimer()  // 启动首次计时
 
  try {
 
    for await (const text of llmIterator) {
 
      resetTimer()  // 每收到一个 token，重置计时器
 
      await stream.writeSSE({
 
        event: 'token',
 
        data: JSON.stringify({ content: text })
 
      })
 
    }
 
  } finally {
 
    clearTimeout(timer)
 
  }
 
}

// recovery.ts
// 在 while 循环结束后追加
 
if (!hasDone) {
 
  // 流异常终止，没有收到 done / error 事件
 
  if (partialReply.length > 0) {
 
    // 已经有部分内容：保留已显示的文字，提示可重试
 
    setReply(prev => prev + '\n\n[回复未完成，点击重试]')
 
  } else {
 
    // 一个字都没收到：直接提示网络问题
 
    setReply('网络不太稳定，请重新发送消息')
 
  }
 
  setStatus('done')
 
}

// idempotent.ts
async function postProcess(env: Bindings, requestId: string, /* ... */) {
 
  // 用 KV 做去重锁，TTL 5 分钟
 
  const lockKey = `post_process_lock:${requestId}`
 
  const existing = await env.KV.get(lockKey)
 
  if (existing) return  // 已经处理过，跳过
 
  await env.KV.put(lockKey, '1', { expirationTtl: 300 })
 
  await Promise.all([
 
    updateEmotion(env, /* ... */),
 
    writeMemories(env, /* ... */),
 
    updateIntimacy(env, /* ... */)
 
  ])
 
}