Spaces:
Sleeping
Sleeping
File size: 5,025 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 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """Consolidation operations for skill deduplication."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import TYPE_CHECKING, List, Literal, Union
if TYPE_CHECKING:
from ..core.skillbook import Skillbook
logger = logging.getLogger(__name__)
class ConsolidationOpType(str, Enum):
"""Valid consolidation operation types from SkillManager responses."""
MERGE = "MERGE"
DELETE = "DELETE"
KEEP = "KEEP"
UPDATE = "UPDATE"
@dataclass
class MergeOp:
"""Merge multiple skills into one.
Combines helpful/harmful counts from all source skills into the kept skill.
Other skills are soft-deleted.
"""
type: Literal["MERGE"] = "MERGE"
source_ids: List[str] = field(default_factory=list)
merged_content: str = ""
keep_id: str = ""
reasoning: str = ""
@dataclass
class DeleteOp:
"""Soft-delete a skill as redundant."""
type: Literal["DELETE"] = "DELETE"
skill_id: str = ""
reasoning: str = ""
@dataclass
class KeepOp:
"""Keep both skills separate (they serve different purposes)."""
type: Literal["KEEP"] = "KEEP"
skill_ids: List[str] = field(default_factory=list)
differentiation: str = ""
reasoning: str = ""
@dataclass
class UpdateOp:
"""Update a skill's content to differentiate it."""
type: Literal["UPDATE"] = "UPDATE"
skill_id: str = ""
new_content: str = ""
reasoning: str = ""
ConsolidationOperation = Union[MergeOp, DeleteOp, KeepOp, UpdateOp]
# ---------------------------------------------------------------------------
# Apply helpers
# ---------------------------------------------------------------------------
def apply_consolidation_operations(
operations: List[ConsolidationOperation],
skillbook: "Skillbook",
) -> None:
"""Apply a list of consolidation operations to a skillbook."""
for op in operations:
if isinstance(op, MergeOp):
_apply_merge(op, skillbook)
elif isinstance(op, DeleteOp):
_apply_delete(op, skillbook)
elif isinstance(op, KeepOp):
_apply_keep(op, skillbook)
elif isinstance(op, UpdateOp):
_apply_update(op, skillbook)
else:
logger.warning("Unknown operation type: %s", type(op))
def _apply_merge(op: MergeOp, skillbook: "Skillbook") -> None:
keep_skill = skillbook.get_skill(op.keep_id)
if keep_skill is None:
logger.warning("MERGE: Keep skill %s not found", op.keep_id)
return
for source_id in op.source_ids:
if source_id == op.keep_id:
continue
source = skillbook.get_skill(source_id)
if source is None:
logger.warning("MERGE: Source skill %s not found", source_id)
continue
skillbook.remove_skill(source_id, soft=True)
logger.info("MERGE: Soft-deleted %s into %s", source_id, op.keep_id)
if op.merged_content:
if keep_skill.section == "context":
keep_skill.insight = op.merged_content
else:
keep_skill.issue = op.merged_content
keep_skill.embedding = None
keep_skill.updated_at = datetime.now(timezone.utc).isoformat()
logger.info("MERGE: Completed merge into %s", op.keep_id)
def _apply_delete(op: DeleteOp, skillbook: "Skillbook") -> None:
skill = skillbook.get_skill(op.skill_id)
if skill is None:
logger.warning("DELETE: Skill %s not found", op.skill_id)
return
skillbook.remove_skill(op.skill_id, soft=True)
logger.info("DELETE: Soft-deleted %s", op.skill_id)
def _apply_keep(op: KeepOp, skillbook: "Skillbook") -> None:
if len(op.skill_ids) < 2:
logger.warning("KEEP: Need at least 2 skill IDs")
return
from ..core.skillbook import SimilarityDecision
for i, id_a in enumerate(op.skill_ids):
for id_b in op.skill_ids[i + 1 :]:
decision = SimilarityDecision(
decision="KEEP",
reasoning=op.reasoning or op.differentiation,
decided_at=datetime.now(timezone.utc).isoformat(),
similarity_at_decision=0.0,
)
skillbook.set_similarity_decision(id_a, id_b, decision)
logger.info("KEEP: Stored decision for (%s, %s)", id_a, id_b)
def _apply_update(op: UpdateOp, skillbook: "Skillbook") -> None:
skill = skillbook.get_skill(op.skill_id)
if skill is None:
logger.warning("UPDATE: Skill %s not found", op.skill_id)
return
if skill.section == "context":
skill.insight = op.new_content
else:
skill.issue = op.new_content
skill.embedding = None
skill.updated_at = datetime.now(timezone.utc).isoformat()
logger.info("UPDATE: Updated content of %s", op.skill_id)
|