| """Real Estate Intelligence Layer - Tool Registry & Dynamic Tool Discovery.""" |
|
|
| from __future__ import annotations |
|
|
| from typing import Any, Dict, List, Optional |
| import numpy as np |
|
|
| from services.tool_schemas import ( |
| COMMERCIAL_TOOLS, |
| GENERATE_CONTACT_BUTTONS_TOOL, |
| GENERATE_DATA_EXPORT_TOOL, |
| REAL_ESTATE_TOOLS, |
| SUGGEST_ACTIONS_TOOL, |
| ) |
|
|
| __all__ = [ |
| "REAL_ESTATE_TOOLS", |
| "COMMERCIAL_TOOLS", |
| "SUGGEST_ACTIONS_TOOL", |
| "GENERATE_DATA_EXPORT_TOOL", |
| "GENERATE_CONTACT_BUTTONS_TOOL", |
| "TOOL_REGISTRY", |
| "discover_tools", |
| "select_tools", |
| "invalidate_tool_embeddings", |
| ] |
| TOOL_REGISTRY: List[Dict[str, Any]] = [ |
| { |
| "name": "get_real_estate_kpis", |
| "category": "MARKET", |
| "retrieval_description": "High-level summary cards, aggregate real estate portfolio metrics, active properties count, availability count, 7-day rate changes, scrape health status, overall KPI numbers.", |
| "schema": REAL_ESTATE_TOOLS[0], |
| }, |
| { |
| "name": "get_market_averages", |
| "category": "MARKET", |
| "retrieval_description": "Average nightly rate and price baselines for any tracked market region over 7-day, 14-day, or 30-day periods, market benchmark pricing.", |
| "schema": REAL_ESTATE_TOOLS[1], |
| }, |
| { |
| "name": "get_market_snapshot", |
| "category": "MARKET", |
| "retrieval_description": "Daily overview snapshot of market performance, market condition in any tracked city or region, nightly rates range, minimum maximum rates, daily summary and availability.", |
| "schema": REAL_ESTATE_TOOLS[2], |
| }, |
| { |
| "name": "get_market_trend", |
| "category": "MARKET", |
| "retrieval_description": "Historical pricing trend direction, rising or falling rates over time, week-over-week rate trajectory in any tracked market region.", |
| "schema": REAL_ESTATE_TOOLS[3], |
| }, |
| { |
| "name": "get_market_rate_changes", |
| "category": "ANOMALY", |
| "categories": ["ANOMALY", "MARKET", "PROPERTY"], |
| "retrieval_description": ( |
| "Count of properties with any nightly rate increase or decrease in a market, " |
| "number of price increases, number of price decreases, properties that changed rates, " |
| "rate movements not just spikes, any rate change at all, how many listings adjusted rates, " |
| "properties with rate increases or decreases in Nigeria Abuja Lagos, how many had price changes, " |
| "nightly rate went up down in last few days." |
| ), |
| "schema": REAL_ESTATE_TOOLS[4], |
| }, |
| { |
| "name": "get_spike_alerts", |
| "category": "ANOMALY", |
| "retrieval_description": "Sudden sharp price jumps, 25% plus rate increases, price spikes, surge pricing alerts, unseasonal rate surges in any tracked market, ranked slices.", |
| "schema": REAL_ESTATE_TOOLS[5], |
| }, |
| { |
| "name": "get_rate_anomaly_report", |
| "category": "ANOMALY", |
| "retrieval_description": "Anomalous pricing patterns, drastic rate drops, price crashes, listings deviating significantly from market baseline.", |
| "schema": REAL_ESTATE_TOOLS[6], |
| }, |
| { |
| "name": "get_most_volatile_properties", |
| "category": "ANOMALY", |
| "retrieval_description": "Listings with the most frequent price fluctuations, highest number of rate adjustments, unstable dynamic pricing, ranked slices.", |
| "schema": REAL_ESTATE_TOOLS[7], |
| }, |
| { |
| "name": "get_property_snapshot", |
| "category": "PROPERTY", |
| "retrieval_description": "Detailed current status and single listing profile for a specific property UUID, property name, or listing URL.", |
| "schema": REAL_ESTATE_TOOLS[8], |
| }, |
| { |
| "name": "get_property_detail", |
| "category": "PROPERTY", |
| "retrieval_description": "Deep single-property exploration, full listing profile, active tracking status, coordinates, URL, and daily-aggregated historical rate revision log.", |
| "schema": REAL_ESTATE_TOOLS[9], |
| }, |
| { |
| "name": "get_property_rate_changes", |
| "category": "PROPERTY", |
| "retrieval_description": "Historical log of price changes, chronological rate revisions, daily past rate adjustments for a single property.", |
| "schema": REAL_ESTATE_TOOLS[10], |
| }, |
| { |
| "name": "compare_properties", |
| "category": "PROPERTY", |
| "retrieval_description": "Side-by-side head-to-head comparison of multiple property listings, prices, bedrooms, and metrics.", |
| "schema": REAL_ESTATE_TOOLS[11], |
| }, |
| { |
| "name": "search_properties", |
| "category": "PROPERTY", |
| "retrieval_description": "Find, list, and filter property listings by bedroom count, price range, market region, availability status, listing name, or ranked slices.", |
| "schema": REAL_ESTATE_TOOLS[12], |
| }, |
| { |
| "name": "get_availability_rate", |
| "category": "PROPERTY", |
| "retrieval_description": "Percentage of units booked versus available, occupancy rates, reservation calendar status in any tracked market or city.", |
| "schema": REAL_ESTATE_TOOLS[13], |
| }, |
| { |
| "name": "geocode_address", |
| "category": "GEO", |
| "retrieval_description": "Convert street address, landmark, or neighborhood name into latitude and longitude geographic coordinates.", |
| "schema": REAL_ESTATE_TOOLS[14], |
| }, |
| { |
| "name": "get_nearby_properties", |
| "category": "GEO", |
| "retrieval_description": "Find listings closest to a specific geographic coordinate, landmark, or neighborhood radius.", |
| "schema": REAL_ESTATE_TOOLS[15], |
| }, |
| { |
| "name": "get_distance_km", |
| "category": "GEO", |
| "retrieval_description": "Calculate straight-line distance in kilometers between two properties or locations by UUID.", |
| "schema": REAL_ESTATE_TOOLS[16], |
| }, |
| { |
| "name": "get_tracked_markets", |
| "category": "MARKET", |
| "retrieval_description": "List all cities and regions currently supported and monitored in the database, fetch the live market registry.", |
| "schema": REAL_ESTATE_TOOLS[17], |
| }, |
| { |
| "name": "get_recently_changed_tracking", |
| "category": "MARKET", |
| "retrieval_description": "Properties recently added to or removed from active monitoring tracking status.", |
| "schema": REAL_ESTATE_TOOLS[18], |
| }, |
| ] |
|
|
| _TOOL_EMBEDDINGS: Optional[np.ndarray] = None |
|
|
|
|
| def _get_tool_embeddings() -> np.ndarray: |
| """Compute and cache normalized vector embeddings for all retrieval descriptions.""" |
| global _TOOL_EMBEDDINGS |
| if _TOOL_EMBEDDINGS is None: |
| from services.embedding_service import get_embedding_model |
| embed_model = get_embedding_model() |
| descs = [entry["retrieval_description"] for entry in TOOL_REGISTRY] |
| _TOOL_EMBEDDINGS = embed_model.encode(descs, normalize_embeddings=True) |
| return _TOOL_EMBEDDINGS |
|
|
|
|
| def _filter_candidate_indices(categories: Optional[List[str]]) -> List[int]: |
| """Filter tool registry indices by category gating if categories specified.""" |
| if not categories: |
| return list(range(len(TOOL_REGISTRY))) |
|
|
| allowed_cats = {c.upper() for c in categories} |
| candidates: List[int] = [] |
| for i, entry in enumerate(TOOL_REGISTRY): |
| entry_cats = entry.get("categories") |
| if entry_cats: |
| if any(c.upper() in allowed_cats for c in entry_cats): |
| candidates.append(i) |
| elif entry.get("category", "").upper() in allowed_cats: |
| candidates.append(i) |
|
|
| return candidates or list(range(len(TOOL_REGISTRY))) |
|
|
|
|
| def _rank_candidate_tools( |
| embed_model: Any, |
| tool_vectors: np.ndarray, |
| candidate_indices: List[int], |
| user_query: str, |
| top_k: int, |
| ) -> List[Dict[str, Any]]: |
| """Rank candidate tools by vector similarity against the user query.""" |
| q_vec = embed_model.encode([user_query], normalize_embeddings=True)[0] |
| candidate_vecs = tool_vectors[candidate_indices] |
| sims = candidate_vecs @ q_vec |
| ranked_local_indices = np.argsort(sims)[::-1][:top_k] |
| return [ |
| TOOL_REGISTRY[candidate_indices[local_idx]]["schema"] |
| for local_idx in ranked_local_indices |
| ] |
|
|
|
|
| def _attach_universal_tools( |
| selected_tools: List[Dict[str, Any]], |
| include_export: bool, |
| ) -> List[Dict[str, Any]]: |
| """Deduplicate tools and append universal export tool if requested.""" |
| results: List[Dict[str, Any]] = [] |
| seen_names: set[str] = set() |
|
|
| for tool in selected_tools: |
| name = tool["function"]["name"] |
| if name not in seen_names: |
| seen_names.add(name) |
| results.append(tool) |
|
|
| if include_export and "generate_data_export" not in seen_names: |
| results.append(GENERATE_DATA_EXPORT_TOOL) |
|
|
| return results |
|
|
|
|
| def discover_tools( |
| user_query: str, |
| categories: Optional[List[str]] = None, |
| top_k: int = 4, |
| include_export: bool = True, |
| ) -> List[Dict[str, Any]]: |
| """Hierarchical dynamic tool discovery ranking top-K tools for user query.""" |
| try: |
| from services.embedding_service import get_embedding_model |
| embed_model = get_embedding_model() |
| tool_vectors = _get_tool_embeddings() |
| candidate_indices = _filter_candidate_indices(categories) |
| selected_tools = _rank_candidate_tools( |
| embed_model, tool_vectors, candidate_indices, user_query, top_k |
| ) |
| except Exception: |
| selected_tools = [ |
| t for t in REAL_ESTATE_TOOLS |
| if t["function"]["name"] in ("get_market_averages", "get_market_snapshot", "get_spike_alerts") |
| ] |
|
|
| return _attach_universal_tools(selected_tools, include_export) |
|
|
|
|
| def invalidate_tool_embeddings() -> None: |
| """Reset the tool embedding cache upon market registry refreshes.""" |
| global _TOOL_EMBEDDINGS |
| _TOOL_EMBEDDINGS = None |
|
|
|
|
| def select_tools(user_query: str) -> List[Dict[str, Any]]: |
| """Backwards-compatible alias forwarding to discover_tools().""" |
| return discover_tools(user_query) |
|
|
