180 lines
8.0 KiB
Python
180 lines
8.0 KiB
Python
"""Analysis template models for customizable multi-agent perspectives."""
|
|
|
|
from typing import Dict, List, Optional, Any, Union
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
|
|
|
|
class TemplateType(str, Enum):
|
|
"""Types of analysis templates."""
|
|
EDUCATIONAL = "educational" # Beginner/Expert/Scholarly progression
|
|
DOMAIN = "domain" # Technical/Business/UX perspectives
|
|
AUDIENCE = "audience" # Different target audiences
|
|
PURPOSE = "purpose" # Different analysis purposes
|
|
CUSTOM = "custom" # User-defined custom templates
|
|
|
|
|
|
class ComplexityLevel(str, Enum):
|
|
"""Complexity levels for educational templates."""
|
|
BEGINNER = "beginner"
|
|
INTERMEDIATE = "intermediate"
|
|
EXPERT = "expert"
|
|
SCHOLARLY = "scholarly"
|
|
|
|
|
|
class AnalysisTemplate(BaseModel):
|
|
"""Template for configuring analysis agent behavior."""
|
|
|
|
# Template identification
|
|
id: str = Field(..., description="Unique template identifier")
|
|
name: str = Field(..., description="Human-readable template name")
|
|
description: str = Field(..., description="Template description and use case")
|
|
template_type: TemplateType = Field(..., description="Template category")
|
|
version: str = Field(default="1.0.0", description="Template version")
|
|
|
|
# Core template configuration
|
|
system_prompt: str = Field(..., description="Base system prompt for the agent")
|
|
analysis_focus: List[str] = Field(..., description="Key areas of focus for analysis")
|
|
output_format: str = Field(..., description="Expected output format and structure")
|
|
|
|
# Behavioral parameters
|
|
complexity_level: Optional[ComplexityLevel] = Field(None, description="Complexity level for educational templates")
|
|
target_audience: str = Field(default="general", description="Target audience for the analysis")
|
|
tone: str = Field(default="professional", description="Communication tone (professional, casual, academic, etc.)")
|
|
depth: str = Field(default="standard", description="Analysis depth (surface, standard, deep, comprehensive)")
|
|
|
|
# Template variables for customization
|
|
variables: Dict[str, Any] = Field(default_factory=dict, description="Template variables for customization")
|
|
|
|
# Content generation parameters
|
|
min_insights: int = Field(default=3, description="Minimum number of key insights to generate")
|
|
max_insights: int = Field(default=7, description="Maximum number of key insights to generate")
|
|
include_examples: bool = Field(default=True, description="Whether to include examples in analysis")
|
|
include_recommendations: bool = Field(default=True, description="Whether to include actionable recommendations")
|
|
|
|
# Metadata
|
|
tags: List[str] = Field(default_factory=list, description="Template tags for categorization")
|
|
author: str = Field(default="system", description="Template author")
|
|
created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")
|
|
is_active: bool = Field(default=True, description="Whether template is active and usable")
|
|
usage_count: int = Field(default=0, description="Number of times template has been used")
|
|
|
|
@field_validator('variables')
|
|
@classmethod
|
|
def validate_variables(cls, v):
|
|
"""Ensure variables are JSON-serializable."""
|
|
import json
|
|
try:
|
|
json.dumps(v)
|
|
return v
|
|
except (TypeError, ValueError) as e:
|
|
raise ValueError(f"Template variables must be JSON-serializable: {e}")
|
|
|
|
def render_prompt(self, content_context: Dict[str, Any] = None) -> str:
|
|
"""Render the system prompt with template variables and context."""
|
|
context = {**self.variables}
|
|
if content_context:
|
|
context.update(content_context)
|
|
|
|
try:
|
|
return self.system_prompt.format(**context)
|
|
except KeyError as e:
|
|
raise ValueError(f"Missing template variable: {e}")
|
|
|
|
def to_perspective_config(self) -> Dict[str, Any]:
|
|
"""Convert template to existing PerspectivePrompt format for compatibility."""
|
|
return {
|
|
"system_prompt": self.system_prompt,
|
|
"analysis_focus": self.analysis_focus,
|
|
"output_format": self.output_format
|
|
}
|
|
|
|
|
|
class TemplateSet(BaseModel):
|
|
"""Collection of templates for multi-perspective analysis."""
|
|
|
|
id: str = Field(..., description="Template set identifier")
|
|
name: str = Field(..., description="Template set name")
|
|
description: str = Field(..., description="Template set description")
|
|
template_type: TemplateType = Field(..., description="Type of templates in this set")
|
|
|
|
templates: Dict[str, AnalysisTemplate] = Field(..., description="Templates in this set")
|
|
synthesis_template: Optional[AnalysisTemplate] = Field(None, description="Template for synthesizing results")
|
|
|
|
# Configuration for multi-agent orchestration
|
|
execution_order: List[str] = Field(default_factory=list, description="Order of template execution")
|
|
parallel_execution: bool = Field(default=True, description="Whether templates can run in parallel")
|
|
|
|
# Metadata
|
|
created_at: datetime = Field(default_factory=datetime.utcnow)
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow)
|
|
is_active: bool = Field(default=True)
|
|
|
|
@field_validator('templates')
|
|
@classmethod
|
|
def validate_templates(cls, v):
|
|
"""Ensure all templates are properly configured."""
|
|
if not v:
|
|
raise ValueError("Template set must contain at least one template")
|
|
|
|
for template_id, template in v.items():
|
|
if template.id != template_id:
|
|
raise ValueError(f"Template ID mismatch: {template.id} != {template_id}")
|
|
|
|
return v
|
|
|
|
def get_template(self, template_id: str) -> Optional[AnalysisTemplate]:
|
|
"""Get a specific template from the set."""
|
|
return self.templates.get(template_id)
|
|
|
|
def add_template(self, template: AnalysisTemplate) -> None:
|
|
"""Add a template to the set."""
|
|
self.templates[template.id] = template
|
|
self.updated_at = datetime.utcnow()
|
|
|
|
def remove_template(self, template_id: str) -> bool:
|
|
"""Remove a template from the set."""
|
|
if template_id in self.templates:
|
|
del self.templates[template_id]
|
|
self.updated_at = datetime.utcnow()
|
|
return True
|
|
return False
|
|
|
|
|
|
class TemplateRegistry(BaseModel):
|
|
"""Registry for managing analysis templates and template sets."""
|
|
|
|
templates: Dict[str, AnalysisTemplate] = Field(default_factory=dict)
|
|
template_sets: Dict[str, TemplateSet] = Field(default_factory=dict)
|
|
|
|
def register_template(self, template: AnalysisTemplate) -> None:
|
|
"""Register a new template."""
|
|
self.templates[template.id] = template
|
|
|
|
def register_template_set(self, template_set: TemplateSet) -> None:
|
|
"""Register a new template set."""
|
|
self.template_sets[template_set.id] = template_set
|
|
|
|
def get_template(self, template_id: str) -> Optional[AnalysisTemplate]:
|
|
"""Get a template by ID."""
|
|
return self.templates.get(template_id)
|
|
|
|
def get_template_set(self, set_id: str) -> Optional[TemplateSet]:
|
|
"""Get a template set by ID."""
|
|
return self.template_sets.get(set_id)
|
|
|
|
def list_templates(self, template_type: Optional[TemplateType] = None) -> List[AnalysisTemplate]:
|
|
"""List templates, optionally filtered by type."""
|
|
templates = list(self.templates.values())
|
|
if template_type:
|
|
templates = [t for t in templates if t.template_type == template_type]
|
|
return templates
|
|
|
|
def list_template_sets(self, template_type: Optional[TemplateType] = None) -> List[TemplateSet]:
|
|
"""List template sets, optionally filtered by type."""
|
|
sets = list(self.template_sets.values())
|
|
if template_type:
|
|
sets = [s for s in sets if s.template_type == template_type]
|
|
return sets |