# AI 流式响应实战：从同步等待到实时推送
在 IM 系统中集成 AI 时，流式响应能显著提升性能。本文介绍 AQChat 如何实现 AI 流式响应，从同步等待到实时推送。

## 一、为什么需要流式响应？
### 同步等待的问题
**传统同步方式的问题**：
```java
// ❌ 同步方式：用户需要等待AI完整响应
String aiResponse = aiService.getAnswer(userMessage);
// 如果AI响应需要10秒，用户就要等待10秒
sendMessage(aiResponse);
```
问题：
1. 等待时间长：AI 生成可能需要 5-10 秒，用户长时间等待
2. 体验差：无法看到生成过程，感觉卡顿
3. 资源占用：长时间占用连线和线程

**流式响应的优势**：
1. 实时反馈：逐字显示，用户可立即看到内容
2. 体验更好：类似 ChatGPT 的打字机效果
3. 资源利用：边生成边推送，不阻塞

**对比**：
| **方式** | **首字延迟** | **完整响应时间** | **用户体验** |
|------|-------|----------|--------|
| 同步等待 | 10秒 | 10秒 | 差 |
| 流式响应 | 1-2秒 | 10秒 | 好 |

## 回调函数模式的设计
### 统一接口设计
**定义统一的 AI 服务接口**：
```java
public interface IAiService {
    /**
     * 流式调用AI服务
     * @param userMsg 用户消息
     * @param consumer 回调函数，处理每个数据块
     */
    void streamCallWithMessage(String userMsg, Consumer<AIResult> consumer);
    
    /**
     * 多轮对话
     */
    default void chat(String message, List<MessageRecord> messages, Consumer<AIResult> consumer) {
    }
}
```

**关键点**：
* 使用 `Consumer<AIResult>` 作为回调
* 每个数据块通过回调处理
* 支持多轮对话

**AIResult 设计**
```java
public interface AIResult {
    String getContent();  // 当前数据块的内容
    int getStatus();      // 状态：WAIT(0-进行中)、END(1-结束)、FAIL(2-失败)
}
```
**状态枚举**：
```java
public enum AIMessageStatusEnum {
    WAIT(0, "wait"),   // 流式响应进行中
    END(1, "end"),     // 流式响应结束
    FAIL(2, "fail");   // 流式响应失败
}
```

## 三、WebSocket 实时推送的实现
### 整体流程
```java
用户发送消息
    ↓
RocketMQ 异步处理
    ↓
AI服务流式调用
    ↓
回调函数处理每个数据块
    ↓
封装为 STREAM_MSG_NOTIFY
    ↓
WebSocket 实时推送
```

### 代码实现
1. **RocketMQ 消费者接收消息**
```java
@Component
public class AIHelperReceiver implements InitializingBean {
    
    @Resource
    private IAiService aiService;
    
    @Resource
    private GlobalChannelHolder globalChannelHolder;
    
    public void initConsumer() {
        defaultMQPushConsumer.setMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            for (MessageExt messageExt : messageExtList) {
                MessageDto messageDto = JSONObject.parseObject(msgStr, MessageDto.class);
                
                // 提交到独立线程池，不阻塞MQ消费线程
                threadPoolUtil.submitTask(() -> {
                    StringBuilder fullContent = new StringBuilder();
                    
                    try {
                        // 流式调用AI服务
                        aiService.streamCallWithMessage(
                            messageDto.getMessageContent(), 
                            aiResult -> {
                                // 回调函数：处理每个数据块
                                AIMessageDto aiMessageDto = new AIMessageDto();
                                aiMessageDto.setMessageId(messageDto.getMessageId());
                                aiMessageDto.setRoomId(messageDto.getRoomId());
                                aiMessageDto.setContent(aiResult.getContent());
                                aiMessageDto.setStatus(aiResult.getStatus());
                                
                                // 实时推送
                                globalChannelHolder.sendBroadcastAIMessage(
                                    aiMessageDto, 
                                    AQBusinessConstant.AI_HELPER_ID
                                );
                                
                                // 累积完整内容
                                fullContent.append(aiResult.getContent());
                            }
                        );
                    } catch (Exception e) {
                        // 错误处理
                        LOGGER.error("AI助手处理消息失败", e);
                        AIMessageDto failMessage = new AIMessageDto();
                        failMessage.setStatus(AIMessageStatusEnum.FAIL.getCode());
                        globalChannelHolder.sendBroadcastAIMessage(failMessage, AQBusinessConstant.AI_HELPER_ID);
                    } finally {
                        // 流式响应结束后，持久化完整消息
                        MessageDto storeMessage = buildStoreMessage(messageDto, fullContent);
                        messageService.saveMessage(storeMessage);
                    }
                });
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }
}
```

2. **封装流式消息并推送**
```java
@Component
public class GlobalChannelHolder {
    
    public void sendBroadcastAIMessage(AIMessageDto aiMessageDto, String aiId) {
        // 1. 获取AI助手信息
        UserGlobalInfoDto userInfo = userHolder.getUserInfo(aiId);
        
        // 2. 构建流式消息
        AQChatMsgProtocol.StreamMsgNotify streamMsgNotify = 
            AQChatMsgProtocol.StreamMsgNotify.newBuilder()
                .setUser(userBuilder)
                .setMsgId(aiMessageDto.getMessageId())
                .setRoomId(aiMessageDto.getRoomId())
                .setContent(aiMessageDto.getContent() == null ? "" : aiMessageDto.getContent())
                .setStreamType(aiMessageDto.getStatus())  // 0-进行中，1-结束，2-失败
                .build();
        
        // 3. 广播到房间内所有用户
        messageBroadcaster.broadcast(aiMessageDto.getRoomId(), streamMsgNotify);
    }
}
```

3. **消息广播**
```java
@Component
public class MessageBroadcaster {
    
    private final Map<String, ChannelGroup> channelGroupMap = new ConcurrentHashMap<>();
    
    public <T extends GeneratedMessageV3> void broadcast(String roomId, T msg) {
        ChannelGroup channelGroup = channelGroupMap.get(roomId);
        if (channelGroup != null) {
            // 批量发送，高效
            channelGroup.writeAndFlush(msg);
        }
    }
}
```

## 四、流式消息的封装（STREM_MSG_NOTIFY）
**Protobuf 消息定义**
```java
// 流式消息通知
message StreamMsgNotify {
    string roomId = 1;      // 房间ID
    string msgId = 2;       // 消息ID
    User user = 3;          // AI助手信息
    int32 streamType = 4;   // 流类型：0-进行中，1-结束，2-失败
    string content = 5;     // 当前数据块内容
}
```

**消息类型**：
```java
enum MsgCommand {
    // ...
    STREAM_MSG_NOTIFY = 32;  // 流式消息通知
    // ...
}
```

**消息状态流转**
```java
用户发送消息
    ↓
STREAM_MSG_NOTIFY (streamType=0, content="你")  ← 第一个数据块
    ↓
STREAM_MSG_NOTIFY (streamType=0, content="好")  ← 第二个数据块
    ↓
STREAM_MSG_NOTIFY (streamType=0, content="，")  ← 第三个数据块
    ↓
...
    ↓
STREAM_MSG_NOTIFY (streamType=1, content="")   ← 结束标志
```

**前端处理示例（伪代码）**：
```js
websocket.onmessage = (event) => {
    const message = JSON.parse(event.data);
    
    if (message.command === 'STREAM_MSG_NOTIFY') {
        if (message.streamType === 0) {
            // 进行中：追加内容
            appendContent(message.content);
        } else if (message.streamType === 1) {
            // 结束：显示完整消息
            showCompleteMessage();
        } else if (message.streamType === 2) {
            // 失败：显示错误提示
            showErrorMessage();
        }
    }
};
```

## 五、多 AI 平台集成的统一接口设计
### 问题：不同 AI 平台的 API 不同
* 阿里百炼：使用 `Flowable<GenerationResult>`
* Gitee AI：使用 `MessageHandler<String>`
* 其他平台：可能有不同的流式接口

### 解决方案：统一接口 + 适配器模式
1. **统一接口定义**
```java
public interface IAiService {
    /**
     * 流式调用，统一使用 Consumer<AIResult> 回调
     */
    void streamCallWithMessage(String userMsg, Consumer<AIResult> consumer);
}
```

2. **阿里百炼实现**
```java
@Service
@Primary
public class QWAiService implements IAiService {
    
    @Override
    public void streamCallWithMessage(String userMsg, Consumer<AIResult> consumer) {
        Generation gen = new Generation();
        Message message = Message.builder()
            .role(Role.USER.getValue())
            .content(userMsg)
            .build();
        
        // 调用阿里百炼流式API
        Flowable<GenerationResult> result = gen.streamCall(generationParam);
        
        // 转换为统一格式
        result.blockingForEach(r -> {
            String content = r.getOutput().getChoices().get(0).getMessage().getContent();
            String finishReason = r.getOutput().getChoices().get(0).getFinishReason();
            
            QWResult qwResult = new QWResult();
            qwResult.setContent(content);
            // 判断是否结束
            qwResult.setStatus("stop".equals(finishReason) 
                ? AIMessageStatusEnum.END.getCode() 
                : AIMessageStatusEnum.WAIT.getCode());
            
            // 调用统一回调
            consumer.accept(qwResult);
        });
    }
}
```

3. **Gitee AI 实现**
```java
@Service
public class GiteeAIService implements IAiService {
    
    @Resource
    private GiteeAIClient giteeAIClient;
    
    @Override
    public void streamCallWithMessage(String userMsg, Consumer<AIResult> consumer) {
        // 调用Gitee AI流式API
        giteeAIClient.streamChat(message, messageList, data -> {
            JSONObject parse = JSONObject.parseObject(data);
            JSONArray choices = parse.getJSONArray("choices");
            JSONObject choicesIn = choices.getJSONObject(0);
            String finishReason = choicesIn.getString("finish_reason");
            
            if (finishReason != null && finishReason.equals("stop")) {
                // 结束
                GiteeResult giteeResult = new GiteeResult();
                giteeResult.setStatus(AIMessageStatusEnum.END.getCode());
                consumer.accept(giteeResult);
                return;
            }
            
            // 进行中
            JSONObject delta = choicesIn.getJSONObject("delta");
            String content = delta.getString("content");
            if (content != null && !content.isEmpty()) {
                GiteeResult giteeResult = new GiteeResult();
                giteeResult.setContent(content);
                giteeResult.setStatus(AIMessageStatusEnum.WAIT.getCode());
                consumer.accept(giteeResult);
            }
        });
    }
}
```

### 统一接口的优势
1. 业务代码无需关心具体平台
2. 易于扩展新平台
3. 便于切换平台（通过 `@Prime` 注解）

**使用示例**
```java
// 业务代码只需要调用统一接口
@Resource
private IAiService aiService;  // Spring会自动注入@Primary的实现

aiService.streamCallWithMessage(userMsg, aiResult -> {
    // 处理流式响应，不关心是哪个AI平台
    sendBroadcastAIMessage(aiResult);
});
```

## 六、性能优化
1. **独立线程池**
```java
// AI处理在独立线程池中执行，不阻塞MQ消费线程
threadPoolUtil.submitTask(() -> {
    aiService.streamCallWithMessage(userMsg, consumer);
});
```

**优势**：
* 不阻塞 RocketMQ 消费线程
* AI 处理失败不影响其他消息
* 可控制并发数

2. **异步处理**
```java
// 消息发送到RocketMQ，异步处理
mqSendingAgent.aiHelper(messageDto);
// 立即返回，不等待AI响应
```

**优势**：
* 用户发送消息后立即返回
* AI 响应通过 WebSocket 实时推送
* 提升响应速度

## 七、总结
**关键点**
1. 流式响应：使用回调函数模式，实时推送每个数据块
2. 统一接口：`IAiservice` 统一不同 AI 平台的接口
3. WebSocket 推送：通过 `STREAM_MSG_NOTIFY` 实时推送
4. 异步处理：使用 RocketMQ + 独立线程池，不阻塞主流程

**优化效果**
| **指标** | **同步** | **流式响应** | **提升** |
|------|------|--------|------|
| 首字延迟 | 10秒 | 1-2秒 | 5-10倍 |
| 用户体验 | 差 | 好 | 显著提升 |
| 资源占用 | 高 | 低 | 降低 |

**经验总结**
1. 流式响应能显著提升性能
2. 统一接口便于多平台集成
3. 异步处理避免阻塞
4. 回调函数模式适合流式场景

通过以上实现，AQChat 实现了类似 ChatGPT 的流式响应效果，提升了用户体验。

在 IM 系统中集成 AI 时，流式响应能显著提升性能。本文介绍 AQChat 如何实现 AI 流式响应，从同步等待到实时推送。

AI 流式响应实战：从同步等待到实时推送

人工智能

后端

网络与通信

本文详解AI流式响应在IM系统的实现，通过WebSocket实时推送、统一接口设计和异步处理优化，将首字延迟从10秒降至1-2秒，显著提升用户体验。涵盖回调函数模式、多AI平台集成方案及性能优化策略。

ChatGPT

JSON

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

AI 流式响应实战：从同步等待到实时推送-腾讯云开发者社区-腾讯云

AI 流式响应实战：从同步等待到实时推送

AI 流式响应实战：从同步等待到实时推送

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐