Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 7,447 Bytes
61d29fc | 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 | """
Parser Agent for extracting and structuring data from raw meeting minutes.
"""
import re
from typing import List, Dict, Any, Optional
from datetime import datetime
from loguru import logger
from agents.base import BaseAgent, AgentRole, AgentMessage, MessageType, AgentStatus
class ParserAgent(BaseAgent):
"""
Agent responsible for parsing raw meeting documents into structured data.
Extracts:
- Meeting metadata (date, type, location)
- Attendees and participants
- Agenda items
- Discussion topics
- Votes and decisions
- Action items
"""
def __init__(self, agent_id: str = "parser-001"):
"""Initialize the parser agent."""
super().__init__(agent_id, AgentRole.PARSER)
self._compile_patterns()
def _compile_patterns(self):
"""Compile regex patterns for parsing."""
self.patterns = {
"date": re.compile(
r"(?:January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{1,2},?\s+\d{4}",
re.IGNORECASE
),
"time": re.compile(r"\d{1,2}:\d{2}\s*(?:AM|PM|am|pm)?"),
"attendees": re.compile(r"(?:Present|Attending|Members Present):(.+?)(?:\n\n|\Z)", re.DOTALL | re.IGNORECASE),
"motion": re.compile(r"(?:MOTION|Motion|MOVED)(.+?)(?:CARRIED|PASSED|FAILED|$)", re.DOTALL | re.IGNORECASE),
"vote": re.compile(r"(?:Vote|VOTE):\s*(.+)", re.IGNORECASE),
"agenda_item": re.compile(r"(?:Item|ITEM)\s+#?(\d+|[A-Z])[\.:]\s*(.+?)(?=\n(?:Item|ITEM)|$)", re.DOTALL | re.IGNORECASE)
}
async def process(self, message: AgentMessage) -> List[AgentMessage]:
"""
Process parsing commands.
Args:
message: Message containing raw documents to parse
Returns:
List of messages with parsed data
"""
self.update_status(AgentStatus.PROCESSING, "Parsing meeting documents")
try:
documents = message.payload.get("documents", [])
parsed_documents = []
for doc in documents:
parsed = await self._parse_document(doc)
if parsed:
parsed_documents.append(parsed)
# Send parsed documents to classifier
response = await self.send_message(
AgentRole.CLASSIFIER,
MessageType.DATA,
{
"workflow_id": message.payload.get("workflow_id"),
"documents": parsed_documents,
"count": len(parsed_documents)
}
)
self.log_success()
logger.info(f"Parsed {len(parsed_documents)} documents")
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 _parse_document(self, doc: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Parse a single meeting document.
Args:
doc: Raw document data
Returns:
Parsed document with structured fields
"""
try:
content = doc.get("content", "")
parsed = {
"document_id": doc["document_id"],
"source_url": doc["source_url"],
"municipality": doc["municipality"],
"state": doc["state"],
"raw_title": doc["title"],
"parsed_at": datetime.utcnow().isoformat(),
# Extracted structured data
"meeting_date": self._extract_date(content, doc.get("meeting_date")),
"meeting_time": self._extract_time(content),
"meeting_type": doc.get("meeting_type", "Unknown"),
"attendees": self._extract_attendees(content),
"agenda_items": self._extract_agenda_items(content),
"motions": self._extract_motions(content),
"votes": self._extract_votes(content),
"discussion_sections": self._extract_discussion_sections(content),
# Full text for semantic search
"full_text": content,
# Metadata
"metadata": doc.get("metadata", {})
}
return parsed
except Exception as e:
logger.error(f"Error parsing document {doc.get('document_id')}: {e}")
return None
def _extract_date(self, content: str, fallback_date: Optional[str]) -> str:
"""Extract meeting date from content."""
match = self.patterns["date"].search(content)
if match:
return match.group(0)
return fallback_date or datetime.utcnow().isoformat()
def _extract_time(self, content: str) -> Optional[str]:
"""Extract meeting time from content."""
match = self.patterns["time"].search(content)
return match.group(0) if match else None
def _extract_attendees(self, content: str) -> List[str]:
"""Extract list of meeting attendees."""
match = self.patterns["attendees"].search(content)
if match:
attendees_text = match.group(1)
# Split by comma or newline
attendees = re.split(r'[,\n]', attendees_text)
return [a.strip() for a in attendees if a.strip()]
return []
def _extract_agenda_items(self, content: str) -> List[Dict[str, str]]:
"""Extract agenda items from content."""
items = []
for match in self.patterns["agenda_item"].finditer(content):
items.append({
"number": match.group(1).strip(),
"description": match.group(2).strip()
})
return items
def _extract_motions(self, content: str) -> List[Dict[str, str]]:
"""Extract motions from content."""
motions = []
for match in self.patterns["motion"].finditer(content):
motions.append({
"text": match.group(1).strip(),
"full_match": match.group(0).strip()
})
return motions
def _extract_votes(self, content: str) -> List[Dict[str, str]]:
"""Extract voting records from content."""
votes = []
for match in self.patterns["vote"].finditer(content):
votes.append({
"result": match.group(1).strip()
})
return votes
def _extract_discussion_sections(self, content: str) -> List[Dict[str, str]]:
"""Extract discussion sections from content."""
# Split content into paragraphs
paragraphs = [p.strip() for p in content.split("\n\n") if p.strip()]
sections = []
for i, para in enumerate(paragraphs):
if len(para) > 100: # Only substantial paragraphs
sections.append({
"section_id": i,
"text": para
})
return sections
|