open-navigator / agents /classifier.py
jcbowyer's picture
Clean HuggingFace deployment without binary files
61d29fc
"""
Classifier Agent for identifying oral health policy topics in meeting minutes.
"""
import asyncio
from typing import List, Dict, Any, Optional, Set
from datetime import datetime
from loguru import logger
from agents.base import BaseAgent, AgentRole, AgentMessage, MessageType, AgentStatus
from config import settings
class PolicyTopic:
"""Enumeration of oral health policy topics."""
WATER_FLUORIDATION = "water_fluoridation"
SCHOOL_DENTAL_SCREENING = "school_dental_screening"
MEDICAID_DENTAL = "medicaid_dental"
DENTAL_CLINIC_FUNDING = "dental_clinic_funding"
COMMUNITY_DENTAL_PROGRAM = "community_dental_program"
CHILDREN_DENTAL_HEALTH = "children_dental_health"
DENTAL_CARE_ACCESS = "dental_care_access"
OTHER_ORAL_HEALTH = "other_oral_health"
NOT_RELEVANT = "not_relevant"
class ClassifierAgent(BaseAgent):
"""
Agent responsible for classifying documents by oral health policy topics.
Uses a combination of:
- Keyword matching for high-precision identification
- LLM-based classification for nuanced topics
- Topic modeling for discovering new themes
"""
def __init__(self, agent_id: str = "classifier-001"):
"""Initialize the classifier agent."""
super().__init__(agent_id, AgentRole.CLASSIFIER)
self._initialize_keywords()
self.llm_client = None # Will be initialized when needed
def _initialize_keywords(self):
"""Initialize keyword patterns for each topic."""
self.topic_keywords = {
PolicyTopic.WATER_FLUORIDATION: [
"fluoridation", "fluoride", "water fluoridation",
"fluoridated water", "fluoride level", "fluoride treatment",
"community water fluoridation"
],
PolicyTopic.SCHOOL_DENTAL_SCREENING: [
"school dental", "dental screening", "school screening",
"school health screening", "dental exam", "school nurse",
"student dental"
],
PolicyTopic.MEDICAID_DENTAL: [
"medicaid dental", "medicaid", "medicare dental",
"public assistance dental", "low-income dental",
"dental benefits", "dental coverage"
],
PolicyTopic.DENTAL_CLINIC_FUNDING: [
"dental clinic", "community dental clinic",
"dental center", "dental facility", "clinic funding",
"dental services funding"
],
PolicyTopic.COMMUNITY_DENTAL_PROGRAM: [
"community dental", "dental program", "oral health program",
"dental outreach", "mobile dental", "dental van"
],
PolicyTopic.CHILDREN_DENTAL_HEALTH: [
"children's dental", "pediatric dental", "child dental",
"kids dental", "youth dental", "infant oral health"
],
PolicyTopic.DENTAL_CARE_ACCESS: [
"dental access", "access to dental", "dental care",
"oral health access", "dental services", "dental disparities"
]
}
async def process(self, message: AgentMessage) -> List[AgentMessage]:
"""
Process classification commands.
Args:
message: Message containing parsed documents to classify
Returns:
List of messages with classification results
"""
self.update_status(AgentStatus.PROCESSING, "Classifying policy documents")
try:
documents = message.payload.get("documents", [])
# Classify documents in batches
batch_size = settings.classifier_batch_size
classified_documents = []
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
batch_results = await self._classify_batch(batch)
classified_documents.extend(batch_results)
# Filter to only relevant documents
relevant_documents = [
doc for doc in classified_documents
if doc["classification"]["primary_topic"] != PolicyTopic.NOT_RELEVANT
]
# Send classified documents to sentiment analyzer
response = await self.send_message(
AgentRole.SENTIMENT_ANALYZER,
MessageType.DATA,
{
"workflow_id": message.payload.get("workflow_id"),
"documents": relevant_documents,
"count": len(relevant_documents),
"filtered_count": len(documents) - len(relevant_documents)
}
)
self.log_success()
logger.info(
f"Classified {len(documents)} documents, "
f"{len(relevant_documents)} relevant to oral health policy"
)
return [response]
except Exception as e:
self.log_failure(str(e))
error_msg = await self.send_message(
AgentRole.ORCHESTRATOR,
MessageType.ERROR,
{"error": str(e), "agent": self.agent_id}
)
return [error_msg]
async def _classify_batch(
self,
documents: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""
Classify a batch of documents.
Args:
documents: Batch of documents to classify
Returns:
Documents with classification results
"""
tasks = [self._classify_document(doc) for doc in documents]
results = await asyncio.gather(*tasks, return_exceptions=True)
classified = []
for doc, result in zip(documents, results):
if isinstance(result, Exception):
logger.error(f"Classification error for {doc['document_id']}: {result}")
doc["classification"] = {
"primary_topic": PolicyTopic.NOT_RELEVANT,
"error": str(result)
}
else:
doc["classification"] = result
classified.append(doc)
return classified
async def _classify_document(
self,
doc: Dict[str, Any]
) -> Dict[str, Any]:
"""
Classify a single document.
Args:
doc: Document to classify
Returns:
Classification results
"""
text = self._get_searchable_text(doc)
text_lower = text.lower()
# Keyword-based classification
topic_scores = {}
for topic, keywords in self.topic_keywords.items():
score = sum(1 for keyword in keywords if keyword in text_lower)
if score > 0:
topic_scores[topic] = score
# Determine primary topic
if topic_scores:
primary_topic = max(topic_scores, key=topic_scores.get)
confidence = "high" if topic_scores[primary_topic] >= 3 else "medium"
# Get all topics mentioned
all_topics = list(topic_scores.keys())
else:
primary_topic = PolicyTopic.NOT_RELEVANT
confidence = "high"
all_topics = []
# Extract relevant excerpts
excerpts = self._extract_relevant_excerpts(doc, primary_topic)
classification = {
"primary_topic": primary_topic,
"all_topics": all_topics,
"topic_scores": topic_scores,
"confidence": confidence,
"relevant_excerpts": excerpts,
"classified_at": datetime.utcnow().isoformat()
}
return classification
def _get_searchable_text(self, doc: Dict[str, Any]) -> str:
"""Extract searchable text from document."""
parts = [
doc.get("raw_title", ""),
doc.get("full_text", "")
]
# Add agenda items
for item in doc.get("agenda_items", []):
parts.append(item.get("description", ""))
# Add discussion sections
for section in doc.get("discussion_sections", []):
parts.append(section.get("text", ""))
return " ".join(parts)
def _extract_relevant_excerpts(
self,
doc: Dict[str, Any],
topic: str
) -> List[Dict[str, str]]:
"""Extract text excerpts relevant to the topic."""
if topic == PolicyTopic.NOT_RELEVANT:
return []
keywords = self.topic_keywords.get(topic, [])
excerpts = []
# Check discussion sections
for section in doc.get("discussion_sections", []):
text = section.get("text", "")
text_lower = text.lower()
# Check if any keywords present
if any(keyword in text_lower for keyword in keywords):
excerpts.append({
"source": "discussion",
"text": text[:500], # First 500 chars
"section_id": section.get("section_id")
})
# Check agenda items
for item in doc.get("agenda_items", []):
desc = item.get("description", "")
desc_lower = desc.lower()
if any(keyword in desc_lower for keyword in keywords):
excerpts.append({
"source": "agenda",
"text": desc,
"item_number": item.get("number")
})
return excerpts[:5] # Return top 5 excerpts
async def _llm_classify(
self,
text: str,
preliminary_topics: List[str]
) -> Dict[str, Any]:
"""
Use LLM for nuanced classification when keywords are ambiguous.
Args:
text: Text to classify
preliminary_topics: Topics identified by keyword matching
Returns:
LLM classification results
"""
# This would use OpenAI API or similar
# Placeholder for now
return {
"llm_topic": preliminary_topics[0] if preliminary_topics else PolicyTopic.NOT_RELEVANT,
"llm_confidence": 0.8,
"llm_reasoning": "Based on keyword analysis"
}