From ed1e28978500b8d54a671962ad7681bf7f074301 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 15:35:15 +0800 Subject: [PATCH] fix(board): add first-chunk timeout to _stream_expert_speech MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Commit 36b0296 changed 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 (commit d0fe661). --- src/agentkit/experts/board_orchestrator.py | 73 +++++++++-- tests/unit/experts/test_board_orchestrator.py | 116 ++++++++++++++++++ 2 files changed, 180 insertions(+), 9 deletions(-) diff --git a/src/agentkit/experts/board_orchestrator.py b/src/agentkit/experts/board_orchestrator.py index 2376a90..656304f 100644 --- a/src/agentkit/experts/board_orchestrator.py +++ b/src/agentkit/experts/board_orchestrator.py @@ -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 diff --git a/tests/unit/experts/test_board_orchestrator.py b/tests/unit/experts/test_board_orchestrator.py index 21ad855..0ae3873 100644 --- a/tests/unit/experts/test_board_orchestrator.py +++ b/tests/unit/experts/test_board_orchestrator.py @@ -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()