9.5 KiB

Raw Blame History

module

date

problem_type

component

severity

symptoms

root_cause

resolution_type

related_components

私董会 chat_stream 首 chunk hang + 双重广播修复

Problem

私董会（@board）触发后，moderator 开场发言正常输出（走 gateway.chat() 非流式），但进入 expert 发言环节后整体卡死：前端 progress 1/5 后无限等待，5 个专家一个都不发言，无 moderator 小结、无最终总结。服务端日志无异常，重启服务后复现。

Symptoms

@board 请求后 moderator 开场正常（30s 内返回 367 字）
进入 expert 发言后永久 hang（等 90s+ 无任何输出）
服务端日志无 error、无 timeout、无 warning — LLM 请求已发出但 async for chunk in stream 永远不进入循环体
重启服务、换 topic 均复现
同一 LLM provider 下，非流式 gateway.chat() 正常工作

What Didn't Work

怀疑 _handle_llm_generate 降级 — 看到 llm_generate_no_client 占位文本，以为是 ConfigDrivenAgent 的 LLM 客户端注入问题。修复 _llm_gateway 优先逻辑后（commit d0fe661），单测通过但 board hang 依旧。原因：board 路径根本不调 ConfigDrivenAgent.handle_task，而是 orchestrator 直接调 gateway.chat_stream()。
怀疑 PII filter 接入导致 — commit cb278bb（PII filter 接入 gateway 生产路径）改了 gateway.py +29 行。检查后发现 PII filter 默认禁用（AGENTKIT_PII_FILTER_ENABLED 未设），不是直接原因。
怀疑 handoff 队列阻塞 — InProcessHandoffTransport.send() 的 await queue.put(message) 没设 timeout，前端 WebSocket 慢/断了会永久阻塞。检查后发现 hang 位置在 LLM 调用之前，不是 broadcast。

Root Cause

commit 36b0296（Jul 2）引入的回归 — 该 commit 把 expert 发言从 gateway.chat()（非流式）改为 gateway.chat_stream()（流式）以实现"逐个输出"UI 体验：

# 改前（commit 36b0296 之前）— 非流式，慢但能用
response = await gateway.chat(messages=[...], model="default")

# 改后 — 流式，DashScope/DeepSeek via LiteLLM 偶尔 hang
async for chunk in gateway.chat_stream(messages=[...], model="default"):
    ...

为什么 hang：DashScope (Qwen) + DeepSeek 通过 LiteLLM 的 acompletion(stream=True) 调用时，偶尔接受请求但永不返回首个 SSE chunk。async for chunk in stream_obj 无限阻塞，没有 error、没有 timeout、没有 warning — board 整体死锁。

为什么 moderator 正常：moderator 的 opening / summary / final 都走 gateway.chat()（非流式），不受影响。只有 expert 发言走 chat_stream()，所以"moderator 能说话但 expert 卡死"是精确的症状特征。

为什么以前能用：commit 36b0296 之前所有调用都是 gateway.chat() 同步等待 — 慢但能用。该 commit 引入 streaming 时没给"获取首 chunk"加超时。

Solution

Fix 1：首 chunk 30s 超时 + fallback（commit `ed1e289`）

board_orchestrator.py _stream_expert_speech — 给 chat_stream 的首 chunk 加 30s 超时，超时则 fallback 到非流式 gateway.chat() + _replay_stream 切片（保留"逐个输出"UX）：

_STREAM_FIRST_CHUNK_TIMEOUT_S: float = 30.0

async def _stream_expert_speech(self, expert, round, prompt):
    gateway = self._get_llm_gateway(expert)
    total = ""
    stream_obj = None
    try:
        stream_obj = gateway.chat_stream(messages=[...], model="default")
        if asyncio.iscoroutine(stream_obj):
            raise TypeError("chat_stream returned a coroutine, not an async generator")

        # 关键：首 chunk 30s 超时
        first_chunk = await asyncio.wait_for(
            stream_obj.__anext__(),
            timeout=self._STREAM_FIRST_CHUNK_TIMEOUT_S,
        )
        # 首 chunk 拿到 → 剩余 stream 信任能正常流
        await _emit(first_chunk)
        async for chunk in stream_obj:
            await _emit(chunk)
        return total.strip()

    except asyncio.TimeoutError:
        # 30s 没首 chunk → 关闭 stream，fallback 到非流式
        if stream_obj and hasattr(stream_obj, "aclose"):
            try: await stream_obj.aclose()
            except Exception: pass
    except StopAsyncIteration:
        # 空 stream → fallback
        ...
    except Exception as e:
        # stream 中途断 → fallback
        ...

    # Fallback：非流式 + 切片重播
    response = await gateway.chat(messages=[...], model="default")
    content = (response.content or "").strip()
    if content:
        await self._replay_stream(expert, round, content, delay=0.05, chunk_size=12)
    return content

Fix 2：中间断流时避免双重广播（commit `0daa4d2`，review 发现）

Code review 发现 Fix 1 的 fallback 路径有 P2 bug：当 first_chunk 成功并已广播，但后续 async for 中间抛异常时，fallback 调 gateway.chat() 返回完整内容并再调 _replay_stream 广播 → UI 看到部分流式内容 + 完整非流式内容 = 内容重复。

修复：fallback 路径中，如果 total 已有部分内容（来自 first_chunk），直接返回部分内容，不调 gateway.chat()：

# 如果已经广播了部分内容，不要重复广播完整内容
if total:
    logger.info(
        f"Expert '{expert.config.name}' stream broke after {len(total)} chars "
        f"of partial content — returning partial (avoid double broadcast)"
    )
    return total.strip()
# 只有 total 为空时才 fallback 到非流式
response = await gateway.chat(...)

Fix 3（独立 commit `d0fe661`）：`_handle_llm_generate` 优先用 gateway

config_driven.py _handle_llm_generate — AgentPool 注入 _llm_gateway 但不设 _llm_client，原代码只检查 _llm_client is None 导致错误降级。改为优先用 _llm_gateway。

注：这个 fix 与 board hang 无关（board 路径不经过 ConfigDrivenAgent.handle_task），但解决了"因 LLM 不可用无法发言"占位文本的独立 bug。

Why This Works

首 chunk 超时 把"永久 hang"降级为"30s 延迟 + fallback" — 即使 DashScope/DeepSeek 流式不返回，30s 后自动切到非流式路径，board 能继续执行
__anext__() + wait_for 而非 async for — async for 无法对单次迭代加超时，显式 __anext__() 可以
fallback 用 _replay_stream 切片 而非一次性广播完整内容 — 保留"逐个输出"UX 体验
部分内容优先于重复内容 — 中间断流时返回部分内容（已广播），不调 chat() 重新广播完整内容

Prevention

模式：所有 `async for chunk in stream` 必须给首 chunk 加超时

任何调用 async for chunk in provider.chat_stream(...) 的地方，都应该用 __anext__() + asyncio.wait_for 给首 chunk 加超时。国产 OpenAI-compat endpoint（DashScope/DeepSeek/Qwen）的 LiteLLM 流式适配偶尔不返回首 chunk，async for 会永久阻塞且无 error。

# ❌ 危险 — 永久 hang 无 error
async for chunk in gateway.chat_stream(...):
    ...

# ✅ 安全 — 30s 超时后 fallback
stream = gateway.chat_stream(...)
try:
    first = await asyncio.wait_for(stream.__anext__(), timeout=30.0)
except asyncio.TimeoutError:
    # fallback to non-streaming
    ...
await _emit(first)
async for chunk in stream:
    await _emit(chunk)

测试覆盖

必须覆盖 5 个 streaming 场景（见 TestStreamExpertSpeechFallback）：

正常流式 — chunks 正常返回
首 chunk 超时 — _HungStream 永久阻塞，验证 fallback
空 stream — StopAsyncIteration，验证 fallback
stream 抛异常 — RuntimeError，验证 fallback
中间断流 — 首 chunk 成功但后续抛异常，验证不双重广播

诊断脚本模板

当 board/团队功能 hang 时，用加超时 + 阶段打点的诊断脚本定位（见 diag_board_trace.py 模式）：

每个 await 加 asyncio.wait_for 超时
monkey-patch 关键方法打印 ENTER/EXIT 时间戳
先跑端到端复现，再用超时定位 hang 位置

Cross-References

streaming-event-whitelist-and-accumulation.md — 前端流式双累积 bug（不同根因，同 streaming 子系统）
streaming-event-contract-residuals.md — streaming 事件契约残留
Commit 36b0296 — 引入 streaming 回归的 commit
Commit ed1e289 — Fix 1（首 chunk 超时）
Commit 0daa4d2 — Fix 2（避免双重广播，review 发现）
Commit d0fe661 — Fix 3（_handle_llm_generate gateway 优先，独立 bug）

9.5 KiB Raw Blame History Unescape Escape