723 lines
27 KiB
Python
723 lines
27 KiB
Python
"""Enhanced Template Manager for Story 4.4 Custom AI Models & Enhanced Export.
|
|
|
|
This service extends the basic template manager with:
|
|
- Custom prompt templates with versioning
|
|
- Domain-specific presets (Educational, Business, Technical, etc.)
|
|
- A/B testing framework for prompt optimization
|
|
- Model parameter configuration
|
|
- Template performance analytics
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional, Union
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
import uuid
|
|
|
|
from ..services.deepseek_service import DeepSeekService
|
|
from ..core.exceptions import ServiceError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DomainCategory(str, Enum):
|
|
"""Domain-specific template categories."""
|
|
EDUCATIONAL = "educational"
|
|
BUSINESS = "business"
|
|
TECHNICAL = "technical"
|
|
CONTENT_CREATION = "content_creation"
|
|
RESEARCH = "research"
|
|
GENERAL = "general"
|
|
|
|
|
|
class TemplateStatus(str, Enum):
|
|
"""Template status options."""
|
|
ACTIVE = "active"
|
|
DRAFT = "draft"
|
|
ARCHIVED = "archived"
|
|
TESTING = "testing"
|
|
|
|
|
|
@dataclass
|
|
class ModelConfig:
|
|
"""AI model configuration for templates."""
|
|
temperature: float = 0.7
|
|
max_tokens: int = 1500
|
|
top_p: float = 1.0
|
|
frequency_penalty: float = 0.0
|
|
presence_penalty: float = 0.0
|
|
model_name: str = "deepseek-chat"
|
|
|
|
|
|
@dataclass
|
|
class PromptTemplate:
|
|
"""Enhanced prompt template with full configuration."""
|
|
id: str
|
|
name: str
|
|
description: str
|
|
prompt_text: str
|
|
domain_category: DomainCategory
|
|
model_config: ModelConfig
|
|
is_public: bool = False
|
|
status: TemplateStatus = TemplateStatus.ACTIVE
|
|
usage_count: int = 0
|
|
rating: float = 0.0
|
|
version: str = "1.0.0"
|
|
created_at: datetime = field(default_factory=datetime.now)
|
|
updated_at: datetime = field(default_factory=datetime.now)
|
|
created_by: Optional[str] = None
|
|
tags: List[str] = field(default_factory=list)
|
|
variables: Dict[str, Any] = field(default_factory=dict)
|
|
performance_metrics: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class ABTestExperiment:
|
|
"""A/B testing experiment configuration."""
|
|
id: str
|
|
name: str
|
|
description: str
|
|
baseline_template_id: str
|
|
variant_template_id: str
|
|
status: str = "active" # active, completed, paused
|
|
success_metric: str = "quality_score" # quality_score, user_rating, processing_time
|
|
statistical_significance: Optional[float] = None
|
|
results: Dict[str, Any] = field(default_factory=dict)
|
|
created_at: datetime = field(default_factory=datetime.now)
|
|
|
|
|
|
class EnhancedTemplateManager:
|
|
"""Advanced template manager with AI model integration."""
|
|
|
|
def __init__(self, ai_service: Optional[DeepSeekService] = None):
|
|
"""Initialize enhanced template manager.
|
|
|
|
Args:
|
|
ai_service: AI service for template testing and optimization
|
|
"""
|
|
self.ai_service = ai_service or DeepSeekService()
|
|
|
|
# In-memory storage (in production, this would be database-backed)
|
|
self.templates: Dict[str, PromptTemplate] = {}
|
|
self.experiments: Dict[str, ABTestExperiment] = {}
|
|
|
|
# Performance tracking
|
|
self.template_usage_stats: Dict[str, Dict[str, Any]] = {}
|
|
self.domain_presets_initialized = False
|
|
|
|
logger.info("EnhancedTemplateManager initialized")
|
|
|
|
async def initialize_domain_presets(self):
|
|
"""Initialize domain-specific preset templates."""
|
|
if self.domain_presets_initialized:
|
|
return
|
|
|
|
domain_presets = {
|
|
DomainCategory.EDUCATIONAL: {
|
|
"name": "Educational Analysis",
|
|
"description": "Focus on learning objectives, key concepts, and educational value",
|
|
"prompt": """Analyze this content from an educational perspective. Focus on:
|
|
|
|
Learning Objectives:
|
|
- What specific knowledge or skills does this content teach?
|
|
- What are the key concepts students should understand?
|
|
- How does this content build upon prerequisite knowledge?
|
|
|
|
Educational Structure:
|
|
- How is the information organized for learning?
|
|
- What teaching methods or techniques are used?
|
|
- Are there examples, exercises, or practice opportunities?
|
|
|
|
Target Audience:
|
|
- What level of prior knowledge is assumed?
|
|
- Is the content appropriate for the intended audience?
|
|
- How could this be adapted for different skill levels?
|
|
|
|
Learning Assessment:
|
|
- How could understanding be tested or validated?
|
|
- What are the key takeaways students should remember?
|
|
- What follow-up learning would be beneficial?
|
|
|
|
Content: {content}
|
|
|
|
Provide a comprehensive educational analysis with actionable insights for learners and educators.""",
|
|
"model_config": ModelConfig(temperature=0.5, max_tokens=1200),
|
|
"tags": ["education", "learning", "pedagogy"]
|
|
},
|
|
|
|
DomainCategory.BUSINESS: {
|
|
"name": "Business Strategy Analysis",
|
|
"description": "Emphasize ROI, market implications, and strategic insights",
|
|
"prompt": """Analyze this content from a business strategy perspective. Focus on:
|
|
|
|
Business Value Proposition:
|
|
- What direct business value does this content provide?
|
|
- How could this impact revenue, costs, or efficiency?
|
|
- What competitive advantages could be gained?
|
|
|
|
Market Implications:
|
|
- How does this relate to current market trends?
|
|
- What opportunities or threats are identified?
|
|
- Who are the key stakeholders and target markets?
|
|
|
|
Strategic Implementation:
|
|
- What resources would be required for implementation?
|
|
- What are the potential risks and mitigation strategies?
|
|
- What is the expected ROI and timeline?
|
|
|
|
Decision-Making Framework:
|
|
- What key decisions need to be made?
|
|
- What criteria should guide those decisions?
|
|
- What are the short-term and long-term implications?
|
|
|
|
Content: {content}
|
|
|
|
Provide a strategic business analysis with actionable recommendations for leadership and decision-makers.""",
|
|
"model_config": ModelConfig(temperature=0.4, max_tokens=1500),
|
|
"tags": ["business", "strategy", "ROI", "market-analysis"]
|
|
},
|
|
|
|
DomainCategory.TECHNICAL: {
|
|
"name": "Technical Implementation Analysis",
|
|
"description": "Highlight implementation details, architecture, and technical best practices",
|
|
"prompt": """Analyze this content from a technical implementation perspective. Focus on:
|
|
|
|
Technical Architecture:
|
|
- What technical approaches, frameworks, or technologies are discussed?
|
|
- How do the components work together systematically?
|
|
- What are the key technical design patterns or principles?
|
|
|
|
Implementation Details:
|
|
- What specific technical steps or processes are outlined?
|
|
- What tools, libraries, or platforms are recommended?
|
|
- Are there code examples, configurations, or technical specifications?
|
|
|
|
Performance and Scalability:
|
|
- How do the technical solutions perform at scale?
|
|
- What are the potential bottlenecks or limitations?
|
|
- What optimization opportunities exist?
|
|
|
|
Best Practices and Standards:
|
|
- What technical best practices are demonstrated?
|
|
- How does this align with industry standards?
|
|
- What security, maintainability, or reliability considerations exist?
|
|
|
|
Content: {content}
|
|
|
|
Provide a comprehensive technical analysis with actionable guidance for developers and engineers.""",
|
|
"model_config": ModelConfig(temperature=0.3, max_tokens=1800),
|
|
"tags": ["technical", "implementation", "architecture", "engineering"]
|
|
},
|
|
|
|
DomainCategory.CONTENT_CREATION: {
|
|
"name": "Content Creator Analysis",
|
|
"description": "Analyze engagement patterns, audience insights, and content strategy",
|
|
"prompt": """Analyze this content from a content creator perspective. Focus on:
|
|
|
|
Audience Engagement:
|
|
- What techniques are used to capture and maintain audience attention?
|
|
- How does the content structure support viewer engagement?
|
|
- What emotional or psychological triggers are employed?
|
|
|
|
Content Strategy:
|
|
- What content format and style choices are made?
|
|
- How is information presented to maximize impact?
|
|
- What storytelling or narrative techniques are used?
|
|
|
|
Production Quality:
|
|
- What production values contribute to the content's effectiveness?
|
|
- How do visual, audio, or presentation elements enhance the message?
|
|
- What technical or creative skills are demonstrated?
|
|
|
|
Growth and Distribution:
|
|
- How could this content be optimized for different platforms?
|
|
- What distribution strategies would maximize reach?
|
|
- How could similar content be created or scaled?
|
|
|
|
Content: {content}
|
|
|
|
Provide insights for content creators on audience engagement, production techniques, and content strategy optimization.""",
|
|
"model_config": ModelConfig(temperature=0.6, max_tokens=1400),
|
|
"tags": ["content-creation", "engagement", "audience", "strategy"]
|
|
},
|
|
|
|
DomainCategory.RESEARCH: {
|
|
"name": "Research & Academic Analysis",
|
|
"description": "Academic focus with citations, methodology, and research implications",
|
|
"prompt": """Analyze this content from an academic research perspective. Focus on:
|
|
|
|
Research Methodology:
|
|
- What research methods, approaches, or frameworks are discussed?
|
|
- How rigorous is the methodology and evidence presented?
|
|
- What are the strengths and limitations of the research design?
|
|
|
|
Literature and Context:
|
|
- How does this content relate to existing academic literature?
|
|
- What theoretical frameworks or models are relevant?
|
|
- Where does this fit in the broader academic discourse?
|
|
|
|
Evidence and Analysis:
|
|
- What evidence is presented and how credible is it?
|
|
- Are the conclusions supported by the data or analysis?
|
|
- What assumptions or biases might be present?
|
|
|
|
Research Implications:
|
|
- What are the implications for future research?
|
|
- What research questions or hypotheses emerge?
|
|
- How could this work be extended or validated?
|
|
|
|
Content: {content}
|
|
|
|
Provide a rigorous academic analysis with attention to methodology, evidence quality, and research implications.""",
|
|
"model_config": ModelConfig(temperature=0.2, max_tokens=1600),
|
|
"tags": ["research", "academic", "methodology", "evidence"]
|
|
},
|
|
|
|
DomainCategory.GENERAL: {
|
|
"name": "Comprehensive General Analysis",
|
|
"description": "Balanced analysis suitable for general audiences",
|
|
"prompt": """Provide a comprehensive analysis of this content. Focus on:
|
|
|
|
Key Information:
|
|
- What are the main topics, themes, or subjects covered?
|
|
- What are the most important points or takeaways?
|
|
- How is the information structured and presented?
|
|
|
|
Practical Applications:
|
|
- How can this information be applied practically?
|
|
- What actions or decisions could result from this content?
|
|
- Who would benefit most from this information?
|
|
|
|
Quality and Credibility:
|
|
- How credible and well-supported is the information?
|
|
- What sources or evidence are provided?
|
|
- Are there any potential biases or limitations?
|
|
|
|
Broader Context:
|
|
- How does this relate to current trends or issues?
|
|
- What additional context would be helpful?
|
|
- What questions or topics warrant further exploration?
|
|
|
|
Content: {content}
|
|
|
|
Provide a balanced, comprehensive analysis that would be valuable for a general audience seeking to understand and apply this information.""",
|
|
"model_config": ModelConfig(temperature=0.5, max_tokens=1300),
|
|
"tags": ["general", "comprehensive", "balanced", "practical"]
|
|
}
|
|
}
|
|
|
|
# Create preset templates
|
|
for domain, preset_data in domain_presets.items():
|
|
template_id = f"preset_{domain.value}"
|
|
|
|
template = PromptTemplate(
|
|
id=template_id,
|
|
name=preset_data["name"],
|
|
description=preset_data["description"],
|
|
prompt_text=preset_data["prompt"],
|
|
domain_category=domain,
|
|
model_config=preset_data["model_config"],
|
|
is_public=True,
|
|
status=TemplateStatus.ACTIVE,
|
|
created_by="system",
|
|
tags=preset_data["tags"],
|
|
variables={"content": "Video transcript or summary content"}
|
|
)
|
|
|
|
self.templates[template_id] = template
|
|
|
|
self.domain_presets_initialized = True
|
|
logger.info(f"Initialized {len(domain_presets)} domain-specific preset templates")
|
|
|
|
async def create_template(
|
|
self,
|
|
name: str,
|
|
description: str,
|
|
prompt_text: str,
|
|
domain_category: DomainCategory,
|
|
model_config: Optional[ModelConfig] = None,
|
|
is_public: bool = False,
|
|
created_by: Optional[str] = None,
|
|
tags: Optional[List[str]] = None
|
|
) -> PromptTemplate:
|
|
"""Create a new custom prompt template."""
|
|
|
|
template_id = str(uuid.uuid4())
|
|
|
|
# Validate prompt template
|
|
if not prompt_text.strip():
|
|
raise ServiceError("Prompt text cannot be empty")
|
|
|
|
if len(prompt_text) > 10000:
|
|
raise ServiceError("Prompt text too long (maximum 10,000 characters)")
|
|
|
|
# Test template with AI service
|
|
try:
|
|
await self._validate_template_with_ai(prompt_text, model_config or ModelConfig())
|
|
except Exception as e:
|
|
logger.warning(f"Template validation warning: {e}")
|
|
|
|
template = PromptTemplate(
|
|
id=template_id,
|
|
name=name,
|
|
description=description,
|
|
prompt_text=prompt_text,
|
|
domain_category=domain_category,
|
|
model_config=model_config or ModelConfig(),
|
|
is_public=is_public,
|
|
created_by=created_by,
|
|
tags=tags or [],
|
|
variables=self._extract_template_variables(prompt_text)
|
|
)
|
|
|
|
self.templates[template_id] = template
|
|
|
|
logger.info(f"Created template: {name} (ID: {template_id})")
|
|
return template
|
|
|
|
async def get_template(self, template_id: str) -> Optional[PromptTemplate]:
|
|
"""Get a template by ID."""
|
|
return self.templates.get(template_id)
|
|
|
|
async def list_templates(
|
|
self,
|
|
domain_category: Optional[DomainCategory] = None,
|
|
is_public: Optional[bool] = None,
|
|
created_by: Optional[str] = None,
|
|
status: Optional[TemplateStatus] = None
|
|
) -> List[PromptTemplate]:
|
|
"""List templates with optional filtering."""
|
|
|
|
await self.initialize_domain_presets()
|
|
|
|
templates = list(self.templates.values())
|
|
|
|
# Apply filters
|
|
if domain_category:
|
|
templates = [t for t in templates if t.domain_category == domain_category]
|
|
|
|
if is_public is not None:
|
|
templates = [t for t in templates if t.is_public == is_public]
|
|
|
|
if created_by:
|
|
templates = [t for t in templates if t.created_by == created_by]
|
|
|
|
if status:
|
|
templates = [t for t in templates if t.status == status]
|
|
|
|
# Sort by usage count and rating
|
|
templates.sort(key=lambda t: (t.usage_count, t.rating), reverse=True)
|
|
|
|
return templates
|
|
|
|
async def update_template(
|
|
self,
|
|
template_id: str,
|
|
name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
prompt_text: Optional[str] = None,
|
|
model_config: Optional[ModelConfig] = None,
|
|
tags: Optional[List[str]] = None
|
|
) -> PromptTemplate:
|
|
"""Update an existing template."""
|
|
|
|
template = self.templates.get(template_id)
|
|
if not template:
|
|
raise ServiceError(f"Template not found: {template_id}")
|
|
|
|
# Prevent updating system preset templates
|
|
if template.created_by == "system":
|
|
raise ServiceError("Cannot modify system preset templates")
|
|
|
|
# Update fields
|
|
if name:
|
|
template.name = name
|
|
if description:
|
|
template.description = description
|
|
if prompt_text:
|
|
template.prompt_text = prompt_text
|
|
template.variables = self._extract_template_variables(prompt_text)
|
|
if model_config:
|
|
template.model_config = model_config
|
|
if tags:
|
|
template.tags = tags
|
|
|
|
template.updated_at = datetime.now()
|
|
template.version = self._increment_version(template.version)
|
|
|
|
logger.info(f"Updated template: {template.name} (ID: {template_id})")
|
|
return template
|
|
|
|
async def delete_template(self, template_id: str) -> bool:
|
|
"""Delete a template."""
|
|
|
|
template = self.templates.get(template_id)
|
|
if not template:
|
|
return False
|
|
|
|
# Prevent deleting system preset templates
|
|
if template.created_by == "system":
|
|
raise ServiceError("Cannot delete system preset templates")
|
|
|
|
del self.templates[template_id]
|
|
logger.info(f"Deleted template: {template.name} (ID: {template_id})")
|
|
return True
|
|
|
|
async def execute_template(
|
|
self,
|
|
template_id: str,
|
|
variables: Dict[str, Any],
|
|
override_config: Optional[ModelConfig] = None
|
|
) -> Dict[str, Any]:
|
|
"""Execute a template with provided variables."""
|
|
|
|
template = self.templates.get(template_id)
|
|
if not template:
|
|
raise ServiceError(f"Template not found: {template_id}")
|
|
|
|
# Prepare prompt with variables
|
|
try:
|
|
filled_prompt = template.prompt_text.format(**variables)
|
|
except KeyError as e:
|
|
raise ServiceError(f"Missing required variable: {e}")
|
|
|
|
# Use override config or template's config
|
|
config = override_config or template.model_config
|
|
|
|
# Execute with AI service
|
|
start_time = datetime.now()
|
|
|
|
try:
|
|
response = await self.ai_service.generate_response(
|
|
prompt=filled_prompt,
|
|
model=config.model_name,
|
|
temperature=config.temperature,
|
|
max_tokens=config.max_tokens
|
|
)
|
|
|
|
processing_time = (datetime.now() - start_time).total_seconds()
|
|
|
|
# Update usage statistics
|
|
await self._update_template_usage(template_id, processing_time, len(response))
|
|
|
|
return {
|
|
"response": response,
|
|
"processing_time_seconds": processing_time,
|
|
"template_used": template.name,
|
|
"model_config": config.__dict__,
|
|
"input_variables": variables
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Template execution failed: {e}")
|
|
raise ServiceError(f"Template execution failed: {str(e)}")
|
|
|
|
async def create_ab_test(
|
|
self,
|
|
name: str,
|
|
description: str,
|
|
baseline_template_id: str,
|
|
variant_template_id: str,
|
|
success_metric: str = "quality_score"
|
|
) -> ABTestExperiment:
|
|
"""Create A/B testing experiment."""
|
|
|
|
# Validate templates exist
|
|
if not self.templates.get(baseline_template_id):
|
|
raise ServiceError(f"Baseline template not found: {baseline_template_id}")
|
|
|
|
if not self.templates.get(variant_template_id):
|
|
raise ServiceError(f"Variant template not found: {variant_template_id}")
|
|
|
|
experiment_id = str(uuid.uuid4())
|
|
experiment = ABTestExperiment(
|
|
id=experiment_id,
|
|
name=name,
|
|
description=description,
|
|
baseline_template_id=baseline_template_id,
|
|
variant_template_id=variant_template_id,
|
|
success_metric=success_metric
|
|
)
|
|
|
|
self.experiments[experiment_id] = experiment
|
|
|
|
logger.info(f"Created A/B test experiment: {name} (ID: {experiment_id})")
|
|
return experiment
|
|
|
|
async def get_template_analytics(self, template_id: str) -> Dict[str, Any]:
|
|
"""Get analytics for a specific template."""
|
|
|
|
template = self.templates.get(template_id)
|
|
if not template:
|
|
raise ServiceError(f"Template not found: {template_id}")
|
|
|
|
stats = self.template_usage_stats.get(template_id, {})
|
|
|
|
return {
|
|
"template_id": template_id,
|
|
"template_name": template.name,
|
|
"usage_count": template.usage_count,
|
|
"average_rating": template.rating,
|
|
"domain_category": template.domain_category.value,
|
|
"created_at": template.created_at.isoformat(),
|
|
"last_used": stats.get("last_used"),
|
|
"avg_processing_time": stats.get("avg_processing_time", 0),
|
|
"total_tokens_generated": stats.get("total_tokens_generated", 0),
|
|
"success_rate": stats.get("success_rate", 0),
|
|
"performance_metrics": template.performance_metrics
|
|
}
|
|
|
|
async def get_domain_recommendations(
|
|
self,
|
|
content_sample: str,
|
|
max_recommendations: int = 3
|
|
) -> List[Dict[str, Any]]:
|
|
"""Get domain template recommendations based on content."""
|
|
|
|
await self.initialize_domain_presets()
|
|
|
|
# Simple keyword-based recommendations (in production, use ML)
|
|
domain_keywords = {
|
|
DomainCategory.EDUCATIONAL: ["learn", "teach", "study", "tutorial", "course", "lesson"],
|
|
DomainCategory.BUSINESS: ["revenue", "profit", "market", "strategy", "ROI", "business"],
|
|
DomainCategory.TECHNICAL: ["code", "programming", "development", "API", "technical", "system"],
|
|
DomainCategory.CONTENT_CREATION: ["video", "content", "creator", "audience", "engagement"],
|
|
DomainCategory.RESEARCH: ["research", "study", "analysis", "methodology", "data", "findings"]
|
|
}
|
|
|
|
content_lower = content_sample.lower()
|
|
domain_scores = {}
|
|
|
|
for domain, keywords in domain_keywords.items():
|
|
score = sum(1 for keyword in keywords if keyword in content_lower)
|
|
if score > 0:
|
|
domain_scores[domain] = score
|
|
|
|
# Get top recommendations
|
|
top_domains = sorted(domain_scores.items(), key=lambda x: x[1], reverse=True)[:max_recommendations]
|
|
|
|
recommendations = []
|
|
for domain, score in top_domains:
|
|
preset_id = f"preset_{domain.value}"
|
|
template = self.templates.get(preset_id)
|
|
|
|
if template:
|
|
recommendations.append({
|
|
"template_id": preset_id,
|
|
"template_name": template.name,
|
|
"domain_category": domain.value,
|
|
"confidence_score": min(score / 3.0, 1.0), # Normalize to 0-1
|
|
"description": template.description,
|
|
"reason": f"Content contains {score} relevant keywords for {domain.value}"
|
|
})
|
|
|
|
return recommendations
|
|
|
|
def _extract_template_variables(self, prompt_text: str) -> Dict[str, Any]:
|
|
"""Extract variables from template prompt text."""
|
|
import re
|
|
|
|
# Find {variable} patterns
|
|
variables = re.findall(r'\{(\w+)\}', prompt_text)
|
|
|
|
return {var: f"Variable for {var}" for var in set(variables)}
|
|
|
|
async def _validate_template_with_ai(self, prompt_text: str, model_config: ModelConfig):
|
|
"""Validate template by testing with sample content."""
|
|
|
|
test_content = "This is sample content for template validation."
|
|
|
|
# Try to format and execute template
|
|
try:
|
|
filled_prompt = prompt_text.format(content=test_content)
|
|
|
|
# Test with AI service (shortened for validation)
|
|
await self.ai_service.generate_response(
|
|
prompt=filled_prompt[:1000], # Limit for validation
|
|
temperature=model_config.temperature,
|
|
max_tokens=min(model_config.max_tokens, 300) # Limit for validation
|
|
)
|
|
|
|
except Exception as e:
|
|
raise ServiceError(f"Template validation failed: {str(e)}")
|
|
|
|
async def _update_template_usage(
|
|
self,
|
|
template_id: str,
|
|
processing_time: float,
|
|
response_length: int
|
|
):
|
|
"""Update template usage statistics."""
|
|
|
|
template = self.templates.get(template_id)
|
|
if template:
|
|
template.usage_count += 1
|
|
|
|
# Update detailed stats
|
|
if template_id not in self.template_usage_stats:
|
|
self.template_usage_stats[template_id] = {
|
|
"total_processing_time": 0,
|
|
"total_executions": 0,
|
|
"total_tokens_generated": 0,
|
|
"last_used": None
|
|
}
|
|
|
|
stats = self.template_usage_stats[template_id]
|
|
stats["total_processing_time"] += processing_time
|
|
stats["total_executions"] += 1
|
|
stats["total_tokens_generated"] += response_length
|
|
stats["last_used"] = datetime.now().isoformat()
|
|
stats["avg_processing_time"] = stats["total_processing_time"] / stats["total_executions"]
|
|
|
|
def _increment_version(self, current_version: str) -> str:
|
|
"""Increment template version number."""
|
|
try:
|
|
major, minor, patch = map(int, current_version.split('.'))
|
|
return f"{major}.{minor}.{patch + 1}"
|
|
except ValueError:
|
|
return "1.0.1"
|
|
|
|
async def get_system_stats(self) -> Dict[str, Any]:
|
|
"""Get overall system statistics."""
|
|
|
|
await self.initialize_domain_presets()
|
|
|
|
templates_by_domain = {}
|
|
for template in self.templates.values():
|
|
domain = template.domain_category.value
|
|
templates_by_domain[domain] = templates_by_domain.get(domain, 0) + 1
|
|
|
|
return {
|
|
"total_templates": len(self.templates),
|
|
"templates_by_domain": templates_by_domain,
|
|
"active_experiments": len([e for e in self.experiments.values() if e.status == "active"]),
|
|
"most_used_templates": [
|
|
{"id": t.id, "name": t.name, "usage_count": t.usage_count}
|
|
for t in sorted(self.templates.values(), key=lambda x: x.usage_count, reverse=True)[:5]
|
|
],
|
|
"domain_presets_available": self.domain_presets_initialized
|
|
}
|
|
|
|
async def initialize_domain_templates(self) -> Dict[str, str]:
|
|
"""Initialize domain-specific preset templates.
|
|
|
|
Returns:
|
|
Dict[str, str]: Mapping of domain category to template ID
|
|
"""
|
|
logger.info("Initializing domain-specific preset templates...")
|
|
|
|
# Use the existing initialize_domain_presets method
|
|
await self.initialize_domain_presets()
|
|
|
|
# Build template mapping
|
|
template_map = {}
|
|
for domain in DomainCategory:
|
|
preset_id = f"preset_{domain.value}"
|
|
if preset_id in self.templates:
|
|
template_map[domain.value.upper()] = preset_id
|
|
|
|
logger.info(f"Initialized {len(template_map)} domain-specific templates")
|
|
return template_map |