From d0fe6611c7b9de3a884633be75328d77292ccfc0 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 15:09:55 +0800 Subject: [PATCH 1/3] fix(core): prefer _llm_gateway over _llm_client in _handle_llm_generate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AgentPool.create_agent() injects only llm_gateway (v2 path) into ConfigDrivenAgent and never sets the legacy _llm_client field. The previous _handle_llm_generate implementation gated the entire LLM call on `self._llm_client is None`, so every expert/board agent created via the pool silently fell through to the `llm_generate_no_client` placeholder — which is why the private board moderator/experts reported "LLM 不可用" even when the gateway and providers were healthy. Mirror the same precedence used by `_handle_direct`: prefer `_llm_gateway` (with agent_name/task_type for usage tracking), fall back to `_llm_client` for backward compatibility, and only return the no-client placeholder when both are unavailable. Add a unit test that constructs an agent with `llm_gateway` only (_llm_client is None) and asserts the gateway.chat path is taken. --- src/agentkit/core/config_driven.py | 23 +++++++++++++-- tests/unit/test_config_driven.py | 46 ++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/src/agentkit/core/config_driven.py b/src/agentkit/core/config_driven.py index 0d461aa..0206e5e 100644 --- a/src/agentkit/core/config_driven.py +++ b/src/agentkit/core/config_driven.py @@ -1358,7 +1358,14 @@ class ConfigDrivenAgent(BaseAgent, EvolutionMixin): return gateway async def _handle_llm_generate(self, task: TaskMessage) -> dict: - """LLM 生成模式：渲染 Prompt → 调用 LLM → 解析输出""" + """LLM 生成模式：渲染 Prompt → 调用 LLM → 解析输出 + + ponytail: 优先使用 ``_llm_gateway``（v2 路径，``AgentPool`` 注入的）， + 仅在 gateway 缺失时回退到 ``_llm_client``（legacy 字段）。原实现只检查 + ``_llm_client``，导致 ``AgentPool`` 创建的 expert/board agent 在 + ``_llm_client is None`` 但 ``_llm_gateway`` 已注入时错误降级为 + ``llm_generate_no_client``。 + """ if not self._prompt_template: raise ConfigValidationError( agent_name=self.name, @@ -1371,7 +1378,19 @@ class ConfigDrivenAgent(BaseAgent, EvolutionMixin): variables["task_type"] = task.task_type messages = self._prompt_template.render(variables=variables) - # 调用 LLM + # v2 路径：_llm_gateway 优先（AgentPool 仅注入 gateway，不设 _llm_client） + if self._llm_gateway is not None: + llm_params = self._config.llm.copy() if self._config.llm else {} + model = llm_params.get("model", "default") + response = await self._llm_gateway.chat( + messages=messages, + model=model, + agent_name=self.name, + task_type=task.task_type, + ) + return self._parse_llm_response(response.content) + + # 兼容旧路径：_llm_client（legacy） if self._llm_client is None: # 无 LLM 客户端时返回渲染后的 Prompt（降级模式） return { diff --git a/tests/unit/test_config_driven.py b/tests/unit/test_config_driven.py index a0ed6ad..409b263 100644 --- a/tests/unit/test_config_driven.py +++ b/tests/unit/test_config_driven.py @@ -144,6 +144,52 @@ class TestConfigDrivenAgent: assert result["mode"] == "llm_generate_no_client" assert len(result["messages"]) == 2 # system + user + async def test_llm_generate_with_gateway_only(self): + """仅注入 LLMGateway（_llm_client=None）时应正常调用 LLM。 + + 复现：AgentPool.create_agent() 只注入 llm_gateway，不设 _llm_client。 + 原实现错误降级为 llm_generate_no_client，导致私董会专家返回占位文本。 + """ + + class _StubResponse: + def __init__(self, content: str) -> None: + self.content = content + + class _StubGateway: + def __init__(self) -> None: + self.calls: list[dict] = [] + + async def chat(self, *, messages, model, agent_name, task_type, **kwargs): + self.calls.append( + { + "messages": messages, + "model": model, + "agent_name": agent_name, + "task_type": task_type, + } + ) + return _StubResponse(json.dumps({"title": "FromGateway", "content": "OK"})) + + gateway = _StubGateway() + config = AgentConfig.from_dict(_sample_llm_config()) + agent = ConfigDrivenAgent(config=config, llm_gateway=gateway) + + # 验证前置条件：_llm_client 必须为 None，模拟 AgentPool 注入路径 + assert agent._llm_client is None + assert agent._llm_gateway is gateway + + task = _make_task() + result = await agent.handle_task(task) + + # 不应再走 no_client 降级 + assert "mode" not in result or result.get("mode") != "llm_generate_no_client" + assert result["title"] == "FromGateway" + assert result["content"] == "OK" + # 验证 gateway.chat 被实际调用 + assert len(gateway.calls) == 1 + assert gateway.calls[0]["agent_name"] == config.name + assert gateway.calls[0]["model"] == "gpt-4" + async def test_llm_generate_with_client(self): """有 LLM 客户端时调用 LLM 并解析结果""" From ed1e28978500b8d54a671962ad7681bf7f074301 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 15:35:15 +0800 Subject: [PATCH 2/3] 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() From 0daa4d21b04c90df901ad830efe0e392edd5c8fd Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 15:53:27 +0800 Subject: [PATCH 3/3] fix(review): avoid double broadcast when stream breaks mid-way MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code review finding (P2): when chat_stream successfully emits the first chunk (broadcasted to UI) but then breaks during subsequent chunks, the fallback path called gateway.chat() and broadcasted the FULL content via _replay_stream — causing the UI to see partial streamed content followed by duplicated full content. Fix: if `total` already contains partial content from first_chunk, return it directly without calling gateway.chat(). Partial content is better than duplicated content. The non-streaming fallback only runs when no chunks were broadcasted (total is empty). Add test_stream_breaks_after_first_chunk_returns_partial covering this edge case: first chunk yields "部分内容", second __anext__ raises RuntimeError. Asserts result is the partial content, gateway.chat() is never called, and broadcast_event is called exactly once (for the first chunk only). --- src/agentkit/experts/board_orchestrator.py | 11 +++++ tests/unit/experts/test_board_orchestrator.py | 46 +++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/src/agentkit/experts/board_orchestrator.py b/src/agentkit/experts/board_orchestrator.py index 656304f..7db0504 100644 --- a/src/agentkit/experts/board_orchestrator.py +++ b/src/agentkit/experts/board_orchestrator.py @@ -424,6 +424,17 @@ class BoardOrchestrator: # Fallback: non-streaming path. Emit the whole content as small # chunks so the UI still renders progressively rather than going # silent and then dumping the whole text in one frame. + # + # If we already broadcasted partial content from first_chunk before + # the stream broke, do NOT call gateway.chat() — that would broadcast + # the full content again and the UI would see duplicated text. + # Return the partial content instead; partial is better than duplicated. + 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() try: response = await gateway.chat( messages=[{"role": "user", "content": prompt}], diff --git a/tests/unit/experts/test_board_orchestrator.py b/tests/unit/experts/test_board_orchestrator.py index 0ae3873..24be12c 100644 --- a/tests/unit/experts/test_board_orchestrator.py +++ b/tests/unit/experts/test_board_orchestrator.py @@ -459,3 +459,49 @@ class TestStreamExpertSpeechFallback: assert "fallback 内容" in result gateway.chat.assert_called_once() + + @pytest.mark.asyncio + async def test_stream_breaks_after_first_chunk_returns_partial(self): + """首 chunk 成功但后续断流时返回部分内容，不调 chat() 避免双重广播。""" + + class _BreakAfterFirstStream: + """首个 chunk 正常返回，第二个 __anext__ 抛异常模拟断流。""" + + def __init__(self): + self._first_yielded = False + + async def __anext__(self): + if not self._first_yielded: + self._first_yielded = True + chunk = MagicMock() + chunk.content = "部分内容" + return chunk + raise RuntimeError("stream broke mid-way") + + def __aiter__(self): + return self + + async def aclose(self): + pass + + 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=_BreakAfterFirstStream()) + expert.agent._llm_gateway = gateway + orchestrator._get_llm_gateway = lambda e: gateway + + with patch.object( + orchestrator, "_broadcast_event", new_callable=AsyncMock + ) as mock_broadcast: + result = await orchestrator._stream_expert_speech(expert, 1, "test prompt") + + # 返回部分内容，而不是完整 fallback 内容 + assert "部分内容" in result + assert "完整 fallback" not in result + # 不应调 chat()（避免双重广播） + gateway.chat.assert_not_called() + # 只广播了 first_chunk 的部分内容 + assert mock_broadcast.call_count == 1