用 Gemini API 做流式聊天应用：我踩过的 6 个坑

2025 年 3 月到 6 月，我一个人做完了智能聊天对话应用。调用 Gemini 大模型 API，有多轮上下文、会话历史存储和流式输出。功能清单看起来不长，但我在这上面花的时间远超预期——大部分不是在写功能，而是在处理“外部服务不可靠”和“流式数据不是我想的那样”。

下面这 6 个问题，每一个都是我真实卡住过的。

坑一：分块响应黏包，JSON 解析随机失败

现象： 流式返回里我按 \n 切分后逐条 json.loads，绝大多数时候正常，但偶尔抛出 json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)，同一个输入重跑又不复现。

原因： 网络传输的分块边界和数据的逻辑边界没有任何关系。一次 recv 可能拿到半行 JSON，也可能一次拿到三行；极端情况下还能拿到一个 UTF-8 汉字被切断的前半段字节。

解决： 用一个缓冲区累积，只在遇到完整行时才尝试解析。残缺的部分留在缓冲区里等下一次数据。

def stream_lines(resp):
    """把流式响应按完整行切分，残缺 chunk 留在缓冲区等待下一块。"""
    buffer = ""          # 累积未成行的文本
    decoder = json.JSONDecoder()

    for chunk in resp.iter_content(chunk_size=None, decode_unicode=True):
        if not chunk:
            continue
        buffer += chunk

        while "\n" in buffer:
            line, buffer = buffer.split("\n", 1)   # 只切第一个换行
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                # 这一行不完整，塞回缓冲区，等后续 chunk 补齐
                buffer = line + "\n" + buffer
                break
            yield obj
    # 流结束时缓冲区里可能还剩最后一行
    if buffer.strip():
        try:
            yield json.loads(buffer.strip())
        except json.JSONDecodeError:
            pass

关键点是 buffer.split("\n", 1) 而不是 splitlines()，以及在解析失败时把内容塞回缓冲区并跳出内层循环，而不是丢弃这一行。

坑二：上下文越滚越长，token 和费用一起失控

现象： 会话聊到三四十轮之后，每次请求的耗时明显变长，账单也涨得不正常。

原因： 我最初的做法是把整个会话历史原封不动地拼进每次请求。第 40 轮的时候，前 39 轮的内容会被重新发送一遍——历史被重复计费了几十次，而且是平方级增长。

解决： 分两层处理。保留最近 N 轮（我用的 10 轮）原文，更早的内容压缩成一段摘要塞进 system 提示里。摘要不是每轮都重新生成，而是当被挤出窗口的消息累计超过阈值时才触发一次。这样既保住了长期记忆，又让单次请求的 token 有个上界。

坑三：流式输出时气泡自动滚动不停抖动

现象： 消息逐个字蹦出来的时候，聊天区域会疯狂自动滚到底部，画面抖得没法看；更烦的是我想往上翻看前文，它每次都被拽回底部。

原因： 我在每次收到新 token 的回调里都无脑调用了一次 scrollToBottom。流式输出每秒可能触发几十次，滚动动画根本没结束就又被重启。

解决： 加两个判断。一是合并滚动，用一个节流（比如 100ms）把几十次滚动请求压成几次；二是用户意图检测——记录当前滚动位置，如果用户已经手动往上滑，且距底部超过一屏的阈值，就暂停自动滚动，并在右下角显示一个“回到最新”的悬浮按钮，用户点了再恢复。

这个“尊重用户意图”的原则我是从别的聊天产品上观察到的，自己实现一遍才发现不检测滚动位置根本做不到。

坑四：网络中断后的错误处理和重试

现象： 弱网环境下请求会卡住很久然后报错，报错信息直接甩给用户，界面上就是一个丑陋的堆栈。

原因： 没有设超时，也没有区分错误类型。429（限流）和 401（Key 无效）被我当成同一类处理，前者重试是有意义的，后者重试一百次也没用。

解决： 封装 API 请求层，统一加超时和分类重试。指数退避是我在网上找到的标准做法，自己实现了一版：

import time, random

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def call_with_retry(fn, max_retries=4, base_delay=1.0):
    """指数退避 + 抖动。只重试可恢复的错误，4xx 里的鉴权类直接抛出。"""
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except TimeoutError:
            err = "请求超时"
        except ApiError as e:
            if e.status not in RETRYABLE_STATUS:
                raise                      # 401/400 之类重试无意义
            err = f"服务返回 {e.status}"
        except ConnectionError:
            err = "网络连接失败"

        if attempt == max_retries:
            raise RuntimeError(f"重试 {max_retries} 次后仍失败：{err}")

        # 1s, 2s, 4s, 8s 再加一点随机抖动，避免多个请求同时重试
        delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
        time.sleep(delay)

界面上则统一显示成“网络似乎不太稳定，正在重试（第 2/4 次）“这种可读的提示，而不是原始异常。

坑五：会话本地持久化的结构设计

现象： 一开始我把整个会话（包括所有消息）序列化成一个 JSON 文件，每次有新消息就全量重写。会话一多、消息一长，写文件就明显卡顿，而且中途崩溃会导致文件损坏、整个会话丢失。

原因： 全量重写 + 单文件存储，两个问题叠在一起。

解决： 拆成两张表的结构。sessions 表存 session_id / title / created_at / updated_at，messages 表存 message_id / session_id / role / content / created_at。发送一条消息只做一次 INSERT，不再重写历史。会话列表页只查 sessions，点进去才按 session_id 加载消息。

这个改动带来的额外好处是：删除会话变成了按 session_id 批量删，切换会话也是换一个查询条件，检索历史会话直接对 title 做模糊匹配就行。

坑六：API Key 硬编码进了代码

现象： 差点把带真实 Key 的代码推到远程仓库。是我自己提交前扫了一眼 diff 才发现的，惊出一身冷汗。

原因： 图省事，API_KEY = "AIza..." 直接写在源码顶部。

解决： 三步。第一，Key 挪到环境变量，代码里只写 os.environ["GEMINI_API_KEY"]；第二，本地放一个 .env 文件，并且第一件事就是把它写进 .gitignore；第三，如果已经提交过，光删掉是不够的——Git 历史里还留着，必须把那一次的 Key 作废重发，历史改起来太麻烦也不一定改得干净。

# .gitignore
.env
.env.local
__pycache__/
*.pyc
.venv/

小结

这六个坑里，真正跟“大模型”相关的只有上下文管理那一条，其余五条都是工程问题：数据流的边界、资源的上界、用户意图的尊重、失败的处理、持久化的粒度、秘密的管理。做完这个项目我最大的收获也在这儿——调用一个外部服务，写对请求只是开始，处理它出错的那部分才是工作量的大头。