"""Phase 2 AI layer: FNOL claim intake and schedule Q&A via the Claude API. Two capabilities, both consumed by app.py: 1. extract_claims(fnol_text) - turns free-text First Notice of Loss (emails, call notes) into structured claim records via Claude's structured outputs, ready to append to claims.csv. 2. ScheduleAssistant - a dispatcher chat assistant with tool access to the current solved schedule. It can explain assignments ("why was CLM-007 dropped?"), look up claims/adjusters, and run hypothetical re-solves ("what if ADJ-01 is out sick?") without touching the baseline solution. Authentication: the Anthropic client resolves credentials from the environment (ANTHROPIC_API_KEY, or an `ant auth login` profile). """ from __future__ import annotations import json from typing import Literal, Optional import anthropic from pydantic import BaseModel import config import distance import solver from data_gen import min_to_hhmm MODEL = "claude-opus-4-8" _client: anthropic.Anthropic | None = None def client() -> anthropic.Anthropic: global _client if _client is None: _client = anthropic.Anthropic() return _client class AssistantError(RuntimeError): """User-friendly wrapper for API failures.""" NO_KEY_MSG = ("No Claude API credentials. Set the ANTHROPIC_API_KEY " "environment variable before starting the app " "(https://platform.claude.com -> API keys).") def _friendly(e: Exception) -> AssistantError: # A key-less client raises TypeError('Could not resolve authentication # method...') at call time rather than an APIError subclass. if isinstance(e, TypeError) and "authentication" in str(e).lower(): return AssistantError(NO_KEY_MSG) if isinstance(e, anthropic.AuthenticationError): return AssistantError(NO_KEY_MSG) if isinstance(e, anthropic.APIConnectionError): return AssistantError("Could not reach the Claude API - check your " "network connection.") if isinstance(e, anthropic.RateLimitError): return AssistantError("Claude API rate limit hit - wait a moment " "and try again.") return AssistantError(f"Claude API error: {e}") # --------------------------------------------------------------------------- # 1. FNOL claim intake (structured extraction) # --------------------------------------------------------------------------- class ExtractedClaim(BaseModel): policyholder_name: Optional[str] address: Optional[str] peril: Literal["fire", "flood", "wind", "hail"] priority: Literal[1, 2, 3] window_start: str # "HH:MM" window_end: str # "HH:MM" service_minutes: int lat: Optional[float] lon: Optional[float] notes: str class ExtractionResult(BaseModel): claims: list[ExtractedClaim] EXTRACTION_SYSTEM = """\ You extract structured insurance claim records from First Notice of Loss text (emails, call-center notes) for a field-adjuster routing system. Rules: - peril: classify the cause of loss as one of fire, flood, wind, hail. Water damage from rising water/storm surge is flood; roof/tree damage from storms is wind. - priority: 1 = must be inspected TODAY (home uninhabitable, safety risk, displaced family, or the text demands same-day service); 2 = high (major damage, distressed policyholder, SLA pressure); 3 = normal. - window_start / window_end: the policyholder's availability window in 24h HH:MM. If none is stated, use 08:00 and 17:00. "Mornings" means 08:00-12:00; "afternoons" means 12:00-17:00. - service_minutes: estimated on-site inspection time. Small/localized damage 60; typical 90; extensive or structural 120; total-loss or large multi-structure 180. - lat/lon: ONLY if explicit coordinates appear in the text; never guess coordinates from an address. Use null otherwise. - notes: one short sentence summarizing the loss for the adjuster. - If the text describes multiple properties/claims, return one record each. If it contains no claim at all, return an empty list.""" def extract_claims(fnol_text: str) -> ExtractionResult: try: response = client().messages.parse( model=MODEL, max_tokens=4096, system=EXTRACTION_SYSTEM, messages=[{"role": "user", "content": fnol_text}], output_format=ExtractionResult, ) except (anthropic.APIError, TypeError) as e: raise _friendly(e) from e return response.parsed_output # --------------------------------------------------------------------------- # 2. Schedule Q&A assistant (tool use) # --------------------------------------------------------------------------- TOOLS = [ { "name": "get_schedule", "description": ( "Get the current solved schedule: every adjuster's route with " "stop order, arrival/departure times and drive legs, plus the " "dropped-claim list and fleet totals. Call this before " "answering any question about today's plan."), "input_schema": {"type": "object", "properties": {}}, }, { "name": "get_claims", "description": ("List all claims in the current instance with " "peril, priority, availability window, service " "time, and location."), "input_schema": {"type": "object", "properties": {}}, }, { "name": "get_adjusters", "description": ("List all adjusters with their skills, shift " "hours, and home locations."), "input_schema": {"type": "object", "properties": {}}, }, { "name": "what_if_solve", "description": ( "Run a HYPOTHETICAL re-solve of today's schedule and return " "the resulting plan. Does NOT change the baseline schedule " "shown in the app. The re-solve uses the SAME solver backend " "and lunch/balance toggles as the schedule on screen, so its " "objective is directly comparable to the baseline. In " "pre-assigned (sequence) mode, upstream assignments stay " "binding: claims are never moved between adjusters, an " "excluded adjuster's claims are dropped and reported for " "rescheduling, and an added adjuster receives no claims. " "Use for questions like 'what if ADJ-01 is out sick?', " "'could we serve CLM-007 if it were urgent?', 'would " "extending ADJ-03 to 19:00 fix the MUST-TODAY violation?', " "or 'what if we brought in one extra flood-qualified " "adjuster?'. exclude_adjuster_ids removes adjusters " "(sick/unavailable); must_today_claim_ids escalates claims " "to must-inspect-today priority; shift_changes temporarily " "alters working hours (overtime); add_adjusters brings in " "hypothetical extra adjusters (new hires / contractors)."), "input_schema": { "type": "object", "properties": { "exclude_adjuster_ids": { "type": "array", "items": {"type": "string"}, "description": "Adjuster ids to remove, e.g. ['ADJ-01']", }, "must_today_claim_ids": { "type": "array", "items": {"type": "string"}, "description": "Claim ids to escalate to priority 1", }, "priority_changes": { "type": "array", "items": { "type": "object", "properties": { "claim_id": {"type": "string"}, "new_priority": { "type": "integer", "enum": [1, 2, 3], "description": "1=MUST-TODAY, 2=high, " "3=normal"}, }, "required": ["claim_id", "new_priority"], }, "description": ("Raise OR lower any claim's priority " "- e.g. de-escalate CLM-012 to " "normal so it can wait"), }, "add_adjusters": { "type": "array", "items": { "type": "object", "properties": { "adjuster_id": { "type": "string", "description": "optional; default TEMP-01," " TEMP-02, ..."}, "name": {"type": "string"}, "skills": { "type": "array", "items": {"type": "string"}, "description": "perils they can handle: " "fire, flood, wind, hail"}, "shift_start": { "type": "string", "description": "HH:MM, default 08:00"}, "shift_end": { "type": "string", "description": "HH:MM, default 17:00"}, "home_lat": { "type": "number", "description": "optional; defaults to the " "region center"}, "home_lon": {"type": "number"}, "max_radius_miles": { "type": "number", "description": "optional service " "territory"}, }, "required": ["skills"], }, "description": ("Hypothetical extra adjusters, e.g. " "one flood-qualified contractor " "working 08:00-18:00"), }, "shift_changes": { "type": "array", "items": { "type": "object", "properties": { "adjuster_id": {"type": "string"}, "new_shift_end": { "type": "string", "description": "HH:MM, e.g. '19:00'"}, "new_shift_start": { "type": "string", "description": "HH:MM, e.g. '06:00'"}, }, "required": ["adjuster_id"], }, "description": ("Temporary working-hour changes, " "e.g. extend ADJ-03's day to 19:00"), }, "time_limit_s": { "type": "integer", "description": ("Solver time limit in seconds. " "Defaults to the user's main-solve " "limit capped at 60 for chat " "responsiveness; explicit values " "are clamped to the main-solve " "limit. Pass a smaller value for " "quick checks."), }, }, }, }, ] ASSISTANT_SYSTEM = """\ You are the dispatch assistant for an insurance field-adjuster routing system. You answer questions about today's solved schedule and run hypothetical what-if re-solves on request. How the optimizer works (use this to explain its decisions): - Each adjuster starts and ends at home, works their shift, and visits claims they are skilled for (peril must match a skill) and that lie inside their service territory (an optional max radius in road miles from their home), arriving inside the policyholder's availability window; on-site service time is fixed. - The objective minimizes total driving minutes plus penalties for dropped claims. Penalties: normal=600, high=3000, MUST-TODAY=1,000,000 (in driving-minute units). A claim is dropped when serving it would cost more than its penalty - because of capacity, windows, skills, or distance. Dropped claims are rescheduled to a later day. - A dropped MUST-TODAY claim is a violation requiring human action. When a MUST-TODAY violation appears, you can actually test the fixes: what_if_solve accepts shift_changes (temporary overtime, e.g. extend an adjuster to 19:00), add_adjusters (hypothetical extra adjusters - give them the skills the violated claim needs; home defaults to the region center unless told otherwise), priority_changes (raise or LOWER any claim's priority - de-escalation frees capacity), and exclusions - run the scenario and report whether it clears the violation and at what cost. What-if re-solves run the SAME solver backend and lunch/balance toggles the user picked for the main solve (the scenario block in the result names them), so objectives are directly comparable. If the schedule was solved in pre-assigned (sequence) mode, assignments stay binding in every what-if: claims never move between adjusters, an excluded adjuster's claims are dropped and reported for rescheduling (not redistributed), and add_adjusters will not help because a new adjuster has no assigned claims - say so instead of suggesting it. Note that applying a scenario (the user's Apply button) changes today's working schedule only; permanent hour changes belong in adjusters.csv. Ground every answer in tool results - call get_schedule before answering schedule questions rather than answering from memory. Be concise and concrete: name claims, adjusters, and times. When you run what_if_solve, compare the hypothetical against the baseline and lead with the impact (claims served, miles, any MUST-TODAY violations), and remind the user it has not changed the real schedule.""" class ScheduleAssistant: """Multi-turn chat with tool access to the solved schedule.""" def __init__(self): self.messages: list = [] self.claims = None self.adjusters = None self.sol = None self.last_what_if: dict | None = None # for the apply-scenario flow # What-if re-solves run through resolver - the same backend + # toggles as the user's last Solve (None falls back to ortools). self.resolver = None self.backend_label = "ortools" self.toggles: dict = {} self.default_time_limit = 10 self.mode = "global" self.matrix_builder = None self.distance_label = "haversine" def set_context(self, claims, adjusters, sol, resolver=None, backend_label="ortools", toggles=None, default_time_limit=10, mode="global", matrix_builder=None, distance_label="haversine") -> None: self.claims = claims self.adjusters = adjusters self.sol = sol self.resolver = resolver self.backend_label = backend_label self.toggles = toggles or {} self.default_time_limit = int(default_time_limit or 10) self.mode = mode self.matrix_builder = matrix_builder self.distance_label = distance_label # -- tool implementations ------------------------------------------------ def _schedule_dict(self, sol) -> dict: return { "routes": [{ "adjuster": r.adjuster.adjuster_id, "name": r.adjuster.name, "leaves_home": min_to_hhmm(r.start_min), "back_home": min_to_hhmm(r.end_min), "total_miles": round(r.total_miles, 1), "stops": [{ "seq": i + 1, "claim_id": s.claim.claim_id, "peril": s.claim.peril, "priority": config.PRIORITY_LABEL[s.claim.priority], "window": f"{min_to_hhmm(s.claim.window_start)}-" f"{min_to_hhmm(s.claim.window_end)}", "on_site": f"{min_to_hhmm(s.arrival_min)}-" f"{min_to_hhmm(s.departure_min)}", "drive_miles": round(s.travel_miles_from_prev, 1), } for i, s in enumerate(r.stops)], } for r in sol.routes], "dropped_for_reschedule": [{ "claim_id": c.claim_id, "peril": c.peril, "priority": config.PRIORITY_LABEL[c.priority], "window": f"{min_to_hhmm(c.window_start)}-" f"{min_to_hhmm(c.window_end)}", "service_minutes": c.service_minutes, "no_qualified_adjuster": c in sol.unservable, } for c in sol.dropped], "totals": { "claims_served": sum(len(r.stops) for r in sol.routes), "claims_total": len(self.claims), "fleet_miles": round(sol.total_miles, 1), "driving_minutes": sol.total_travel_min, "objective": sol.objective, "must_today_violations": [c.claim_id for c in sol.dropped_must_today], }, } def _get_claims(self) -> list[dict]: return [{ "claim_id": c.claim_id, "peril": c.peril, "priority": config.PRIORITY_LABEL[c.priority], "window": f"{min_to_hhmm(c.window_start)}-" f"{min_to_hhmm(c.window_end)}", "service_minutes": c.service_minutes, "lat": c.lat, "lon": c.lon, } for c in self.claims] def _get_adjusters(self) -> list[dict]: return [{ "adjuster_id": a.adjuster_id, "name": a.name, "skills": a.skills, "shift": f"{min_to_hhmm(a.shift_start)}-" f"{min_to_hhmm(a.shift_end)}", "home": {"lat": a.home_lat, "lon": a.home_lon}, "territory_radius_miles": a.max_radius_miles, } for a in self.adjusters] def _what_if(self, tool_input: dict) -> dict: exclude = set(tool_input.get("exclude_adjuster_ids") or []) escalate = set(tool_input.get("must_today_claim_ids") or []) # Default to the main solve's budget capped at 60s so a chat # turn stays responsive; explicit requests are clamped to the # user's own slider setting, never beyond it. cap = max(1, self.default_time_limit) requested = tool_input.get("time_limit_s") time_limit = int(requested) if requested else min(cap, 60) time_limit = max(1, min(time_limit, cap)) import copy from data_gen import hhmm_to_min adjusters = copy.deepcopy([a for a in self.adjusters if a.adjuster_id not in exclude]) if not adjusters: return {"error": "cannot exclude every adjuster"} unknown = exclude - {a.adjuster_id for a in self.adjusters} if unknown: return {"error": f"unknown adjuster ids: {sorted(unknown)}"} shift_changes = tool_input.get("shift_changes") or [] by_id = {a.adjuster_id: a for a in adjusters} applied_shifts = [] for ch in shift_changes: a = by_id.get(ch.get("adjuster_id")) if a is None: return {"error": f"unknown or excluded adjuster in " f"shift_changes: {ch.get('adjuster_id')}"} try: if ch.get("new_shift_start"): a.shift_start = hhmm_to_min(ch["new_shift_start"]) if ch.get("new_shift_end"): a.shift_end = hhmm_to_min(ch["new_shift_end"]) except (ValueError, AttributeError): return {"error": "shift times must be HH:MM, e.g. '19:00'"} if a.shift_end <= a.shift_start: return {"error": f"{a.adjuster_id}: shift end must be " f"after shift start"} applied_shifts.append( {"adjuster_id": a.adjuster_id, "new_shift_start": ch.get("new_shift_start"), "new_shift_end": ch.get("new_shift_end")}) from data_gen import Adjuster added_adjusters = [] for n, spec in enumerate(tool_input.get("add_adjusters") or [], start=1): skills = [str(s).strip().lower() for s in (spec.get("skills") or [])] if not skills or any(s not in config.PERILS for s in skills): return {"error": f"add_adjusters skills must be non-empty " f"and from {config.PERILS}"} aid = spec.get("adjuster_id") or f"TEMP-{n:02d}" if any(x.adjuster_id == aid for x in adjusters) \ or aid in {a.adjuster_id for a in self.adjusters}: return {"error": f"adjuster id {aid} already exists"} try: ss = hhmm_to_min(spec.get("shift_start") or "08:00") se = hhmm_to_min(spec.get("shift_end") or "17:00") except (ValueError, AttributeError): return {"error": "shift times must be HH:MM"} if se <= ss: return {"error": f"{aid}: shift end must be after start"} resolved = { "adjuster_id": aid, "name": spec.get("name") or f"Temp Adjuster {n}", "home_lat": float(spec.get("home_lat") or config.REGION_CENTER[0]), "home_lon": float(spec.get("home_lon") or config.REGION_CENTER[1]), "skills": skills, "shift_start": ss, "shift_end": se, "max_radius_miles": (float(spec["max_radius_miles"]) if spec.get("max_radius_miles") else None), } adjusters.append(Adjuster(**resolved)) added_adjusters.append(resolved) claims = copy.deepcopy(self.claims) unknown_c = escalate - {c.claim_id for c in claims} if unknown_c: return {"error": f"unknown claim ids: {sorted(unknown_c)}"} for c in claims: if c.claim_id in escalate: c.priority = config.PRIORITY_MUST_TODAY prio_changes = [] by_claim = {c.claim_id: c for c in claims} for ch in tool_input.get("priority_changes") or []: c = by_claim.get(ch.get("claim_id")) if c is None: return {"error": f"unknown claim id in priority_changes: " f"{ch.get('claim_id')}"} p = ch.get("new_priority") if p not in (1, 2, 3): return {"error": "new_priority must be 1, 2, or 3"} c.priority = int(p) prio_changes.append({"claim_id": c.claim_id, "new_priority": int(p)}) if self.matrix_builder is not None: try: miles, travel_min = self.matrix_builder(adjusters, claims) except Exception as e: return {"error": f"distance matrices failed: {e}"} else: miles, travel_min = distance.build_matrices(adjusters, claims) if self.resolver is not None: sol = self.resolver(adjusters, claims, miles, travel_min, time_limit) else: sol = solver.solve(adjusters, claims, miles, travel_min, time_limit_s=time_limit) if sol is None: return {"error": "no feasible solution found"} self.last_what_if = {"exclude_adjuster_ids": sorted(exclude), "must_today_claim_ids": sorted(escalate), "shift_changes": applied_shifts, "add_adjusters": added_adjusters, "priority_changes": prio_changes} result = self._schedule_dict(sol) result["scenario"] = { "solver_backend": self.backend_label, "toggles": dict(self.toggles), "mode": self.mode, "distance_model": self.distance_label, "time_limit_s": time_limit, "excluded_adjusters": sorted(exclude), "escalated_to_must_today": sorted(escalate), "shift_changes": applied_shifts, "priority_changes": prio_changes, "added_adjusters": [ {"adjuster_id": r["adjuster_id"], "skills": r["skills"], "shift": f"{min_to_hhmm(r['shift_start'])}-" f"{min_to_hhmm(r['shift_end'])}"} for r in added_adjusters], "note": "hypothetical only - baseline schedule unchanged", } return result def _dispatch(self, name: str, tool_input: dict): if name == "get_schedule": return self._schedule_dict(self.sol) if name == "get_claims": return self._get_claims() if name == "get_adjusters": return self._get_adjusters() if name == "what_if_solve": return self._what_if(tool_input) raise ValueError(f"unknown tool: {name}") # -- the agentic loop ---------------------------------------------------- def ask(self, user_text: str) -> str: if self.sol is None: return ("No solved schedule yet - click Solve first, then ask " "me about the plan.") checkpoint = len(self.messages) self.messages.append({"role": "user", "content": user_text}) response = None try: for _ in range(8): # tool-round guard response = client().messages.create( model=MODEL, max_tokens=16000, thinking={"type": "adaptive"}, system=ASSISTANT_SYSTEM, tools=TOOLS, messages=self.messages, ) self.messages.append({"role": "assistant", "content": response.content}) if response.stop_reason != "tool_use": break results = [] for block in response.content: if block.type != "tool_use": continue try: out = self._dispatch(block.name, dict(block.input)) results.append({ "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(out), }) except Exception as e: results.append({ "type": "tool_result", "tool_use_id": block.id, "content": f"Tool error: {e}", "is_error": True, }) self.messages.append({"role": "user", "content": results}) except (anthropic.APIError, TypeError) as e: del self.messages[checkpoint:] # roll back the failed turn raise _friendly(e) from e if response is None: return "Something went wrong - no response from the model." text = "\n".join(b.text for b in response.content if b.type == "text") return text or "(no text response)" def ask_stream(self, user_text: str): """Streaming version of ask(): yields the growing reply text. Tool rounds run silently; the final round streams token by token.""" if self.sol is None: yield ("No solved schedule yet - click Solve first, then ask " "me about the plan.") return checkpoint = len(self.messages) self.messages.append({"role": "user", "content": user_text}) try: for _ in range(8): with client().messages.stream( model=MODEL, max_tokens=16000, thinking={"type": "adaptive"}, system=ASSISTANT_SYSTEM, tools=TOOLS, messages=self.messages, ) as stream: partial = "" for text in stream.text_stream: partial += text yield partial response = stream.get_final_message() self.messages.append({"role": "assistant", "content": response.content}) if response.stop_reason != "tool_use": return results = [] for block in response.content: if block.type != "tool_use": continue try: out = self._dispatch(block.name, dict(block.input)) results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(out)}) except Exception as e: results.append({"type": "tool_result", "tool_use_id": block.id, "content": f"Tool error: {e}", "is_error": True}) self.messages.append({"role": "user", "content": results}) except (anthropic.APIError, TypeError) as e: del self.messages[checkpoint:] yield f"Error: {_friendly(e)}" BRIEFING_SYSTEM = """\ You write morning briefings for insurance field adjusters. You receive today's solved schedule as JSON. Write one short briefing per adjuster with routes, in Markdown: a '## ' heading, then a friendly 2-3 sentence overview of their day (how many stops, total driving, when they're done), then a numbered stop list - each line with the claim id, damage type, the time to be on site, and anything notable (MUST-TODAY urgency, tight windows, long drives). Close each briefing with one practical reminder if warranted. Plain language, no jargon, no invented facts - use only what the JSON contains.""" def generate_briefings(assistant_state: "ScheduleAssistant") -> str: """One Claude call: turn the solved schedule into per-adjuster morning briefings (Markdown).""" if assistant_state.sol is None: raise AssistantError("Solve a schedule first.") schedule = assistant_state._schedule_dict(assistant_state.sol) try: with client().messages.stream( model=MODEL, max_tokens=16000, system=BRIEFING_SYSTEM, messages=[{"role": "user", "content": json.dumps(schedule)}], ) as stream: response = stream.get_final_message() except (anthropic.APIError, TypeError) as e: raise _friendly(e) from e return "\n".join(b.text for b in response.content if b.type == "text")