File size: 5,172 Bytes
a9b7eae c5ab8f7 9f94f09 a9b7eae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """
Agent handler for LockIn AI.
Main entry point for agent execution - the run_agent_handler() function.
"""
import time
import uuid
from typing import Dict, Any
from app.agent.intent_router import intent_router
from app.agent.agent_service import agent_service
from app.guardrails import input_guardrails, profile_guardrails, output_guardrails
from app.services.profile_service import profile_service
from app.models.enums import RequestStatus, Intent
from app.schemas.chat import ChatResponse
def run_agent_handler(user_id: str, message: str) -> ChatResponse:
"""
Main handler function for agent execution.
This is the required handler function that orchestrates the entire
agent pipeline from input validation to response generation.
Args:
user_id: User identifier
message: User message
Returns:
ChatResponse with result or error
"""
start_time = time.time()
request_id = f"req_{uuid.uuid4().hex[:12]}"
# Step 1: Input Guardrails
is_valid, guardrail_code, user_message = input_guardrails.validate(message)
if not is_valid:
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.BLOCKED,
response=user_message,
guardrail_triggered=guardrail_code,
latency_ms=latency_ms
)
# Step 2: Profile Guardrails
profile, missing_fields = profile_guardrails.get_profile_or_error(user_id)
if not profile:
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.PROFILE_REQUIRED,
missing_fields=missing_fields,
latency_ms=latency_ms
)
# Step 3: Intent Classification
intent = intent_router.classify(message)
# Step 4: Agent Execution
try:
agent_result = agent_service.run(
message=message,
profile=profile,
intent=intent,
user_id=user_id
)
# Check if agent returned an error
if agent_result.get('error'):
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.ERROR,
response=agent_result.get('response', 'An error occurred'),
latency_ms=latency_ms
)
response_text = agent_result.get('response', '')
tool_calls = agent_result.get('tool_calls', [])
tool_results = agent_result.get('tool_results', [])
# Step 5: Output Guardrails
is_valid_output, output_error = output_guardrails.validate(
response=response_text,
tool_results=tool_results,
intent=intent,
allow_profile_numbers=profile is not None
)
if not is_valid_output:
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.BLOCKED,
response="I can't provide unsafe or medical advice.",
guardrail_triggered="output_validation_failed",
latency_ms=latency_ms
)
# Clean response to remove internal reflection text
cleaned_response = output_guardrails.clean_response(response_text)
# Extract structured data from tool results for meal plans
structured_data = None
if intent == Intent.MEAL_PLAN and tool_results:
for result in tool_results:
if result.get('tool_name') == 'daily_planner' and result.get('success'):
structured_data = result.get('result')
break
# Step 6: Build Response
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.SUCCESS,
intent=intent,
response=cleaned_response,
data=structured_data,
latency_ms=latency_ms,
tool_calls=tool_calls if tool_calls else None
)
except Exception as e:
latency_ms = int((time.time() - start_time) * 1000)
return ChatResponse(
request_id=request_id,
status=RequestStatus.ERROR,
response=f"An error occurred: {str(e)}",
latency_ms=latency_ms
)
def get_handler_metadata() -> Dict[str, Any]:
"""
Get metadata about the handler.
Returns:
Dict with handler information
"""
return {
'handler_name': 'run_agent_handler',
'version': '1.0.0',
'description': 'Main agent execution handler with full pipeline',
'pipeline_steps': [
'Input Guardrails',
'Profile Guardrails',
'Intent Classification',
'Agent Execution',
'Output Guardrails',
'Response Building'
]
}
|