From 1970751b2f40f2395543c77df250315e8f6084f7 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Sat, 13 Dec 2025 12:59:20 +0100 Subject: [PATCH] feat: add orchestration loop with think update streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Creates orchestration module for multi-expert coordination: - parse_delegation_from_steward_note(): Extracts delegation task - execute_delegation(): Routes to appropriate expert agent - orchestrate_with_think_updates(): Streams updates around delegation calls while using run() internally This enables real-time user feedback while avoiding Ollama's streaming+tool call bugs. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/agents/orchestration.py | 220 ++++++++++++++++++++++++++++++++++++ 1 file changed, 220 insertions(+) create mode 100644 src/agents/orchestration.py diff --git a/src/agents/orchestration.py b/src/agents/orchestration.py new file mode 100644 index 0000000..f3fb1d2 --- /dev/null +++ b/src/agents/orchestration.py @@ -0,0 +1,220 @@ +""" +Orchestration module for multi-expert agent coordination. + +Provides infrastructure for Tatlock to orchestrate expert agents +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. +""" +from dataclasses import dataclass +from typing import AsyncGenerator, Optional, Callable, Any + +from src.agents.delegation import DelegationTask, DelegationResult, delegate_to_librarian +from src.core.logging_config import get_logger + +logger = get_logger(__name__) + + +@dataclass +class OrchestrationContext: + """ + Context for an orchestration session. + + Tracks the user's request, delegation tasks, and results. + """ + user_message: str + steward_note: str + conversation_id: Optional[str] = None + + +def parse_delegation_from_steward_note(steward_note: str) -> Optional[DelegationTask]: + """ + Parse a delegation task from Steward's note. + + Looks for the DELEGATE: pattern in the Steward's recommendation. + + Args: + steward_note: Formatted note from Steward + + Returns: + DelegationTask if delegation found, None otherwise + + Example: + >>> note = "DELEGATE: librarian to create a wiki page about CI/CD" + >>> task = parse_delegation_from_steward_note(note) + >>> task.expert_name + 'librarian' + >>> task.task + 'create a wiki page about CI/CD' + """ + import re + + # Look for DELEGATE: pattern + # Match: "DELEGATE: expert_name to action description" + match = re.search( + r'DELEGATE:\s*(\w+)\s+to\s+(.+?)(?:\n|REASON:|COMPLEXITY:|CONTEXT:|$)', + steward_note, + re.IGNORECASE | re.MULTILINE + ) + + if match: + expert_name = match.group(1).lower() + task_description = match.group(2).strip() + + # Handle "none" case + if expert_name == "none": + return None + + return DelegationTask( + expert_name=expert_name, + task=task_description, + ) + + return None + + +async def execute_delegation( + task: DelegationTask, +) -> DelegationResult: + """ + Execute a delegation task. + + Routes to the appropriate expert agent based on expert_name. + + Args: + task: Delegation task to execute + + Returns: + DelegationResult from the expert agent + """ + logger.info( + "executing_delegation", + expert=task.expert_name, + task=task.task[:50], + ) + + if task.expert_name == "librarian": + return await delegate_to_librarian( + task=task.task, + context=task.context, + ) + + # Future experts would be added here: + # elif task.expert_name == "memory": + # return await delegate_to_memory(task.task, task.context) + # elif task.expert_name == "home_automation": + # return await delegate_to_home_automation(task.task, task.context) + + # Unknown expert - return error result + logger.warning("unknown_expert", expert=task.expert_name) + return DelegationResult( + expert_name=task.expert_name, + task=task.task, + success=False, + output="", + error=f"Unknown expert: {task.expert_name}", + ) + + +async def orchestrate_with_think_updates( + user_message: str, + steward_note: str, + delegation_task: Optional[DelegationTask] = None, +) -> AsyncGenerator[str, None]: + """ + Orchestrate expert delegation with streaming think updates. + + Emits updates before and after delegation calls to + keep the user informed of progress. Expert calls use run() + internally to avoid Ollama streaming bugs. + + Args: + user_message: Original user message + steward_note: Steward's analysis and instructions + delegation_task: Optional pre-parsed delegation task + + Yields: + Think update strings and final expert output + + Example: + >>> async for update in orchestrate_with_think_updates( + ... "Create a wiki page about CI/CD", + ... "DELEGATE: librarian to create wiki page", + ... ): + ... print(update) + Consulting The Librarian... + Delegation complete. + [Wiki page created successfully...] + """ + # Parse delegation if not provided + if delegation_task is None: + delegation_task = parse_delegation_from_steward_note(steward_note) + + if delegation_task is None: + # No delegation needed - nothing to orchestrate + logger.debug("no_delegation_needed") + return + + # Stream: About to delegate + expert_display_name = delegation_task.expert_name.title() + if delegation_task.expert_name == "librarian": + expert_display_name = "The Librarian" + + yield f"🤝 Consulting {expert_display_name}...\n" + + # Execute delegation (uses run() internally) + result = await execute_delegation(delegation_task) + + if result.success: + yield f"✅ {expert_display_name} completed research.\n" + + # Yield the expert's findings + if result.output: + yield f"\n{result.output}" + else: + yield f"⚠️ {expert_display_name} encountered an issue: {result.error}\n" + + logger.info( + "orchestration_complete", + expert=delegation_task.expert_name, + success=result.success, + ) + + +def extract_delegation_context( + steward_note: str, +) -> dict[str, str]: + """ + Extract context fields from Steward's note. + + Args: + steward_note: Formatted note from Steward + + Returns: + Dict with reason, complexity, and context + """ + import re + + result = { + "reason": "", + "complexity": "", + "context": "", + } + + # Extract REASON: + reason_match = re.search(r'REASON:\s*(.+?)(?:\n|COMPLEXITY:|CONTEXT:|$)', steward_note, re.IGNORECASE) + if reason_match: + result["reason"] = reason_match.group(1).strip() + + # Extract COMPLEXITY: + complexity_match = re.search(r'COMPLEXITY:\s*(.+?)(?:\n|CONTEXT:|$)', steward_note, re.IGNORECASE) + if complexity_match: + result["complexity"] = complexity_match.group(1).strip() + + # Extract CONTEXT: + context_match = re.search(r'CONTEXT:\s*(.+?)$', steward_note, re.IGNORECASE | re.MULTILINE) + if context_match: + result["context"] = context_match.group(1).strip() + + return result