ALwrity Chatbot, SEO, Social media, Settings, Dashboard UI styling changes
This commit is contained in:
ajaysi
21
lib/chatbot_custom/core/__init__.py
Normal file
21
lib/chatbot_custom/core/__init__.py
Normal file
@@ -0,0 +1,21 @@
|
||||
"""
|
||||
Core modules for the Enhanced ALwrity Chatbot.
|
||||
|
||||
This package contains the core functionality split into manageable modules:
|
||||
- workflow_engine: Handles multi-tool workflows and automation
|
||||
- tool_router: Intelligent tool routing based on user intent
|
||||
- intent_analyzer: Advanced user intent analysis
|
||||
- context_manager: Conversation context and state management
|
||||
"""
|
||||
|
||||
from .workflow_engine import WorkflowEngine
|
||||
from .tool_router import SmartToolRouter
|
||||
from .intent_analyzer import IntentAnalyzer
|
||||
from .context_manager import ContextManager
|
||||
|
||||
__all__ = [
|
||||
'WorkflowEngine',
|
||||
'SmartToolRouter',
|
||||
'IntentAnalyzer',
|
||||
'ContextManager'
|
||||
]
|
||||
413
lib/chatbot_custom/core/context_manager.py
Normal file
413
lib/chatbot_custom/core/context_manager.py
Normal file
@@ -0,0 +1,413 @@
|
||||
"""
|
||||
Context Manager for Enhanced ALwrity Chatbot.
|
||||
|
||||
Manages conversation context, state, and user preferences with persistence.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Any, Optional
|
||||
from dataclasses import dataclass, asdict
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversationTurn:
|
||||
"""Represents a single conversation turn."""
|
||||
timestamp: str
|
||||
user_input: str
|
||||
intent: str
|
||||
tools_used: List[str]
|
||||
response_summary: str
|
||||
satisfaction_score: Optional[float] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class UserPreferences:
|
||||
"""User preferences and settings."""
|
||||
content_preferences: List[str]
|
||||
preferred_tone: str
|
||||
preferred_length: str
|
||||
industry_focus: List[str]
|
||||
language: str
|
||||
timezone: str
|
||||
notification_settings: Dict[str, bool]
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkflowState:
|
||||
"""Represents the state of an active workflow."""
|
||||
workflow_id: str
|
||||
workflow_name: str
|
||||
current_step: int
|
||||
total_steps: int
|
||||
step_data: Dict[str, Any]
|
||||
started_at: str
|
||||
last_updated: str
|
||||
is_paused: bool = False
|
||||
|
||||
|
||||
class ContextManager:
|
||||
"""Advanced conversation context and state management."""
|
||||
|
||||
def __init__(self, user_id: str = "default", context_file: str = None):
|
||||
self.user_id = user_id
|
||||
self.context_file = context_file or f"user_context_{user_id}.json"
|
||||
self.context_dir = "lib/chatbot_custom/user_contexts"
|
||||
|
||||
# Ensure context directory exists
|
||||
os.makedirs(self.context_dir, exist_ok=True)
|
||||
self.context_path = os.path.join(self.context_dir, self.context_file)
|
||||
|
||||
# Initialize context data
|
||||
self.conversation_history: List[ConversationTurn] = []
|
||||
self.user_preferences: UserPreferences = UserPreferences(
|
||||
content_preferences=[],
|
||||
preferred_tone="professional",
|
||||
preferred_length="medium",
|
||||
industry_focus=[],
|
||||
language="en",
|
||||
timezone="UTC",
|
||||
notification_settings={}
|
||||
)
|
||||
self.active_workflows: List[WorkflowState] = []
|
||||
self.tool_usage_history: List[Dict[str, Any]] = []
|
||||
self.session_data: Dict[str, Any] = {}
|
||||
self.analytics_data: Dict[str, Any] = {
|
||||
"total_interactions": 0,
|
||||
"tools_used_count": {},
|
||||
"workflows_completed": 0,
|
||||
"average_session_length": 0,
|
||||
"last_active": None
|
||||
}
|
||||
|
||||
# Load existing context
|
||||
self.load_context()
|
||||
|
||||
def add_conversation_turn(self, user_input: str, intent: str,
|
||||
tools_used: List[str], response_summary: str,
|
||||
satisfaction_score: Optional[float] = None):
|
||||
"""Add a new conversation turn to history."""
|
||||
turn = ConversationTurn(
|
||||
timestamp=datetime.now().isoformat(),
|
||||
user_input=user_input,
|
||||
intent=intent,
|
||||
tools_used=tools_used,
|
||||
response_summary=response_summary,
|
||||
satisfaction_score=satisfaction_score
|
||||
)
|
||||
|
||||
self.conversation_history.append(turn)
|
||||
|
||||
# Keep only last 50 turns to manage memory
|
||||
if len(self.conversation_history) > 50:
|
||||
self.conversation_history = self.conversation_history[-50:]
|
||||
|
||||
# Update analytics
|
||||
self.analytics_data["total_interactions"] += 1
|
||||
self.analytics_data["last_active"] = datetime.now().isoformat()
|
||||
|
||||
# Update tool usage statistics
|
||||
for tool in tools_used:
|
||||
if tool in self.analytics_data["tools_used_count"]:
|
||||
self.analytics_data["tools_used_count"][tool] += 1
|
||||
else:
|
||||
self.analytics_data["tools_used_count"][tool] = 1
|
||||
|
||||
self.save_context()
|
||||
|
||||
def update_user_preferences(self, preferences: Dict[str, Any]):
|
||||
"""Update user preferences."""
|
||||
for key, value in preferences.items():
|
||||
if hasattr(self.user_preferences, key):
|
||||
setattr(self.user_preferences, key, value)
|
||||
|
||||
self.save_context()
|
||||
|
||||
def get_recent_context(self, turns: int = 5) -> List[ConversationTurn]:
|
||||
"""Get recent conversation turns for context."""
|
||||
return self.conversation_history[-turns:] if self.conversation_history else []
|
||||
|
||||
def get_recent_topics(self, hours: int = 24) -> List[str]:
|
||||
"""Get topics discussed in recent hours."""
|
||||
cutoff_time = datetime.now() - timedelta(hours=hours)
|
||||
recent_topics = []
|
||||
|
||||
for turn in self.conversation_history:
|
||||
turn_time = datetime.fromisoformat(turn.timestamp)
|
||||
if turn_time > cutoff_time:
|
||||
# Extract topics from intent and tools used
|
||||
recent_topics.append(turn.intent)
|
||||
recent_topics.extend(turn.tools_used)
|
||||
|
||||
# Return unique topics
|
||||
return list(set(recent_topics))
|
||||
|
||||
def get_tool_usage_history(self, limit: int = 10) -> List[str]:
|
||||
"""Get recent tool usage history."""
|
||||
recent_tools = []
|
||||
for turn in self.conversation_history[-limit:]:
|
||||
recent_tools.extend(turn.tools_used)
|
||||
|
||||
return recent_tools
|
||||
|
||||
def start_workflow(self, workflow_id: str, workflow_name: str, total_steps: int):
|
||||
"""Start a new workflow."""
|
||||
workflow_state = WorkflowState(
|
||||
workflow_id=workflow_id,
|
||||
workflow_name=workflow_name,
|
||||
current_step=0,
|
||||
total_steps=total_steps,
|
||||
step_data={},
|
||||
started_at=datetime.now().isoformat(),
|
||||
last_updated=datetime.now().isoformat()
|
||||
)
|
||||
|
||||
self.active_workflows.append(workflow_state)
|
||||
self.save_context()
|
||||
|
||||
return workflow_state
|
||||
|
||||
def update_workflow_step(self, workflow_id: str, step_data: Dict[str, Any]):
|
||||
"""Update workflow step data."""
|
||||
for workflow in self.active_workflows:
|
||||
if workflow.workflow_id == workflow_id:
|
||||
workflow.current_step += 1
|
||||
workflow.step_data.update(step_data)
|
||||
workflow.last_updated = datetime.now().isoformat()
|
||||
|
||||
# Check if workflow is completed
|
||||
if workflow.current_step >= workflow.total_steps:
|
||||
self.complete_workflow(workflow_id)
|
||||
|
||||
self.save_context()
|
||||
return workflow
|
||||
|
||||
return None
|
||||
|
||||
def complete_workflow(self, workflow_id: str):
|
||||
"""Mark workflow as completed and remove from active workflows."""
|
||||
self.active_workflows = [w for w in self.active_workflows if w.workflow_id != workflow_id]
|
||||
self.analytics_data["workflows_completed"] += 1
|
||||
self.save_context()
|
||||
|
||||
def pause_workflow(self, workflow_id: str):
|
||||
"""Pause an active workflow."""
|
||||
for workflow in self.active_workflows:
|
||||
if workflow.workflow_id == workflow_id:
|
||||
workflow.is_paused = True
|
||||
workflow.last_updated = datetime.now().isoformat()
|
||||
self.save_context()
|
||||
return True
|
||||
return False
|
||||
|
||||
def resume_workflow(self, workflow_id: str):
|
||||
"""Resume a paused workflow."""
|
||||
for workflow in self.active_workflows:
|
||||
if workflow.workflow_id == workflow_id:
|
||||
workflow.is_paused = False
|
||||
workflow.last_updated = datetime.now().isoformat()
|
||||
self.save_context()
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_active_workflows(self) -> List[WorkflowState]:
|
||||
"""Get all active workflows."""
|
||||
return [w for w in self.active_workflows if not w.is_paused]
|
||||
|
||||
def get_paused_workflows(self) -> List[WorkflowState]:
|
||||
"""Get all paused workflows."""
|
||||
return [w for w in self.active_workflows if w.is_paused]
|
||||
|
||||
def set_session_data(self, key: str, value: Any):
|
||||
"""Set session-specific data."""
|
||||
self.session_data[key] = value
|
||||
|
||||
def get_session_data(self, key: str, default: Any = None) -> Any:
|
||||
"""Get session-specific data."""
|
||||
return self.session_data.get(key, default)
|
||||
|
||||
def clear_session_data(self):
|
||||
"""Clear all session data."""
|
||||
self.session_data.clear()
|
||||
|
||||
def get_context_for_intent_analysis(self) -> Dict[str, Any]:
|
||||
"""Get context data for intent analysis."""
|
||||
return {
|
||||
"recent_topics": self.get_recent_topics(),
|
||||
"user_preferences": asdict(self.user_preferences),
|
||||
"active_workflows": [w.workflow_name for w in self.get_active_workflows()],
|
||||
"tool_usage_history": self.get_tool_usage_history(),
|
||||
"session_data": self.session_data
|
||||
}
|
||||
|
||||
def get_user_analytics(self) -> Dict[str, Any]:
|
||||
"""Get user analytics and usage statistics."""
|
||||
# Calculate average session length
|
||||
if self.conversation_history:
|
||||
session_starts = []
|
||||
current_session_start = None
|
||||
|
||||
for turn in self.conversation_history:
|
||||
turn_time = datetime.fromisoformat(turn.timestamp)
|
||||
if not current_session_start:
|
||||
current_session_start = turn_time
|
||||
elif (turn_time - current_session_start).total_seconds() > 3600: # 1 hour gap = new session
|
||||
session_starts.append(current_session_start)
|
||||
current_session_start = turn_time
|
||||
|
||||
if current_session_start:
|
||||
session_starts.append(current_session_start)
|
||||
|
||||
# Most used tools
|
||||
most_used_tools = sorted(
|
||||
self.analytics_data["tools_used_count"].items(),
|
||||
key=lambda x: x[1],
|
||||
reverse=True
|
||||
)[:5]
|
||||
|
||||
# Recent activity pattern
|
||||
recent_activity = {}
|
||||
for turn in self.conversation_history[-20:]: # Last 20 turns
|
||||
date = turn.timestamp.split('T')[0] # Get date part
|
||||
if date in recent_activity:
|
||||
recent_activity[date] += 1
|
||||
else:
|
||||
recent_activity[date] = 1
|
||||
|
||||
return {
|
||||
**self.analytics_data,
|
||||
"most_used_tools": most_used_tools,
|
||||
"recent_activity_pattern": recent_activity,
|
||||
"active_workflows_count": len(self.get_active_workflows()),
|
||||
"paused_workflows_count": len(self.get_paused_workflows()),
|
||||
"conversation_turns": len(self.conversation_history)
|
||||
}
|
||||
|
||||
def export_conversation_history(self, format: str = "json") -> str:
|
||||
"""Export conversation history in specified format."""
|
||||
if format.lower() == "json":
|
||||
return json.dumps([asdict(turn) for turn in self.conversation_history], indent=2)
|
||||
elif format.lower() == "txt":
|
||||
text_export = []
|
||||
for turn in self.conversation_history:
|
||||
text_export.append(f"[{turn.timestamp}] User: {turn.user_input}")
|
||||
text_export.append(f"Intent: {turn.intent}, Tools: {', '.join(turn.tools_used)}")
|
||||
text_export.append(f"Response: {turn.response_summary}")
|
||||
text_export.append("-" * 50)
|
||||
return "\n".join(text_export)
|
||||
else:
|
||||
raise ValueError("Unsupported export format. Use 'json' or 'txt'.")
|
||||
|
||||
def cleanup_old_data(self, days: int = 30):
|
||||
"""Clean up old conversation data beyond specified days."""
|
||||
cutoff_date = datetime.now() - timedelta(days=days)
|
||||
|
||||
self.conversation_history = [
|
||||
turn for turn in self.conversation_history
|
||||
if datetime.fromisoformat(turn.timestamp) > cutoff_date
|
||||
]
|
||||
|
||||
self.save_context()
|
||||
|
||||
def save_context(self):
|
||||
"""Save context data to file."""
|
||||
try:
|
||||
context_data = {
|
||||
"user_id": self.user_id,
|
||||
"conversation_history": [asdict(turn) for turn in self.conversation_history],
|
||||
"user_preferences": asdict(self.user_preferences),
|
||||
"active_workflows": [asdict(workflow) for workflow in self.active_workflows],
|
||||
"analytics_data": self.analytics_data,
|
||||
"last_saved": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
with open(self.context_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(context_data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error saving context: {e}")
|
||||
|
||||
def load_context(self):
|
||||
"""Load context data from file."""
|
||||
try:
|
||||
if os.path.exists(self.context_path):
|
||||
with open(self.context_path, 'r', encoding='utf-8') as f:
|
||||
context_data = json.load(f)
|
||||
|
||||
# Load conversation history
|
||||
self.conversation_history = [
|
||||
ConversationTurn(**turn_data)
|
||||
for turn_data in context_data.get("conversation_history", [])
|
||||
]
|
||||
|
||||
# Load user preferences
|
||||
prefs_data = context_data.get("user_preferences", {})
|
||||
if prefs_data:
|
||||
self.user_preferences = UserPreferences(**prefs_data)
|
||||
|
||||
# Load active workflows
|
||||
self.active_workflows = [
|
||||
WorkflowState(**workflow_data)
|
||||
for workflow_data in context_data.get("active_workflows", [])
|
||||
]
|
||||
|
||||
# Load analytics data
|
||||
self.analytics_data.update(context_data.get("analytics_data", {}))
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error loading context: {e}")
|
||||
# Continue with default values if loading fails
|
||||
|
||||
def reset_context(self):
|
||||
"""Reset all context data (use with caution)."""
|
||||
self.conversation_history.clear()
|
||||
self.active_workflows.clear()
|
||||
self.session_data.clear()
|
||||
self.analytics_data = {
|
||||
"total_interactions": 0,
|
||||
"tools_used_count": {},
|
||||
"workflows_completed": 0,
|
||||
"average_session_length": 0,
|
||||
"last_active": None
|
||||
}
|
||||
|
||||
# Reset user preferences to defaults
|
||||
self.user_preferences = UserPreferences(
|
||||
content_preferences=[],
|
||||
preferred_tone="professional",
|
||||
preferred_length="medium",
|
||||
industry_focus=[],
|
||||
language="en",
|
||||
timezone="UTC",
|
||||
notification_settings={}
|
||||
)
|
||||
|
||||
self.save_context()
|
||||
|
||||
def get_context_summary(self) -> str:
|
||||
"""Get a human-readable summary of the current context."""
|
||||
summary_parts = []
|
||||
|
||||
# Basic stats
|
||||
summary_parts.append(f"Total interactions: {self.analytics_data['total_interactions']}")
|
||||
summary_parts.append(f"Conversation turns: {len(self.conversation_history)}")
|
||||
|
||||
# Active workflows
|
||||
active_workflows = self.get_active_workflows()
|
||||
if active_workflows:
|
||||
workflow_names = [w.workflow_name for w in active_workflows]
|
||||
summary_parts.append(f"Active workflows: {', '.join(workflow_names)}")
|
||||
|
||||
# Recent topics
|
||||
recent_topics = self.get_recent_topics(hours=6) # Last 6 hours
|
||||
if recent_topics:
|
||||
summary_parts.append(f"Recent topics: {', '.join(recent_topics[:5])}")
|
||||
|
||||
# User preferences
|
||||
if self.user_preferences.content_preferences:
|
||||
summary_parts.append(f"Content preferences: {', '.join(self.user_preferences.content_preferences)}")
|
||||
|
||||
summary_parts.append(f"Preferred tone: {self.user_preferences.preferred_tone}")
|
||||
|
||||
return "\n".join(summary_parts)
|
||||
413
lib/chatbot_custom/core/intent_analyzer.py
Normal file
413
lib/chatbot_custom/core/intent_analyzer.py
Normal file
@@ -0,0 +1,413 @@
|
||||
"""
|
||||
Intent Analyzer for Enhanced ALwrity Chatbot.
|
||||
|
||||
Advanced user intent analysis with context awareness and multi-intent detection.
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Any
|
||||
|
||||
|
||||
class IntentAnalyzer:
|
||||
"""Advanced user intent analysis with context awareness."""
|
||||
|
||||
def __init__(self):
|
||||
self.intent_keywords = {
|
||||
"write": {
|
||||
"keywords": ["write", "create", "generate", "compose", "draft", "author", "produce", "craft"],
|
||||
"sub_intents": ["blog", "article", "story", "social", "product", "email", "copy", "script"]
|
||||
},
|
||||
"analyze": {
|
||||
"keywords": ["analyze", "review", "check", "examine", "evaluate", "audit", "assess", "study"],
|
||||
"sub_intents": ["seo", "competitor", "website", "content", "performance", "traffic", "keywords"]
|
||||
},
|
||||
"seo": {
|
||||
"keywords": ["seo", "optimize", "rank", "keyword", "search", "meta", "visibility", "serp"],
|
||||
"sub_intents": ["on_page", "technical", "content_gap", "backlinks", "local", "mobile"]
|
||||
},
|
||||
"social": {
|
||||
"keywords": ["social", "facebook", "twitter", "linkedin", "instagram", "youtube", "tiktok"],
|
||||
"sub_intents": ["post", "campaign", "engagement", "hashtags", "stories", "ads"]
|
||||
},
|
||||
"research": {
|
||||
"keywords": ["research", "competitor", "market", "trend", "keyword", "analysis", "study"],
|
||||
"sub_intents": ["competitor", "keyword", "market", "content_gap", "audience", "trends"]
|
||||
},
|
||||
"plan": {
|
||||
"keywords": ["plan", "strategy", "calendar", "schedule", "roadmap", "organize", "structure"],
|
||||
"sub_intents": ["content_calendar", "strategy", "campaign", "workflow", "editorial"]
|
||||
},
|
||||
"workflow": {
|
||||
"keywords": ["workflow", "automate", "process", "step", "guide", "complete", "pipeline"],
|
||||
"sub_intents": ["blog_creation", "seo_audit", "social_campaign", "content_strategy"]
|
||||
},
|
||||
"optimize": {
|
||||
"keywords": ["optimize", "improve", "enhance", "boost", "increase", "maximize", "refine"],
|
||||
"sub_intents": ["seo", "content", "performance", "conversion", "speed", "engagement"]
|
||||
},
|
||||
"learn": {
|
||||
"keywords": ["learn", "how", "tutorial", "guide", "help", "explain", "teach", "show"],
|
||||
"sub_intents": ["seo", "content", "social", "tools", "strategy", "best_practices"]
|
||||
},
|
||||
"fix": {
|
||||
"keywords": ["fix", "solve", "repair", "troubleshoot", "debug", "resolve", "correct"],
|
||||
"sub_intents": ["seo_issues", "technical", "content", "performance", "errors"]
|
||||
}
|
||||
}
|
||||
|
||||
self.content_type_keywords = {
|
||||
"blog": ["blog", "article", "post", "content"],
|
||||
"social": ["social", "post", "tweet", "update", "story"],
|
||||
"email": ["email", "newsletter", "campaign", "sequence"],
|
||||
"video": ["video", "youtube", "script", "transcript"],
|
||||
"ad": ["ad", "advertisement", "promotion", "campaign"],
|
||||
"product": ["product", "description", "listing", "catalog"],
|
||||
"news": ["news", "press", "announcement", "release"],
|
||||
"story": ["story", "narrative", "fiction", "creative"],
|
||||
"technical": ["technical", "documentation", "manual", "guide"],
|
||||
"academic": ["academic", "research", "paper", "thesis"]
|
||||
}
|
||||
|
||||
self.urgency_keywords = {
|
||||
"high": ["urgent", "asap", "immediately", "emergency", "critical", "now"],
|
||||
"medium": ["soon", "quickly", "fast", "priority", "important"],
|
||||
"low": ["eventually", "when possible", "later", "sometime"]
|
||||
}
|
||||
|
||||
self.complexity_indicators = {
|
||||
"high": ["comprehensive", "detailed", "complete", "full", "extensive", "thorough"],
|
||||
"medium": ["moderate", "standard", "regular", "normal", "typical"],
|
||||
"low": ["simple", "basic", "quick", "brief", "short", "minimal"]
|
||||
}
|
||||
|
||||
def analyze_user_intent(self, prompt: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
|
||||
"""Enhanced user intent analysis with context awareness."""
|
||||
prompt_lower = prompt.lower()
|
||||
|
||||
# Detect primary and secondary intents
|
||||
detected_intents = self._detect_intents(prompt_lower)
|
||||
|
||||
# Detect sub-intents
|
||||
sub_intents = self._detect_sub_intents(prompt_lower, detected_intents)
|
||||
|
||||
# Determine content types
|
||||
content_types = self._detect_content_types(prompt_lower)
|
||||
|
||||
# Assess urgency
|
||||
urgency = self._assess_urgency(prompt_lower)
|
||||
|
||||
# Determine complexity
|
||||
complexity = self._assess_complexity(prompt_lower)
|
||||
|
||||
# Calculate confidence scores
|
||||
confidence_scores = self._calculate_confidence_scores(prompt_lower, detected_intents)
|
||||
|
||||
# Context-aware enhancements
|
||||
if context:
|
||||
detected_intents, confidence_scores = self._enhance_with_context(
|
||||
detected_intents, confidence_scores, context, prompt_lower
|
||||
)
|
||||
|
||||
# Determine primary intent
|
||||
primary_intent = self._determine_primary_intent(detected_intents, confidence_scores)
|
||||
|
||||
# Generate suggestions
|
||||
suggested_workflows = self._suggest_workflows(detected_intents, content_types)
|
||||
suggested_tools = self._suggest_tools(detected_intents, sub_intents, content_types)
|
||||
|
||||
return {
|
||||
"primary_intent": primary_intent,
|
||||
"all_intents": detected_intents,
|
||||
"sub_intents": sub_intents,
|
||||
"content_types": content_types,
|
||||
"confidence_scores": confidence_scores,
|
||||
"urgency": urgency,
|
||||
"complexity": complexity,
|
||||
"suggested_workflows": suggested_workflows,
|
||||
"suggested_tools": suggested_tools,
|
||||
"intent_strength": self._calculate_intent_strength(confidence_scores),
|
||||
"multi_intent": len(detected_intents) > 1,
|
||||
"context_enhanced": context is not None
|
||||
}
|
||||
|
||||
def _detect_intents(self, prompt_lower: str) -> List[str]:
|
||||
"""Detect all intents in the user prompt."""
|
||||
detected_intents = []
|
||||
|
||||
for intent, data in self.intent_keywords.items():
|
||||
matches = sum(1 for keyword in data["keywords"] if keyword in prompt_lower)
|
||||
if matches > 0:
|
||||
detected_intents.append(intent)
|
||||
|
||||
return detected_intents
|
||||
|
||||
def _detect_sub_intents(self, prompt_lower: str, detected_intents: List[str]) -> List[str]:
|
||||
"""Detect sub-intents based on primary intents."""
|
||||
sub_intents = []
|
||||
|
||||
for intent in detected_intents:
|
||||
if intent in self.intent_keywords:
|
||||
for sub_intent in self.intent_keywords[intent]["sub_intents"]:
|
||||
if sub_intent in prompt_lower:
|
||||
sub_intents.append(sub_intent)
|
||||
|
||||
return list(set(sub_intents)) # Remove duplicates
|
||||
|
||||
def _detect_content_types(self, prompt_lower: str) -> List[str]:
|
||||
"""Detect content types mentioned in the prompt."""
|
||||
content_types = []
|
||||
|
||||
for content_type, keywords in self.content_type_keywords.items():
|
||||
if any(keyword in prompt_lower for keyword in keywords):
|
||||
content_types.append(content_type)
|
||||
|
||||
return content_types
|
||||
|
||||
def _assess_urgency(self, prompt_lower: str) -> Dict[str, Any]:
|
||||
"""Assess the urgency level of the request."""
|
||||
urgency_level = "normal"
|
||||
urgency_score = 0.5
|
||||
|
||||
for level, keywords in self.urgency_keywords.items():
|
||||
matches = sum(1 for keyword in keywords if keyword in prompt_lower)
|
||||
if matches > 0:
|
||||
if level == "high":
|
||||
urgency_level = "high"
|
||||
urgency_score = 0.9
|
||||
break
|
||||
elif level == "medium" and urgency_level == "normal":
|
||||
urgency_level = "medium"
|
||||
urgency_score = 0.7
|
||||
elif level == "low" and urgency_level == "normal":
|
||||
urgency_level = "low"
|
||||
urgency_score = 0.3
|
||||
|
||||
return {
|
||||
"level": urgency_level,
|
||||
"score": urgency_score,
|
||||
"is_urgent": urgency_level in ["high", "medium"]
|
||||
}
|
||||
|
||||
def _assess_complexity(self, prompt_lower: str) -> Dict[str, Any]:
|
||||
"""Assess the complexity level of the request."""
|
||||
complexity_level = "medium"
|
||||
complexity_score = 0.5
|
||||
|
||||
for level, keywords in self.complexity_indicators.items():
|
||||
matches = sum(1 for keyword in keywords if keyword in prompt_lower)
|
||||
if matches > 0:
|
||||
complexity_level = level
|
||||
complexity_score = {"high": 0.9, "medium": 0.5, "low": 0.3}[level]
|
||||
break
|
||||
|
||||
# Additional complexity indicators
|
||||
word_count = len(prompt_lower.split())
|
||||
if word_count > 50:
|
||||
complexity_score = min(complexity_score + 0.2, 1.0)
|
||||
elif word_count < 10:
|
||||
complexity_score = max(complexity_score - 0.2, 0.1)
|
||||
|
||||
return {
|
||||
"level": complexity_level,
|
||||
"score": complexity_score,
|
||||
"word_count": word_count
|
||||
}
|
||||
|
||||
def _calculate_confidence_scores(self, prompt_lower: str, detected_intents: List[str]) -> Dict[str, float]:
|
||||
"""Calculate confidence scores for detected intents."""
|
||||
confidence_scores = {}
|
||||
|
||||
for intent in detected_intents:
|
||||
if intent in self.intent_keywords:
|
||||
keywords = self.intent_keywords[intent]["keywords"]
|
||||
matches = sum(1 for keyword in keywords if keyword in prompt_lower)
|
||||
confidence = matches / len(keywords)
|
||||
|
||||
# Boost confidence for exact matches
|
||||
if intent in prompt_lower:
|
||||
confidence += 0.3
|
||||
|
||||
# Boost confidence for multiple keyword matches
|
||||
if matches > 2:
|
||||
confidence += 0.2
|
||||
|
||||
confidence_scores[intent] = min(confidence, 1.0)
|
||||
|
||||
return confidence_scores
|
||||
|
||||
def _enhance_with_context(self, detected_intents: List[str], confidence_scores: Dict[str, float],
|
||||
context: Dict[str, Any], prompt_lower: str) -> tuple:
|
||||
"""Enhance intent detection with conversation context."""
|
||||
enhanced_intents = detected_intents.copy()
|
||||
enhanced_scores = confidence_scores.copy()
|
||||
|
||||
# Recent conversation topics
|
||||
recent_topics = context.get("recent_topics", [])
|
||||
for topic in recent_topics:
|
||||
if topic.lower() in prompt_lower:
|
||||
# Boost related intents
|
||||
for intent in self.intent_keywords:
|
||||
if topic.lower() in self.intent_keywords[intent]["keywords"]:
|
||||
if intent in enhanced_scores:
|
||||
enhanced_scores[intent] += 0.1
|
||||
else:
|
||||
enhanced_intents.append(intent)
|
||||
enhanced_scores[intent] = 0.4
|
||||
|
||||
# User preferences
|
||||
user_prefs = context.get("user_preferences", {})
|
||||
if user_prefs.get("content_preferences"):
|
||||
for pref in user_prefs["content_preferences"]:
|
||||
if pref in prompt_lower:
|
||||
# Boost content creation intents
|
||||
if "write" in enhanced_scores:
|
||||
enhanced_scores["write"] += 0.15
|
||||
|
||||
# Active workflows
|
||||
active_workflows = context.get("active_workflows", [])
|
||||
if active_workflows:
|
||||
# Boost workflow-related intents
|
||||
if "workflow" in enhanced_scores:
|
||||
enhanced_scores["workflow"] += 0.2
|
||||
else:
|
||||
enhanced_intents.append("workflow")
|
||||
enhanced_scores["workflow"] = 0.6
|
||||
|
||||
# Tool usage history
|
||||
tool_history = context.get("tool_usage_history", [])
|
||||
if tool_history:
|
||||
last_tools = tool_history[-3:] # Last 3 tools
|
||||
for tool in last_tools:
|
||||
# Map tools to intents and boost related intents
|
||||
tool_intent_mapping = {
|
||||
"ai_blog_writer": "write",
|
||||
"content_gap_analysis": "analyze",
|
||||
"technical_seo": "seo",
|
||||
"linkedin_writer": "social"
|
||||
}
|
||||
|
||||
if tool in tool_intent_mapping:
|
||||
intent = tool_intent_mapping[tool]
|
||||
if intent in enhanced_scores:
|
||||
enhanced_scores[intent] += 0.1
|
||||
|
||||
return enhanced_intents, enhanced_scores
|
||||
|
||||
def _determine_primary_intent(self, detected_intents: List[str], confidence_scores: Dict[str, float]) -> str:
|
||||
"""Determine the primary intent from detected intents."""
|
||||
if not detected_intents:
|
||||
return "general"
|
||||
|
||||
if len(detected_intents) == 1:
|
||||
return detected_intents[0]
|
||||
|
||||
# Return intent with highest confidence
|
||||
primary_intent = max(detected_intents, key=lambda x: confidence_scores.get(x, 0))
|
||||
return primary_intent
|
||||
|
||||
def _suggest_workflows(self, detected_intents: List[str], content_types: List[str]) -> List[str]:
|
||||
"""Suggest relevant workflows based on intents and content types."""
|
||||
suggested_workflows = []
|
||||
|
||||
# Intent-based workflow suggestions
|
||||
workflow_mapping = {
|
||||
"write": ["blog_creation_workflow", "content_strategy_workflow"],
|
||||
"analyze": ["competitor_analysis_workflow", "seo_audit_workflow"],
|
||||
"seo": ["seo_audit_workflow", "content_gap_workflow"],
|
||||
"social": ["social_media_workflow", "content_repurposing_workflow"],
|
||||
"plan": ["content_strategy_workflow", "editorial_calendar_workflow"]
|
||||
}
|
||||
|
||||
for intent in detected_intents:
|
||||
if intent in workflow_mapping:
|
||||
suggested_workflows.extend(workflow_mapping[intent])
|
||||
|
||||
# Content type specific workflows
|
||||
if "blog" in content_types:
|
||||
suggested_workflows.append("blog_creation_workflow")
|
||||
if "social" in content_types:
|
||||
suggested_workflows.append("social_media_workflow")
|
||||
|
||||
return list(set(suggested_workflows)) # Remove duplicates
|
||||
|
||||
def _suggest_tools(self, detected_intents: List[str], sub_intents: List[str],
|
||||
content_types: List[str]) -> List[str]:
|
||||
"""Suggest relevant tools based on intents, sub-intents, and content types."""
|
||||
suggested_tools = []
|
||||
|
||||
# Intent-based tool suggestions
|
||||
tool_mapping = {
|
||||
"write": ["ai_blog_writer", "story_writer", "email_writer"],
|
||||
"analyze": ["content_gap_analysis", "website_analyzer", "competitor_analyzer"],
|
||||
"seo": ["technical_seo", "on_page_seo", "keyword_research"],
|
||||
"social": ["linkedin_writer", "facebook_writer", "social_campaign"],
|
||||
"research": ["competitor_analysis", "keyword_research", "market_research"],
|
||||
"optimize": ["seo_optimizer", "content_optimizer", "performance_optimizer"]
|
||||
}
|
||||
|
||||
for intent in detected_intents:
|
||||
if intent in tool_mapping:
|
||||
suggested_tools.extend(tool_mapping[intent])
|
||||
|
||||
# Sub-intent specific tools
|
||||
sub_intent_tools = {
|
||||
"blog": ["ai_blog_writer", "seo_optimizer"],
|
||||
"competitor": ["competitor_analysis", "content_gap_analysis"],
|
||||
"technical": ["technical_seo", "performance_analyzer"],
|
||||
"social": ["linkedin_writer", "facebook_writer"]
|
||||
}
|
||||
|
||||
for sub_intent in sub_intents:
|
||||
if sub_intent in sub_intent_tools:
|
||||
suggested_tools.extend(sub_intent_tools[sub_intent])
|
||||
|
||||
# Content type specific tools
|
||||
content_tools = {
|
||||
"blog": ["ai_blog_writer", "seo_optimizer"],
|
||||
"social": ["linkedin_writer", "facebook_writer"],
|
||||
"email": ["email_writer", "campaign_creator"],
|
||||
"video": ["youtube_writer", "script_generator"]
|
||||
}
|
||||
|
||||
for content_type in content_types:
|
||||
if content_type in content_tools:
|
||||
suggested_tools.extend(content_tools[content_type])
|
||||
|
||||
return list(set(suggested_tools)) # Remove duplicates
|
||||
|
||||
def _calculate_intent_strength(self, confidence_scores: Dict[str, float]) -> str:
|
||||
"""Calculate overall intent strength."""
|
||||
if not confidence_scores:
|
||||
return "weak"
|
||||
|
||||
max_confidence = max(confidence_scores.values())
|
||||
avg_confidence = sum(confidence_scores.values()) / len(confidence_scores)
|
||||
|
||||
if max_confidence >= 0.8 and avg_confidence >= 0.6:
|
||||
return "strong"
|
||||
elif max_confidence >= 0.6 or avg_confidence >= 0.4:
|
||||
return "moderate"
|
||||
else:
|
||||
return "weak"
|
||||
|
||||
def get_intent_explanation(self, intent_analysis: Dict[str, Any]) -> str:
|
||||
"""Generate a human-readable explanation of the intent analysis."""
|
||||
primary = intent_analysis["primary_intent"]
|
||||
confidence = intent_analysis["confidence_scores"].get(primary, 0)
|
||||
urgency = intent_analysis["urgency"]["level"]
|
||||
complexity = intent_analysis["complexity"]["level"]
|
||||
|
||||
explanation = f"Primary intent: {primary} (confidence: {confidence:.2f})\n"
|
||||
|
||||
if intent_analysis["multi_intent"]:
|
||||
other_intents = [i for i in intent_analysis["all_intents"] if i != primary]
|
||||
explanation += f"Additional intents: {', '.join(other_intents)}\n"
|
||||
|
||||
if intent_analysis["content_types"]:
|
||||
explanation += f"Content types: {', '.join(intent_analysis['content_types'])}\n"
|
||||
|
||||
explanation += f"Urgency: {urgency}, Complexity: {complexity}\n"
|
||||
|
||||
if intent_analysis["suggested_tools"]:
|
||||
explanation += f"Recommended tools: {', '.join(intent_analysis['suggested_tools'][:3])}"
|
||||
|
||||
return explanation
|
||||
285
lib/chatbot_custom/core/tool_router.py
Normal file
285
lib/chatbot_custom/core/tool_router.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""
|
||||
Smart Tool Router for Enhanced ALwrity Chatbot.
|
||||
|
||||
Intelligent tool routing based on user intent and context.
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Any
|
||||
|
||||
|
||||
class SmartToolRouter:
|
||||
"""Intelligent tool routing based on user intent and context."""
|
||||
|
||||
def __init__(self):
|
||||
self.tool_categories = {
|
||||
"content_creation": [
|
||||
"ai_blog_writer", "story_writer", "essay_writer",
|
||||
"product_description", "email_writer", "news_writer"
|
||||
],
|
||||
"seo_tools": [
|
||||
"content_gap_analysis", "technical_seo", "on_page_seo",
|
||||
"competitor_analysis", "keyword_research", "meta_generator"
|
||||
],
|
||||
"social_media": [
|
||||
"linkedin_writer", "facebook_writer", "youtube_writer",
|
||||
"instagram_writer", "twitter_writer", "social_campaign"
|
||||
],
|
||||
"analysis": [
|
||||
"website_analyzer", "content_analyzer", "competitor_analyzer",
|
||||
"performance_analyzer", "seo_analyzer"
|
||||
],
|
||||
"planning": [
|
||||
"content_calendar", "content_repurposing", "strategy_planner",
|
||||
"campaign_planner", "editorial_calendar"
|
||||
],
|
||||
"optimization": [
|
||||
"seo_optimizer", "content_optimizer", "performance_optimizer",
|
||||
"conversion_optimizer", "speed_optimizer"
|
||||
]
|
||||
}
|
||||
|
||||
self.intent_tool_mapping = {
|
||||
"write": ["ai_blog_writer", "story_writer", "essay_writer", "email_writer"],
|
||||
"analyze": ["content_gap_analysis", "technical_seo", "website_analyzer", "competitor_analyzer"],
|
||||
"seo": ["on_page_seo", "technical_seo", "content_gap_analysis", "seo_optimizer"],
|
||||
"social": ["linkedin_writer", "facebook_writer", "youtube_writer", "social_campaign"],
|
||||
"plan": ["content_calendar", "content_repurposing", "strategy_planner", "campaign_planner"],
|
||||
"research": ["competitor_analysis", "content_gap_analysis", "keyword_research", "market_research"],
|
||||
"optimize": ["seo_optimizer", "content_optimizer", "performance_optimizer"],
|
||||
"create": ["ai_blog_writer", "content_creator", "social_content_creation"],
|
||||
"audit": ["technical_seo", "seo_analyzer", "website_analyzer", "performance_analyzer"]
|
||||
}
|
||||
|
||||
# Tool confidence weights based on effectiveness
|
||||
self.tool_weights = {
|
||||
"ai_blog_writer": 0.9,
|
||||
"content_gap_analysis": 0.85,
|
||||
"technical_seo": 0.8,
|
||||
"linkedin_writer": 0.85,
|
||||
"competitor_analysis": 0.8,
|
||||
"seo_optimizer": 0.75,
|
||||
"content_calendar": 0.7
|
||||
}
|
||||
|
||||
def route_to_tools(self, user_intent: str, context: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""Route user intent to relevant tools with confidence scoring."""
|
||||
suggested_tools = []
|
||||
user_intent_lower = user_intent.lower()
|
||||
|
||||
# Primary intent matching
|
||||
for intent, tools in self.intent_tool_mapping.items():
|
||||
if intent in user_intent_lower:
|
||||
for tool in tools:
|
||||
confidence = self._calculate_confidence(intent, user_intent, context)
|
||||
suggested_tools.append({
|
||||
"tool": tool,
|
||||
"category": self._get_tool_category(tool),
|
||||
"confidence": confidence,
|
||||
"intent_match": intent,
|
||||
"reason": f"Matches '{intent}' intent"
|
||||
})
|
||||
|
||||
# Context-based suggestions
|
||||
context_tools = self._get_context_based_suggestions(context, user_intent)
|
||||
suggested_tools.extend(context_tools)
|
||||
|
||||
# Remove duplicates and sort by confidence
|
||||
unique_tools = {}
|
||||
for tool in suggested_tools:
|
||||
tool_name = tool["tool"]
|
||||
if tool_name not in unique_tools or tool["confidence"] > unique_tools[tool_name]["confidence"]:
|
||||
unique_tools[tool_name] = tool
|
||||
|
||||
# Sort by confidence and return top suggestions
|
||||
sorted_tools = sorted(unique_tools.values(), key=lambda x: x["confidence"], reverse=True)
|
||||
return sorted_tools[:8] # Return top 8 suggestions
|
||||
|
||||
def _get_tool_category(self, tool: str) -> str:
|
||||
"""Get category for a tool."""
|
||||
for category, tools in self.tool_categories.items():
|
||||
if tool in tools:
|
||||
return category
|
||||
return "general"
|
||||
|
||||
def _calculate_confidence(self, intent: str, user_text: str, context: Dict[str, Any]) -> float:
|
||||
"""Calculate confidence score for tool suggestion."""
|
||||
base_score = 0.5
|
||||
user_text_lower = user_text.lower()
|
||||
|
||||
# Intent match bonus
|
||||
if intent in user_text_lower:
|
||||
base_score += 0.3
|
||||
|
||||
# Keyword bonuses
|
||||
keyword_bonuses = {
|
||||
"write": ["create", "generate", "compose", "draft", "author", "produce"],
|
||||
"analyze": ["check", "review", "examine", "evaluate", "assess", "study"],
|
||||
"seo": ["optimize", "rank", "search", "keywords", "meta", "visibility"],
|
||||
"social": ["post", "share", "engage", "campaign", "viral", "audience"],
|
||||
"plan": ["schedule", "organize", "strategy", "roadmap", "timeline"],
|
||||
"research": ["study", "investigate", "explore", "discover", "find"]
|
||||
}
|
||||
|
||||
if intent in keyword_bonuses:
|
||||
for keyword in keyword_bonuses[intent]:
|
||||
if keyword in user_text_lower:
|
||||
base_score += 0.1
|
||||
|
||||
# Context bonuses
|
||||
if context:
|
||||
# Recent tool usage
|
||||
recent_tools = context.get('tool_usage_history', [])[-3:]
|
||||
if any(tool in user_text_lower for tool in recent_tools):
|
||||
base_score += 0.15
|
||||
|
||||
# User preferences
|
||||
user_prefs = context.get('user_preferences', {})
|
||||
if user_prefs.get('industry') and user_prefs['industry'].lower() in user_text_lower:
|
||||
base_score += 0.1
|
||||
|
||||
# Urgency bonus
|
||||
urgency_keywords = ["urgent", "asap", "quickly", "fast", "immediate", "now"]
|
||||
if any(keyword in user_text_lower for keyword in urgency_keywords):
|
||||
base_score += 0.1
|
||||
|
||||
return min(base_score, 1.0)
|
||||
|
||||
def _get_context_based_suggestions(self, context: Dict[str, Any], user_intent: str) -> List[Dict[str, Any]]:
|
||||
"""Get tool suggestions based on conversation context."""
|
||||
context_tools = []
|
||||
|
||||
if not context:
|
||||
return context_tools
|
||||
|
||||
# Recent tool usage patterns
|
||||
recent_tools = context.get('tool_usage_history', [])
|
||||
if recent_tools:
|
||||
# Suggest complementary tools
|
||||
last_tool = recent_tools[-1] if recent_tools else None
|
||||
complementary_tools = self._get_complementary_tools(last_tool)
|
||||
|
||||
for tool in complementary_tools:
|
||||
context_tools.append({
|
||||
"tool": tool,
|
||||
"category": self._get_tool_category(tool),
|
||||
"confidence": 0.6,
|
||||
"intent_match": "context",
|
||||
"reason": f"Complements recent use of {last_tool}"
|
||||
})
|
||||
|
||||
# Active workflows
|
||||
active_workflows = context.get('active_workflows', [])
|
||||
if active_workflows:
|
||||
# Suggest tools for current workflow steps
|
||||
for workflow in active_workflows:
|
||||
workflow_tools = self._get_workflow_tools(workflow)
|
||||
for tool in workflow_tools:
|
||||
context_tools.append({
|
||||
"tool": tool,
|
||||
"category": self._get_tool_category(tool),
|
||||
"confidence": 0.7,
|
||||
"intent_match": "workflow",
|
||||
"reason": f"Next step in {workflow} workflow"
|
||||
})
|
||||
|
||||
# User preferences
|
||||
user_prefs = context.get('user_preferences', {})
|
||||
if user_prefs.get('content_preferences'):
|
||||
pref_tools = self._get_preference_based_tools(user_prefs['content_preferences'])
|
||||
for tool in pref_tools:
|
||||
context_tools.append({
|
||||
"tool": tool,
|
||||
"category": self._get_tool_category(tool),
|
||||
"confidence": 0.65,
|
||||
"intent_match": "preference",
|
||||
"reason": "Based on your content preferences"
|
||||
})
|
||||
|
||||
return context_tools
|
||||
|
||||
def _get_complementary_tools(self, last_tool: str) -> List[str]:
|
||||
"""Get tools that complement the last used tool."""
|
||||
complementary_mapping = {
|
||||
"ai_blog_writer": ["seo_optimizer", "meta_generator", "content_gap_analysis"],
|
||||
"content_gap_analysis": ["ai_blog_writer", "keyword_research", "competitor_analysis"],
|
||||
"technical_seo": ["on_page_seo", "content_optimizer", "performance_analyzer"],
|
||||
"linkedin_writer": ["social_campaign", "content_calendar", "hashtag_research"],
|
||||
"competitor_analysis": ["content_gap_analysis", "keyword_research", "strategy_planner"],
|
||||
"keyword_research": ["ai_blog_writer", "content_gap_analysis", "seo_optimizer"]
|
||||
}
|
||||
|
||||
return complementary_mapping.get(last_tool, [])
|
||||
|
||||
def _get_workflow_tools(self, workflow: str) -> List[str]:
|
||||
"""Get tools associated with a specific workflow."""
|
||||
workflow_tools = {
|
||||
"blog_creation_workflow": ["keyword_research", "ai_blog_writer", "seo_optimizer"],
|
||||
"competitor_analysis_workflow": ["competitor_analysis", "content_gap_analysis"],
|
||||
"social_media_workflow": ["linkedin_writer", "facebook_writer", "social_campaign"],
|
||||
"seo_audit_workflow": ["technical_seo", "on_page_seo", "competitor_analysis"]
|
||||
}
|
||||
|
||||
return workflow_tools.get(workflow, [])
|
||||
|
||||
def _get_preference_based_tools(self, content_preferences: List[str]) -> List[str]:
|
||||
"""Get tools based on user content preferences."""
|
||||
preference_tools = []
|
||||
|
||||
for pref in content_preferences:
|
||||
if pref in ["blog", "article"]:
|
||||
preference_tools.extend(["ai_blog_writer", "seo_optimizer"])
|
||||
elif pref in ["social", "post"]:
|
||||
preference_tools.extend(["linkedin_writer", "facebook_writer"])
|
||||
elif pref in ["seo", "optimization"]:
|
||||
preference_tools.extend(["technical_seo", "on_page_seo"])
|
||||
|
||||
return list(set(preference_tools)) # Remove duplicates
|
||||
|
||||
def get_tool_info(self, tool_name: str) -> Dict[str, Any]:
|
||||
"""Get detailed information about a specific tool."""
|
||||
tool_info = {
|
||||
"ai_blog_writer": {
|
||||
"name": "AI Blog Writer",
|
||||
"description": "Create comprehensive, SEO-optimized blog posts",
|
||||
"category": "content_creation",
|
||||
"use_cases": ["Blog posts", "Articles", "Long-form content"],
|
||||
"estimated_time": "5-10 minutes"
|
||||
},
|
||||
"content_gap_analysis": {
|
||||
"name": "Content Gap Analysis",
|
||||
"description": "Identify content opportunities vs competitors",
|
||||
"category": "seo_tools",
|
||||
"use_cases": ["Competitor research", "Content strategy", "SEO planning"],
|
||||
"estimated_time": "10-15 minutes"
|
||||
},
|
||||
"technical_seo": {
|
||||
"name": "Technical SEO Crawler",
|
||||
"description": "Comprehensive technical SEO audit",
|
||||
"category": "seo_tools",
|
||||
"use_cases": ["Site audits", "Technical issues", "Performance analysis"],
|
||||
"estimated_time": "15-20 minutes"
|
||||
},
|
||||
"linkedin_writer": {
|
||||
"name": "LinkedIn Writer",
|
||||
"description": "Create professional LinkedIn content",
|
||||
"category": "social_media",
|
||||
"use_cases": ["LinkedIn posts", "Professional articles", "Networking content"],
|
||||
"estimated_time": "3-5 minutes"
|
||||
}
|
||||
}
|
||||
|
||||
return tool_info.get(tool_name, {
|
||||
"name": tool_name.replace('_', ' ').title(),
|
||||
"description": f"ALwrity {tool_name.replace('_', ' ')} tool",
|
||||
"category": self._get_tool_category(tool_name),
|
||||
"use_cases": ["Content creation", "Analysis", "Optimization"],
|
||||
"estimated_time": "5-10 minutes"
|
||||
})
|
||||
|
||||
def get_category_tools(self, category: str) -> List[str]:
|
||||
"""Get all tools in a specific category."""
|
||||
return self.tool_categories.get(category, [])
|
||||
|
||||
def get_all_categories(self) -> List[str]:
|
||||
"""Get all available tool categories."""
|
||||
return list(self.tool_categories.keys())
|
||||
171
lib/chatbot_custom/core/workflow_engine.py
Normal file
171
lib/chatbot_custom/core/workflow_engine.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
Workflow Engine for Enhanced ALwrity Chatbot.
|
||||
|
||||
Handles multi-tool workflows and automation for complex content creation tasks.
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Any
|
||||
|
||||
|
||||
class WorkflowEngine:
|
||||
"""Handles multi-tool workflows and automation."""
|
||||
|
||||
def __init__(self):
|
||||
self.workflows = {
|
||||
"blog_creation_workflow": {
|
||||
"name": "Complete Blog Creation",
|
||||
"description": "From idea to published blog post",
|
||||
"steps": [
|
||||
{"tool": "keyword_research", "name": "Keyword Research"},
|
||||
{"tool": "content_gap_analysis", "name": "Content Gap Analysis"},
|
||||
{"tool": "blog_writing", "name": "Blog Writing"},
|
||||
{"tool": "seo_optimization", "name": "SEO Optimization"},
|
||||
{"tool": "meta_generation", "name": "Meta Tags Generation"}
|
||||
]
|
||||
},
|
||||
"competitor_analysis_workflow": {
|
||||
"name": "Competitor Content Strategy",
|
||||
"description": "Analyze competitors and create content plan",
|
||||
"steps": [
|
||||
{"tool": "competitor_analysis", "name": "Competitor Analysis"},
|
||||
{"tool": "content_gap_analysis", "name": "Content Gap Analysis"},
|
||||
{"tool": "content_calendar", "name": "Content Calendar Creation"},
|
||||
{"tool": "content_ideas", "name": "Content Ideas Generation"}
|
||||
]
|
||||
},
|
||||
"social_media_workflow": {
|
||||
"name": "Social Media Campaign",
|
||||
"description": "Create comprehensive social media content",
|
||||
"steps": [
|
||||
{"tool": "audience_analysis", "name": "Audience Analysis"},
|
||||
{"tool": "content_planning", "name": "Content Planning"},
|
||||
{"tool": "social_content_creation", "name": "Social Content Creation"},
|
||||
{"tool": "hashtag_research", "name": "Hashtag Research"}
|
||||
]
|
||||
},
|
||||
"seo_audit_workflow": {
|
||||
"name": "Complete SEO Audit",
|
||||
"description": "Comprehensive website SEO analysis and optimization",
|
||||
"steps": [
|
||||
{"tool": "technical_seo", "name": "Technical SEO Analysis"},
|
||||
{"tool": "on_page_seo", "name": "On-Page SEO Review"},
|
||||
{"tool": "content_gap_analysis", "name": "Content Gap Analysis"},
|
||||
{"tool": "competitor_seo", "name": "Competitor SEO Analysis"},
|
||||
{"tool": "optimization_plan", "name": "SEO Optimization Plan"}
|
||||
]
|
||||
},
|
||||
"content_strategy_workflow": {
|
||||
"name": "Content Strategy Development",
|
||||
"description": "Develop comprehensive content strategy from research to execution",
|
||||
"steps": [
|
||||
{"tool": "market_research", "name": "Market Research"},
|
||||
{"tool": "audience_analysis", "name": "Audience Analysis"},
|
||||
{"tool": "competitor_analysis", "name": "Competitor Analysis"},
|
||||
{"tool": "content_pillars", "name": "Content Pillars Definition"},
|
||||
{"tool": "content_calendar", "name": "Content Calendar Creation"}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
def suggest_workflows(self, user_intent: str) -> List[Dict[str, Any]]:
|
||||
"""Suggest relevant workflows based on user intent."""
|
||||
relevant_workflows = []
|
||||
user_intent_lower = user_intent.lower()
|
||||
|
||||
# Blog and content creation
|
||||
if any(word in user_intent_lower for word in ['blog', 'article', 'post', 'write', 'content']):
|
||||
relevant_workflows.append(self.workflows["blog_creation_workflow"])
|
||||
|
||||
# Competitor and market analysis
|
||||
if any(word in user_intent_lower for word in ['competitor', 'analysis', 'research', 'market']):
|
||||
relevant_workflows.append(self.workflows["competitor_analysis_workflow"])
|
||||
|
||||
# Social media
|
||||
if any(word in user_intent_lower for word in ['social', 'facebook', 'linkedin', 'campaign', 'instagram', 'twitter']):
|
||||
relevant_workflows.append(self.workflows["social_media_workflow"])
|
||||
|
||||
# SEO related
|
||||
if any(word in user_intent_lower for word in ['seo', 'optimize', 'rank', 'search', 'audit']):
|
||||
relevant_workflows.append(self.workflows["seo_audit_workflow"])
|
||||
|
||||
# Strategy and planning
|
||||
if any(word in user_intent_lower for word in ['strategy', 'plan', 'roadmap', 'framework']):
|
||||
relevant_workflows.append(self.workflows["content_strategy_workflow"])
|
||||
|
||||
return relevant_workflows
|
||||
|
||||
def get_workflow(self, workflow_id: str) -> Dict[str, Any]:
|
||||
"""Get a specific workflow by ID."""
|
||||
return self.workflows.get(workflow_id)
|
||||
|
||||
def get_all_workflows(self) -> Dict[str, Dict[str, Any]]:
|
||||
"""Get all available workflows."""
|
||||
return self.workflows
|
||||
|
||||
def create_custom_workflow(self, name: str, description: str, steps: List[Dict[str, str]]) -> str:
|
||||
"""Create a custom workflow."""
|
||||
workflow_id = f"custom_{name.lower().replace(' ', '_')}"
|
||||
self.workflows[workflow_id] = {
|
||||
"name": name,
|
||||
"description": description,
|
||||
"steps": steps,
|
||||
"custom": True
|
||||
}
|
||||
return workflow_id
|
||||
|
||||
def get_workflow_progress(self, workflow_id: str, completed_steps: List[str]) -> Dict[str, Any]:
|
||||
"""Get progress information for a workflow."""
|
||||
workflow = self.workflows.get(workflow_id)
|
||||
if not workflow:
|
||||
return {"error": "Workflow not found"}
|
||||
|
||||
total_steps = len(workflow["steps"])
|
||||
completed_count = len(completed_steps)
|
||||
progress_percentage = (completed_count / total_steps) * 100 if total_steps > 0 else 0
|
||||
|
||||
next_step = None
|
||||
if completed_count < total_steps:
|
||||
next_step = workflow["steps"][completed_count]
|
||||
|
||||
return {
|
||||
"workflow_name": workflow["name"],
|
||||
"total_steps": total_steps,
|
||||
"completed_steps": completed_count,
|
||||
"progress_percentage": progress_percentage,
|
||||
"next_step": next_step,
|
||||
"is_complete": completed_count >= total_steps
|
||||
}
|
||||
|
||||
def get_step_details(self, workflow_id: str, step_index: int) -> Dict[str, Any]:
|
||||
"""Get detailed information about a specific workflow step."""
|
||||
workflow = self.workflows.get(workflow_id)
|
||||
if not workflow or step_index >= len(workflow["steps"]):
|
||||
return {"error": "Workflow or step not found"}
|
||||
|
||||
step = workflow["steps"][step_index]
|
||||
|
||||
# Add detailed descriptions for each tool
|
||||
step_descriptions = {
|
||||
"keyword_research": "Research and identify target keywords for your content",
|
||||
"content_gap_analysis": "Analyze competitor content to find opportunities",
|
||||
"blog_writing": "Create high-quality, SEO-optimized blog content",
|
||||
"seo_optimization": "Optimize content for search engines",
|
||||
"meta_generation": "Generate meta titles and descriptions",
|
||||
"competitor_analysis": "Analyze competitor strategies and performance",
|
||||
"content_calendar": "Plan and schedule content publication",
|
||||
"content_ideas": "Generate creative content ideas and topics",
|
||||
"audience_analysis": "Research and define target audience",
|
||||
"content_planning": "Plan content strategy and themes",
|
||||
"social_content_creation": "Create platform-specific social media content",
|
||||
"hashtag_research": "Research relevant hashtags for social media",
|
||||
"technical_seo": "Analyze technical SEO aspects of website",
|
||||
"on_page_seo": "Review and optimize on-page SEO elements"
|
||||
}
|
||||
|
||||
return {
|
||||
"tool": step["tool"],
|
||||
"name": step["name"],
|
||||
"description": step_descriptions.get(step["tool"], "Execute this workflow step"),
|
||||
"step_number": step_index + 1,
|
||||
"total_steps": len(workflow["steps"])
|
||||
}
|
||||
Reference in New Issue
Block a user