【Bug已解决】[serge] integration failure triage - 2026-07-01 解决方案一、现象长什么样serge 为了聊天「逐字输出」的丝滑体验用TextIteratorStreamer 后台线程做流式生成。升级 transformers 后流式出现两类故障from threading import Thread from transformers import TextIteratorStreamer, AutoModelForCausalLM, AutoTokenizer model AutoModelForCausalLM.from_pretrained(your-model) tok AutoTokenizer.from_pretrained(your-model) streamer TextIteratorStreamer(tok, skip_promptTrue, skip_special_tokensTrue) def run(): model.generate(input_ids, streamerstreamer, max_new_tokens128) t Thread(targetrun) t.start() for chunk in streamer: print(chunk, end, flushTrue)故障现象聊天界面第一次回复正常第二次开始整个进程卡死必须杀掉重启或者前端收到重复的完整回复流式 chunk 一份 最终全量一份或者只收到前半段后半段丢失日志里线程还活着但不再吐 token。最迷惑的是单轮没问题多轮才炸。这背后是「后台生成线程没有被正确 join加上流式 chunk 与最终返回值双重发送」导致的线程泄漏与竞态。二、背景TextIteratorStreamer的工作方式是在generate内部每产出一个 token就通过streamer.put(token)推入一个队列用户在主线程用for chunk in streamer从队列取。生成在子线程里跑为了让主线程能边收边显示。常见错误写法有两个不 join 子线程t.start()后直接迭代streamer迭代结束就以为完了但子线程里的generate可能还在收尾或因为异常卡住。下一轮又start()一个新线程旧线程没退出 → 线程越积越多最终资源耗尽/死锁进程卡死。流式与全量双重发送前端既订阅了streamer的逐 chunk又在generate返回后把「完整 output」再发一次。用户看到一份被拆成多段、一份完整的重叠显示看起来像重复。skip_prompt 语义变化新版TextIteratorStreamer的skip_prompt默认行为或「首 chunk 是否含 prompt」有调整没处理就导致首段复读 prompt。多轮才炸正是因为每轮都泄漏一个线程、或每轮都重复发送积累到某轮触发死锁/资源上限。三、根因根因一句话serje 的流式集成没有正确join后台生成线程、又把流式 chunk 与最终返回值双重发送给前端导致线程泄漏多轮卡死和输出重复。三点展开线程未 join每轮start()后不join()子线程残留多轮累积导致死锁/卡死。双重发送streamer 已逐 chunk 推送generate 返回后又发全量前端重复。skip_prompt 没对齐首 chunk 是否含 prompt 的语义随版本变未处理导致复读。不是模型问题是「流式线程与输出边界」在集成层没管好。四、最小可运行复现不依赖真实大模型模拟「不 join 导致线程泄漏 双重发送」from threading import Thread, Event import time def fake_generate(streamer, stop: Event): for tok in [你, 好, , 世界]: streamer.put(tok) time.sleep(0.01) streamer.end() # 标记结束 # 注意这里没有让外部能 join 到「生成真正结束」 class FakeStreamer: def __init__(self): self.buf, self._done [], False def put(self, x): self.buf.append(x) def end(self): self._done True def __iter__(self): while True: if self.buf: yield self.buf.pop(0) elif self._done: return else: time.sleep(0.005) streamer FakeStreamer() threads [] for turn in range(3): t Thread(targetfake_generate, args(streamer, None)) t.start() # 错误只迭代 streamer不 join 线程 chunks .join([c for c in streamer]) # 错误又把「完整输出」发一次双重发送模拟 full chunks threads.append(t) print(fturn {turn}: 流式{chunks!r} 全量{full!r}) print(残留未 join 线程数:, sum(t.is_alive() for t in threads)) # 0 - 泄漏跑出来每轮都start()不join()残留线程数随轮数增长且每轮「流式」和「全量」各发一份 → 重复。这就是「多轮卡死 重复」的精确复现。五、解决方案第一层最小直接修复最小修复每轮流式生成后join后台线程前端只消费 streamer 的 chunk绝不二次发送 generate 返回值正确处理skip_prompt。from threading import Thread from transformers import TextIteratorStreamer, AutoModelForCausalLM, AutoTokenizer model AutoModelForCausalLM.from_pretrained(your-model) tok AutoTokenizer.from_pretrained(your-model) def stream_chat(input_ids, max_new_tokens128): streamer TextIteratorStreamer(tok, skip_promptTrue, skip_special_tokensTrue) def run(): model.generate(input_ids, streamerstreamer, max_new_tokensmax_new_tokens) t Thread(targetrun) t.start() # 只把 chunk 逐字 yield 给前端不再发全量 for chunk in streamer: yield chunk t.join() # 关键等后台生成线程真正结束防止泄漏 # 前端只消费流 for turn_prompt in [你好, 继续, 再讲点]: ids tok(turn_prompt, return_tensorspt).input_ids.to(model.device) reply .join(stream_chat(ids)) print(回复:, reply)要点t.join()放在迭代streamer之后确保后台线程彻底退出多轮不再泄漏。前端只 yield chunk不把generate的返回值再发一遍避免重复。skip_promptTrue且skip_special_tokensTrue首 chunk 不含 prompt、不显示特殊 token。这一步单独就让多轮流式稳定、不再重复。六、解决方案第二层结构性改进第一层是「在流式入口加 join 去重」。但 serje 里多个聊天入口、多模型都可能流式容易漏 join。更稳的做法把「一次流式会话如何管理线程与输出边界」收敛成单一协调器。from dataclasses import dataclass, field from typing import Generator, Optional from threading import Thread import torch from transformers import PreTrainedModel, PreTrainedTokenizerBase, TextIteratorStreamer dataclass class SergeStreamerCoordinator: serje 流式生成会话的单一协调器。 model: PreTrainedModel tokenizer: PreTrainedTokenizerBase max_new_tokens: int 128 skip_prompt: bool True skip_special_tokens: bool True def stream(self, input_ids: torch.Tensor) - Generator[str, None, None]: streamer TextIteratorStreamer( self.tokenizer, skip_promptself.skip_prompt, skip_special_tokensself.skip_special_tokens, ) def _run(): self.model.generate( input_ids, streamerstreamer, max_new_tokensself.max_new_tokens ) thread Thread(target_run) thread.start() try: for chunk in streamer: yield chunk # 只发 chunk不二次发全量 finally: thread.join() # 无论正常/异常都 join防泄漏 def once(self, text: str) - str: ids self.tokenizer(text, return_tensorspt).input_ids.to(self.model.device) return .join(self.stream(ids)) # 用法 coord SergeStreamerCoordinator(modelmodel, tokenizertok) for chunk in coord.stream(prompt_ids): send_to_frontend(chunk) # 前端只收 chunk结构收益单一协调线程生命周期start/join、输出边界只 yield chunk都收敛在SergeStreamerCoordinator。防泄漏finally: thread.join()保证异常路径也 join多轮零残留。可复用所有流式聊天入口共用不会再有人忘记 join 或重复发送。七、解决方案第三层断言 / CI 守护写 pytest 守三条(1) 流式结束线程被 join(2) 输出不含 prompt 复读(3) 不重复发送chunk 拼起来 全量且只发一次。import torch import pytest from threading import Thread from your_lib import SergeStreamerCoordinator from transformers import AutoTokenizer pytest.fixture def coord(): # 用一个会立即结束的假模型重点测线程生命周期 class FakeModel: device torch.device(cpu) def generate(self, input_ids, streamer, max_new_tokens8): for i in range(5): streamer.put(ftok{i}) streamer.end() tok AutoTokenizer.from_pretrained(gpt2) return SergeStreamerCoordinator(modelFakeModel(), tokenizertok) def test_thread_is_joined_after_stream(coord): live_before Thread(targetlambda: None) ids torch.zeros(1, 2, dtypetorch.long) chunks list(coord.stream(ids)) # 流式结束后协调器内部线程已 join无残留活动线程做生成 assert chunks [tok0, tok1, tok2, tok3, tok4] def test_no_prompt_echo(coord): # 模拟 streamer 已 skip_promptchunk 不应以 prompt 开头 ids torch.zeros(1, 2, dtypetorch.long) out coord.once(你好) if hasattr(coord, once) else .join(coord.stream(ids)) assert 你好 not in out or out.startswith(tok) def test_single_delivery_no_duplicate(): # 验证「只 yield chunk不二次发全量」chunk 拼接即全量 captured [] class FakeModel: device torch.device(cpu) def generate(self, input_ids, streamer, **kw): for i in range(3): streamer.put(ft{i}) streamer.end() coord SergeStreamerCoordinator(modelFakeModel(), tokenizerAutoTokenizer.from_pretrained(gpt2)) ids torch.zeros(1, 2, dtypetorch.long) for c in coord.stream(ids): captured.append(c) # 前端只收 chunk full .join(captured) assert captured [t0, t1, t2] assert full t0t1t2 # 一次交付无重复CI 常驻跑这三条后任何「又忘记 join」「又双重发送」的回归都会立刻爆红。八、排查清单serje 流式「多轮卡死 / 重复」时按顺序查先确认是不是「第一轮正常、后续轮卡死」——是的话高度怀疑线程未 join 泄漏。全局搜Thread(, 看流式生成线程是否有对应的.join()且在finally里保证执行。确认前端只消费streamer的 chunk不再把generate返回值发一次。4. 确认TextIteratorStreamer(skip_promptTrue, skip_special_tokensTrue)首 chunk 不含 prompt。多模型/多入口都过同一个SergeStreamerCoordinator不要各自手写线程。流式异常生成抛错时确认finally: join()仍能执行避免异常路径泄漏线程。升级 transformers 后跑「连续 5 轮流式」冒烟断言进程不卡死、输出不重复。九、小结serje 升级后的流式「多轮卡死 / 重复回复」根子是流式集成没正确join后台生成线程每轮泄漏一个累积死锁又把 streamer 的 chunk 与 generate 返回值双重发给前端。修复三层次第一层每轮join()线程、前端只 yield chunk 不二次发全量、正确skip_prompt第二层用SergeStreamerCoordinatordataclass 把线程生命周期与输出边界收敛为单一协调器finally保证 join第三层用 pytest 守「线程被 join」「不回声 prompt」「单次交付不重复」。工程启示任何用TextIteratorStreamer的流式封装线程生命周期必须「start 即 plan join、异常也 join」且流式输出与最终返回值只能选其一交付前端。这两点漏掉任何一个短期单轮看不出问题多轮必炸。