Files
FusionAGI/fusionagi/schemas/recommendation.py
defiQUG c052b07662
Some checks failed
Tests / test (3.10) (push) Has been cancelled
Tests / test (3.11) (push) Has been cancelled
Tests / test (3.12) (push) Has been cancelled
Tests / lint (push) Has been cancelled
Tests / docker (push) Has been cancelled
Initial commit: add .gitignore and README
2026-02-09 21:51:42 -08:00

93 lines
3.0 KiB
Python

"""Schemas for self-improvement: recommendations and training suggestions."""
from datetime import datetime, timezone
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field, field_validator
def _utc_now() -> datetime:
"""Return current UTC time (timezone-aware)."""
return datetime.now(timezone.utc)
class RecommendationKind(str, Enum):
"""Type of auto-generated recommendation."""
NEXT_ACTION = "next_action"
TRAINING_TARGET = "training_target"
TOOL_ADDITION = "tool_addition"
STRATEGY_CHANGE = "strategy_change"
GUARDRAIL_UPDATE = "guardrail_update"
OTHER = "other"
class Recommendation(BaseModel):
"""
Auto-generated recommendation from reflection and lessons.
Used by the auto-recommend/suggest pipeline.
"""
kind: RecommendationKind = Field(
default=RecommendationKind.OTHER,
description="Category of recommendation",
)
title: str = Field(..., min_length=1, description="Short title")
description: str = Field(default="", description="Detailed description")
payload: dict[str, Any] = Field(default_factory=dict, description="Structured data")
source_task_id: str | None = Field(default=None, description="Task that triggered this")
priority: int = Field(default=0, ge=0, le=10, description="0=low, 10=critical")
created_at: datetime = Field(default_factory=_utc_now, description="Creation time")
@field_validator("title")
@classmethod
def validate_title(cls, v: str) -> str:
"""Ensure title is not just whitespace."""
if not v.strip():
raise ValueError("title cannot be empty or whitespace")
return v
class TrainingSuggestionKind(str, Enum):
"""Type of auto-training suggestion."""
HEURISTIC_UPDATE = "heuristic_update"
PROMPT_TUNING = "prompt_tuning"
FINE_TUNE_DATASET = "fine_tune_dataset"
STRATEGY_PARAM = "strategy_param"
OTHER = "other"
class TrainingSuggestion(BaseModel):
"""
Auto-generated training suggestion from reflection and failures.
Can be applied (e.g. heuristic update) or emitted for external training pipelines.
"""
kind: TrainingSuggestionKind = Field(
default=TrainingSuggestionKind.OTHER,
description="Type of training",
)
key: str = Field(..., min_length=1, description="Target key (e.g. heuristic name)")
value: Any = Field(..., description="Value to apply or dataset description")
source_task_id: str | None = Field(default=None, description="Task that triggered this")
reason: str = Field(default="", description="Why this suggestion was generated")
created_at: datetime = Field(default_factory=_utc_now, description="Creation time")
@field_validator("key")
@classmethod
def validate_key(cls, v: str) -> str:
"""Ensure key is not just whitespace."""
if not v.strip():
raise ValueError("key cannot be empty or whitespace")
return v
__all__ = [
"Recommendation",
"RecommendationKind",
"TrainingSuggestion",
"TrainingSuggestionKind",
]