File size: 7,759 Bytes
45efbb3 | 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 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | import json
from datetime import datetime, timedelta
from typing import Optional
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
import google.generativeai as genai
from Backend.core.config import settings
from Backend.core.events import event_bus, Event
from Backend.core.logging import get_logger
from Backend.database.models import Issue, IssueEvent, Department, Member, Classification
from Backend.orchestration.base import BaseAgent
logger = get_logger(__name__, agent_name="RoutingAgent")
if settings.gemini_api_key:
genai.configure(api_key=settings.gemini_api_key)
PRIORITY_SLA_HOURS = {
1: 4,
2: 12,
3: 48,
4: 168,
}
class IssueAssigned(Event):
department_code: str
member_id: Optional[UUID] = None
member_name: Optional[str] = None
sla_deadline: datetime
sla_hours: int
class RoutingAgent(BaseAgent):
def __init__(self, db: AsyncSession):
super().__init__("RoutingAgent")
self.db = db
if settings.gemini_api_key:
self.model = genai.GenerativeModel('gemma-3-27b-it')
else:
self.model = None
async def find_department(self, category: Optional[str], description: Optional[str] = None) -> Optional[Department]:
query = select(Department).where(Department.is_active == True)
result = await self.db.execute(query)
departments = result.scalars().all()
if not departments:
return None
if not self.model or not category:
return departments[0]
dept_info = "\n".join([f"- {d.code}: {d.name} ({d.categories})" for d in departments])
prompt = f"""Route civic issue to correct department:
Issue Category: {category}
Description: {description[:150] if description else 'N/A'}
Available Departments:
{dept_info}
Return ONLY the department CODE (e.g., PWD, TRAFFIC, SANITATION)"""
try:
response = self.model.generate_content(prompt)
dept_code = response.text.strip().upper()
for dept in departments:
if dept.code == dept_code:
return dept
except Exception as e:
logger.error(f"Gemini routing failed: {e}")
return departments[0]
async def find_available_member(
self,
department_id: UUID,
city: Optional[str] = None,
locality: Optional[str] = None
) -> Optional[Member]:
base_query = (
select(Member)
.where(Member.department_id == department_id)
.where(Member.is_active == True)
.where(Member.current_workload < Member.max_workload)
)
if city:
city_query = base_query.where(Member.city.ilike(f"%{city}%"))
result = await self.db.execute(city_query.order_by(Member.current_workload.asc()))
member = result.scalars().first()
if member:
logger.info(f"Found member in city: {city}")
return member
if locality:
locality_query = base_query.where(Member.locality.ilike(f"%{locality}%"))
result = await self.db.execute(locality_query.order_by(Member.current_workload.asc()))
member = result.scalars().first()
if member:
logger.info(f"Found member in locality: {locality}")
return member
result = await self.db.execute(base_query.order_by(Member.current_workload.asc()))
member = result.scalars().first()
if member:
logger.info(f"Assigned to available member (no location match)")
return member
def calculate_sla(self, priority: int, department: Optional[Department]) -> tuple[int, datetime]:
base_hours = PRIORITY_SLA_HOURS.get(priority, 48)
if department and department.default_sla_hours:
base_hours = min(base_hours, department.default_sla_hours)
deadline = datetime.utcnow() + timedelta(hours=base_hours)
return base_hours, deadline
async def process_issue(self, issue_id: UUID) -> dict:
query = (
select(Issue)
.options(selectinload(Issue.classification))
.where(Issue.id == issue_id)
)
result = await self.db.execute(query)
issue = result.scalar_one_or_none()
if not issue:
return {"error": "Issue not found"}
if issue.is_duplicate:
self.log_decision(
issue_id=issue_id,
decision="Skipped routing",
reasoning="Issue is a duplicate"
)
return {"skipped": True, "reason": "duplicate"}
category = issue.classification.primary_category if issue.classification else None
priority = issue.priority or 3
department = await self.find_department(category, issue.description)
member = None
if department:
member = await self.find_available_member(
department.id,
city=issue.city,
locality=issue.locality
)
if member:
member.current_workload += 1
sla_hours, sla_deadline = self.calculate_sla(priority, department)
issue.department_id = department.id if department else None
issue.assigned_member_id = member.id if member else None
issue.sla_hours = sla_hours
issue.sla_deadline = sla_deadline
issue.state = "assigned"
dept_code = department.code if department else "UNASSIGNED"
member_name = member.name if member else "Unassigned"
member_city = member.city if member else "N/A"
reasoning = f"Category '{category}' → {dept_code}"
if issue.city:
reasoning += f", Issue location: {issue.city}"
if member:
reasoning += f", Member location: {member_city}"
reasoning += f", SLA: {sla_hours}h"
self.log_decision(
issue_id=issue_id,
decision=f"Routed to {dept_code} → {member_name}",
reasoning=reasoning
)
event_record = IssueEvent(
issue_id=issue_id,
event_type="assigned",
agent_name=self.name,
event_data=json.dumps({
"department_code": dept_code,
"member_id": str(member.id) if member else None,
"member_name": member_name,
"issue_city": issue.city,
"issue_locality": issue.locality,
"member_city": member.city if member else None,
"sla_hours": sla_hours,
"sla_deadline": sla_deadline.isoformat(),
})
)
self.db.add(event_record)
await self.db.flush()
assign_event = IssueAssigned(
issue_id=issue_id,
department_code=dept_code,
member_id=member.id if member else None,
member_name=member_name,
sla_deadline=sla_deadline,
sla_hours=sla_hours,
)
await event_bus.publish(assign_event)
return {
"department": dept_code,
"member": member_name,
"issue_city": issue.city,
"issue_locality": issue.locality,
"sla_hours": sla_hours,
"sla_deadline": sla_deadline.isoformat(),
}
async def handle(self, event) -> None:
await self.process_issue(event.issue_id)
|