From 81a35dac27b546687d6db6eb96a0e4a8bb68b534 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 13:43:31 +0800 Subject: [PATCH] feat(experts): U4 TeamOrchestrator parallel independent subtasks (R1-R5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add explicit support for parallel execution of dependency-free subtasks: - TeamPlan.get_independent_subtasks(): returns phases with depends_on==[] (introspection entry — topological_sort already groups them in layer 0) - TeamOrchestrator.MAX_INDEPENDENT_SUBTASKS=10: aligns with router.MAX_EXPERTS - _rebalance_independent_subtasks(): when Lead over-decomposes (>10 independent subtasks), re-decompose once with merge hint. Fallbacks: no gateway → keep original; LLM error → keep original; single-phase fallback → keep original (don't collapse 11→1); still over → return new (MAX_PHASES truncation handles it) Tests: 15 new tests covering get_independent_subtasks, topological_sort layer contract, SharedWorkspace path uniqueness, rebalance (5 paths), and full execute() integration. Existing TeamOrchestrator tests pass. --- src/agentkit/experts/orchestrator.py | 72 +++++ src/agentkit/experts/plan.py | 9 + tests/unit/test_team_parallel.py | 464 +++++++++++++++++++++++++++ 3 files changed, 545 insertions(+) create mode 100644 tests/unit/test_team_parallel.py diff --git a/src/agentkit/experts/orchestrator.py b/src/agentkit/experts/orchestrator.py index f325bc1..8aa501a 100644 --- a/src/agentkit/experts/orchestrator.py +++ b/src/agentkit/experts/orchestrator.py @@ -62,6 +62,9 @@ class TeamOrchestrator( MAX_DEBATE_ROUNDS = 4 # Hard cap on debate rounds per phase MAX_DEBATES = 3 # Hard cap on auto-inserted debate phases per execution DEFAULT_MAX_CONCURRENT_PHASES = 3 # 同层最大并发阶段数，避免 LLM 限流洪峰 + # IQ-Boost/U4 (R4): aligns with router.MAX_EXPERTS — if Lead decomposes + # more independent subtasks than this, re-decompose once with a merge hint. + MAX_INDEPENDENT_SUBTASKS = 10 STOP_COMMANDS = frozenset({"/stop", "停止", "stop", "结束"}) # G9/U4: RollbackExecutor default timeout for validation_command / rollback_command. # Override via constructor `rollback_timeout` from `rollback.default_timeout` config. @@ -197,6 +200,11 @@ class TeamOrchestrator( PlanPhase(name="执行", assigned_expert=lead.config.name, task_description=task) ] + # IQ-Boost/U4 (R4): if Lead over-decomposed independent subtasks beyond + # MAX_INDEPENDENT_SUBTASKS, ask Lead to re-decompose with a merge hint + # (one retry — further overflow falls through to MAX_PHASES truncation). + phases = await self._rebalance_independent_subtasks(lead, task, phases) + plan.phases = phases[: self.MAX_PHASES] # U3: Optionally add plan review debate before execution @@ -531,6 +539,70 @@ class TeamOrchestrator( self._team.set_status(TeamStatus.EXECUTING) return await self._run_pipeline(lead, plan, phase_results, task) + async def _rebalance_independent_subtasks( + self, lead: Expert, task: str, phases: list[PlanPhase] + ) -> list[PlanPhase]: + """IQ-Boost/U4 (R4): if phases contain more independent subtasks + (depends_on == []) than MAX_INDEPENDENT_SUBTASKS, ask Lead to + re-decompose with a merge hint. One retry only — further overflow + is handled by MAX_PHASES truncation in execute(). + + Returns the original phases if count is within limit or retry fails. + """ + # Count independent subtasks via a transient TeamPlan (avoids mutating + # the real plan before execute() finalizes phase list). + independent_count = sum(1 for ph in phases if not ph.depends_on) + if independent_count <= self.MAX_INDEPENDENT_SUBTASKS: + return phases + + logger.info( + f"U4: Lead decomposed {independent_count} independent subtasks " + f"(> {self.MAX_INDEPENDENT_SUBTASKS}), requesting re-decompose with merge hint" + ) + + gateway = self._get_llm_gateway(lead) + if not gateway: + # No gateway — can't re-decompose; rely on MAX_PHASES truncation. + return phases + + # Re-decompose with explicit merge hint + original_hint = ( + f"Previous decomposition produced {independent_count} independent subtasks " + f"(no dependencies), exceeding the {self.MAX_INDEPENDENT_SUBTASKS} limit. " + f"Please merge related subtasks so the total independent (depends_on=[]) " + f"subtasks is at most {self.MAX_INDEPENDENT_SUBTASKS}. Keep dependencies " + f"where natural." + ) + # Temporarily wrap task with hint — call _decompose_task with augmented task. + augmented_task = f"{task}\n\n[Re-decomposition hint]: {original_hint}" + try: + new_phases = await self._decompose_task(lead, augmented_task) + except (LLMProviderError, asyncio.TimeoutError, ConnectionError, ValueError) as e: + logger.warning(f"U4 re-decompose failed: {e}, keeping original phases") + return phases + + if not new_phases: + return phases + + new_independent = sum(1 for ph in new_phases if not ph.depends_on) + if new_independent <= self.MAX_INDEPENDENT_SUBTASKS: + # ponytail: detect single-phase fallback (LLM returned invalid + # JSON → _decompose_task returned degenerate single phase). + # Collapsing 11 subtasks to 1 is worse than truncation; keep + # original so MAX_PHASES handles it. + if len(new_phases) == 1 and new_independent < independent_count: + logger.info("U4: re-decompose fell back to single phase, keeping original") + return phases + logger.info(f"U4: re-decompose succeeded ({new_independent} independent subtasks)") + return new_phases + + # Still over limit — take the new decomposition anyway (truncation will cap it) + logger.info( + f"U4: re-decompose still has {new_independent} independent subtasks, " + f"proceeding with MAX_PHASES truncation" + ) + return new_phases + async def _decompose_task(self, lead: Expert, task: str) -> list[PlanPhase]: """Lead Expert decomposes task into phases using LLM. diff --git a/src/agentkit/experts/plan.py b/src/agentkit/experts/plan.py index 06f7b81..02160ea 100644 --- a/src/agentkit/experts/plan.py +++ b/src/agentkit/experts/plan.py @@ -443,3 +443,12 @@ class TeamPlan: in_degree[dep_id] -= 1 return layers + + def get_independent_subtasks(self) -> list[PlanPhase]: + """返回无依赖的子任务（depends_on == []）。 + + IQ-Boost/U4 (R1): 用于 Lead 分解后检查独立子任务数量是否超过 + MAX_EXPERTS。这些子任务会被 topological_sort 派发到 layer 0 并行执行 + （同层并行已有，本方法仅提供显式 introspection 入口）。 + """ + return [ph for ph in self.phases if not ph.depends_on] diff --git a/tests/unit/test_team_parallel.py b/tests/unit/test_team_parallel.py new file mode 100644 index 0000000..8bd8f1e --- /dev/null +++ b/tests/unit/test_team_parallel.py @@ -0,0 +1,464 @@ +"""U4: TeamOrchestrator 无依赖子任务并行模式单元测试 (R1-R5). + +Covers: +- TeamPlan.get_independent_subtasks() introspection +- _rebalance_independent_subtasks() MAX_EXPERTS enforcement (R4) +- topological_sort 同层并行 (existing behavior, validated here for U4 contract) +- SharedWorkspace 路径唯一性 (phase_id is UUID → no collision) +- 综合阶段等待所有并行子任务完成 (layer-sequential contract) +""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from agentkit.core.handoff_transport import InProcessHandoffTransport +from agentkit.core.protocol import TaskResult, TaskStatus +from agentkit.experts.config import ExpertConfig +from agentkit.experts.expert import Expert +from agentkit.experts.orchestrator import TeamOrchestrator +from agentkit.experts.plan import PlanPhase, TeamPlan +from agentkit.experts.team import ExpertTeam + +from tests.unit.experts._helpers import ( + make_chat_stream_mock, + make_execute_stream_mock, +) + + +# ── Helpers ──────────────────────────────────────────────────────────── + + +def _make_expert_config(name: str, is_lead: bool = False) -> ExpertConfig: + return ExpertConfig( + name=name, + agent_type="expert", + persona="测试专家", + thinking_style="逻辑推理", + bound_skills=["skill_a"], + is_lead=is_lead, + task_mode="llm_generate", + prompt={"identity": "测试"}, + llm={"model": "default"}, + ) + + +def _make_mock_expert(name: str, is_lead: bool = False) -> MagicMock: + config = _make_expert_config(name=name, is_lead=is_lead) + expert = MagicMock(spec=Expert) + expert.config = config + expert.is_active = True + expert.team_id = None + expert.get_capabilities_summary.return_value = { + "name": name, + "persona": config.persona, + "thinking_style": config.thinking_style, + "bound_skills": config.bound_skills, + "is_lead": is_lead, + } + mock_agent = MagicMock() + mock_agent.execute = AsyncMock( + return_value=TaskResult( + task_id="test", + agent_name=name, + status=TaskStatus.COMPLETED.value, + output_data={"content": f"Result from {name}"}, + error_message=None, + started_at=None, + completed_at=None, + ) + ) + mock_agent.execute_stream = make_execute_stream_mock(f"Result from {name}") + mock_agent._llm_gateway = None + expert.agent = mock_agent + return expert + + +def _make_team_with_experts( + expert_names: list[str] | None = None, + lead_name: str = "lead", +) -> ExpertTeam: + team = ExpertTeam() + transport = AsyncMock(spec=InProcessHandoffTransport) + team._handoff_transport = transport + if expert_names is None: + expert_names = [lead_name, "member1", "member2"] + for name in expert_names: + is_lead = name == lead_name + expert = _make_mock_expert(name=name, is_lead=is_lead) + team._experts[name] = expert + if is_lead: + team._lead_expert_name = name + return team + + +def _make_phase( + id: str, name: str, assigned_expert: str, depends_on: list[str] | None = None +) -> PlanPhase: + return PlanPhase( + id=id, + name=name, + assigned_expert=assigned_expert, + task_description=f"task for {name}", + depends_on=depends_on or [], + ) + + +def _make_mock_llm_gateway_with_phases( + phases_responses: list[list[dict]], + synthesis_content: str = "综合结果", +) -> MagicMock: + """Gateway that returns successive decompositions then synthesis. + + phases_responses: each element is a list of phase dicts to return on + successive decomposition calls. After exhausted, returns synthesis. + """ + gateway = AsyncMock() + decomp_responses = [MagicMock(content=json.dumps(phases)) for phases in phases_responses] + synth_response = MagicMock(content=synthesis_content) + call_count = [0] + + async def chat_side_effect(messages, model=None, **kwargs): + call_count[0] += 1 + if call_count[0] <= len(decomp_responses): + return decomp_responses[call_count[0] - 1] + return synth_response + + gateway.chat = AsyncMock(side_effect=chat_side_effect) + gateway.chat_stream = make_chat_stream_mock(synthesis_content) + return gateway + + +# ── TeamPlan.get_independent_subtasks ───────────────────────────────── + + +class TestGetIndependentSubtasks: + """U4/R1: TeamPlan.get_independent_subtasks() returns phases with no + depends_on.""" + + def test_returns_empty_for_no_phases(self): + plan = TeamPlan(task="test", lead_expert="lead") + assert plan.get_independent_subtasks() == [] + + def test_returns_only_independent_phases(self): + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + _make_phase("p1", "A", "lead", depends_on=[]), + _make_phase("p2", "B", "member1", depends_on=["p1"]), + _make_phase("p3", "C", "member2", depends_on=[]), + _make_phase("p4", "D", "member1", depends_on=["p1", "p3"]), + ], + ) + independent = plan.get_independent_subtasks() + assert len(independent) == 2 + assert {ph.id for ph in independent} == {"p1", "p3"} + + def test_returns_all_when_no_dependencies(self): + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + _make_phase("p1", "A", "lead", depends_on=[]), + _make_phase("p2", "B", "member1", depends_on=[]), + _make_phase("p3", "C", "member2", depends_on=[]), + ], + ) + independent = plan.get_independent_subtasks() + assert len(independent) == 3 + + +# ── topological_sort 同层并行 contract (U4/R1, R3) ──────────────────── + + +class TestParallelLayerContract: + """U4/R1, R3: independent subtasks land in layer 0 (parallel); synthesis + waits for all layers to complete (layer-sequential).""" + + def test_three_independent_subtasks_in_layer_0(self): + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + _make_phase("p1", "A", "lead", depends_on=[]), + _make_phase("p2", "B", "member1", depends_on=[]), + _make_phase("p3", "C", "member2", depends_on=[]), + ], + ) + layers = plan.topological_sort() + assert len(layers) == 1 + assert len(layers[0]) == 3 + assert {ph.id for ph in layers[0]} == {"p1", "p2", "p3"} + + def test_shared_dependency_degrades_to_layered(self): + """R3: subtasks sharing an upstream dependency are NOT parallel with + that upstream — they land in later layers.""" + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + _make_phase("p1", "Base", "lead", depends_on=[]), + _make_phase("p2", "A", "member1", depends_on=["p1"]), + _make_phase("p3", "B", "member2", depends_on=["p1"]), + ], + ) + layers = plan.topological_sort() + assert len(layers) == 2 + assert len(layers[0]) == 1 # only Base + assert layers[0][0].id == "p1" + assert len(layers[1]) == 2 # A and B parallel (same layer) + assert {ph.id for ph in layers[1]} == {"p2", "p3"} + + def test_synthesis_waits_for_all_layers(self): + """R3: dependent phase in last layer — synthesis cannot run until + its layer completes (validated via topological_sort layer count).""" + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + _make_phase("p1", "A", "lead", depends_on=[]), + _make_phase("p2", "B", "member1", depends_on=["p1"]), + _make_phase("p3", "C", "member2", depends_on=["p2"]), + ], + ) + layers = plan.topological_sort() + assert len(layers) == 3 # strict sequential — no parallelism + + +# ── SharedWorkspace 路径唯一性 (U4/R2) ───────────────────────────────── + + +class TestSharedWorkspacePathUniqueness: + """U4/R2: parallel subtask output paths must not collide. PlanPhase.id + is UUID by default → unique per phase, ensuring + ``{plan_id}/phase/{phase_id}/output`` never overlaps.""" + + def test_default_phase_ids_are_unique(self): + phases = [ + PlanPhase(name=f"phase_{i}", assigned_expert="lead", task_description="t") + for i in range(5) + ] + ids = [ph.id for ph in phases] + assert len(set(ids)) == len(ids), "phase_ids must be unique" + + def test_independent_subtasks_have_unique_output_paths(self): + plan = TeamPlan( + task="test", + lead_expert="lead", + phases=[ + PlanPhase(name="A", assigned_expert="lead", task_description="t"), + PlanPhase(name="B", assigned_expert="member1", task_description="t"), + PlanPhase(name="C", assigned_expert="member2", task_description="t"), + ], + ) + independent = plan.get_independent_subtasks() + paths = {f"{plan.id}/phase/{ph.id}/output" for ph in independent} + assert len(paths) == len(independent) + + +# ── _rebalance_independent_subtasks (U4/R4 MAX_EXPERTS) ──────────────── + + +class TestRebalanceIndependentSubtasks: + """U4/R4: when Lead over-decomposes independent subtasks beyond + MAX_INDEPENDENT_SUBTASKS, request re-decompose with merge hint.""" + + @pytest.mark.asyncio + async def test_no_rebalance_when_within_limit(self): + """5 independent subtasks ≤ 10 → no re-decompose.""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + # 5 independent phases + phases = [_make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(5)] + original_phases = list(phases) + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=phases + ) + assert result is original_phases or result == original_phases + + @pytest.mark.asyncio + async def test_rebalance_triggers_when_over_limit(self): + """11 independent subtasks > 10 → re-decompose with merge hint.""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + + # Original: 11 independent (over limit) + original_phases = [ + _make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(11) + ] + # Re-decomposed: 8 independent (within limit) — merge succeeded + gateway = _make_mock_llm_gateway_with_phases( + phases_responses=[ + [ # first call (original _decompose_task already happened — + # but _rebalance calls _decompose_task again, returning this) + { + "name": f"merged_{i}", + "assigned_expert": "lead", + "task_description": "t", + "depends_on": [], + } + for i in range(8) + ], + ] + ) + team._experts["lead"].agent._llm_gateway = gateway + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=original_phases + ) + # Should return the re-decomposed phases (8 independent) + assert len(result) == 8 + assert all(not ph.depends_on for ph in result) + # Original phases should NOT be returned + assert {ph.id for ph in result}.isdisjoint({ph.id for ph in original_phases}) + + @pytest.mark.asyncio + async def test_rebalance_no_gateway_returns_original(self): + """No LLM gateway → can't re-decompose, return original.""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + # gateway is None by default in _make_mock_expert + original_phases = [ + _make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(11) + ] + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=original_phases + ) + assert result is original_phases + + @pytest.mark.asyncio + async def test_rebalance_still_over_limit_returns_new_anyway(self): + """Re-decompose still produces >MAX independent → return new phases + (MAX_PHASES truncation handles it in execute()).""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + + original_phases = [ + _make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(11) + ] + # Re-decompose returns 12 independent (still over) + gateway = _make_mock_llm_gateway_with_phases( + phases_responses=[ + [ + { + "name": f"q{i}", + "assigned_expert": "lead", + "task_description": "t", + "depends_on": [], + } + for i in range(12) + ], + ] + ) + team._experts["lead"].agent._llm_gateway = gateway + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=original_phases + ) + # Should return the 12-phase re-decomposition (truncation happens later) + assert len(result) == 12 + + @pytest.mark.asyncio + async def test_rebalance_decompose_failure_returns_original(self): + """Re-decompose raises LLMProviderError → return original phases.""" + from agentkit.core.exceptions import LLMProviderError + + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + + original_phases = [ + _make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(11) + ] + gateway = AsyncMock() + gateway.chat = AsyncMock(side_effect=LLMProviderError("LLM down")) + gateway.chat_stream = make_chat_stream_mock("err") + team._experts["lead"].agent._llm_gateway = gateway + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=original_phases + ) + assert result is original_phases + + @pytest.mark.asyncio + async def test_rebalance_empty_redecompose_returns_original(self): + """Re-decompose returns empty list → return original phases.""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + + original_phases = [ + _make_phase(f"p{i}", f"phase_{i}", "lead", depends_on=[]) for i in range(11) + ] + # Gateway returns malformed JSON → _parse_phases returns [] + gateway = AsyncMock() + bad_response = MagicMock(content="not json") + gateway.chat = AsyncMock(return_value=bad_response) + gateway.chat_stream = make_chat_stream_mock("err") + team._experts["lead"].agent._llm_gateway = gateway + + result = await orchestrator._rebalance_independent_subtasks( + lead=team.lead_expert, task="test", phases=original_phases + ) + assert result is original_phases + + +# ── Integration: full execute() with parallel subtasks ──────────────── + + +class TestExecuteParallelSubtasks: + """U4/R1, R3, R5: full execute() path with independent subtasks running + in parallel + synthesis waits for all.""" + + @pytest.mark.asyncio + async def test_three_parallel_subtasks_complete_then_synthesis(self): + """3 independent subtasks → all run in layer 0 → synthesis runs + after all complete.""" + team = _make_team_with_experts() + orchestrator = TeamOrchestrator(team) + + gateway = _make_mock_llm_gateway_with_phases( + phases_responses=[ + [ + { + "name": "A", + "assigned_expert": "lead", + "task_description": "a", + "depends_on": [], + }, + { + "name": "B", + "assigned_expert": "member1", + "task_description": "b", + "depends_on": [], + }, + { + "name": "C", + "assigned_expert": "member2", + "task_description": "c", + "depends_on": [], + }, + ], + ] + ) + team._experts["lead"].agent._llm_gateway = gateway + + result = await orchestrator.execute("并行任务") + + assert result["status"] == "completed" + plan: TeamPlan = result["plan"] + layers = plan.topological_sort() + assert len(layers) == 1 # all 3 in layer 0 (parallel) + assert len(layers[0]) == 3 + + # All phases completed + completed = plan.completed_phases + assert len(completed) == 3 + + # phase_results has all 3 + phase_results: dict = result["phase_results"] + assert len(phase_results) == 3