为什么需要流式输出？#

1
class AiChatController extends Controller
2
{
3
    public function stream(Request $request)
4
    {
5
        $param = $this->validate($request, ChatValidate::stream());
6

7
        // 设置 SSE 响应头
8
        $connection = $request->connection;
9
        $connection->send(new Response(200, [
10
            'Content-Type' => 'text/event-stream',
11
            'Cache-Control' => 'no-cache',
12
            'Connection' => 'keep-alive',
13
            'X-Accel-Buffering' => 'no', // 告诉 Nginx 不要缓冲
14
        ]));
15

16
        // 获取智能体和模型
17
        $agent = AiAgents::find($param['agent_id']);
18
        $model = AiModels::find($agent->model_id);
19

20
        // 构建消息上下文
21
        $messages = $this->buildMessages($agent, $param);
22

23
        // 流式调用大模型
24
        $fullContent = '';
25
        $this->aiService->streamChat($messages, $model, function ($chunk) use ($connection, &$fullContent) {
26
            $fullContent .= $chunk;
27

28
            // SSE 数据格式：data: {json}\n\n
29
            $connection->send("data: " . json_encode([
30
                'type' => 'content',
31
                'content' => $chunk,
32
            ]) . "\n\n");
33
        });
34

35
        // 发送结束信号
36
        $connection->send("data: " . json_encode([
37
            'type' => 'done',
38
            'content' => $fullContent,
39
        ]) . "\n\n");
40

41
        // 异步保存对话记录（不阻塞响应）
42
        $this->asyncSaveMessage($param, $fullContent);
43
    }
44
}

1
interface StreamOptions {
2
  url: string
3
  data: Record<string, any>
4
  onMessage: (chunk: string) => void
5
  onDone?: (fullContent: string) => void
6
  onError?: (error: Error) => void
7
}
8

9
export function streamPost(options: StreamOptions): AbortController {
10
  const controller = new AbortController()
11

12
  fetch(options.url, {
13
    method: 'POST',
14
    headers: {
15
      'Content-Type': 'application/json',
16
      'Authorization': `Bearer ${getToken()}`,
17
    },
18
    body: JSON.stringify(options.data),
19
    signal: controller.signal,
20
  }).then(async (response) => {
21
    if (!response.ok) {
22
      throw new Error(`HTTP ${response.status}`)
23
    }
24

25
    const reader = response.body!.getReader()
26
    const decoder = new TextDecoder()
27
    let buffer = ''
28

29
    while (true) {
30
      const { done, value } = await reader.read()
31
      if (done) break
32

33
      buffer += decoder.decode(value, { stream: true })
34

35
      // 按 SSE 协议解析：以 \n\n 分割消息
36
      const messages = buffer.split('\n\n')
37
      buffer = messages.pop() || '' // 最后一个可能是不完整的
38

39
      for (const msg of messages) {
40
        if (!msg.startsWith('data: ')) continue
41
        const jsonStr = msg.slice(6)
42

43
        try {
44
          const data = JSON.parse(jsonStr)
45
          if (data.type === 'content') {
46
            options.onMessage(data.content)
47
          } else if (data.type === 'done') {
48
            options.onDone?.(data.content)
49
          }
50
        } catch {
51
          // 忽略解析失败的消息
52
        }
53
      }
54
    }
55
  }).catch((err) => {
56
    if (err.name === 'AbortError') return // 用户主动中断，不算错误
57
    options.onError?.(err)
58
  })
59

60
  return controller // 返回 controller，用于中断
61
}

1
const controller = streamPost({
2
  url: '/api/ai/chat/stream',
3
  data: { agent_id: 1, messages: [...] },
4
  onMessage(chunk) {
5
    // 逐字追加到界面
6
    currentMessage.value += chunk
7
  },
8
  onDone(fullContent) {
9
    // 完成后刷新对话列表
10
    refreshMessages()
11
  },
12
  onError(err) {
13
    ElNotification.error({ message: '对话失败：' + err.message })
14
  },
15
})
16

17
// 用户点击"停止生成"
18
stopButton.onclick = () => controller.abort()

1
$stopped = false;
2

3
$connection->onClose = function () use (&$stopped) {
4
    $stopped = true;
5
};
6

7
$this->aiService->streamChat($messages, $model, function ($chunk) use ($connection, &$stopped, &$fullContent) {
8
    if ($stopped) {
9
        throw new StreamInterruptedException();
10
    }
11

12
    $fullContent .= $chunk;
13
    $connection->send("data: " . json_encode([
14
        'type' => 'content',
15
        'content' => $chunk,
16
    ]) . "\n\n");
17
});

1
try {
2
    // 流式输出...
3
} catch (StreamInterruptedException $e) {
4
    // 中断不是错误，保存已生成的内容
5
} finally {
6
    if ($fullContent) {
7
        $this->saveMessage($param, $fullContent, $stopped ? 'interrupted' : 'completed');
8
    }
9
}

1
// 对话完成后，异步生成标题
2
if ($isFirstMessage) {
3
    Redis::send('chat-queue', [
4
        'action' => 'generate_title',
5
        'conversation_id' => $param['conversation_id'],
6
        'first_message' => $param['messages'][0]['content'],
7
    ]);
8
}

1
private function generateTitle(array $data): void
2
{
3
    $result = $this->aiService->chat([
4
        ['role' => 'system', 'content' => '用10个字以内概括以下对话的主题，直接输出标题，不要解释'],
5
        ['role' => 'user', 'content' => $data['first_message']],
6
    ], $this->getLightModel());
7

8
    Conversation::where('id', $data['conversation_id'])
9
        ->update(['title' => $result->content]);
10
}

1
// 1. 超时控制
2
$this->aiService->streamChat($messages, $model, $callback, [
3
    'timeout' => 120, // 最长等待 2 分钟
4
]);
5

6
// 2. 重试机制（仅对可重试的错误）
7
$retryableErrors = [429, 500, 502, 503];
8
$maxRetries = 2;
9

10
for ($i = 0; $i <= $maxRetries; $i++) {
11
    try {
12
        return $this->callModel($messages, $model);
13
    } catch (ApiException $e) {
14
        if (!in_array($e->getCode(), $retryableErrors) || $i === $maxRetries) {
15
            throw $e;
16
        }
17
        sleep(pow(2, $i)); // 指数退避：1s, 2s, 4s
18
    }
19
}
20

21
// 3. 前端友好提示
22
// 不要把原始错误信息暴露给用户
23
$userMessage = match (true) {
24
    $e->getCode() === 429 => '当前使用人数较多，请稍后再试',
25
    $e->getCode() >= 500 => 'AI 服务暂时不可用，请稍后再试',
26
    default => '对话失败，请重试',
27
};