| """
|
| Task: T026, T085, T086
|
| Spec: 011-chatbot-issues-fixes/spec.md - Agent Runner with proper logging
|
|
|
| Agent runner for processing messages through the AI agent and managing tool executions.
|
| Handles message persistence and conversation state management with proper logging.
|
| """
|
|
|
| import asyncio
|
| import structlog
|
| from typing import Dict, Any, List
|
| from datetime import datetime
|
| from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
| from app.services.conversation_service import ConversationService
|
| from app.services.message_service import MessageService
|
| from app.services.task_service import TaskService
|
| from .ai_agent import AIAgent
|
|
|
|
|
| logger = structlog.get_logger(__name__)
|
|
|
|
|
| class AgentRunner:
|
| """Runs the AI agent and manages the conversation lifecycle."""
|
|
|
| def __init__(self):
|
| """Initialize the agent runner with required services."""
|
| self.ai_agent = AIAgent()
|
|
|
| async def process_conversation(
|
| self,
|
| session: AsyncSession,
|
| user_id: str,
|
| conversation_id: str,
|
| message_content: str
|
| ) -> Dict[str, Any]:
|
| """
|
| Process a conversation with a user message and return the AI response.
|
|
|
| Args:
|
| session: Database session
|
| user_id: ID of the authenticated user
|
| conversation_id: ID of the conversation (None for new conversation)
|
| message_content: Content of the user's message
|
|
|
| Returns:
|
| Dictionary containing the AI response and conversation details
|
| """
|
| try:
|
|
|
| if conversation_id:
|
|
|
| conversation = await ConversationService.get_conversation_by_id(
|
| session, conversation_id, user_id
|
| )
|
| if not conversation:
|
| return {
|
| "error": "Conversation not found or doesn't belong to user",
|
| "status": "error"
|
| }
|
| else:
|
|
|
| conversation = await ConversationService.create_conversation(
|
| session, user_id
|
| )
|
| conversation_id = conversation.id
|
|
|
|
|
| user_message = await MessageService.create_message(
|
| session=session,
|
| user_id=user_id,
|
| conversation_id=conversation_id,
|
| role="user",
|
| content=message_content
|
| )
|
|
|
|
|
| recent_messages = await MessageService.get_latest_messages_by_conversation(
|
| session, conversation_id, user_id, limit=10
|
| )
|
|
|
|
|
| conversation_history = []
|
| for msg in recent_messages[:-1]:
|
| conversation_history.append({
|
| "role": msg.role,
|
| "content": msg.content
|
| })
|
|
|
|
|
| logger.info("Processing message with AI agent", user_id=user_id, message_content=message_content)
|
| ai_response = await self.ai_agent.process_message(
|
| user_id=user_id,
|
| message_content=message_content,
|
| conversation_history=conversation_history,
|
| session=session
|
| )
|
| logger.debug("AI agent response received", response=ai_response)
|
|
|
| if not ai_response["success"]:
|
| logger.warning("AI agent returned failure", response=ai_response)
|
|
|
| await MessageService.create_message(
|
| session=session,
|
| user_id=user_id,
|
| conversation_id=conversation_id,
|
| role="assistant",
|
| content=ai_response["response"]
|
| )
|
| return ai_response
|
|
|
| logger.debug("AI agent successful, checking for tool calls", tool_calls=ai_response.get('tool_calls', []))
|
|
|
|
|
| logger.debug("About to execute tool calls", tool_calls=ai_response.get('tool_calls', []))
|
| tool_responses = []
|
| tool_results = []
|
|
|
| if ai_response.get("tool_calls"):
|
| logger.debug(f"Processing {len(ai_response['tool_calls'])} tool calls")
|
| for tool_call in ai_response["tool_calls"]:
|
| logger.debug(f"Processing tool call", tool_call=tool_call)
|
| try:
|
|
|
| func_name = tool_call["function"]["name"]
|
| import json
|
| args = json.loads(tool_call["function"]["arguments"])
|
| logger.debug(f"Tool function: {func_name}, Args: {args}")
|
|
|
|
|
| if func_name == "add_task":
|
| from app.services.task_service import TaskService
|
| logger.debug(f"Calling TaskService.create_task", user_id=args['user_id'], title=args['title'])
|
| task = await TaskService.create_task(
|
| session=session,
|
| user_id=args["user_id"],
|
| title=args["title"],
|
| description=args.get("description")
|
| )
|
| logger.debug(f"Task created", task_id=task.id, title=task.title)
|
| result = {
|
| "task_id": task.id,
|
| "status": "created",
|
| "title": task.title
|
| }
|
|
|
| tool_results.append(f"Added task: '{task.title}' (ID: {task.id})")
|
|
|
| elif func_name == "list_tasks":
|
| from app.services.task_service import TaskService
|
| logger.debug(f"Calling TaskService.get_user_tasks", user_id=args['user_id'])
|
| tasks = await TaskService.get_user_tasks(
|
| session=session,
|
| user_id=args["user_id"],
|
| status=args.get("status", "all")
|
| )
|
| logger.debug(f"Retrieved tasks", count=len(tasks))
|
| result = [
|
| {
|
| "id": task.id,
|
| "title": task.title,
|
| "completed": task.completed
|
| }
|
| for task in tasks
|
| ]
|
|
|
| if tasks:
|
|
|
| pending_tasks = [task for task in tasks if not task.completed]
|
| completed_tasks = [task for task in tasks if task.completed]
|
|
|
| response_parts = []
|
|
|
| if pending_tasks:
|
| pending_list = "\n".join([f" • {task.title} (ID: {task.id})" for task in pending_tasks])
|
| response_parts.append(f"Pending Tasks ({len(pending_tasks)}):\n{pending_list}")
|
|
|
| if completed_tasks:
|
| completed_list = "\n".join([f" • {task.title} (ID: {task.id})" for task in completed_tasks])
|
| response_parts.append(f"Completed Tasks ({len(completed_tasks)}):\n{completed_list}")
|
|
|
| if response_parts:
|
| tool_results.append("Here are your tasks:\n" + "\n\n".join(response_parts))
|
| else:
|
| tool_results.append("You don't have any tasks yet.")
|
| else:
|
| tool_results.append("You don't have any tasks yet.")
|
|
|
| elif func_name == "complete_task":
|
| from app.services.task_service import TaskService
|
| logger.debug(f"Calling TaskService.complete_task", task_id=args['task_id'])
|
| task = await TaskService.complete_task(
|
| session=session,
|
| task_id=args["task_id"],
|
| user_id=args["user_id"]
|
| )
|
| if task:
|
| logger.debug(f"Task completed", task_id=task.id)
|
| result = {
|
| "task_id": task.id,
|
| "status": "completed",
|
| "title": task.title
|
| }
|
| tool_results.append(f"Completed task: '{task.title}' (ID: {task.id})")
|
| else:
|
| logger.warning(f"Could not complete task", task_id=args['task_id'])
|
| result = {"error": f"Task {args['task_id']} not found or doesn't belong to user"}
|
| tool_results.append(f"Could not complete task: {result['error']}")
|
|
|
| elif func_name == "delete_task":
|
| from app.services.task_service import TaskService
|
| logger.debug(f"Calling TaskService.delete_task", task_id=args['task_id'])
|
| success = await TaskService.delete_task(
|
| session=session,
|
| task_id=args["task_id"],
|
| user_id=args["user_id"]
|
| )
|
| if success:
|
| logger.debug(f"Task deleted", task_id=args['task_id'])
|
| result = {
|
| "task_id": args["task_id"],
|
| "status": "deleted"
|
| }
|
| tool_results.append(f"Deleted task with ID: {args['task_id']}")
|
| else:
|
| logger.warning(f"Could not delete task", task_id=args['task_id'])
|
| result = {"error": f"Task {args['task_id']} not found or doesn't belong to user"}
|
| tool_results.append(f"Could not delete task: {result['error']}")
|
|
|
| elif func_name == "update_task":
|
| from app.services.task_service import TaskService
|
| logger.debug(f"Calling TaskService.update_task", task_id=args['task_id'])
|
| updated_task = await TaskService.update_task(
|
| session=session,
|
| task_id=args["task_id"],
|
| user_id=args["user_id"],
|
| title=args.get("title"),
|
| description=args.get("description")
|
| )
|
| if updated_task:
|
| logger.debug(f"Task updated", task_id=updated_task.id)
|
| result = {
|
| "task_id": updated_task.id,
|
| "status": "updated",
|
| "title": updated_task.title
|
| }
|
| tool_results.append(f"Updated task: '{updated_task.title}' (ID: {updated_task.id})")
|
| else:
|
| logger.warning(f"Could not update task", task_id=args['task_id'])
|
| result = {"error": f"Task {args['task_id']} not found or doesn't belong to user"}
|
| tool_results.append(f"Could not update task: {result['error']}")
|
|
|
| else:
|
| logger.warning(f"Unknown tool function", func_name=func_name)
|
| result = {"error": f"Unknown tool: {func_name}"}
|
| tool_results.append(f"Error: {result['error']}")
|
|
|
| tool_responses.append({
|
| "tool_call_id": tool_call["id"],
|
| "output": result
|
| })
|
| logger.debug(f"Tool call executed successfully", tool_call_id=tool_call["id"])
|
| except Exception as e:
|
| logger.error(f"Failed to execute tool call", tool_call=tool_call, error=str(e))
|
| import traceback
|
| traceback.print_exc()
|
| tool_responses.append({
|
| "tool_call_id": tool_call["id"],
|
| "output": {"error": f"Error executing tool {tool_call['function']['name']}: {str(e)}"}
|
| })
|
| tool_results.append(f"Error executing tool: {str(e)}")
|
| else:
|
| logger.debug("No tool calls to execute")
|
|
|
|
|
| logger.debug("Tool responses prepared", tool_responses=tool_responses)
|
| ai_response["tool_responses"] = tool_responses
|
|
|
|
|
| original_response = ai_response["response"]
|
| if tool_results:
|
|
|
| enhanced_response = original_response + "\n\n" + "\n".join(tool_results)
|
| content_to_save = enhanced_response
|
| else:
|
| content_to_save = original_response
|
|
|
|
|
| ai_message = await MessageService.create_message(
|
| session=session,
|
| user_id=user_id,
|
| conversation_id=conversation_id,
|
| role="assistant",
|
| content=content_to_save,
|
| tool_calls=ai_response.get("tool_calls"),
|
| tool_responses=tool_responses
|
| )
|
|
|
|
|
| await ConversationService.update_conversation_updated_at(
|
| session, conversation_id, user_id
|
| )
|
|
|
|
|
| await MessageService.truncate_old_messages(
|
| session, conversation_id, user_id
|
| )
|
|
|
|
|
| result = {
|
| "conversation_id": conversation_id,
|
| "response": content_to_save,
|
| "tool_calls": ai_response.get("tool_calls", []),
|
| "success": True
|
| }
|
|
|
| return result
|
|
|
| except Exception as e:
|
| return {
|
| "error": f"Error processing conversation: {str(e)}",
|
| "status": "error",
|
| "success": False
|
| }
|
|
|
| async def get_conversation_history(
|
| self,
|
| session: AsyncSession,
|
| conversation_id: str,
|
| user_id: str
|
| ) -> List[Dict[str, Any]]:
|
| """
|
| Retrieve the conversation history for a specific conversation.
|
|
|
| Args:
|
| session: Database session
|
| conversation_id: ID of the conversation to retrieve
|
| user_id: ID of the authenticated user
|
|
|
| Returns:
|
| List of messages in the conversation
|
| """
|
| try:
|
| messages = await MessageService.get_messages_by_conversation(
|
| session, conversation_id, user_id
|
| )
|
|
|
| return [
|
| {
|
| "role": msg.role,
|
| "content": msg.content,
|
| "timestamp": msg.created_at.isoformat() if msg.created_at else None,
|
| "tool_calls": msg.tool_calls,
|
| "tool_responses": msg.tool_responses
|
| }
|
| for msg in messages
|
| ]
|
| except Exception as e:
|
| raise e
|
|
|
| async def list_user_conversations(
|
| self,
|
| session: AsyncSession,
|
| user_id: str
|
| ) -> List[Dict[str, Any]]:
|
| """
|
| List all conversations for a user.
|
|
|
| Args:
|
| session: Database session
|
| user_id: ID of the authenticated user
|
|
|
| Returns:
|
| List of user's conversations
|
| """
|
| try:
|
| conversations = await ConversationService.get_user_conversations(
|
| session, user_id
|
| )
|
|
|
| return [
|
| {
|
| "id": conv.id,
|
| "created_at": conv.created_at.isoformat() if conv.created_at else None,
|
| "updated_at": conv.updated_at.isoformat() if conv.updated_at else None
|
| }
|
| for conv in conversations
|
| ]
|
| except Exception as e:
|
| raise e |
