# AI Workflow Agent - Chat API """ Conversational interface for the AI Workflow Agent. Supports multi-turn conversations with session management. """ import uuid import json import logging from datetime import datetime from typing import Dict, Any, Optional, List from dataclasses import dataclass, field, asdict from enum import Enum logger = logging.getLogger(__name__) class MessageRole(Enum): """Message roles in conversation.""" USER = "user" ASSISTANT = "assistant" SYSTEM = "system" class ConversationState(Enum): """Current state of conversation.""" INITIAL = "initial" ANALYZING = "analyzing" CLARIFYING = "clarifying" PLANNING = "planning" BUILDING = "building" COMPLETE = "complete" ERROR = "error" @dataclass class Message: """Single message in conversation.""" role: str content: str timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class Session: """Conversation session.""" session_id: str created_at: str state: str = ConversationState.INITIAL.value messages: List[Dict[str, Any]] = field(default_factory=list) context: Dict[str, Any] = field(default_factory=dict) project_type: Optional[str] = None workflow: Optional[Dict[str, Any]] = None pending_questions: List[str] = field(default_factory=list) def add_message(self, role: str, content: str, metadata: Dict = None): """Add a message to the conversation.""" self.messages.append({ "role": role, "content": content, "timestamp": datetime.now().isoformat(), "metadata": metadata or {} }) def get_history_text(self, limit: int = 10) -> str: """Get conversation history as text for LLM context.""" recent = self.messages[-limit:] lines = [] for msg in recent: role = msg["role"].upper() content = msg["content"] lines.append(f"{role}: {content}") return "\n".join(lines) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary.""" return asdict(self) class SessionManager: """Manages conversation sessions.""" def __init__(self, max_sessions: int = 100): self.sessions: Dict[str, Session] = {} self.max_sessions = max_sessions def create_session(self) -> Session: """Create a new conversation session.""" # Cleanup old sessions if limit reached if len(self.sessions) >= self.max_sessions: self._cleanup_old_sessions() session_id = str(uuid.uuid4())[:8] session = Session( session_id=session_id, created_at=datetime.now().isoformat() ) # Add system message session.add_message( MessageRole.SYSTEM.value, "AI Workflow Agent initialized. Ready to help build n8n, ComfyUI, " "or hybrid workflows. Describe what you want to create." ) self.sessions[session_id] = session logger.info(f"Created session: {session_id}") return session def get_session(self, session_id: str) -> Optional[Session]: """Get existing session by ID.""" return self.sessions.get(session_id) def get_or_create(self, session_id: Optional[str] = None) -> Session: """Get existing session or create new one.""" if session_id and session_id in self.sessions: return self.sessions[session_id] return self.create_session() def update_state(self, session_id: str, state: ConversationState): """Update session state.""" if session_id in self.sessions: self.sessions[session_id].state = state.value def delete_session(self, session_id: str) -> bool: """Delete a session.""" if session_id in self.sessions: del self.sessions[session_id] logger.info(f"Deleted session: {session_id}") return True return False def list_sessions(self) -> List[Dict[str, Any]]: """List all active sessions.""" return [ { "session_id": s.session_id, "created_at": s.created_at, "state": s.state, "message_count": len(s.messages), "project_type": s.project_type } for s in self.sessions.values() ] def _cleanup_old_sessions(self): """Remove oldest sessions to make room.""" if not self.sessions: return # Sort by creation time and remove oldest 20% sorted_sessions = sorted( self.sessions.items(), key=lambda x: x[1].created_at ) to_remove = len(sorted_sessions) // 5 for session_id, _ in sorted_sessions[:to_remove]: del self.sessions[session_id] logger.info(f"Cleaned up {to_remove} old sessions") class ChatHandler: """Handles chat interactions with the agent system.""" def __init__(self): self.session_manager = SessionManager() self._agent_system = None # Lazy load @property def agent_system(self): """Lazy load agent system to avoid circular imports.""" if self._agent_system is None: from crew_agents import crew_agent_system self._agent_system = crew_agent_system return self._agent_system async def chat( self, message: str, session_id: Optional[str] = None ) -> Dict[str, Any]: """ Process a chat message and return response. Args: message: User message session_id: Optional existing session ID Returns: Dict with response, session_id, state, and optionally questions/workflow """ # Get or create session session = self.session_manager.get_or_create(session_id) # Add user message session.add_message(MessageRole.USER.value, message) try: # Handle based on current state if session.state == ConversationState.CLARIFYING.value: # User is answering clarifying questions return await self._handle_clarification(session, message) else: # New request or continuation return await self._handle_request(session, message) except Exception as e: logger.error(f"Chat error: {e}") session.state = ConversationState.ERROR.value session.add_message( MessageRole.ASSISTANT.value, f"Sorry, I encountered an error: {str(e)}. Please try again." ) return { "success": False, "session_id": session.session_id, "response": f"Error: {str(e)}", "state": session.state } async def _handle_request(self, session: Session, message: str) -> Dict[str, Any]: """Handle a new or continuing request.""" session.state = ConversationState.ANALYZING.value # Analyze the request analysis = await self.agent_system.analyze_request( query=message, session_id=session.session_id, context={"history": session.get_history_text()} ) if not analysis.get("success"): error_msg = analysis.get("error", "Analysis failed") session.add_message(MessageRole.ASSISTANT.value, f"Error: {error_msg}") return { "success": False, "session_id": session.session_id, "response": error_msg, "state": session.state } # Check if clarification needed if analysis.get("needs_clarification") and analysis.get("confidence", 0) < 0.7: session.state = ConversationState.CLARIFYING.value questions = analysis.get("questions", []) session.pending_questions = questions # Build response with questions response_parts = [analysis.get("analysis", "I need some clarification:")] for i, q in enumerate(questions, 1): response_parts.append(f"\n{i}. {q}") response = "\n".join(response_parts) session.add_message(MessageRole.ASSISTANT.value, response) return { "success": True, "session_id": session.session_id, "response": response, "state": session.state, "needs_clarification": True, "questions": questions, "project_type": analysis.get("project_type") } # Proceed to build return await self._build_workflow(session, analysis) async def _handle_clarification(self, session: Session, answer: str) -> Dict[str, Any]: """Handle user's answer to clarifying questions.""" # Store the clarification if session.pending_questions: question = session.pending_questions[0] self.agent_system.add_clarification( session.session_id, question, answer ) session.pending_questions = session.pending_questions[1:] # If more questions pending, ask next one if session.pending_questions: next_question = session.pending_questions[0] response = f"Thanks! Next question: {next_question}" session.add_message(MessageRole.ASSISTANT.value, response) return { "success": True, "session_id": session.session_id, "response": response, "state": session.state, "needs_clarification": True, "questions": session.pending_questions } # All questions answered, proceed to build session.add_message( MessageRole.ASSISTANT.value, "Great, I have all the information I need. Building your workflow..." ) # Re-analyze with new information conv_context = self.agent_system.get_session(session.session_id) if conv_context: analysis = { "project_type": conv_context.project_type, "confidence": 0.9, "requirements": conv_context.requirements } return await self._build_workflow(session, analysis) return await self._handle_request(session, session.messages[-2]["content"]) async def _build_workflow(self, session: Session, analysis: Dict[str, Any]) -> Dict[str, Any]: """Build the workflow based on analysis.""" session.state = ConversationState.PLANNING.value session.project_type = analysis.get("project_type") # Use the simple builders for reliability (CrewAI for complex cases) from tools.n8n_builder import N8NWorkflowBuilder from tools.comfyui_builder import ComfyUIWorkflowBuilder from tools.github_search import GitHubSearchTool project_type = analysis.get("project_type", "unknown") original_query = session.messages[1]["content"] if len(session.messages) > 1 else "" session.state = ConversationState.BUILDING.value try: if project_type == "n8n": builder = N8NWorkflowBuilder() workflow = await builder.generate_workflow(original_query) response = "I've generated an n8n workflow for you. Here's the configuration:" elif project_type == "comfyui": builder = ComfyUIWorkflowBuilder() workflow = await builder.generate_workflow(original_query) response = "I've generated a ComfyUI workflow. Here's the configuration:" elif project_type == "hybrid": n8n_builder = N8NWorkflowBuilder() comfyui_builder = ComfyUIWorkflowBuilder() n8n_wf = await n8n_builder.generate_workflow(original_query) comfyui_wf = await comfyui_builder.generate_workflow(original_query) workflow = { "type": "hybrid", "n8n_workflow": n8n_wf, "comfyui_workflow": comfyui_wf, "integration_note": "n8n can call ComfyUI via HTTP Request node to /prompt endpoint" } response = "I've generated a hybrid workflow combining n8n automation with ComfyUI for AI generation." elif project_type == "external_repo": github = GitHubSearchTool() repos = await github.search(original_query, max_results=3) recommendation = await github.generate_recommendation(repos) workflow = { "type": "external_repo", "repositories": repos, "recommendation": recommendation } response = f"I found some relevant repositories:\n\n{recommendation}" else: workflow = None response = "I couldn't determine the project type. Could you provide more details?" session.workflow = workflow session.state = ConversationState.COMPLETE.value if workflow and project_type not in ["external_repo"]: response += f"\n\n```json\n{json.dumps(workflow, indent=2)[:2000]}\n```" session.add_message(MessageRole.ASSISTANT.value, response[:500] + "..." if len(response) > 500 else response) return { "success": True, "session_id": session.session_id, "response": response, "state": session.state, "project_type": project_type, "workflow": workflow } except Exception as e: logger.error(f"Build error: {e}") session.state = ConversationState.ERROR.value response = f"Error building workflow: {str(e)}" session.add_message(MessageRole.ASSISTANT.value, response) return { "success": False, "session_id": session.session_id, "response": response, "state": session.state, "error": str(e) } def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]: """Get session information.""" session = self.session_manager.get_session(session_id) if session: return session.to_dict() return None def list_sessions(self) -> List[Dict[str, Any]]: """List all sessions.""" return self.session_manager.list_sessions() def clear_session(self, session_id: str) -> bool: """Clear a session.""" # Also clear from agent system if hasattr(self, '_agent_system') and self._agent_system: self._agent_system.clear_session(session_id) return self.session_manager.delete_session(session_id) # Singleton instance chat_handler = ChatHandler()