fix(board): add first-chunk timeout to _stream_expert_speech
Commit36b0296changed expert speech generation from gateway.chat() (non-streaming) to gateway.chat_stream() (streaming) for progressive UI output. However, DashScope/DeepSeek via LiteLLM occasionally accept the streaming request but never emit the first SSE chunk — the async generator hangs indefinitely with no error and no timeout, freezing the entire board. Root cause: `async for chunk in gateway.chat_stream(...)` blocks forever when the provider silently stalls. The moderator path (gateway.chat) still works because it's synchronous — that's why the moderator opening always succeeds but experts hang. Fix: pull the first chunk via `__anext__()` wrapped in `asyncio.wait_for(timeout=30s)`. If no chunk arrives within 30s, close the stream and fall back to non-streaming gateway.chat() + _replay_stream() (which splits the response into small chunks for progressive UI rendering). This preserves the "逐个输出" UX while guaranteeing the board never hangs on a stalled streaming provider. Add 4 unit tests covering: - Normal streaming works (existing path) - First-chunk timeout → fallback to chat() - Empty stream → fallback to chat() - chat_stream raising → fallback to chat() E2E verification: 5-expert board with DashScope completes in ~277s (status=completed), all experts produce content, no hang. Same branch as the earlier _handle_llm_gateway fix (commitd0fe661).
This commit is contained in:
Chiguyong
parent
d0fe6611c7
commit
ed1e289785
|
|
@ -311,12 +311,20 @@ class BoardOrchestrator:
|
|||
# loop is unchanged.
|
||||
return await self._stream_expert_speech(expert, round, prompt)
|
||||
|
||||
# First-chunk timeout for chat_stream. Some OpenAI-compatible providers
|
||||
# (DashScope/DeepSeek via LiteLLM) occasionally accept the request but
|
||||
# never emit the first SSE chunk, causing the board to hang silently.
|
||||
# 30s is the P95 of moderator (non-streaming) chat on this provider —
|
||||
# if streaming can't beat that, we fall back to a working path.
|
||||
_STREAM_FIRST_CHUNK_TIMEOUT_S: float = 30.0
|
||||
|
||||
async def _stream_expert_speech(self, expert: Expert, round: int, prompt: str) -> str:
|
||||
"""Stream an expert's speech via chat_stream, emitting chunks.
|
||||
|
||||
Falls back to non-streaming ``chat()`` when ``chat_stream`` is
|
||||
unavailable (e.g. an LLM provider without streaming support) or
|
||||
raises before any chunk is produced.
|
||||
unavailable (e.g. an LLM provider without streaming support), raises
|
||||
before any chunk is produced, or **fails to emit the first chunk
|
||||
within ``_STREAM_FIRST_CHUNK_TIMEOUT_S`` seconds**.
|
||||
|
||||
ponytail: when the LLM does not actually stream (returns a single
|
||||
big chunk), we still want the UI to see content appearing
|
||||
|
|
@ -324,22 +332,45 @@ class BoardOrchestrator:
|
|||
sentence/line chunks and emit them with a small delay. The
|
||||
``expert_speech_chunk`` event already handles duplicate-sender
|
||||
dedup, so emitting many small chunks is safe.
|
||||
|
||||
Regression guard: commit 36b0296 introduced streaming here, but
|
||||
DashScope/DeepSeek via LiteLLM occasionally hang on the first SSE
|
||||
chunk with no error and no timeout. The first-chunk timeout below
|
||||
ensures we fall back to the proven non-streaming path instead of
|
||||
blocking the board indefinitely.
|
||||
"""
|
||||
gateway = self._get_llm_gateway(expert)
|
||||
assert gateway is not None # checked by caller
|
||||
total = ""
|
||||
# Emit an opening chunk-less event so the UI can create the streaming
|
||||
# placeholder before the first token arrives (keeps the first paint
|
||||
# aligned with the streaming indicator).
|
||||
|
||||
# Try streaming with a hard first-chunk deadline. If the provider
|
||||
# accepts the request but never emits the first chunk (observed on
|
||||
# DashScope/DeepSeek via LiteLLM), fall through to non-streaming.
|
||||
stream_obj = None
|
||||
try:
|
||||
streamed_chunk_count = 0
|
||||
async for chunk in gateway.chat_stream(
|
||||
stream_obj = gateway.chat_stream(
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
model="default",
|
||||
):
|
||||
)
|
||||
# Defensive: provider returning a coroutine instead of an async
|
||||
# generator indicates an implementation bug — raise so the
|
||||
# except below picks up the non-streaming fallback.
|
||||
if asyncio.iscoroutine(stream_obj):
|
||||
raise TypeError("chat_stream returned a coroutine, not an async generator")
|
||||
|
||||
# Pull the first chunk with a timeout. If it arrives, the
|
||||
# remainder of the stream is trusted to flow normally.
|
||||
first_chunk = await asyncio.wait_for(
|
||||
stream_obj.__anext__(),
|
||||
timeout=self._STREAM_FIRST_CHUNK_TIMEOUT_S,
|
||||
)
|
||||
streamed_chunk_count = 0
|
||||
|
||||
async def _emit(chunk) -> None:
|
||||
nonlocal total, streamed_chunk_count
|
||||
delta = chunk.content or ""
|
||||
if not delta:
|
||||
continue
|
||||
return
|
||||
total += delta
|
||||
streamed_chunk_count += 1
|
||||
await self._broadcast_event(
|
||||
|
|
@ -353,6 +384,10 @@ class BoardOrchestrator:
|
|||
"role": "expert",
|
||||
},
|
||||
)
|
||||
|
||||
await _emit(first_chunk)
|
||||
async for chunk in stream_obj:
|
||||
await _emit(chunk)
|
||||
# If the LLM "streamed" but only delivered one big chunk, still
|
||||
# let the UI see content arrive progressively.
|
||||
if streamed_chunk_count <= 1 and total:
|
||||
|
|
@ -363,8 +398,28 @@ class BoardOrchestrator:
|
|||
f"Provider for '{expert.config.name}' lacks chat_stream, "
|
||||
f"falling back to non-streaming: {e}"
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(
|
||||
f"Expert '{expert.config.name}' stream produced no chunk in "
|
||||
f"{self._STREAM_FIRST_CHUNK_TIMEOUT_S}s — falling back to "
|
||||
f"non-streaming (provider may be DashScope/DeepSeek via LiteLLM)"
|
||||
)
|
||||
# Close the partial stream to avoid resource leak
|
||||
if stream_obj is not None and hasattr(stream_obj, "aclose"):
|
||||
try:
|
||||
await stream_obj.aclose()
|
||||
except Exception: # noqa: BLE001 — best-effort cleanup
|
||||
pass
|
||||
except StopAsyncIteration:
|
||||
# Empty stream — no chunks at all, fall through to fallback
|
||||
logger.info(f"Expert '{expert.config.name}' stream produced no chunks")
|
||||
except Exception as e:
|
||||
logger.warning(f"Expert '{expert.config.name}' stream failed: {e}")
|
||||
if stream_obj is not None and hasattr(stream_obj, "aclose"):
|
||||
try:
|
||||
await stream_obj.aclose()
|
||||
except Exception: # noqa: BLE001 — best-effort cleanup
|
||||
pass
|
||||
|
||||
# Fallback: non-streaming path. Emit the whole content as small
|
||||
# chunks so the UI still renders progressively rather than going
|
||||
|
|
|
|||
|
|
@ -343,3 +343,119 @@ class TestBoardOrchestratorBroadcast:
|
|||
|
||||
# 不应抛出异常
|
||||
await orchestrator._broadcast_event("board_started", {"topic": "测试"})
|
||||
|
||||
|
||||
# ── BoardOrchestrator._stream_expert_speech 测试 ─────────
|
||||
|
||||
|
||||
class TestStreamExpertSpeechFallback:
|
||||
"""_stream_expert_speech 首 chunk 超时 fallback 测试。
|
||||
|
||||
回归 commit 36b0296：该 commit 把 expert 发言从 gateway.chat() 改为
|
||||
gateway.chat_stream() 以实现"逐个输出"UI 体验，但 DashScope/DeepSeek
|
||||
via LiteLLM 偶尔不返回首个 SSE chunk，导致 board 整体 hang。
|
||||
修复：首 chunk 30s 超时 → fallback 到非流式 chat() + _replay_stream。
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_normal_works(self):
|
||||
"""chat_stream 正常返回 chunks 时走流式路径。"""
|
||||
team = BoardTeam()
|
||||
orchestrator = BoardOrchestrator(team=team)
|
||||
expert = _make_mock_expert("tester", is_lead=False)
|
||||
|
||||
# chat_stream 立刻返回 chunks
|
||||
gateway = _make_mock_gateway("正常流式回复")
|
||||
expert.agent._llm_gateway = gateway
|
||||
orchestrator._get_llm_gateway = lambda e: gateway
|
||||
|
||||
with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
|
||||
result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
|
||||
|
||||
assert "正常流式回复" in result
|
||||
# chat_stream 应被调用（流式路径）
|
||||
gateway.chat_stream.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_first_chunk_timeout_falls_back_to_chat(self):
|
||||
"""chat_stream 首 chunk 超时时 fallback 到 gateway.chat()。"""
|
||||
|
||||
class _HungStream:
|
||||
"""模拟 DashScope via LiteLLM 不返回首 chunk 的 hang。"""
|
||||
|
||||
async def __anext__(self):
|
||||
# 永远不返回 — 触发 wait_for 超时
|
||||
import asyncio as _asyncio
|
||||
|
||||
await _asyncio.Event().wait() # 永久阻塞
|
||||
raise StopAsyncIteration # pragma: no cover
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def aclose(self):
|
||||
pass
|
||||
|
||||
team = BoardTeam()
|
||||
# 把超时调小让测试快速通过
|
||||
orchestrator = BoardOrchestrator(team=team)
|
||||
orchestrator._STREAM_FIRST_CHUNK_TIMEOUT_S = 0.1
|
||||
expert = _make_mock_expert("tester", is_lead=False)
|
||||
|
||||
gateway = _make_mock_gateway("fallback 内容")
|
||||
# chat_stream 返回 hang 的 stream
|
||||
gateway.chat_stream = MagicMock(return_value=_HungStream())
|
||||
expert.agent._llm_gateway = gateway
|
||||
orchestrator._get_llm_gateway = lambda e: gateway
|
||||
|
||||
with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
|
||||
result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
|
||||
|
||||
# 应该 fallback 到 chat() 并返回内容
|
||||
assert "fallback 内容" in result
|
||||
gateway.chat.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_empty_falls_back_to_chat(self):
|
||||
"""chat_stream 立即结束（无 chunks）时 fallback 到 chat()。"""
|
||||
|
||||
async def _empty_stream():
|
||||
return
|
||||
yield # pragma: no cover — make this an async generator
|
||||
|
||||
team = BoardTeam()
|
||||
orchestrator = BoardOrchestrator(team=team)
|
||||
expert = _make_mock_expert("tester", is_lead=False)
|
||||
|
||||
gateway = _make_mock_gateway("fallback 内容")
|
||||
gateway.chat_stream = MagicMock(return_value=_empty_stream())
|
||||
expert.agent._llm_gateway = gateway
|
||||
orchestrator._get_llm_gateway = lambda e: gateway
|
||||
|
||||
with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
|
||||
result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
|
||||
|
||||
assert "fallback 内容" in result
|
||||
gateway.chat.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_chat_stream_raising_falls_back_to_chat(self):
|
||||
"""chat_stream 抛异常时 fallback 到 chat()。"""
|
||||
|
||||
def _raising_stream(*a, **kw):
|
||||
raise RuntimeError("provider error")
|
||||
|
||||
team = BoardTeam()
|
||||
orchestrator = BoardOrchestrator(team=team)
|
||||
expert = _make_mock_expert("tester", is_lead=False)
|
||||
|
||||
gateway = _make_mock_gateway("fallback 内容")
|
||||
gateway.chat_stream = MagicMock(side_effect=_raising_stream)
|
||||
expert.agent._llm_gateway = gateway
|
||||
orchestrator._get_llm_gateway = lambda e: gateway
|
||||
|
||||
with patch.object(orchestrator, "_broadcast_event", new_callable=AsyncMock):
|
||||
result = await orchestrator._stream_expert_speech(expert, 1, "test prompt")
|
||||
|
||||
assert "fallback 内容" in result
|
||||
gateway.chat.assert_called_once()
|
||||
|
|
|
|||
Loading…
Reference in New Issue