File size: 7,659 Bytes
42d88ae | 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 224 225 226 | import json
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, IssueClassified, Event
from Backend.core.logging import get_logger
from Backend.database.models import Issue, IssueEvent, Classification
from Backend.utils.geo import haversine_distance, get_bounding_box
from Backend.orchestration.base import BaseAgent
logger = get_logger(__name__, agent_name="GeoDeduplicateAgent")
if settings.gemini_api_key:
genai.configure(api_key=settings.gemini_api_key)
class IssueDeduplicated(Event):
is_duplicate: bool
parent_issue_id: Optional[UUID] = None
cluster_id: Optional[str] = None
nearby_count: int = 0
class GeoDeduplicateAgent(BaseAgent):
def __init__(self, db: AsyncSession):
super().__init__("GeoDeduplicateAgent")
self.db = db
self.radius_meters = settings.duplicate_radius_meters
if settings.gemini_api_key:
self.model = genai.GenerativeModel('gemma-3-27b-it')
else:
self.model = None
async def semantic_similarity(self, desc1: str, desc2: str, cat1: str, cat2: str) -> float:
if not self.model:
return 0.5
prompt = f"""Rate semantic similarity (0.0-1.0) between civic issue reports:
Issue A:
Category: {cat1}
Description: {desc1[:200] if desc1 else 'N/A'}
Issue B:
Category: {cat2}
Description: {desc2[:200] if desc2 else 'N/A'}
Consider:
- Same problem type?
- Same physical location context?
- Same infrastructure element?
Return ONLY a decimal number between 0.0 and 1.0."""
try:
response = self.model.generate_content(prompt)
score = float(response.text.strip())
return max(0.0, min(1.0, score))
except Exception as e:
logger.error(f"Gemini similarity failed: {e}")
return 0.5
async def find_nearby_issues(
self,
latitude: float,
longitude: float,
exclude_id: UUID,
category: Optional[str] = None
) -> list[tuple[Issue, float]]:
min_lat, max_lat, min_lon, max_lon = get_bounding_box(
latitude, longitude, self.radius_meters
)
query = (
select(Issue)
.options(selectinload(Issue.classification))
.where(Issue.latitude >= min_lat)
.where(Issue.latitude <= max_lat)
.where(Issue.longitude >= min_lon)
.where(Issue.longitude <= max_lon)
.where(Issue.id != exclude_id)
.where(Issue.state.in_(["reported", "validated", "assigned", "in_progress"]))
.where(Issue.is_duplicate == False)
)
result = await self.db.execute(query)
candidates = result.scalars().all()
nearby = []
for issue in candidates:
distance = haversine_distance(
latitude, longitude,
issue.latitude, issue.longitude
)
if distance <= self.radius_meters:
if category and issue.classification:
if issue.classification.primary_category == category:
nearby.append((issue, distance))
else:
nearby.append((issue, distance))
return sorted(nearby, key=lambda x: x[1])
async def check_duplicate(
self,
issue_id: UUID,
latitude: float,
longitude: float,
category: Optional[str] = None,
description: Optional[str] = None
) -> tuple[bool, Optional[UUID], list[tuple[Issue, float]]]:
nearby = await self.find_nearby_issues(
latitude, longitude, issue_id, category
)
if not nearby:
return False, None, []
best_match = None
highest_score = 0.0
for issue, distance in nearby:
if issue.classification and category:
cat1 = category
cat2 = issue.classification.primary_category
desc1 = description or ""
desc2 = issue.description or ""
similarity = await self.semantic_similarity(desc1, desc2, cat1, cat2)
if similarity > highest_score:
highest_score = similarity
best_match = issue
if highest_score > 0.75 and best_match:
return True, best_match.id, nearby
return False, None, nearby
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"}
category = None
if issue.classification:
category = issue.classification.primary_category
is_duplicate, parent_id, nearby = await self.check_duplicate(
issue.id,
issue.latitude,
issue.longitude,
category,
issue.description
)
if is_duplicate and parent_id:
issue.is_duplicate = True
issue.parent_issue_id = parent_id
issue.geo_status = "duplicate"
issue.geo_cluster_id = str(parent_id)
parent = await self.db.get(Issue, parent_id)
if parent and issue.priority and parent.priority:
if issue.priority < parent.priority:
parent.priority = issue.priority
self.log_decision(
issue_id=issue_id,
decision="Marked as duplicate",
reasoning=f"Found {len(nearby)} nearby issues within {self.radius_meters}m, linked to parent {parent_id}"
)
else:
issue.is_duplicate = False
issue.geo_status = "unique"
self.log_decision(
issue_id=issue_id,
decision="Marked as unique",
reasoning=f"No similar issues found within {self.radius_meters}m radius"
)
event_record = IssueEvent(
issue_id=issue_id,
event_type="geo_deduplicated",
agent_name=self.name,
event_data=json.dumps({
"is_duplicate": is_duplicate,
"parent_issue_id": str(parent_id) if parent_id else None,
"nearby_count": len(nearby),
"radius_meters": self.radius_meters,
})
)
self.db.add(event_record)
await self.db.flush()
dedup_event = IssueDeduplicated(
issue_id=issue_id,
is_duplicate=is_duplicate,
parent_issue_id=parent_id,
cluster_id=str(parent_id) if parent_id else None,
nearby_count=len(nearby),
)
await event_bus.publish(dedup_event)
return {
"is_duplicate": is_duplicate,
"parent_issue_id": str(parent_id) if parent_id else None,
"nearby_count": len(nearby),
"geo_status": issue.geo_status,
}
async def handle(self, event: IssueClassified) -> None:
await self.process_issue(event.issue_id)
|