23.09-流式聊天UI对接

// src/index.ts
import { Hono } from 'hono'
import { streamSSE } from 'hono/streaming'
 
const app = new Hono()
 
app.post('/v1/chat/completions/stream', async (c) => {
  const body = await c.req.json()
 
  return streamSSE(c, async (stream) => {
    // 调用 LLM API（流式）
    const llmStream = await callLLMStream(body)
 
    for await (const chunk of llmStream) {
      // 发送 SSE 事件
      await stream.writeSSE({
        event: 'message',
        data: JSON.stringify({
          choices: [{
            delta: {
              content: chunk.content,
            },
          }],
        }),
      })
    }
 
    // 发送结束事件
    await stream.writeSSE({
      event: 'done',
      data: '[DONE]',
    })
  })
})
 
export default app

// 模拟 LLM 流式响应
async function* mockLLMStream(text: string) {
  const words = text.split(' ')
  for (const word of words) {
    await new Promise(resolve => setTimeout(resolve, 100))
    yield { content: word + ' ' }
  }
}
 
app.post('/v1/chat/completions/stream', async (c) => {
  const body = await c.req.json()
  const message = body.messages[body.messages.length - 1].content
 
  return streamSSE(c, async (stream) => {
    const mockText = `这是一个模拟的 AI 回复，你说了：${message}`
 
    for await (const chunk of mockLLMStream(mockText)) {
      await stream.writeSSE({
        event: 'message',
        data: JSON.stringify({
          choices: [{
            delta: {
              content: chunk.content,
            },
          }],
        }),
      })
    }
 
    await stream.writeSSE({
      event: 'done',
      data: '[DONE]',
    })
  })
})

// src/lib/stream-chat.ts
export async function streamChat(
  message: string,
  onChunk: (content: string) => void,
  onDone: () => void,
  onError: (error: Error) => void
) {
  const response = await fetch('/api/chat/stream', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      messages: [{ role: 'user', content: message }],
    }),
  })
 
  if (!response.ok) {
    onError(new Error(`Request failed: ${response.status}`))
    return
  }
 
  const reader = response.body!.getReader()
  const decoder = new TextDecoder()
  let buffer = ''
 
  while (true) {
    const { done, value } = await reader.read()
 
    if (done) break
 
    buffer += decoder.decode(value, { stream: true })
 
    // 解析 SSE 数据
    const lines = buffer.split('\n')
    buffer = lines.pop() || ''
 
    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6)
 
        if (data === '[DONE]') {
          onDone()
          return
        }
 
        try {
          const parsed = JSON.parse(data)
          const content = parsed.choices?.[0]?.delta?.content
 
          if (content) {
            onChunk(content)
          }
        } catch (err) {
          console.error('Failed to parse SSE data:', err)
        }
      }
    }
  }
 
  onDone()
}

// src/components/Chat.tsx
import { useState } from 'react'
import { streamChat } from '@/lib/stream-chat'
 
export function Chat() {
  const [input, setInput] = useState('')
  const [messages, setMessages] = useState([])
  const [streaming, setStreaming] = useState(false)
 
  async function handleSend() {
    const userMessage = { role: 'user', content: input }
    setMessages(prev => [...prev, userMessage])
    setInput('')
    setStreaming(true)
 
    let assistantMessage = ''
 
    await streamChat(
      input,
      // onChunk
      (content) => {
        assistantMessage += content
        setMessages(prev => {
          const last = prev[prev.length - 1]
          if (last?.role === 'assistant') {
            return [...prev.slice(0, -1), { ...last, content: assistantMessage }]
          }
          return [...prev, { role: 'assistant', content: assistantMessage }]
        })
      },
      // onDone
      () => {
        setStreaming(false)
      },
      // onError
      (error) => {
        console.error(error)
        setStreaming(false)
      }
    )
  }
 
  return (
    <div>
      <div className="messages">
        {messages.map((msg, i) => (
          <div key={i} className={msg.role}>
            {msg.content}
          </div>
        ))}
      </div>
      <input
        value={input}
        onChange={(e) => setInput(e.target.value)}
        onKeyPress={(e) => e.key === 'Enter' && handleSend()}
        disabled={streaming}
      />
      <button onClick={handleSend} disabled={streaming}>
        {streaming ? '回复中...' : '发送'}
      </button>
    </div>
  )
}

// 注意：EventSource 只支持 GET 请求，POST 请求需要用 fetch
// 如果后端支持 GET 方式的流式响应，可以使用 EventSource
 
const eventSource = new EventSource('/api/chat/stream?message=你好')
 
eventSource.addEventListener('message', (event) => {
  const data = JSON.parse(event.data)
  const content = data.choices?.[0]?.delta?.content
 
  if (content) {
    console.log('Chunk:', content)
  }
})
 
eventSource.addEventListener('done', () => {
  console.log('Stream done')
  eventSource.close()
})
 
eventSource.onerror = (error) => {
  console.error('Stream error:', error)
  eventSource.close()
}

// src/hooks/useStreamChat.ts
import { useState, useCallback } from 'react'
 
interface Message {
  role: 'user' | 'assistant'
  content: string
}
 
export function useStreamChat() {
  const [messages, setMessages] = useState<Message[]>([])
  const [streaming, setStreaming] = useState(false)
  const [error, setError] = useState<Error | null>(null)
 
  const sendMessage = useCallback(async (content: string) => {
    setMessages(prev => [...prev, { role: 'user', content }])
    setStreaming(true)
    setError(null)
 
    let assistantContent = ''
 
    try {
      const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          messages: [...messages, { role: 'user', content }],
        }),
      })
 
      if (!response.ok) {
        throw new Error(`Request failed: ${response.status}`)
      }
 
      const reader = response.body!.getReader()
      const decoder = new TextDecoder()
      let buffer = ''
 
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
 
        buffer += decoder.decode(value, { stream: true })
        const lines = buffer.split('\n')
        buffer = lines.pop() || ''
 
        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = line.slice(6)
 
            if (data === '[DONE]') {
              setStreaming(false)
              return
            }
 
            try {
              const parsed = JSON.parse(data)
              const chunk = parsed.choices?.[0]?.delta?.content
 
              if (chunk) {
                assistantContent += chunk
                setMessages(prev => {
                  const last = prev[prev.length - 1]
                  if (last?.role === 'assistant') {
                    return [...prev.slice(0, -1), { role: 'assistant', content: assistantContent }]
                  }
                  return [...prev, { role: 'assistant', content: assistantContent }]
                })
              }
            } catch (err) {
              console.error('Failed to parse SSE:', err)
            }
          }
        }
      }
    } catch (err) {
      setError(err as Error)
      setStreaming(false)
    }
  }, [messages])
 
  return {
    messages,
    streaming,
    error,
    sendMessage,
  }
}

// 使用
import { useStreamChat } from '@/hooks/useStreamChat'
 
function Chat() {
  const { messages, streaming, error, sendMessage } = useStreamChat()
  const [input, setInput] = useState('')
 
  return (
    <div>
      {messages.map((msg, i) => (
        <div key={i}>{msg.role}: {msg.content}</div>
      ))}
      {error && <div>错误：{error.message}</div>}
      <input
        value={input}
        onChange={(e) => setInput(e.target.value)}
        disabled={streaming}
      />
      <button onClick={() => sendMessage(input)} disabled={streaming}>
        {streaming ? '回复中...' : '发送'}
      </button>
    </div>
  )
}

export async function streamChatWithRetry(
  message: string,
  maxRetries = 3
): Promise<string> {
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      let content = ''
 
      await streamChat(
        message,
        (chunk) => { content += chunk },
        () => {},
        (error) => { throw error }
      )
 
      return content
    } catch (error) {
      if (attempt === maxRetries) {
        throw error
      }
 
      console.warn(`Attempt ${attempt} failed, retrying...`)
      await new Promise(resolve => setTimeout(resolve, 1000 * attempt))
    }
  }
 
  throw new Error('Max retries exceeded')
}

export function useStreamChat() {
  const [streaming, setStreaming] = useState(false)
  const abortControllerRef = useRef<AbortController | null>(null)
 
  const sendMessage = useCallback(async (content: string) => {
    abortControllerRef.current = new AbortController()
 
    try {
      const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ messages: [{ role: 'user', content }] }),
        signal: abortControllerRef.current.signal,
      })
 
      // ... 处理流式响应
    } catch (error) {
      if (error.name === 'AbortError') {
        console.log('Request cancelled')
      } else {
        throw error
      }
    } finally {
      setStreaming(false)
    }
  }, [])
 
  const cancel = useCallback(() => {
    abortControllerRef.current?.abort()
    setStreaming(false)
  }, [])
 
  return {
    streaming,
    sendMessage,
    cancel,
  }
}