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:
parent
d0fe6611c7
commit
ed1e289785
|
|
@ -311,12 +311,20 @@ class BoardOrchestrator:
|
||||||
# loop is unchanged.
|
# loop is unchanged.
|
||||||
return await self._stream_expert_speech(expert, round, prompt)
|
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:
|
async def _stream_expert_speech(self, expert: Expert, round: int, prompt: str) -> str:
|
||||||
"""Stream an expert's speech via chat_stream, emitting chunks.
|
"""Stream an expert's speech via chat_stream, emitting chunks.
|
||||||
|
|
||||||
Falls back to non-streaming ``chat()`` when ``chat_stream`` is
|
Falls back to non-streaming ``chat()`` when ``chat_stream`` is
|
||||||
unavailable (e.g. an LLM provider without streaming support) or
|
unavailable (e.g. an LLM provider without streaming support), raises
|
||||||
raises before any chunk is produced.
|
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
|
ponytail: when the LLM does not actually stream (returns a single
|
||||||
big chunk), we still want the UI to see content appearing
|
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
|
sentence/line chunks and emit them with a small delay. The
|
||||||
``expert_speech_chunk`` event already handles duplicate-sender
|
``expert_speech_chunk`` event already handles duplicate-sender
|
||||||
dedup, so emitting many small chunks is safe.
|
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)
|
gateway = self._get_llm_gateway(expert)
|
||||||
assert gateway is not None # checked by caller
|
assert gateway is not None # checked by caller
|
||||||
total = ""
|
total = ""
|
||||||
# Emit an opening chunk-less event so the UI can create the streaming
|
|
||||||
# placeholder before the first token arrives (keeps the first paint
|
# Try streaming with a hard first-chunk deadline. If the provider
|
||||||
# aligned with the streaming indicator).
|
# accepts the request but never emits the first chunk (observed on
|
||||||
|
# DashScope/DeepSeek via LiteLLM), fall through to non-streaming.
|
||||||
|
stream_obj = None
|
||||||
try:
|
try:
|
||||||
streamed_chunk_count = 0
|
stream_obj = gateway.chat_stream(
|
||||||
async for chunk in gateway.chat_stream(
|
|
||||||
messages=[{"role": "user", "content": prompt}],
|
messages=[{"role": "user", "content": prompt}],
|
||||||
model="default",
|
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 ""
|
delta = chunk.content or ""
|
||||||
if not delta:
|
if not delta:
|
||||||
continue
|
return
|
||||||
total += delta
|
total += delta
|
||||||
streamed_chunk_count += 1
|
streamed_chunk_count += 1
|
||||||
await self._broadcast_event(
|
await self._broadcast_event(
|
||||||
|
|
@ -353,6 +384,10 @@ class BoardOrchestrator:
|
||||||
"role": "expert",
|
"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
|
# If the LLM "streamed" but only delivered one big chunk, still
|
||||||
# let the UI see content arrive progressively.
|
# let the UI see content arrive progressively.
|
||||||
if streamed_chunk_count <= 1 and total:
|
if streamed_chunk_count <= 1 and total:
|
||||||
|
|
@ -363,8 +398,28 @@ class BoardOrchestrator:
|
||||||
f"Provider for '{expert.config.name}' lacks chat_stream, "
|
f"Provider for '{expert.config.name}' lacks chat_stream, "
|
||||||
f"falling back to non-streaming: {e}"
|
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:
|
except Exception as e:
|
||||||
logger.warning(f"Expert '{expert.config.name}' stream failed: {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
|
# Fallback: non-streaming path. Emit the whole content as small
|
||||||
# chunks so the UI still renders progressively rather than going
|
# chunks so the UI still renders progressively rather than going
|
||||||
|
|
|
||||||
|
|
@ -343,3 +343,119 @@ class TestBoardOrchestratorBroadcast:
|
||||||
|
|
||||||
# 不应抛出异常
|
# 不应抛出异常
|
||||||
await orchestrator._broadcast_event("board_started", {"topic": "测试"})
|
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