more_dots/workflows/workflow_manager.py

from typing import Dict, Any, Optional, List
from enum import Enum
from agent.conversation import ConversationAgent
from agent.tool import ToolAgent


class WorkflowType(Enum):
    """可用的工作流类型"""
    CONVERSATION = "conversation"
    TOOL_USING = "tool_using"
    

class WorkflowManager:
    """管理不同工作流类型及其执行"""
    
    def __init__(self, default_model_section: Optional[str] = None):
        self.workflows = {
            WorkflowType.CONVERSATION: ConversationAgent(model_section=default_model_section),
            WorkflowType.TOOL_USING: ToolAgent(model_section=default_model_section)
        }
        self.active_sessions: Dict[str, Any] = {}
    
    def get_workflow(self, workflow_type: WorkflowType):
        """获取工作流实例"""
        return self.workflows.get(workflow_type)
    
    def execute_workflow(self, workflow_type: WorkflowType, user_input: str, 
                        session_id: Optional[str] = None, **kwargs) -> Dict[str, Any]:
        """执行指定工作流"""
        workflow = self.get_workflow(workflow_type)
        
        if not workflow:
            return {"error": f"Workflow {workflow_type.value} not found"}
        
        # 未提供会话 ID 时生成
        if not session_id:
            session_id = f"session_{len(self.active_sessions) + 1}"
        
        # 执行工作流
        result = workflow.run(user_input, **kwargs)
        
        # 存储会话数据
        self.active_sessions[session_id] = {
            "workflow_type": workflow_type,
            "last_result": result,
            "timestamp": self._get_timestamp()
        }
        
        return {
            "session_id": session_id,
            "workflow_type": workflow_type.value,
            "result": result
        }
    
    def get_available_workflows(self) -> List[str]:
        """获取可用工作流列表"""
        return [workflow.value for workflow in WorkflowType]
    
    def _get_timestamp(self) -> str:
        """获取当前时间戳"""
        from datetime import datetime
        return datetime.now().isoformat()
    
    def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]:
        """获取会话信息"""
        return self.active_sessions.get(session_id)
    
    def cleanup_sessions(self, older_than_hours: int = 24):
        """清理过期会话"""
        from datetime import datetime, timedelta
        
        cutoff_time = datetime.now() - timedelta(hours=older_than_hours)
        
        sessions_to_remove = []
        for session_id, session_data in self.active_sessions.items():
            session_time = datetime.fromisoformat(session_data["timestamp"])
            if session_time < cutoff_time:
                sessions_to_remove.append(session_id)
        
        for session_id in sessions_to_remove:
            del self.active_sessions[session_id]
        
        return len(sessions_to_remove)