From 1e4bba24222d03e80baf54eddd5d811aebde2b11 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Sat, 13 Dec 2025 13:22:11 +0100 Subject: [PATCH] feat: add sequential multi-expert execution to orchestration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds multi-expert coordination infrastructure: - ExecutionMode enum (SEQUENTIAL, PARALLEL) - MultiExpertResult dataclass for aggregating results - execute_sequential(): Tasks run one after another - execute_parallel(): Tasks run concurrently via asyncio.gather - orchestrate_multi_expert(): Streaming think updates during multi-expert work Supports: - Stop-on-failure mode for sequential execution - Partial failure handling (some succeed, some fail) - Result aggregation with combined output formatting - Exception handling in parallel execution 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/agents/orchestration.py | 299 +++++++++++++++++++++++++++++++++++- 1 file changed, 298 insertions(+), 1 deletion(-) diff --git a/src/agents/orchestration.py b/src/agents/orchestration.py index f3fb1d2..32ffd80 100644 --- a/src/agents/orchestration.py +++ b/src/agents/orchestration.py @@ -6,8 +6,17 @@ with streaming think updates to keep users informed of progress. Key pattern: Stream user-facing interactions, use run() internally to avoid Ollama streaming+tool call bugs. + +Supports: +- Single expert delegation with think updates +- Sequential multi-expert execution (task A → task B → task C) +- Parallel multi-expert execution (tasks A, B, C concurrently) +- Result aggregation from multiple experts +- Partial failure handling """ -from dataclasses import dataclass +import asyncio +from dataclasses import dataclass, field +from enum import Enum from typing import AsyncGenerator, Optional, Callable, Any from src.agents.delegation import DelegationTask, DelegationResult, delegate_to_librarian @@ -16,6 +25,12 @@ from src.core.logging_config import get_logger logger = get_logger(__name__) +class ExecutionMode(str, Enum): + """Execution mode for multi-expert coordination.""" + SEQUENTIAL = "sequential" # One at a time, in order + PARALLEL = "parallel" # All at once, concurrently + + @dataclass class OrchestrationContext: """ @@ -218,3 +233,285 @@ def extract_delegation_context( result["context"] = context_match.group(1).strip() return result + + +# ============================================================================ +# Multi-Expert Coordination +# ============================================================================ + +@dataclass +class MultiExpertResult: + """ + Aggregated result from multiple expert delegations. + + Attributes: + results: Dict mapping expert name to their result + all_succeeded: True if all delegations succeeded + failed_experts: List of expert names that failed + combined_output: Aggregated output from all successful experts + """ + results: dict[str, DelegationResult] = field(default_factory=dict) + all_succeeded: bool = True + failed_experts: list[str] = field(default_factory=list) + combined_output: str = "" + + def add_result(self, result: DelegationResult) -> None: + """Add a result and update aggregation state.""" + self.results[result.expert_name] = result + if not result.success: + self.all_succeeded = False + self.failed_experts.append(result.expert_name) + + def aggregate_outputs(self, separator: str = "\n\n---\n\n") -> str: + """Combine all successful outputs into one string.""" + outputs = [] + for expert_name, result in self.results.items(): + if result.success and result.output: + outputs.append(f"**{expert_name.title()}**: {result.output}") + + self.combined_output = separator.join(outputs) + return self.combined_output + + +async def execute_sequential( + tasks: list[DelegationTask], + stop_on_failure: bool = False, +) -> MultiExpertResult: + """ + Execute multiple delegation tasks sequentially. + + Tasks run one after another in order. Later tasks can depend on + earlier results (though this function doesn't handle passing + results between tasks - that's the orchestrator's job). + + Args: + tasks: List of delegation tasks to execute in order + stop_on_failure: If True, stop execution if any task fails + + Returns: + MultiExpertResult with all task results + + Example: + >>> tasks = [ + ... DelegationTask(expert_name="memory", task="get user location"), + ... DelegationTask(expert_name="librarian", task="search weather"), + ... ] + >>> result = await execute_sequential(tasks) + >>> result.all_succeeded + True + """ + multi_result = MultiExpertResult() + + logger.info( + "sequential_execution_started", + task_count=len(tasks), + experts=[t.expert_name for t in tasks], + ) + + for i, task in enumerate(tasks): + logger.debug( + "sequential_task_executing", + index=i, + expert=task.expert_name, + task=task.task[:50], + ) + + result = await execute_delegation(task) + multi_result.add_result(result) + + if not result.success and stop_on_failure: + logger.warning( + "sequential_execution_stopped", + failed_at=i, + expert=task.expert_name, + error=result.error, + ) + break + + multi_result.aggregate_outputs() + + logger.info( + "sequential_execution_complete", + total_tasks=len(tasks), + succeeded=len(tasks) - len(multi_result.failed_experts), + failed=len(multi_result.failed_experts), + ) + + return multi_result + + +async def execute_parallel( + tasks: list[DelegationTask], +) -> MultiExpertResult: + """ + Execute multiple delegation tasks in parallel. + + All tasks run concurrently using asyncio.gather. Use this when + tasks are independent and don't depend on each other's results. + + Args: + tasks: List of delegation tasks to execute concurrently + + Returns: + MultiExpertResult with all task results + + Example: + >>> tasks = [ + ... DelegationTask(expert_name="librarian", task="search wiki"), + ... DelegationTask(expert_name="memory", task="get preferences"), + ... ] + >>> result = await execute_parallel(tasks) + >>> len(result.results) + 2 + """ + multi_result = MultiExpertResult() + + logger.info( + "parallel_execution_started", + task_count=len(tasks), + experts=[t.expert_name for t in tasks], + ) + + # Execute all tasks concurrently + results = await asyncio.gather( + *[execute_delegation(task) for task in tasks], + return_exceptions=True, + ) + + # Process results + for i, result in enumerate(results): + if isinstance(result, Exception): + # Handle exceptions as failed delegations + error_result = DelegationResult( + expert_name=tasks[i].expert_name, + task=tasks[i].task, + success=False, + output="", + error=str(result), + ) + multi_result.add_result(error_result) + logger.error( + "parallel_task_exception", + expert=tasks[i].expert_name, + error=str(result), + ) + else: + multi_result.add_result(result) + + multi_result.aggregate_outputs() + + logger.info( + "parallel_execution_complete", + total_tasks=len(tasks), + succeeded=len(tasks) - len(multi_result.failed_experts), + failed=len(multi_result.failed_experts), + ) + + return multi_result + + +async def orchestrate_multi_expert( + tasks: list[DelegationTask], + mode: ExecutionMode = ExecutionMode.SEQUENTIAL, + stop_on_failure: bool = False, +) -> AsyncGenerator[str, None]: + """ + Orchestrate multiple expert delegations with streaming think updates. + + Emits updates for each delegation phase and yields + combined results at the end. + + Args: + tasks: List of delegation tasks + mode: SEQUENTIAL or PARALLEL execution + stop_on_failure: For sequential mode, stop if a task fails + + Yields: + Think updates and combined expert output + + Example: + >>> tasks = [ + ... DelegationTask(expert_name="memory", task="get location"), + ... DelegationTask(expert_name="librarian", task="search weather"), + ... ] + >>> async for update in orchestrate_multi_expert(tasks): + ... print(update) + Starting multi-expert coordination (2 tasks)... + Consulting Memory... + Memory completed. + Consulting The Librarian... + The Librarian completed. + All experts completed successfully. + [Combined output from all experts...] + """ + if not tasks: + logger.debug("no_tasks_to_orchestrate") + return + + # Stream: Starting multi-expert coordination + yield f"🎯 Starting multi-expert coordination ({len(tasks)} tasks, {mode.value})...\n" + + if mode == ExecutionMode.PARALLEL: + # Parallel execution - emit one update then run all at once + expert_names = ", ".join(_get_display_name(t.expert_name) for t in tasks) + yield f"🔄 Consulting in parallel: {expert_names}...\n" + + result = await execute_parallel(tasks) + + # Emit completion updates for each + for expert_name, expert_result in result.results.items(): + display_name = _get_display_name(expert_name) + if expert_result.success: + yield f"✅ {display_name} completed.\n" + else: + yield f"⚠️ {display_name} failed: {expert_result.error}\n" + + else: + # Sequential execution - emit updates for each task + result = MultiExpertResult() + + for task in tasks: + display_name = _get_display_name(task.expert_name) + yield f"🤝 Consulting {display_name}...\n" + + task_result = await execute_delegation(task) + result.add_result(task_result) + + if task_result.success: + yield f"✅ {display_name} completed.\n" + else: + yield f"⚠️ {display_name} failed: {task_result.error}\n" + if stop_on_failure: + yield "🛑 Stopping due to failure.\n" + break + + result.aggregate_outputs() + + # Stream: Summary + if result.all_succeeded: + yield "🎉 All experts completed successfully.\n" + else: + failed_names = ", ".join(_get_display_name(e) for e in result.failed_experts) + yield f"⚠️ Some experts failed: {failed_names}\n" + + # Yield combined output + if result.combined_output: + yield f"\n{result.combined_output}" + + logger.info( + "multi_expert_orchestration_complete", + task_count=len(tasks), + mode=mode.value, + all_succeeded=result.all_succeeded, + ) + + +def _get_display_name(expert_name: str) -> str: + """Get user-friendly display name for an expert.""" + display_names = { + "librarian": "The Librarian", + "memory": "Memory", + "home_automation": "Home Automation", + "tatlock_core": "Core Tools", + } + return display_names.get(expert_name, expert_name.title())