如何在Android中基于Retrofit实现OpenAI GPT-3.5 Turbo API流式响应

核心原理 #

OpenAI流式响应采用SSE（Server-Sent Events）协议，返回内容为分块文本，每块格式为data: {JSON对象}\n\n，最后以data: [DONE]标记响应结束。要实现实时展示，需逐行读取响应流、解析单块数据并即时更新UI，而非等待完整响应返回。

优化后代码示例 #

1. 确认API接口定义（无需修改） #

public interface ChatApiService {
    @POST("chat/completions")
    Call<ResponseBody> getChatResponse(@Body RequestBody requestBody);
}

2. 修改callAPI方法实现流式处理 #

public void callAPI(String question) {
    OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
    httpClientBuilder.connectTimeout(60, TimeUnit.SECONDS);
    // 流式响应可能持续较久，设置无读取超时或更长时间
    httpClientBuilder.readTimeout(0, TimeUnit.SECONDS); 
    httpClientBuilder.writeTimeout(60, TimeUnit.SECONDS);

    Retrofit retrofit = new Retrofit.Builder()
            .baseUrl("https://api.openai.com/v1/")
            .client(httpClientBuilder.build())
            .addConverterFactory(GsonConverterFactory.create())
            .build();

    ChatApiService chatApiService = retrofit.create(ChatApiService.class);

    JSONObject jsonBody = new JSONObject();
    try {
        jsonBody.put("model", "gpt-3.5-turbo");
        jsonBody.put("max_tokens", 4000);
        jsonBody.put("temperature", 0);
        jsonBody.put("stream", true); // 必须开启流式模式
        JSONArray messageArray = new JSONArray();

        JSONObject userMessage = new JSONObject();
        userMessage.put("role", "user");
        userMessage.put("content", question);
        messageArray.put(userMessage);

        JSONObject assistantMessage = new JSONObject();
        assistantMessage.put("role", "assistant");
        assistantMessage.put("content", SharedPreference.getString(context, BaseUrl.Key_last_answer));
        messageArray.put(assistantMessage);

        jsonBody.put("messages", messageArray);
    } catch (JSONException e) {
        e.printStackTrace();
    }

    RequestBody requestBody = RequestBody.create(MediaType.parse("application/json"), jsonBody.toString());

    Call<ResponseBody> call = chatApiService.getChatResponse(requestBody);
    call.enqueue(new Callback<ResponseBody>() {
        private final StringBuilder fullAnswer = new StringBuilder();
        private final Handler mainHandler = new Handler(Looper.getMainLooper());

        @Override
        public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
            if (!response.isSuccessful()) {
                handleErrorResponse(response);
                return;
            }

            // 后台线程处理响应流，避免阻塞主线程
            new Thread(() -> {
                BufferedReader reader = null;
                try {
                    ResponseBody body = response.body();
                    if (body == null) {
                        mainHandler.post(() -> addResponse("Empty response from server"));
                        return;
                    }
                    reader = new BufferedReader(new InputStreamReader(body.source().inputStream()));
                    String line;
                    while ((line = reader.readLine()) != null) {
                        // 过滤空行和非data开头的无效行
                        if (line.isEmpty() || !line.startsWith("data: ")) {
                            continue;
                        }
                        String dataStr = line.substring(6).trim();
                        // 检测响应结束标记
                        if ("[DONE]".equals(dataStr)) {
                            break;
                        }
                        // 解析单块响应数据
                        JSONObject jsonObject = new JSONObject(dataStr);
                        JSONArray choices = jsonObject.getJSONArray("choices");
                        JSONObject delta = choices.getJSONObject(0).getJSONObject("delta");
                        if (delta.has("content")) {
                            String chunk = delta.getString("content");
                            fullAnswer.append(chunk);
                            // 切换到主线程更新UI
                            mainHandler.post(() -> appendResponseChunk(chunk));
                        }
                    }
                    // 响应完成后的收尾操作
                    mainHandler.post(() -> {
                        String finalAnswer = fullAnswer.toString().trim();
                        addToChatHistory();
                        speakAnswer();
                        SharedPreference.putString(context, BaseUrl.Key_last_answer, finalAnswer);
                    });
                } catch (IOException | JSONException e) {
                    e.printStackTrace();
                    mainHandler.post(() -> addResponse("Failed to process stream: " + e.getMessage()));
                } finally {
                    // 释放资源
                    if (reader != null) {
                        try {
                            reader.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (response.body() != null) {
                        response.body().close();
                    }
                }
            }).start();
        }

        @Override
        public void onFailure(Call<ResponseBody> call, Throwable t) {
            addResponse("Failed to load response due to " + t.getMessage());
        }

        private void handleErrorResponse(Response<ResponseBody> response) {
            if (response.code() == 429) {
                addResponse("Oops, something went wrong. Please try again in a little while.");
            } else {
                if (response.errorBody() != null) {
                    try {
                        addResponse("Failed to load response due to " + response.errorBody().string());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    });
}

// 新增：实时追加响应内容到UI的方法
private void appendResponseChunk(String chunk) {
    // 这里实现向聊天窗口追加文本的逻辑，例如：
    // tvAnswer.setText(tvAnswer.getText().toString() + chunk);
}

关键优化点说明 #

• 超时配置：将readTimeout设为0（无超时），避免流式响应过程中因长时间无数据返回被判定为超时。
• 线程分离：用单独线程读取响应流，防止阻塞主线程导致ANR。
• 逐块解析：过滤无效行，提取data:字段内容，解析每块delta中的content实现实时展示。
• UI线程切换：通过Handler将UI更新操作切换到主线程，符合Android线程规范。
• 资源释放：完成流读取后关闭BufferedReader和ResponseBody，避免内存泄漏。

额外优化建议 #

• 添加请求取消逻辑：用户退出聊天界面时调用call.cancel()，终止未完成的流式请求。
• 网络异常处理：在流读取过程中捕获IOException，及时提示用户网络中断。
• 体验优化：配合流式文本输出添加打字动画，提升用户感知。

内容的提问来源于stack exchange，提问作者Nikhil Mali