Hamza4100/ai-workflow-agent
0
1# AI Workflow Agent - Chat API
2"""
3Conversational interface for the AI Workflow Agent.
4Supports multi-turn conversations with session management.
5"""
6
7import uuid
8import json
9import logging
10from datetime import datetime
11from typing import Dict, Any, Optional, List
12from dataclasses import dataclass, field, asdict
13from enum import Enum
14
15logger = logging.getLogger(__name__)
16
17
18class MessageRole(Enum):
19 """Message roles in conversation."""
20 USER = "user"
21 ASSISTANT = "assistant"
22 SYSTEM = "system"
23
24
25class ConversationState(Enum):
26 """Current state of conversation."""
27 INITIAL = "initial"
28 ANALYZING = "analyzing"
29 CLARIFYING = "clarifying"
30 PLANNING = "planning"
31 BUILDING = "building"
32 COMPLETE = "complete"
33 ERROR = "error"
34
35
36@dataclass
37class Message:
38 """Single message in conversation."""
39 role: str
40 content: str
41 timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
42 metadata: Dict[str, Any] = field(default_factory=dict)
43
44
45@dataclass
46class Session:
47 """Conversation session."""
48 session_id: str
49 created_at: str
50 state: str = ConversationState.INITIAL.value
51 messages: List[Dict[str, Any]] = field(default_factory=list)
52 context: Dict[str, Any] = field(default_factory=dict)
53 project_type: Optional[str] = None
54 workflow: Optional[Dict[str, Any]] = None
55 pending_questions: List[str] = field(default_factory=list)
56
57 def add_message(self, role: str, content: str, metadata: Dict = None):
58 """Add a message to the conversation."""
59 self.messages.append({
60 "role": role,
61 "content": content,
62 "timestamp": datetime.now().isoformat(),
63 "metadata": metadata or {}
64 })
65
66 def get_history_text(self, limit: int = 10) -> str:
67 """Get conversation history as text for LLM context."""
68 recent = self.messages[-limit:]
69 lines = []
70 for msg in recent:
71 role = msg["role"].upper()
72 content = msg["content"]
73 lines.append(f"{role}: {content}")
74 return "\n".join(lines)
75
76 def to_dict(self) -> Dict[str, Any]:
77 """Convert to dictionary."""
78 return asdict(self)
79
80
81class SessionManager:
82 """Manages conversation sessions."""
83
84 def __init__(self, max_sessions: int = 100):
85 self.sessions: Dict[str, Session] = {}
86 self.max_sessions = max_sessions
87
88 def create_session(self) -> Session:
89 """Create a new conversation session."""
90 # Cleanup old sessions if limit reached
91 if len(self.sessions) >= self.max_sessions:
92 self._cleanup_old_sessions()
93
94 session_id = str(uuid.uuid4())[:8]
95 session = Session(
96 session_id=session_id,
97 created_at=datetime.now().isoformat()
98 )
99
100 # Add system message
101 session.add_message(
102 MessageRole.SYSTEM.value,
103 "AI Workflow Agent initialized. Ready to help build n8n, ComfyUI, "
104 "or hybrid workflows. Describe what you want to create."
105 )
106
107 self.sessions[session_id] = session
108 logger.info(f"Created session: {session_id}")
109 return session
110
111 def get_session(self, session_id: str) -> Optional[Session]:
112 """Get existing session by ID."""
113 return self.sessions.get(session_id)
114
115 def get_or_create(self, session_id: Optional[str] = None) -> Session:
116 """Get existing session or create new one."""
117 if session_id and session_id in self.sessions:
118 return self.sessions[session_id]
119 return self.create_session()
120
121 def update_state(self, session_id: str, state: ConversationState):
122 """Update session state."""
123 if session_id in self.sessions:
124 self.sessions[session_id].state = state.value
125
126 def delete_session(self, session_id: str) -> bool:
127 """Delete a session."""
128 if session_id in self.sessions:
129 del self.sessions[session_id]
130 logger.info(f"Deleted session: {session_id}")
131 return True
132 return False
133
134 def list_sessions(self) -> List[Dict[str, Any]]:
135 """List all active sessions."""
136 return [
137 {
138 "session_id": s.session_id,
139 "created_at": s.created_at,
140 "state": s.state,
141 "message_count": len(s.messages),
142 "project_type": s.project_type
143 }
144 for s in self.sessions.values()
145 ]
146
147 def _cleanup_old_sessions(self):
148 """Remove oldest sessions to make room."""
149 if not self.sessions:
150 return
151
152 # Sort by creation time and remove oldest 20%
153 sorted_sessions = sorted(
154 self.sessions.items(),
155 key=lambda x: x[1].created_at
156 )
157
158 to_remove = len(sorted_sessions) // 5
159 for session_id, _ in sorted_sessions[:to_remove]:
160 del self.sessions[session_id]
161
162 logger.info(f"Cleaned up {to_remove} old sessions")
163
164
165class ChatHandler:
166 """Handles chat interactions with the agent system."""
167
168 def __init__(self):
169 self.session_manager = SessionManager()
170 self._agent_system = None # Lazy load
171
172 @property
173 def agent_system(self):
174 """Lazy load agent system to avoid circular imports."""
175 if self._agent_system is None:
176 from crew_agents import crew_agent_system
177 self._agent_system = crew_agent_system
178 return self._agent_system
179
180 async def chat(
181 self,
182 message: str,
183 session_id: Optional[str] = None
184 ) -> Dict[str, Any]:
185 """
186 Process a chat message and return response.
187
188 Args:
189 message: User message
190 session_id: Optional existing session ID
191
192 Returns:
193 Dict with response, session_id, state, and optionally questions/workflow
194 """
195 # Get or create session
196 session = self.session_manager.get_or_create(session_id)
197
198 # Add user message
199 session.add_message(MessageRole.USER.value, message)
200
201 try:
202 # Handle based on current state
203 if session.state == ConversationState.CLARIFYING.value:
204 # User is answering clarifying questions
205 return await self._handle_clarification(session, message)
206 else:
207 # New request or continuation
208 return await self._handle_request(session, message)
209
210 except Exception as e:
211 logger.error(f"Chat error: {e}")
212 session.state = ConversationState.ERROR.value
213 session.add_message(
214 MessageRole.ASSISTANT.value,
215 f"Sorry, I encountered an error: {str(e)}. Please try again."
216 )
217 return {
218 "success": False,
219 "session_id": session.session_id,
220 "response": f"Error: {str(e)}",
221 "state": session.state
222 }
223
224 async def _handle_request(self, session: Session, message: str) -> Dict[str, Any]:
225 """Handle a new or continuing request."""
226 session.state = ConversationState.ANALYZING.value
227
228 # Analyze the request
229 analysis = await self.agent_system.analyze_request(
230 query=message,
231 session_id=session.session_id,
232 context={"history": session.get_history_text()}
233 )
234
235 if not analysis.get("success"):
236 error_msg = analysis.get("error", "Analysis failed")
237 session.add_message(MessageRole.ASSISTANT.value, f"Error: {error_msg}")
238 return {
239 "success": False,
240 "session_id": session.session_id,
241 "response": error_msg,
242 "state": session.state
243 }
244
245 # Check if clarification needed
246 if analysis.get("needs_clarification") and analysis.get("confidence", 0) < 0.7:
247 session.state = ConversationState.CLARIFYING.value
248 questions = analysis.get("questions", [])
249 session.pending_questions = questions
250
251 # Build response with questions
252 response_parts = [analysis.get("analysis", "I need some clarification:")]
253 for i, q in enumerate(questions, 1):
254 response_parts.append(f"\n{i}. {q}")
255
256 response = "\n".join(response_parts)
257 session.add_message(MessageRole.ASSISTANT.value, response)
258
259 return {
260 "success": True,
261 "session_id": session.session_id,
262 "response": response,
263 "state": session.state,
264 "needs_clarification": True,
265 "questions": questions,
266 "project_type": analysis.get("project_type")
267 }
268
269 # Proceed to build
270 return await self._build_workflow(session, analysis)
271
272 async def _handle_clarification(self, session: Session, answer: str) -> Dict[str, Any]:
273 """Handle user's answer to clarifying questions."""
274 # Store the clarification
275 if session.pending_questions:
276 question = session.pending_questions[0]
277 self.agent_system.add_clarification(
278 session.session_id,
279 question,
280 answer
281 )
282 session.pending_questions = session.pending_questions[1:]
283
284 # If more questions pending, ask next one
285 if session.pending_questions:
286 next_question = session.pending_questions[0]
287 response = f"Thanks! Next question: {next_question}"
288 session.add_message(MessageRole.ASSISTANT.value, response)
289
290 return {
291 "success": True,
292 "session_id": session.session_id,
293 "response": response,
294 "state": session.state,
295 "needs_clarification": True,
296 "questions": session.pending_questions
297 }
298
299 # All questions answered, proceed to build
300 session.add_message(
301 MessageRole.ASSISTANT.value,
302 "Great, I have all the information I need. Building your workflow..."
303 )
304
305 # Re-analyze with new information
306 conv_context = self.agent_system.get_session(session.session_id)
307 if conv_context:
308 analysis = {
309 "project_type": conv_context.project_type,
310 "confidence": 0.9,
311 "requirements": conv_context.requirements
312 }
313 return await self._build_workflow(session, analysis)
314
315 return await self._handle_request(session, session.messages[-2]["content"])
316
317 async def _build_workflow(self, session: Session, analysis: Dict[str, Any]) -> Dict[str, Any]:
318 """Build the workflow based on analysis."""
319 session.state = ConversationState.PLANNING.value
320 session.project_type = analysis.get("project_type")
321
322 # Use the simple builders for reliability (CrewAI for complex cases)
323 from tools.n8n_builder import N8NWorkflowBuilder
324 from tools.comfyui_builder import ComfyUIWorkflowBuilder
325 from tools.github_search import GitHubSearchTool
326
327 project_type = analysis.get("project_type", "unknown")
328 original_query = session.messages[1]["content"] if len(session.messages) > 1 else ""
329
330 session.state = ConversationState.BUILDING.value
331
332 try:
333 if project_type == "n8n":
334 builder = N8NWorkflowBuilder()
335 workflow = await builder.generate_workflow(original_query)
336 response = "I've generated an n8n workflow for you. Here's the configuration:"
337
338 elif project_type == "comfyui":
339 builder = ComfyUIWorkflowBuilder()
340 workflow = await builder.generate_workflow(original_query)
341 response = "I've generated a ComfyUI workflow. Here's the configuration:"
342
343 elif project_type == "hybrid":
344 n8n_builder = N8NWorkflowBuilder()
345 comfyui_builder = ComfyUIWorkflowBuilder()
346
347 n8n_wf = await n8n_builder.generate_workflow(original_query)
348 comfyui_wf = await comfyui_builder.generate_workflow(original_query)
349
350 workflow = {
351 "type": "hybrid",
352 "n8n_workflow": n8n_wf,
353 "comfyui_workflow": comfyui_wf,
354 "integration_note": "n8n can call ComfyUI via HTTP Request node to /prompt endpoint"
355 }
356 response = "I've generated a hybrid workflow combining n8n automation with ComfyUI for AI generation."
357
358 elif project_type == "external_repo":
359 github = GitHubSearchTool()
360 repos = await github.search(original_query, max_results=3)
361 recommendation = await github.generate_recommendation(repos)
362
363 workflow = {
364 "type": "external_repo",
365 "repositories": repos,
366 "recommendation": recommendation
367 }
368 response = f"I found some relevant repositories:\n\n{recommendation}"
369
370 else:
371 workflow = None
372 response = "I couldn't determine the project type. Could you provide more details?"
373
374 session.workflow = workflow
375 session.state = ConversationState.COMPLETE.value
376
377 if workflow and project_type not in ["external_repo"]:
378 response += f"\n\n```json\n{json.dumps(workflow, indent=2)[:2000]}\n```"
379
380 session.add_message(MessageRole.ASSISTANT.value, response[:500] + "..." if len(response) > 500 else response)
381
382 return {
383 "success": True,
384 "session_id": session.session_id,
385 "response": response,
386 "state": session.state,
387 "project_type": project_type,
388 "workflow": workflow
389 }
390
391 except Exception as e:
392 logger.error(f"Build error: {e}")
393 session.state = ConversationState.ERROR.value
394 response = f"Error building workflow: {str(e)}"
395 session.add_message(MessageRole.ASSISTANT.value, response)
396
397 return {
398 "success": False,
399 "session_id": session.session_id,
400 "response": response,
401 "state": session.state,
402 "error": str(e)
403 }
404
405 def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]:
406 """Get session information."""
407 session = self.session_manager.get_session(session_id)
408 if session:
409 return session.to_dict()
410 return None
411
412 def list_sessions(self) -> List[Dict[str, Any]]:
413 """List all sessions."""
414 return self.session_manager.list_sessions()
415
416 def clear_session(self, session_id: str) -> bool:
417 """Clear a session."""
418 # Also clear from agent system
419 if hasattr(self, '_agent_system') and self._agent_system:
420 self._agent_system.clear_session(session_id)
421 return self.session_manager.delete_session(session_id)
422
423
424# Singleton instance
425chat_handler = ChatHandler()
426