Spaces:
Sleeping
Sleeping
File size: 1,599 Bytes
116524e | 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 | """DeduplicateStep — periodically consolidates similar skills."""
from __future__ import annotations
import logging
from ..core.context import ACEStepContext
from ..protocols import DeduplicationManagerLike
from ..core.skillbook import Skillbook
logger = logging.getLogger(__name__)
class DeduplicateStep:
"""Consolidate similar skills in the skillbook at a configurable interval.
Optional side-effect step — appended to the pipeline by factory methods
when ``dedup_config`` is provided.
Stateless — uses ``ctx.global_sample_index`` with ``self.interval`` to
skip most invocations. Deduplication involves O(n^2) similarity
comparisons, so running on every sample would be expensive.
"""
requires: frozenset[str] = frozenset({"global_sample_index"})
provides: frozenset[str] = frozenset()
max_workers = 1
def __init__(
self,
manager: DeduplicationManagerLike,
skillbook: Skillbook,
*,
interval: int = 10,
) -> None:
self.manager = manager
self.skillbook = skillbook
self.interval = interval
def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
if ctx.global_sample_index % self.interval != 0:
return ctx
report = self.manager.get_similarity_report(self.skillbook)
if report:
logger.info(
"DeduplicateStep: similarity report at sample %d:\n%s",
ctx.global_sample_index,
report,
)
return ctx
|