feat: add DelegationTask dataclass and delegate_to_librarian wrapper
Introduces agent-as-tool pattern infrastructure: - DelegationTask: Structured representation of expert work - DelegationResult: Typed result from expert delegation - delegate_to_librarian(): Wrapper for Librarian agent calls This implements PydanticAI's recommended delegation pattern where parent agents call child agents via tool wrappers. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,152 @@
|
||||
"""
|
||||
Delegation infrastructure for expert agent calls.
|
||||
|
||||
Provides delegation wrappers that Tatlock uses to call expert agents.
|
||||
Each wrapper encapsulates the complexity of calling an expert and
|
||||
returns a structured result for synthesis.
|
||||
|
||||
This implements the agent-as-tool pattern recommended by PydanticAI:
|
||||
agents call other agents via tool wrappers, keeping each agent focused.
|
||||
"""
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Optional, Any
|
||||
|
||||
from src.core.logging_config import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DelegationTask:
|
||||
"""
|
||||
A task to be delegated to an expert agent.
|
||||
|
||||
Represents a unit of work that Tatlock delegates to a specialist.
|
||||
Used for tracking and orchestration of multi-expert workflows.
|
||||
|
||||
Attributes:
|
||||
expert_name: Name of the expert agent (e.g., "librarian", "memory")
|
||||
task: Clear description of what needs to be done
|
||||
context: Additional context from the conversation
|
||||
action: Specific action verb (create, search, update, etc.)
|
||||
priority: Execution priority (lower = higher priority)
|
||||
depends_on: List of task IDs this task depends on
|
||||
result: Result from expert after execution
|
||||
"""
|
||||
expert_name: str
|
||||
task: str
|
||||
context: str = ""
|
||||
action: str = ""
|
||||
priority: int = 0
|
||||
depends_on: list[str] = field(default_factory=list)
|
||||
result: Optional[str] = None
|
||||
task_id: str = ""
|
||||
|
||||
def __post_init__(self):
|
||||
"""Generate task ID if not provided."""
|
||||
if not self.task_id:
|
||||
import uuid
|
||||
self.task_id = f"{self.expert_name}_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class DelegationResult:
|
||||
"""
|
||||
Result from an expert agent delegation.
|
||||
|
||||
Attributes:
|
||||
expert_name: Which expert handled the task
|
||||
task: Original task description
|
||||
success: Whether the delegation succeeded
|
||||
output: Expert's response/findings
|
||||
error: Error message if failed
|
||||
"""
|
||||
expert_name: str
|
||||
task: str
|
||||
success: bool
|
||||
output: str
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
async def delegate_to_librarian(
|
||||
task: str,
|
||||
context: str = "",
|
||||
) -> DelegationResult:
|
||||
"""
|
||||
Delegate a research or wiki task to The Librarian.
|
||||
|
||||
The Librarian handles:
|
||||
- Wiki creation (smart_create_wiki_page for topic-based)
|
||||
- Wiki updates (update_wiki_page for modifications)
|
||||
- Research queries (hybrid_search for comprehensive search)
|
||||
- Knowledge graph exploration
|
||||
- Document lookups and semantic search
|
||||
|
||||
This wrapper uses run() not run_stream() to avoid Ollama's
|
||||
streaming + tool call bug (PydanticAI issues #1292, #2256).
|
||||
|
||||
Args:
|
||||
task: Clear description of what needs to be done.
|
||||
Include the action verb (create, search, update, etc.)
|
||||
Example: "Create a wiki page about CI/CD pipelines"
|
||||
Example: "Search for information about Docker networking"
|
||||
context: Additional context from the user's request or
|
||||
conversation history
|
||||
|
||||
Returns:
|
||||
DelegationResult with the Librarian's findings
|
||||
|
||||
Example:
|
||||
>>> result = await delegate_to_librarian(
|
||||
... task="Create a wiki page about Kubernetes deployments",
|
||||
... context="User is setting up a homelab cluster",
|
||||
... )
|
||||
>>> if result.success:
|
||||
... print(result.output)
|
||||
"""
|
||||
from src.agents.librarian.agent import run_librarian
|
||||
|
||||
logger.info(
|
||||
"delegation_to_librarian_started",
|
||||
task=task[:100],
|
||||
has_context=bool(context),
|
||||
)
|
||||
|
||||
try:
|
||||
# Use run() not run_stream() - avoids Ollama bug
|
||||
output = await run_librarian(task=task, context=context)
|
||||
|
||||
logger.info(
|
||||
"delegation_to_librarian_completed",
|
||||
task=task[:50],
|
||||
output_length=len(output),
|
||||
)
|
||||
|
||||
return DelegationResult(
|
||||
expert_name="librarian",
|
||||
task=task,
|
||||
success=True,
|
||||
output=output,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"delegation_to_librarian_error",
|
||||
task=task[:50],
|
||||
error=str(e),
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
return DelegationResult(
|
||||
expert_name="librarian",
|
||||
task=task,
|
||||
success=False,
|
||||
output="",
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
|
||||
# Future expert delegation wrappers will be added here:
|
||||
# - delegate_to_memory(task, context) -> DelegationResult
|
||||
# - delegate_to_home_automation(task, context) -> DelegationResult
|
||||
# - delegate_to_developer(task, context) -> DelegationResult
|
||||
Reference in New Issue
Block a user