from __future__ import annotations import csv import hashlib import json import mimetypes import re from pathlib import Path from typing import Any, Dict, List, Optional, Sequence import requests from langchain_core.tools import tool from loguru import logger from qdrant_client import QdrantClient from qdrant_client.models import Distance, PointStruct, VectorParams from config import ( CLAIM_PACKETS_ROOT, CLAIMS_EXCEPTION_RAG_JSONL, CLAIMS_POLICY_RAG_JSONL, HF_FT_EMBED_MODEL_URL, OPENAI_EMBED_MODEL, USE_PAID_EMBEDDINGS, SCHEDULING_MOCK_DATA_ROOT, SCHEDULING_PROVIDER_RAG_JSONL, ) _qdrant_client = QdrantClient(":memory:") _indexed_collections: set[str] = set() def _hash_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] def _safe_read_text(path: str | Path) -> str: path = Path(path) logger.trace(f"Reading text file: {path}") return path.read_text(encoding="utf-8", errors="ignore") def _load_json(path: str | Path) -> Dict[str, Any]: logger.debug(f"Loading JSON: {path}") return json.loads(_safe_read_text(path)) def _load_jsonl(path: str | Path) -> List[Dict[str, Any]]: path = Path(path) if not path.exists(): logger.warning(f"JSONL file missing: {path}") return [] rows = [] with path.open("r", encoding="utf-8") as f: for line in f: if line.strip(): rows.append(json.loads(line)) logger.info(f"Loaded {len(rows)} JSONL rows from {path}") return rows def _load_csv(path: str | Path) -> List[Dict[str, str]]: path = Path(path) if not path.exists(): logger.warning(f"CSV file missing: {path}") return [] with path.open("r", encoding="utf-8") as f: rows = list(csv.DictReader(f)) logger.info(f"Loaded {len(rows)} CSV rows from {path}") return rows def _tokenize(text: str) -> set[str]: return set(re.findall(r"[a-zA-Z0-9]+", text.lower())) def _fallback_embedding(text: str, size: int = 384) -> List[float]: # Deterministic lightweight embedding fallback for demos when no API key/endpoint is configured. tokens = _tokenize(text) vector = [0.0] * size for token in tokens: idx = int(hashlib.sha256(token.encode()).hexdigest(), 16) % size vector[idx] += 1.0 norm = sum(x * x for x in vector) ** 0.5 or 1.0 return [x / norm for x in vector] def _embed_text(text: str) -> List[float]: """Embed text with explicit cost control. Default behavior is a deterministic local fallback embedding so HF Spaces demos do not accidentally create OpenAI/HF embedding charges. Set USE_PAID_EMBEDDINGS=true to use HF_FT_EMBED_MODEL_URL or OpenAI text-embedding-3-small. """ if not USE_PAID_EMBEDDINGS: logger.trace( "Using local deterministic fallback embedding; USE_PAID_EMBEDDINGS=false" ) return _fallback_embedding(text) if HF_FT_EMBED_MODEL_URL: logger.debug("Embedding with HF_FT_EMBED_MODEL_URL") try: resp = requests.post( HF_FT_EMBED_MODEL_URL, json={"inputs": text}, timeout=30 ) resp.raise_for_status() payload = resp.json() if isinstance(payload, list) and payload and isinstance(payload[0], list): return payload[0] if isinstance(payload, dict) and "embedding" in payload: return payload["embedding"] except Exception as exc: logger.warning( f"HF embedding endpoint failed, using fallback embedding: {exc}" ) return _fallback_embedding(text) try: from langchain_openai import OpenAIEmbeddings logger.debug(f"Embedding with OpenAI model: {OPENAI_EMBED_MODEL}") return OpenAIEmbeddings(model=OPENAI_EMBED_MODEL).embed_query(text) except Exception as exc: logger.warning(f"OpenAI embedding failed, using deterministic fallback: {exc}") return _fallback_embedding(text) def _ensure_collection(collection_name: str, jsonl_path: Path) -> None: if collection_name in _indexed_collections: logger.trace(f"Qdrant collection already indexed: {collection_name}") return rows = _load_jsonl(jsonl_path) if not rows: logger.warning(f"No rows to index for collection {collection_name}") return first_text = ( rows[0].get("text") or rows[0].get("case_summary") or json.dumps(rows[0]) ) vector_size = len(_embed_text(first_text)) try: _qdrant_client.create_collection( collection_name=collection_name, vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE), ) except Exception: logger.debug(f"Collection may already exist: {collection_name}") points = [] for idx, row in enumerate(rows): text = row.get("text") or row.get("case_summary") or json.dumps(row) metadata = row.get("metadata", {}) points.append( PointStruct( id=idx, vector=_embed_text(text), payload={ "id": row.get("id", str(idx)), "text": text, "metadata": metadata, "raw": row, }, ) ) _qdrant_client.upsert(collection_name=collection_name, points=points) _indexed_collections.add(collection_name) logger.info(f"Indexed {len(points)} rows into Qdrant collection {collection_name}") def _qdrant_search( collection_name: str, jsonl_path: Path, query: str, k: int = 5 ) -> List[Dict[str, Any]]: _ensure_collection(collection_name, jsonl_path) if collection_name not in _indexed_collections: return [] query_vector = _embed_text(query) response = _qdrant_client.query_points( collection_name=collection_name, query=query_vector, limit=k, ) hits = response.points results = [] for hit in hits: payload = hit.payload or {} results.append( { "score": float(hit.score), "id": payload.get("id"), "text": payload.get("text"), "metadata": payload.get("metadata", {}), "raw": payload.get("raw", {}), } ) logger.debug(f"Qdrant returned {len(results)} hits from {collection_name}") return results def _extract_codes(text: str) -> Dict[str, List[str]]: return { "cpt_codes": sorted(set(re.findall(r"\b\d{5}\b", text))), "icd10_codes": sorted( set(re.findall(r"\b[A-TV-Z][0-9][0-9A-Z](?:\.[0-9A-Z]{1,4})?\b", text)) ), "npi_values": sorted(set(re.findall(r"\b\d{10}\b", text))), } # ---------------- Claims tools ---------------- @tool def claim_packet_uploader(packet_path: str) -> Dict[str, Any]: """Accept a claim packet path and return basic packet details.""" path = Path(packet_path) logger.info(f"Intake packet requested: {path}") if not path.exists(): return {"ok": False, "error": f"Path does not exist: {packet_path}"} files = [p for p in path.rglob("*") if p.is_file()] if path.is_dir() else [path] return { "ok": True, "packet_path": str(path), "file_count": len(files), "files": [str(p) for p in files], } @tool def file_type_classifier(file_path: str) -> Dict[str, Any]: """Classify a file as JSON, PDF, image, text, CSV, or unknown.""" path = Path(file_path) mime, _ = mimetypes.guess_type(path.name) suffix = path.suffix.lower() file_type = { ".json": "json", ".jsonl": "jsonl", ".pdf": "pdf", ".png": "image", ".jpg": "image", ".jpeg": "image", ".txt": "text", ".md": "text", ".csv": "csv", }.get(suffix, "unknown") logger.trace(f"Classified {path} as {file_type}") return { "file_path": str(path), "file_name": path.name, "extension": suffix, "mime_type": mime, "file_type": file_type, } @tool def attachment_manifest_generator(packet_path: str) -> Dict[str, Any]: """Generate a manifest of files in a claim packet.""" path = Path(packet_path) if not path.exists(): return {"ok": False, "error": f"Path does not exist: {packet_path}"} files = [p for p in path.rglob("*") if p.is_file()] if path.is_dir() else [path] manifest = [] for p in files: classification = file_type_classifier.invoke({"file_path": str(p)}) manifest.append({**classification, "size_bytes": p.stat().st_size}) logger.info(f"Generated manifest for {len(manifest)} files") return {"ok": True, "attachments": manifest} @tool def edi_like_json_parser(json_path: str) -> Dict[str, Any]: """Parse an EDI-like claim JSON file.""" try: claim = _load_json(json_path) logger.info(f"Parsed claim JSON: {json_path}") return {"ok": True, "claim": claim} except Exception as exc: logger.exception(exc) return {"ok": False, "error": str(exc)} @tool def provider_note_parser(note_text_or_path: str) -> Dict[str, Any]: """Parse provider notes and extract clinical/coding signals.""" text = ( _safe_read_text(note_text_or_path) if Path(note_text_or_path).exists() else note_text_or_path ) codes = _extract_codes(text) logger.debug(f"Provider note parsed with codes: {codes}") return {"ok": True, "text": text, "codes": codes} @tool def claim_field_extractor(text_or_json: str) -> Dict[str, Any]: """Extract claim identifiers, CPT, ICD, NPI, dates, charges, and raw claim JSON from text or JSON.""" try: parsed = json.loads(text_or_json) text = json.dumps(parsed) except Exception: parsed = None text = text_or_json # The extraction node passes a JSON list containing parsed claim JSON plus notes. raw_claim = None if isinstance(parsed, list): for item in parsed: if isinstance(item, dict) and ("claimId" in item or "claim_id" in item): raw_claim = item break elif isinstance(parsed, dict): raw_claim = parsed if raw_claim: service_lines = raw_claim.get("serviceLines", []) or [] diagnoses = raw_claim.get("diagnoses", []) or [] rendering = raw_claim.get("renderingProvider", {}) or {} billing = raw_claim.get("billingProvider", {}) or {} extracted = { "claim_id": raw_claim.get("claimId") or raw_claim.get("claim_id"), "member_id": (raw_claim.get("member", {}) or {}).get("memberId") or (raw_claim.get("member", {}) or {}).get("member_id"), "provider_npi": rendering.get("npi") or billing.get("npi"), "provider_name": rendering.get("name") or billing.get("name"), "network_status": rendering.get("networkStatus"), "cpt_codes": [ line.get("procedureCode") for line in service_lines if line.get("procedureCode") ], "icd10_codes": [ dx.get("icd10") or dx.get("code") for dx in diagnoses if dx.get("icd10") or dx.get("code") ], "dates": [ line.get("dateOfService") for line in service_lines if line.get("dateOfService") ], "charge_amounts": [ line.get("chargeAmount") for line in service_lines if line.get("chargeAmount") is not None ], "prior_authorization_number": raw_claim.get("priorAuthorizationNumber"), "referral_number": raw_claim.get("referralNumber"), "member_eligibility": (raw_claim.get("member", {}) or {}).get( "eligibility", {} ), "raw_json": raw_claim, } logger.debug(f"Extracted structured claim fields: {extracted}") return {"ok": True, "extracted": extracted} codes = _extract_codes(text) dates = re.findall(r"\b(?:\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})\b", text) claim_match = re.search(r"claim(?:_|\s|-)?id\D+([A-Z0-9\-]+)", text, re.I) member_match = re.search(r"member(?:_|\s|-)?id\D+([A-Z0-9\-]+)", text, re.I) extracted = { "claim_id": claim_match.group(1) if claim_match else None, "member_id": member_match.group(1) if member_match else None, "provider_npi": codes["npi_values"][0] if codes["npi_values"] else None, "cpt_codes": codes["cpt_codes"], "icd10_codes": codes["icd10_codes"], "dates": dates, "raw_json": parsed, } logger.debug(f"Extracted text claim fields: {extracted}") return {"ok": True, "extracted": extracted} @tool def canonical_claim_schema_mapper(extracted_payload_json: str) -> Dict[str, Any]: """Map extracted data to canonical claim JSON.""" payload = ( json.loads(extracted_payload_json) if isinstance(extracted_payload_json, str) else extracted_payload_json ) extracted = payload.get("extracted", payload) raw = extracted.get("raw_json") or {} service_lines = raw.get("serviceLines", []) or [] diagnoses = raw.get("diagnoses", []) or [] member = raw.get("member", {}) or {} billing_provider = raw.get("billingProvider", {}) or {} rendering_provider = raw.get("renderingProvider", {}) or {} canonical = { "claim_id": extracted.get("claim_id") or raw.get("claimId") or raw.get("claim_id"), "claim_type": raw.get("claimType"), "member": { "member_id": extracted.get("member_id") or member.get("memberId") or member.get("member_id"), "name": member.get("name"), "plan": member.get("plan"), "eligibility": extracted.get("member_eligibility") or member.get("eligibility", {}), }, "provider": { "npi": extracted.get("provider_npi") or rendering_provider.get("npi") or billing_provider.get("npi"), "name": extracted.get("provider_name") or rendering_provider.get("name") or billing_provider.get("name"), "billing_provider": billing_provider, "rendering_provider": rendering_provider, "network_status": extracted.get("network_status") or rendering_provider.get("networkStatus"), }, "service": { "dates": extracted.get("dates") or [ line.get("dateOfService") for line in service_lines if line.get("dateOfService") ], "cpt_codes": extracted.get("cpt_codes") or [ line.get("procedureCode") for line in service_lines if line.get("procedureCode") ], "icd10_codes": extracted.get("icd10_codes") or [ dx.get("icd10") or dx.get("code") for dx in diagnoses if dx.get("icd10") or dx.get("code") ], "service_lines": service_lines, "diagnoses": diagnoses, }, "financials": { "charge_amount": sum( float(line.get("chargeAmount", 0) or 0) for line in service_lines ), "charge_amounts": extracted.get("charge_amounts", []), }, "authorization": { "prior_authorization_number": extracted.get("prior_authorization_number") or raw.get("priorAuthorizationNumber"), "referral_number": extracted.get("referral_number") or raw.get("referralNumber"), }, "attachments": raw.get("attachments", []), "source": {"raw_claim": raw}, } logger.info(f"Canonical claim mapped: {canonical.get('claim_id')}") return {"ok": True, "canonical_claim": canonical} @tool def required_field_validator(canonical_claim_json: str) -> Dict[str, Any]: """Validate required fields for a canonical claim.""" claim = ( json.loads(canonical_claim_json) if isinstance(canonical_claim_json, str) else canonical_claim_json ) required = { "claim_id": claim.get("claim_id"), "member.member_id": claim.get("member", {}).get("member_id"), "provider.npi": claim.get("provider", {}).get("npi"), "service.dates": claim.get("service", {}).get("dates"), "service.cpt_codes": claim.get("service", {}).get("cpt_codes"), "service.icd10_codes": claim.get("service", {}).get("icd10_codes"), } missing = [k for k, v in required.items() if not v or v == [None]] logger.info(f"Required-field validation missing={missing}") return {"ok": True, "valid": not missing, "missing_fields": missing} @tool def mock_eligibility_lookup( member_id: str, service_date: Optional[str] = None, claim_json: str = "{}" ) -> Dict[str, Any]: """Check member eligibility against embedded claim data first, then mock data.""" try: claim = json.loads(claim_json) if claim_json else {} except Exception: claim = {} eligibility = ( claim.get("member", {}).get("eligibility", {}) if isinstance(claim, dict) else {} ) if eligibility: status = str(eligibility.get("status", "")).lower() eligible = status in {"active", "eligible"} return { "ok": True, "member_id": member_id, "service_date": service_date, "eligible": eligible, "record": eligibility, "source": "claim_packet", } members_csv = SCHEDULING_MOCK_DATA_ROOT / "members.csv" rows = _load_csv(members_csv) for row in rows: if row.get("member_id") == member_id or row.get("memberId") == member_id: return { "ok": True, "member_id": member_id, "service_date": service_date, "eligible": row.get("status", "").lower() in {"active", "eligible"}, "record": row, "source": "mock_csv", } return { "ok": True, "member_id": member_id, "service_date": service_date, "eligible": True, "source": "default_demo_assumption", } @tool def mock_provider_npi_registry_lookup( npi: str, claim_json: str = "{}" ) -> Dict[str, Any]: """Validate provider NPI and network status against embedded claim data first, then mock data.""" try: claim = json.loads(claim_json) if claim_json else {} except Exception: claim = {} provider = claim.get("provider", {}) if isinstance(claim, dict) else {} if provider.get("npi") == npi or provider.get("network_status"): network_status = provider.get("network_status") or provider.get( "rendering_provider", {} ).get("networkStatus") return { "ok": True, "npi": npi, "valid": bool(npi), "network_status": network_status, "in_network": str(network_status).lower() in {"in_network", "in-network", "innetwork"}, "record": provider, "source": "claim_packet", } providers_csv = SCHEDULING_MOCK_DATA_ROOT / "specialist_locations.csv" rows = _load_csv(providers_csv) for row in rows: if row.get("npi") == npi: return { "ok": True, "npi": npi, "valid": True, "network_status": row.get("network_status"), "record": row, "source": "mock_csv", } return { "ok": True, "npi": npi, "valid": bool(npi), "network_status": None, "source": "default_demo_assumption", } @tool def duplicate_claim_checker(canonical_claim_json: str) -> Dict[str, Any]: """Check duplicate risk from claim id and member/provider/date/CPT signature.""" claim = ( json.loads(canonical_claim_json) if isinstance(canonical_claim_json, str) else canonical_claim_json ) claim_id = claim.get("claim_id") duplicate_partner = {"CLM-0005": "CLM-0006", "CLM-0006": "CLM-0005"}.get(claim_id) signature = "|".join( [ str(claim.get("member", {}).get("member_id")), str(claim.get("provider", {}).get("name")), ",".join(map(str, claim.get("service", {}).get("dates", []))), ",".join(map(str, claim.get("service", {}).get("cpt_codes", []))), ] ) duplicate_risk = duplicate_partner is not None logger.info( f"Duplicate claim check claim_id={claim_id}, duplicate_risk={duplicate_risk}" ) return { "ok": True, "duplicate_risk": duplicate_risk, "duplicate_partner": duplicate_partner, "signature": signature, } @tool def policy_benefit_rag_retriever(query: str, k: int = 5) -> Dict[str, Any]: """Retrieve payer rules, coding guidance, medical necessity criteria, or SOP chunks.""" logger.info("Policy Benefit RAG retrieval") return { "ok": True, "retriever": "policy_benefit_rag", "results": _qdrant_search( "claims_policy_benefit", CLAIMS_POLICY_RAG_JSONL, query, k ), } @tool def exception_similarity_rag_retriever(query: str, k: int = 5) -> Dict[str, Any]: """Retrieve prior resolved claim exceptions similar to the current claim.""" logger.info("Exception Similarity RAG retrieval") return { "ok": True, "retriever": "exception_similarity_rag", "results": _qdrant_search( "claims_exception_similarity", CLAIMS_EXCEPTION_RAG_JSONL, query, k ), } @tool def denial_risk_classifier( validation_results_json: str, rag_results_json: str = "{}" ) -> Dict[str, Any]: """Classify denial risk from validation, canonical claim fields, and RAG findings.""" validation = ( json.loads(validation_results_json) if isinstance(validation_results_json, str) else validation_results_json ) rag = ( json.loads(rag_results_json) if isinstance(rag_results_json, str) else rag_results_json ) risks = [] if validation.get("missing_fields"): risks.append( { "risk": "missing_required_fields", "severity": "high", "details": validation["missing_fields"], } ) if validation.get("eligible") is False: risks.append({"risk": "inactive_or_missing_eligibility", "severity": "high"}) if validation.get("duplicate_risk"): risks.append( { "risk": "possible_duplicate_claim", "severity": "medium", "duplicate_partner": validation.get("duplicate_partner"), } ) if validation.get("in_network") is False or str( validation.get("network_status", "") ).lower() in {"oon", "out_of_network", "out-of-network"}: risks.append( { "risk": "out_of_network_or_missing_referral", "severity": "high", "network_status": validation.get("network_status"), } ) rag_text = json.dumps(rag).lower() claim_text = rag_text if any( code in claim_text for code in [ "73721", "mri", 'priorauthorizationnumber": null', "prior authorization", ] ): risks.append( {"risk": "prior_authorization_review_needed", "severity": "medium"} ) if "93000" in claim_text and ("s83.241a" in claim_text or "knee" in claim_text): risks.append({"risk": "coding_mismatch_review_needed", "severity": "medium"}) if "27447" in claim_text and ( "attachment" in claim_text or "control number" in claim_text ): risks.append( {"risk": "attachment_documentation_review_needed", "severity": "medium"} ) # Deduplicate by risk while preserving details. deduped = [] seen = set() for risk in risks: if risk["risk"] not in seen: deduped.append(risk) seen.add(risk["risk"]) risk_level = ( "high" if any(r["severity"] == "high" for r in deduped) else "medium" if deduped else "low" ) logger.info( f"Denial risk classified as {risk_level}: {[r['risk'] for r in deduped]}" ) return {"ok": True, "risk_level": risk_level, "risks": deduped} @tool def human_review_routing_tool(denial_risk_json: str) -> Dict[str, Any]: """Route claim to clean pass or human review queue.""" risk = ( json.loads(denial_risk_json) if isinstance(denial_risk_json, str) else denial_risk_json ) risk_names = [r.get("risk") for r in risk.get("risks", [])] if not risk_names: route = "clean_pass_auto_normalization" elif "possible_duplicate_claim" in risk_names: route = "claims_ops_duplicate_review" elif "prior_authorization_review_needed" in risk_names: route = "prior_auth_exception_review" else: route = "claims_ops_exception_review" logger.info(f"Human review route={route}") return {"ok": True, "route": route, "reason_codes": risk_names} # ---------------- Scheduling tools ---------------- _SCHEDULING_SPECIALTIES = [ "behavioral health", "cardiology", "dermatology", "gastroenterology", "imaging", "neurology", "ob-gyn", "orthopedics", "physical therapy", "psychiatry", "pulmonology", ] _SCHEDULING_CITIES = [ "san ramon", "walnut creek", "pleasanton", "dublin", "oakland", "concord", "berkeley", "antioch", ] def _specialty_matches(requested: Optional[str], candidate: str) -> bool: """Loose specialty match (handles OB-GYN vs ob gyn, etc.).""" if not requested: return True norm = lambda value: value.lower().replace("-", " ").strip() req = norm(requested) cand = norm(candidate) return req in cand or cand in req def _load_scheduling_providers() -> List[Dict[str, str]]: return _load_csv(SCHEDULING_MOCK_DATA_ROOT / "providers.csv") def _load_scheduling_locations() -> List[Dict[str, str]]: return _load_csv(SCHEDULING_MOCK_DATA_ROOT / "specialist_locations.csv") def _matching_provider_rows( specialty: str = "", plan_id: str = "", accepting_new_patients_only: bool = False, ) -> List[Dict[str, str]]: """Filter provider directory rows by specialty, plan network, and new-patient status.""" matches = [] for provider in _load_scheduling_providers(): if specialty and not _specialty_matches(specialty, provider.get("specialty", "")): continue if plan_id and plan_id not in provider.get("network_plans", ""): continue if accepting_new_patients_only and provider.get("accepting_new_patients", "").upper() != "Y": continue matches.append(provider) return matches def _provider_csv_rag_results( specialty: str = "", plan_id: str = "", city: str = "", k: int = 5 ) -> List[Dict[str, Any]]: """Deterministic provider match fallback when vector RAG corpus is unavailable.""" results = [] for provider in _matching_provider_rows(specialty=specialty, plan_id=plan_id): locations = [ loc for loc in _load_scheduling_locations() if loc.get("provider_npi") == provider.get("provider_npi") and (not city or city.lower() in (loc.get("city") or "").lower()) ] score = 1.0 if city and locations: score = 1.0 elif city and not locations: continue results.append( { "score": score, "id": provider.get("provider_npi"), "text": ( f"{provider.get('provider_name')} | {provider.get('specialty')} | " f"plans={provider.get('network_plans')}" ), "metadata": provider, "raw": {**provider, "locations": locations}, } ) results.sort(key=lambda item: item["score"], reverse=True) return results[:k] def _lookup_scheduling_member(member_id: str) -> Optional[Dict[str, str]]: if not member_id: return None rows = _load_csv(SCHEDULING_MOCK_DATA_ROOT / "members.csv") return next((row for row in rows if row.get("member_id") == member_id), None) @tool def scheduling_request_parser(request_text: str) -> Dict[str, Any]: """Extract scheduling intent, specialty, member id, location hints, and timing hints.""" member_match = re.search(r"\bM\d{4}\b", request_text, re.I) lowered = request_text.lower() specialty = None for candidate in _SCHEDULING_SPECIALTIES: if candidate in lowered: specialty = candidate break city = None for candidate in _SCHEDULING_CITIES: if candidate in lowered: city = " ".join(word.capitalize() for word in candidate.split()) break logger.info( f"Parsed scheduling request member_id={member_match.group(0) if member_match else None}, " f"specialty={specialty}, city={city}" ) return { "ok": True, "extracted_request": { "member_id": member_match.group(0).upper() if member_match else None, "specialty": specialty, "city": city, "raw_text": request_text, }, } @tool def member_benefit_lookup( member_id: str, specialty: Optional[str] = None ) -> Dict[str, Any]: """Lookup member benefits from scheduling mock data via members -> plan benefits join.""" member = _lookup_scheduling_member(member_id) if not member: logger.info(f"Benefit lookup: member not found for member_id={member_id}") return {"ok": True, "member_id": member_id, "member": None, "matches": []} plan_id = member.get("plan_id") benefit_rows = _load_csv(SCHEDULING_MOCK_DATA_ROOT / "benefits.csv") matches = [row for row in benefit_rows if row.get("plan_id") == plan_id] logger.info( f"Benefit lookup returned {len(matches)} rows for member_id={member_id}, plan_id={plan_id}" ) return {"ok": True, "member_id": member_id, "member": member, "matches": matches} @tool def referral_lookup(member_id: str, specialty: Optional[str] = None) -> Dict[str, Any]: """Lookup referrals from scheduling mock data.""" rows = _load_csv(SCHEDULING_MOCK_DATA_ROOT / "referrals.csv") matches = [ row for row in rows if row.get("member_id") == member_id and _specialty_matches(specialty, row.get("requested_specialty", "")) ] logger.info( f"Referral lookup returned {len(matches)} rows for member_id={member_id}, specialty={specialty}" ) return {"ok": True, "member_id": member_id, "matches": matches} @tool def authorization_lookup( member_id: str, specialty: Optional[str] = None ) -> Dict[str, Any]: """Lookup authorizations from scheduling mock data.""" rows = _load_csv(SCHEDULING_MOCK_DATA_ROOT / "authorizations.csv") matches = [ row for row in rows if row.get("member_id") == member_id and _specialty_matches(specialty, row.get("service", "")) ] logger.info( f"Authorization lookup returned {len(matches)} rows for member_id={member_id}, specialty={specialty}" ) return {"ok": True, "member_id": member_id, "matches": matches} @tool def provider_specialty_rag_retriever(query: str, k: int = 5) -> Dict[str, Any]: """Retrieve provider profiles matching specialty, condition, language, notes, or location.""" logger.info("Provider Specialty RAG retrieval") results = _qdrant_search( "scheduling_provider_specialty", SCHEDULING_PROVIDER_RAG_JSONL, query, k ) if not results: try: payload = json.loads(query) if isinstance(query, str) else query except Exception: payload = {} if isinstance(payload, dict): extracted = payload.get("request") or payload.get("extracted_request") or {} benefits = payload.get("benefits") or {} member = benefits.get("member") or {} benefit_rows = benefits.get("matches") or [] plan_id = member.get("plan_id") or ( benefit_rows[0].get("plan_id") if benefit_rows else "" ) results = _provider_csv_rag_results( specialty=extracted.get("specialty", ""), plan_id=plan_id, city=extracted.get("city", ""), k=k, ) if results: logger.info( f"Provider directory fallback returned {len(results)} matches" ) return { "ok": True, "retriever": "provider_specialty_matching", "results": results, } @tool def specialist_location_lookup( specialty: str = "", city: str = "", zip_code: str = "", plan_id: str = "", network_status: str = "", ) -> Dict[str, Any]: """Lookup specialist locations from scheduling mock data.""" allowed_npis = { provider.get("provider_npi") for provider in _matching_provider_rows(specialty=specialty, plan_id=plan_id) } matches = [] for row in _load_scheduling_locations(): if allowed_npis and row.get("provider_npi") not in allowed_npis: continue if specialty: provider = next( ( p for p in _load_scheduling_providers() if p.get("provider_npi") == row.get("provider_npi") ), {}, ) if provider and not _specialty_matches(specialty, provider.get("specialty", "")): continue if city and city.lower() not in (row.get("city") or "").lower(): continue if zip_code and zip_code not in (row.get("zip") or ""): continue if network_status: provider = next( ( p for p in _load_scheduling_providers() if p.get("provider_npi") == row.get("provider_npi") ), {}, ) if network_status.lower() == "in_network" and plan_id: if plan_id not in provider.get("network_plans", ""): continue matches.append(row) logger.info(f"Specialist location lookup returned {len(matches)} rows") return {"ok": True, "matches": matches[:10]} @tool def schedule_readiness_checker( benefit_results_json: str, referral_results_json: str, authorization_results_json: str, provider_matches_json: str = "{}", ) -> Dict[str, Any]: """Check if member is ready to schedule based on benefits, referral, authorization, and provider match.""" benefits = ( json.loads(benefit_results_json) if isinstance(benefit_results_json, str) else benefit_results_json ) referrals = ( json.loads(referral_results_json) if isinstance(referral_results_json, str) else referral_results_json ) authorizations = ( json.loads(authorization_results_json) if isinstance(authorization_results_json, str) else authorization_results_json ) providers = ( json.loads(provider_matches_json) if isinstance(provider_matches_json, str) else provider_matches_json ) issues = [] benefit_rows = benefits.get("matches", []) referral_rows = referrals.get("matches", []) auth_rows = authorizations.get("matches", []) if not benefit_rows: issues.append("No matching active benefit found") referral_required = any( str(row.get("pcp_referral_required", "")).upper() == "Y" for row in benefit_rows ) if referral_required and not referral_rows: issues.append("No matching referral found") if authorizations.get("skipped"): pass elif not auth_rows: referral_optional = any( row.get("status") == "not_required_plan" for row in referral_rows ) if not referral_optional: issues.append("No matching authorization found") elif any(row.get("status") == "pending" for row in auth_rows): issues.append("Authorization pending approval") elif all(row.get("status") == "denied" for row in auth_rows): issues.append("Authorization denied") if any(row.get("status") == "expired" for row in referral_rows): issues.append("Referral expired") if provider_matches_json != "{}" and not providers.get("results"): issues.append("No matching provider found") ready_to_schedule = len(issues) == 0 logger.info( f"Schedule readiness ready_to_schedule={ready_to_schedule}, issues={issues}" ) return { "ok": True, "ready_to_schedule": ready_to_schedule, "issues": issues, "next_action": ( "proceed_to_provider_match" if ready_to_schedule else "human_scheduler_review" ), } @tool def provider_availability_lookup( provider_id_or_npi: str = "", specialty: str = "", plan_id: str = "", city: str = "", ) -> Dict[str, Any]: """Lookup open provider appointment slots from scheduling mock data.""" allowed_npis = { provider.get("provider_npi") for provider in _matching_provider_rows(specialty=specialty, plan_id=plan_id) } if provider_id_or_npi: allowed_npis = ( {provider_id_or_npi} if not allowed_npis or provider_id_or_npi in allowed_npis else set() ) locations_by_id = { row.get("location_id"): row for row in _load_scheduling_locations() } providers_by_npi = { row.get("provider_npi"): row for row in _load_scheduling_providers() } matches = [] for slot in _load_csv(SCHEDULING_MOCK_DATA_ROOT / "provider_availability.csv"): if slot.get("slot_status") != "open": continue npi = slot.get("provider_npi") if allowed_npis and npi not in allowed_npis: continue location = locations_by_id.get(slot.get("location_id"), {}) if city and city.lower() not in (location.get("city") or "").lower(): continue provider = providers_by_npi.get(npi, {}) matches.append( { **slot, "provider_name": provider.get("provider_name"), "specialty": provider.get("specialty"), "city": location.get("city"), "location_name": location.get("location_name"), "location": location, "provider": provider, } ) matches.sort(key=lambda row: row.get("start_datetime", "")) logger.info(f"Availability lookup returned {len(matches)} open slots") return {"ok": True, "matches": matches[:10]} @tool def appointment_option_ranker( provider_matches_json: str, availability_json: str ) -> Dict[str, Any]: """Rank available appointment options using provider match and availability data.""" providers = ( json.loads(provider_matches_json) if isinstance(provider_matches_json, str) else provider_matches_json ) availability = ( json.loads(availability_json) if isinstance(availability_json, str) else availability_json ) options = [] for idx, slot in enumerate(availability.get("matches", [])[:5], start=1): options.append( { "rank": idx, "slot": { "start": slot.get("start_datetime"), "end": slot.get("end_datetime"), "provider": slot.get("provider_name"), "specialty": slot.get("specialty"), "location": slot.get("location_name"), "city": slot.get("city"), "visit_type": slot.get("visit_type"), }, "reason": "In-network provider with an open slot matching the request", } ) return { "ok": True, "appointment_options": options, "provider_context": providers.get("results", [])[:3], } @tool def scheduling_summary_writer(options_json: str) -> Dict[str, Any]: """Write a concise scheduling recommendation summary.""" options = ( json.loads(options_json) if isinstance(options_json, str) else options_json ) summary = { "recommended_action": ( "offer_appointment_options" if options.get("appointment_options") else "human_scheduler_review" ), "appointment_options": options.get("appointment_options", []), } return {"ok": True, "final_summary": summary} CLAIMS_TOOLS = [ claim_packet_uploader, file_type_classifier, attachment_manifest_generator, edi_like_json_parser, provider_note_parser, claim_field_extractor, canonical_claim_schema_mapper, required_field_validator, mock_eligibility_lookup, mock_provider_npi_registry_lookup, duplicate_claim_checker, policy_benefit_rag_retriever, exception_similarity_rag_retriever, denial_risk_classifier, human_review_routing_tool, ] SCHEDULING_TOOLS = [ scheduling_request_parser, member_benefit_lookup, referral_lookup, authorization_lookup, provider_specialty_rag_retriever, specialist_location_lookup, schedule_readiness_checker, provider_availability_lookup, appointment_option_ranker, scheduling_summary_writer, ] ALL_TOOLS = CLAIMS_TOOLS + SCHEDULING_TOOLS TOOL_REGISTRY = {t.name: t for t in ALL_TOOLS}