import re import json from datetime import datetime, timedelta from typing import Any from app.config import get_settings from app.domain.schemas import PlannerDecision, ToolDefinition, TranscriptChunk class HybridPlanner: """Claude planner with deterministic fallback for offline demos.""" def __init__(self) -> None: self.settings = get_settings() async def plan( self, chunk: TranscriptChunk, context: dict[str, Any], memories: list[dict[str, Any]], tools: list[ToolDefinition], ) -> PlannerDecision: if self.settings.anthropic_api_key: try: return await self._plan_with_claude(chunk, context, memories, tools) except Exception: return self._deterministic_plan(chunk) return self._deterministic_plan(chunk) async def _plan_with_claude( self, chunk: TranscriptChunk, context: dict[str, Any], memories: list[dict[str, Any]], tools: list[ToolDefinition], ) -> PlannerDecision: from anthropic import AsyncAnthropic client = AsyncAnthropic(api_key=self.settings.anthropic_api_key) response = await client.messages.create( model=self.settings.anthropic_model, max_tokens=1200, system=( "You are an event-driven meeting planner. Produce exactly one structured action plan " "or no_action. Never execute tools. Do not invent missing required fields." ), messages=[ { "role": "user", "content": json.dumps( { "transcript_chunk": chunk.model_dump(mode="json"), "context": context, "memories": memories, "available_tools": [tool.model_dump() for tool in tools], "business_rules": [ "All actions must start as pending review.", "Only select tools from available_tools.", "Do not include parameters that are not in the tool schema.", ], }, default=str, ), } ], tools=[ { "name": "emit_plan", "description": "Emit the planner decision for the current transcript chunk.", "input_schema": PlannerDecision.model_json_schema(), } ], tool_choice={"type": "tool", "name": "emit_plan"}, ) for block in response.content: if getattr(block, "type", None) == "tool_use" and getattr(block, "name", None) == "emit_plan": return PlannerDecision.model_validate(block.input) raise ValueError("Claude did not emit the required planner tool output.") def _deterministic_plan(self, chunk: TranscriptChunk) -> PlannerDecision: text = chunk.text.strip() lowered = text.lower() if "task" in lowered or lowered.startswith("i'll ") or "i will " in lowered: assignee = self._extract_assignee(text) or chunk.speaker title = self._task_title(text) args: dict[str, Any] = { "title": title, "description": f"Follow-up captured from meeting transcript. Suggested owner: {assignee}.", "notes": text, "is_completed": False, } if "tomorrow" in lowered: args["due_date"] = (chunk.timestamp + timedelta(days=1)).isoformat() return PlannerDecision( intent="create_task", confidence=0.94, reasoning="The speaker described an explicit follow-up task.", tool="create_task", arguments=args, extracted_entities={"assignee": assignee, "title": title}, ) if "note" in lowered or "prefers" in lowered or "preference" in lowered: category = "preference" if "prefer" in lowered else "general" return PlannerDecision( intent="create_note", confidence=0.91, reasoning="The utterance captures information that should be preserved as a note.", tool="create_note", arguments={"content": text}, extracted_entities={"category": category}, ) if "reserve" in lowered or "reservation" in lowered or "venue" in lowered: if not self.settings.wed_default_partner_id: return PlannerDecision( intent="no_action", confidence=0.65, reasoning="A reservation request was detected, but WED_DEFAULT_PARTNER_ID is required to target the Wed backend.", extracted_entities={"missing_configuration": "WED_DEFAULT_PARTNER_ID"}, ) date = self._extract_date(text, chunk.timestamp) event_type = "Wedding" match = re.search(r"reserve (?:the )?([a-zA-Z ]+?)(?: for| on|$)", text, re.I) if match: event_type = match.group(1).strip().title() return PlannerDecision( intent="create_guest_reservation", confidence=0.89, reasoning="The utterance requests a reservation that requires review before execution.", tool="create_guest_reservation", arguments={ "partner_id": self.settings.wed_default_partner_id, "guest_first_name": self.settings.wed_guest_first_name, "guest_last_name": self.settings.wed_guest_last_name, "guest_email": self.settings.wed_guest_email, "guest_phone": self.settings.wed_guest_phone, "event_date": date.isoformat(), "details": text, "event_type": event_type, "interested_dates": date.date().isoformat(), }, extracted_entities={"event_type": event_type, "event_date": date.isoformat()}, ) return PlannerDecision(intent="no_action", confidence=0.55, reasoning="No actionable backend intent was detected.") def _extract_assignee(self, text: str) -> str | None: for pattern in [r"for ([A-Z][a-z]+) to", r"([A-Z][a-z]+) to"]: match = re.search(pattern, text) if match: return match.group(1) return None def _task_title(self, text: str) -> str: cleaned = re.sub(r"create a task for [A-Z][a-z]+ to ", "", text, flags=re.I) cleaned = re.sub(r"^i(?:'ll| will) ", "", cleaned, flags=re.I) cleaned = cleaned.rstrip(".") return cleaned[:1].upper() + cleaned[1:] def _extract_date(self, text: str, fallback: datetime) -> datetime: lowered = text.lower() if "tomorrow" in lowered: return fallback + timedelta(days=1) match = re.search(r"(january|february|march|april|may|june|july|august|september|october|november|december)\s+(\d{1,2})", lowered) if match: month = datetime.strptime(match.group(1), "%B").month return datetime(fallback.year, month, int(match.group(2)), 12, 0) return fallback + timedelta(days=7)