《C++ 项目：手写大模型接入 SDK》Gemini接口设计（下）：流式响应实现和测试

前言

在上一章中，我们完成了 Gemini 兼容 OpenAI 网关的全量响应接口开发，打通了模型调用链路，熟悉了它独有的代理依赖、鉴权规则以及报文格式规范。而在实际业务场景中，流式输出是大模型交互里非常高频的能力，能够实现逐字增量返回、降低用户等待感知，这也是我们 SDK 框架必须补齐的能力。

本章我们将聚焦sendMessageStream流式接口的落地实现，延续之前统一的回调契约设计，保证和 DeepSeek、ChatGPT 的流式接口调用风格完全一致。我们会从接口签名、回调职责约定开始，按步骤拆解每一段代码的逻辑，理解 SSE 协议在 httplib 流式回调下的处理方式，重点处理缓冲区半包解析、流结束标记识别、代理配置、异常兜底这些工程细节。接口实现完成后，我们还会编写 GTest 单元测试，完成配置、调用、结果展示的完整验证。

一、Gemini 流式响应接口实现

在上一篇文章我们已经完成了 Gemini 全量响应接口的开发，打通了 Gemini OpenAI 兼容网关的调用链路，掌握了它独有的代理配置、鉴权规范。本章我们继续基于这套兼容网关，落地编码实现 sendMessageStream 流式增量接口，同样实现大模型逐片输出的打字机效果。

1.1 接口定义与职责边界

1.1.1 方法签名

namespace ai_chat_sdk {
class GeminiProvider : public LLMProvider
{
public:
    /**
     * @brief 发送消息并接收流式增量响应
     * @param messages 历史对话消息列表
     * @param requestParams 请求参数映射（temperature、max_tokens、model等）
     * @param callback 增量数据回调函数，由调用方实现业务逻辑
     * @return std::optional<std::string> 完整的响应文本（所有增量拼接结果）
     */
    std::optional<std::string> sendMessageStream(
        const std::vector<Message>& messages,
        const std::unordered_map<std::string, std::string>& requestParams,
        std::function<void(const std::string &response, bool isEnd)> callback
    );
};
}

这个接口作为上层对外调用入口，返回值是拼接完成后的完整回答内容；核心交互依靠传入的回调函数向外推送分片数据。

1.1.2 回调契约

回调函数 std::function<void(const std::string &response, bool isEnd)> 是流式 SDK 解耦的关键，约定如下：

第一个入参：模型返回的增量文本片段，仅携带本次新生成的内容，不是完整回复；
第二个入参：流结束标记。true 代表所有增量数据推送完毕，流正常结束；false 代表还有后续分片；
职责划分：SDK 只负责完成 HTTP 请求、SSE 协议解析、提取增量文本并触发回调；打印、界面渲染、持久化存储这类业务逻辑全部交给上层调用者实现，底层 Provider 不耦合业务。

回调接口签名和 DeepSeek、ChatGPT 保持一致，上层业务代码可以无缝切换不同 Provider，这也是我们抽象 LLMProvider 基类的价值。

1.2 完整实现流程与代码逐步解析

1.2.1 检查模型是否有效

// 1. 检查模型是否有效
if (!IsAvailable())
{
    ERR("GeminiProvider sendMessage model not available");
    return std::nullopt;
}

代码分析：
调用基类继承来的 IsAvailable() 检测当前 Provider 是否初始化完成、配置是否合法。如果模型未就绪，直接打印错误日志并返回空，不再执行后续网络请求逻辑。

1.2.2 构造并获取请求参数

// 2. 构造并获取请求参数
// 2.1 构造默认请求参数：如果requestParams中没有指定temperature、max_tokens，则使用默认值
double temperature = 0.7;
int max_tokens = 2048;
std::string model;

// 2.2 从requestParams中获取temperature、max_tokens参数：如果requestParams中指定了，则使用指定值
auto it = requestParams.find("temperature");
if (it != requestParams.end())
{
    temperature = std::stod(it->second);
}
it = requestParams.find("max_tokens");
if (it != requestParams.end())
{
    max_tokens = std::stoi(it->second);
}
it = requestParams.find("model");
if (it != requestParams.end())
{
    model = it->second;
}
else
{
    ERR("model name not found in requestParams, must config it");
    return std::nullopt;
}

代码分析：
流式接口的参数规则和全量接口保持统一：temperature、max_tokens 设置默认值，允许上层调用时按需覆盖；model 是必传参数，兼容网关需要依靠该参数识别 Gemini 具体模型版本，缺失则直接报错退出。

1.2.3 组装 messages 对话数组

// 2.3 构造并获取历史消息；获取请求体中的 messages 这个Json数组
Json::Value messageArray(Json::arrayValue);
for (auto &message : messages)
{
    Json::Value messageObj;
    messageObj["role"] = message._role;
    messageObj["content"] = message._content;
    messageArray.append(messageObj);
}

代码分析：
遍历我们封装的 Message 会话结构体，转换成网关要求的 JSON 消息数组，字段 role 和 content 和 OpenAI Chat Completions 规范对齐，以此适配兼容层网关。

1.2.4 构造请求体，开启流式开关

// 3. 构造请求体
// 和历史消息构造一样，请求体的字段名称必须与gemini模型的接口文档中要求的一致
Json::Value requestBody = messageArray;
requestBody["messages"] = messageArray;
requestBody["temperature"] = temperature;
requestBody["max_tokens"] = max_tokens;
requestBody["model"] = model;
// 关键：流式响应需要设置stream为true，否则是全量响应会返回完整的响应体，而不是增量响应
requestBody["stream"] = true;

代码分析：
这是流式接口和全量接口最核心的一处区别：stream: true 告诉网关启用 SSE 流式返回，服务端会持续下发增量 chunk；如果不开启该字段，网关会一次性返回完整结果，不会走分片推送逻辑。

1.2.5 JSON 序列化请求体

// 4. 对requestBody进行序列化
Json::StreamWriterBuilder writerbuilder;
std::string requestBodyStr = Json::writeString(writerbuilder, requestBody);
INFO("GeminiProvider sendMessageStream requestBody: {}", requestBodyStr);

代码分析：
利用 JsonCpp 将组装好的 JSON 对象序列成字符串，作为 HTTP POST 的 body，同时打印日志方便调试，排查请求报文格式错误。

1.2.6 初始化 httplib 客户端、超时、代理配置

// 5. 使用cpp-httplib构建HTTP客户端
httplib::Client client(scheme_host_port);

// 5.1 设置超时时间
client.set_read_timeout(300, 0);   // 读取超时时间300秒
client.set_connection_timeout(20, 0); // 连接超时时间20秒

// 5.2 关键点：从环境变量中获取代理配置
char *proxy_env = std::getenv("MY_APP_PROXY");
if (proxy_env != nullptr)
{
    std::string proxy_str = proxy_env;
    auto pos = proxy_str.find(":");
    if (pos == std::string::npos)
    {
        ERR("GeminiProvider sendMessageStream proxy_env format error, must be host:port");
        return std::nullopt;
    }
    else
    {
        std::string proxy_host = proxy_str.substr(0, pos);
        int proxy_port = std::stoi(proxy_str.substr(pos + 1));
        client.set_proxy(proxy_host, proxy_port);
    }
}
else
{
    // 针对Gemini模型的特殊情况：如果用户没有配置代理环境变量，那么请求无法访问到Gemini模型的服务器
    ERR("GeminiProvider sendMessageStream proxy_env not found, must config it: MY_APP_PROXY=host:port");
    return std::nullopt;
}

代码分析：
Gemini 官方接口在国内无法直连，所以这里强制要求代理环境变量。代码解析 host:port 格式的代理字符串，配置到 httplib 客户端；超时时间拉长，因为流式对话生成回答耗时较长，短超时很容易提前断开连接。

1.2.7 请求头构造

// 5.3 构建请求头（关键：相较于非流式响应，流式响应需要设置Accept为text/event-stream）
httplib::Headers headers = {
    {"Content-Type", "application/json"},
    {"Authorization", "Bearer " + _apikey},
    {"Accept", "text/event-stream"}
};

代码分析：
Accept: text/event-stream 是 SSE 流式协议的标准请求头，告诉服务端返回 SSE 事件流；鉴权头使用 Bearer 方式，适配兼容 OpenAI 格式的 Gemini 网关。
如果使用 Gemini 原生官方接口，鉴权头为 x-goog-api-key: {apikey}，报文结构、字段命名和 OpenAI 体系差异很大，这就是两种接入方式的核心差异点。

1.2.8 定义流式全局变量

std::string buffer;
std::string fullResponse;
bool streamEnd = false;
bool isError = false;
std::string errorMsg;

buffer：SSE 半包缓冲区，用来缓存不完整的 chunk，保证每次解析的是完整 SSE 事件；
fullResponse：拼接全部增量文本，函数最终返回这个完整字符串；
streamEnd、isError：流结束标记、异常标记；

1.2.9 构建请求对象、注册响应回调与数据接收回调

// 6. 构建请求对象
httplib::Request request;
request.path = "/v1beta/openai/chat/completions";
request.method = "POST";
request.body = requestBodyStr;
request.headers = headers;

// 6.1 构建响应处理器回调函数: response_handler
request.response_handler = [&](const httplib::Response &resp) -> bool
{
    if (resp.status != 200)
    {
        isError = true;
        ERR("GeminiProvider sendMessageStream status not 200, status: {}, body: {}", resp.status, resp.body);
        return false;
    }
    return true;
};

// 6.2 构建数据接收处理器回调函数: content_receiver
request.content_receiver = [&](const char *data, size_t data_length, uint64_t offset, uint64_t total_length) -> bool
{
    if (isError)
    {
        return false;
    }

    buffer.append(data, data_length);
    INFO("GeminiProvider sendMessageStream data: {}", buffer);

    size_t pos = 0;
    while ((pos = buffer.find("\n\n")) != std::string::npos)
    {
        std::string event = buffer.substr(0, pos);
        buffer.erase(0, pos + 2);

        std::string eventLine;
        std::stringstream eventstream(event);
        while (std::getline(eventstream, eventLine))
        {
            if (eventLine.empty() || eventLine[0] == ':')
            {
                continue;
            }
            if (eventLine.compare(0, 6, "data: ") == 0)
            {
                eventData = eventLine.substr(6);
                BUG("GeminiProvider sendMessageStream eventData: {}", eventData);
            }
        }

        if (eventData == "[DONE]")
        {
            streamEnd = true;
            callback("", true);
            break;
        }

        Json::CharReaderBuilder readerbuilder;
        Json::Value eventDataJson;
        std::istringstream eventDataStream(eventData);
        if (!Json::parseFromStream(readerbuilder, &eventDataStream, &eventDataJson, &errorMsg))
        {
            ERR("GeminiProvider sendMessageStream parse eventData error: {}", errorMsg);
            continue;
        }

        if (eventDataJson.isMember("choices") && eventDataJson["choices"].isArray() && !eventDataJson["choices"].empty())
        {
            Json::Value choice = eventDataJson["choices"][0];
            if (choice.isMember("delta") && choice["delta"].isObject() &&
                choice["delta"].isMember("content") && choice["delta"]["content"].isString())
            {
                std::string datacontent = choice["delta"]["content"].asString();
                fullResponse += datacontent;
                callback(datacontent, streamEnd);
            }
        }
    }
    return true;
};

代码分析：

response_handler：先校验 HTTP 状态码，非 200 直接标记错误，终止流；
content_receiver：httplib 收到 chunk 分片就会触发，这里是 SSE 解析核心；
缓冲区 buffer 用来解决半包问题：单次收到的 chunk 有可能只是半个 SSE 事件，需要缓存，直到读到 \n\n 分隔符才取出一条完整事件；
过滤注释行、提取 data: 后面的 JSON 字符串；当读到 data: [DONE]，标记流结束，调用回调通知上层；
解析 JSON，取出 delta.content 增量文本，拼接进完整结果，同时触发回调向外推送片段。

注意：这里的解析逻辑和 DeepSeek 保持一致，得益于我们选用 OpenAI 兼容网关，网关把 Gemini 原生流式结构翻译成了 Chat Completions 标准 SSE 格式，解析代码可以复用。

1.2.10 发送请求，处理发送结果

// 7. 发送请求，获取服务器响应
auto result = client.send(request);
// 7.1 判断请求发送是否成功
if (!result)
{
    // 发送请求失败，可能出现网络问题，比如DNS解析失败、连接超时等等
    ERR("GeminiProvider sendMessageStream send request error: {}, {}", httplib::to_string(result.error()));
    return std::nullopt;
}

if (!streamEnd)
{
    WARN("GeminiProvider stream ended without [DONE] marker");
    // 主动调用一次结束回调，告诉上层：流已经结束了，但是没有收到[DONE]标记
    callback("\n\n连接中断，请重试", true);
}

return fullResponse;

代码分析：
调用 client.send 发起网络请求；send 会阻塞直到流式传输结束。
增加兜底判断：如果流正常结束但是没有收到 [DONE] 标记（网络中断、网关异常），主动触发结束回调，避免上层业务一直等待。最后返回拼接完成的完整回答文本。

1.3 流式响应接口完整代码（附注释分析）

// 发送消息：增量调用，实时返回响应
// requestParams：针对Gemini模型本次请求专属参数，包含model、temperature、max_output_tokens等
std::optional<std::string> GeminiProvider::sendMessageStream(const std::vector<Message> &messages,
                                                             const std::unordered_map<std::string, std::string> &requestParams,
                                                             std::function<void(const std::string &response, bool isEnd)> callback)
{
    // 1. 检查模型是否有效
    if (!IsAvailable())
    {
        ERR("GeminiProvider sendMessage model not available");
        return std::nullopt;
    }

    // 2、构造并获取请求参数
    // 2.1 构造默认请求参数：如果requestParams中没有指定temperature、max_tokens，则使用默认值
    double temperature = 0.7;
    int max_tokens = 2048;
    std::string model;
    // 2.2 从requestParams中获取temperature、max_tokens参数：如果requestParams中指定了，则使用指定值
    auto it = requestParams.find("temperature");
    if (it != requestParams.end())
    {
        temperature = std::stod(it->second);
    }
    it = requestParams.find("max_tokens");
    if (it != requestParams.end())
    {
        max_tokens = std::stoi(it->second);
    }
    it = requestParams.find("model");
    if (it != requestParams.end())
    {
        model = it->second;
    }
    else
    {
        ERR("model name not found in requestParams, must config it");
        return std::nullopt;
    }

    // 2.3 构造并获取历史消息：获取请求体中的 messages 这个Json数组
    Json::Value messageArray(Json::arrayValue);
    for (auto &message : messages)
    {
        Json::Value messageObj;
        messageObj["role"] = message._role;
        messageObj["content"] = message._content;
        messageArray.append(messageObj);
    }

    // 3、构造请求体
    // 和历史消息构造一样，请求体的字段名称必须与gemini模型的接口文档中要求的一致
    Json::Value requestBody;
    requestBody["messages"] = messageArray;
    requestBody["temperature"] = temperature;
    requestBody["max_tokens"] = max_tokens;
    requestBody["model"] = model;
    // 关键：流式响应需要设置stream为true（否则是全量响应会返回完整的响应体，而不是增量响应）
    requestBody["stream"] = true;

    // 4、对requestBody进行序列化
    Json::StreamWriterBuilder writerbuilder;
    std::string requestBodyStr = Json::writeString(writerbuilder, requestBody);
    INFO("GeminiProvider sendMessage requestBody: {}", requestBodyStr);

    // 5、使用cpp-httplib构建HTTP客户端
    httplib::Client client(_baseUrl);

    // 5.1设置超时时间
    client.set_read_timeout(300, 0);      // 读取超时时间300秒
    client.set_connection_timeout(20, 0); // 连接超时时间20秒
    // 5.2 关键点：从环境变量中获取代理配置
    char *proxy_env = std::getenv("MY_APP_PROXY");
    if (proxy_env != nullptr)
    {
        std::string proxy_str(proxy_env);
        auto pos = proxy_str.find(":");
        if (pos == std::string::npos)
        {
            ERR("GeminiProvider sendMessage proxy_env format error, must be host:port");
            return std::nullopt;
        }
        else
        {
            std::string proxy_host = proxy_str.substr(0, pos);
            int proxy_port = std::stoi(proxy_str.substr(pos + 1));
            client.set_proxy(proxy_host, proxy_port);
        }
    }
    else
    {
        // 针对Gemini模型的特殊情况：如果用户没有配置代理环境变量，那么请求无法访问到Gemini模型的服务器
        ERR("GeminiProvider sendMessage proxy_env not found, must config it: MY_APP_PROXY=host:port");
        return std::nullopt;
    }
    // 5.3 构建请求头（关键：相较于非流式响应，流式响应需要设置Accept为text/event-stream）
    httplib::Headers headers = {
        {"Content-Type", "application/json"},
        {"Authorization", "Bearer " + _apikey},
        {"Accept", "text/event-stream"}};

    // 在构建请求对象之前，我们可以先设置一些流式处理变量，用于处理回调过程中的相关逻辑。
    std::string buffer;       // 用于存储每次读取到的Chunk数据块中的内容
    std::string fullResponce; // 用于累积存储获取到的完整的响应体
    bool streamEnd = false;   // 用于标记流式响应是否结束
    bool isError = false;     // 用于标记是否发生错误
    std::string errorMsg;     // 用于存储错误描述信息

    // 6、构建请求对象
    httplib::Request request;
    request.path = "/v1beta/openai/chat/completions";
    request.method = "POST";
    request.body = requestBodyStr;
    request.headers = headers;

    // 6.1 构建响应处理器回调函数：response_handler
    request.response_handler = [&](const httplib::Response &resp) -> bool
    {
        if (resp.status != 200)
        {
            isError = true;
            ERR("GeminiProvider sendMessageStream status not 200, status: {}, body: {}", resp.status, resp.body);
            return false;
        }
        return true;
    };

    // 6.2 构建数据接收处理器回调函数：content_receiver
    request.content_receiver = [&](const char *data, size_t data_length,
                                   uint64_t offset, uint64_t total_length) -> bool
    {
        if (isError)
        {
            return false;
        }

        buffer.append(data, data_length);
        INFO("GeminiProvider sendMessageStream data: {}", buffer);

        size_t pos;
        while ((pos = buffer.find("\n\n")) != std::string::npos)
        {
            std::string event = buffer.substr(0, pos);
            buffer.erase(0, pos + 2);

            std::string eventline;
            std::string eventData;
            std::stringstream eventstream(event);
            while (std::getline(eventstream, eventline))
            {
                if (eventline.empty() || eventline[0] == ':')
                {
                    continue;
                }
                if (eventline.compare(0, 6, "data: ") == 0)
                {
                    eventData = eventline.substr(6);
                    BUG("GeminiProvider sendMessageStream eventData: {}", eventData);
                }
            }

            if (eventData == "[DONE]")
            {
                streamEnd = true;
                callback("", true);
                break;
            }

            Json::CharReaderBuilder readerbuilder;
            Json::Value eventDataJson;
            std::istringstream eventDataStream(eventData);
            if (!Json::parseFromStream(readerbuilder, eventDataStream, &eventDataJson, &errorMsg))
            {
                ERR("GeminiProvider sendMessageStream parse eventData error: {}", errorMsg);
                continue;
            }

            if (eventDataJson.isMember("choices") && eventDataJson["choices"].isArray() && !eventDataJson["choices"].empty())
            {
                Json::Value choice = eventDataJson["choices"][0];
                if (choice.isMember("delta") && choice["delta"].isObject() &&
                    choice["delta"].isMember("content") && choice["delta"]["content"].isString())
                {
                    std::string datacontent = choice["delta"]["content"].asString();
                    fullResponce += datacontent;
                    callback(datacontent, streamEnd);
                }
            }
        }
        return true;
    };

    // 7、发送请求，获取服务器响应
    auto result = client.send(request);
    // 7.1 判断请求发送是否成功
    if (!result)
    {
        // 发送请求失败，可能出现网络问题，比如DNS解析失败，连接超时等等
        ERR("GeminiProvider sendMessageStream send request error: {}", httplib::to_string(result.error()));
        return std::nullopt;
    }

    if (!streamEnd)
    {
        WARN("GeminiProvider stream ended without [DONE] marker");
        // 手动将第二个参数设置成true，告诉用户我们已经获取到了响应内容，但是没有收到[DONE]结束标记
        callback("\n\n[系统提示: 连接意外中断，请重试]", true);
    }
    return fullResponce;
}

二、流式响应单元测试

2.1 初始化与配置

#include "../SDK/include/ChatGPTProvider.h"
#include "../SDK/include/DeepSeekProvider.h"
#include "../SDK/include/GeminiProvider.h"
#include "../SDK/include/LLMManager.h"
#include "../SDK/include/OllamaLLMProvider.h"
#include <gtest/gtest.h>
#include <spdlog/common.h>

using namespace ai_chat_sdk;

TEST(GeminiProviderTest, sendMessage)
{
    // 1、创建GeminiProvider对象
    std::shared_ptr<LLMProvider> geminiProvider = std::make_shared<GeminiProvider>();
    // 1.1 检查GeminiProvider是否初始化成功
    ASSERT_TRUE(geminiProvider != nullptr);

    // 2、构建模型配置
    std::unordered_map<std::string, std::string> modelConfig;
    modelConfig["api_key"] = getenv("gemini_apikey") ? getenv("gemini_apikey") : "";
    modelConfig["base_url"] = "https://generativelanguage.googleapis.com";

    // 3、初始化GeminiProvider
    geminiProvider->initModel(modelConfig);
    // 3.1 检查模型是否有效
    ASSERT_TRUE(geminiProvider->IsAvailable());

    // 4、构建请求参数配置(不写则使用默认值)
    std::unordered_map<std::string, std::string> requestConfig;
    requestConfig["model"] = "gemini-3.6-flash";
    requestConfig["temperature"] = "0.7";
    requestConfig["max_tokens"] = "2048";

    // 5、构建消息列表
    std::vector<Message> messages;
    messages.push_back({"user", "你好, 请简单介绍一下你自己"});

    // 6、调用ssendMessageStreame发送消息
    auto response = geminiProvider->sendMessageStream(
        messages, requestConfig,
        [&](const std::string &chunk, bool isEnd)
        {
            INFO("chunk: {}", chunk);
            if (isEnd)
            {
                INFO("[DONE]");
            }
        });
    // 6.1 检查响应是否为空
    ASSERT_TRUE(response.has_value());
    INFO("response: {}", response.value());
}

2.2 入口函数

int main(int argc, char **argv)
{
    // 初始化Google Test框架
    testing::InitGoogleTest(&argc, argv);

    // 初始化日志系统
    Logger::logger_init("testLLM", "stdout", spdlog::level::info);

    // 运行所有测试用例
    return RUN_ALL_TESTS();
}

2.3 运行结果展示

# 进入构建目录
cd test/build
 
# 生成构建文件
cmake ..
 
# 清理并编译
make clean
make
 
# 运行测试程序
./testLLM

结束语

至此，Gemini 的流式响应接口编码与单元测试验证工作全部完成。我们沿用了整套 SDK 统一的回调规范，依托 OpenAI 兼容网关，把 Gemini 的流式输出封装到和 DeepSeek、ChatGPT 一致的调用接口之下，同时妥善处理了 SSE 半包缓存、流结束标记、代理环境依赖、异常兜底等工程问题。

从全量接口到流式接口，GeminiProvider 的核心能力已经补齐。也再次印证了我们前期抽象 LLMProvider 基类的价值：不同厂商模型的底层接口规范纵然存在差异，只要适配统一的上层契约，业务层就可以无感知切换模型。