Spaces:
Sleeping
Sleeping
File size: 1,940 Bytes
5df8a73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """
Capability Protocol
===================
Base class for the Capability layer (Level 2).
Capabilities are multi-step agent pipelines invoked when the user selects
a deep mode (e.g. Deep Solve, Deep Question).
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
from .context import UnifiedContext
from .stream_bus import StreamBus
@dataclass
class CapabilityManifest:
"""Static metadata for a capability."""
name: str
description: str
stages: list[str] = field(default_factory=list)
tools_used: list[str] = field(default_factory=list)
cli_aliases: list[str] = field(default_factory=list)
request_schema: dict[str, Any] = field(default_factory=dict)
config_defaults: dict[str, Any] = field(default_factory=dict)
class BaseCapability(ABC):
"""
Abstract base for all capabilities (deep modes).
Subclasses must provide ``manifest`` and implement ``run``.
Example::
class MySolverCapability(BaseCapability):
manifest = CapabilityManifest(
name="deep_solve",
description="Multi-agent problem solving.",
stages=["planning", "reasoning", "writing"],
tools_used=["rag", "web_search", "code_execution"],
)
async def run(self, context, stream):
async with stream.stage("planning", source=self.manifest.name):
plan = await self._plan(context)
...
"""
manifest: CapabilityManifest
@abstractmethod
async def run(self, context: UnifiedContext, stream: StreamBus) -> None:
"""Execute the full capability pipeline, emitting events to *stream*."""
...
@property
def name(self) -> str:
return self.manifest.name
@property
def stages(self) -> list[str]:
return self.manifest.stages
|