Spaces:
Running
Running
File size: 2,092 Bytes
9eecab5 | 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 | from utils.logger import logger
class MetadataAgent:
def __init__(self, registry):
self.registry = registry
def _detect_dataset(self, query, datasets):
q = query.lower()
for d in datasets:
if d.lower() in q:
return d
# fallback to first dataset
return datasets[0]
def handle(self, query):
q = query.lower()
try:
datasets = self.registry.list_datasets()
if not datasets:
return "No datasets available."
dataset = self._detect_dataset(q, datasets)
meta = self.registry.get_info(dataset)
cols = meta.get("columns", [])
nums = meta.get("numeric_columns", [])
cats = meta.get("categorical_columns", [])
miss = meta.get("missing_values", {})
# ---- INTENT DETECTION ----
if "how many column" in q or "number of column" in q:
return f"{dataset} has {len(cols)} columns."
if "numeric" in q:
if not nums:
return f"No numeric columns found in {dataset}."
return f"Numeric columns in {dataset}: {', '.join(nums)}"
if "categorical" in q:
if not cats:
return f"No categorical columns found in {dataset}."
return f"Categorical columns in {dataset}: {', '.join(cats)}"
if "missing" in q:
if not miss:
return f"No missing value info for {dataset}."
return f"Missing values in {dataset}: {miss}"
if "column" in q:
if not cols:
return f"No columns found for {dataset}."
return f"Columns in {dataset}: {', '.join(cols)}"
logger.info(f"MetadataAgent | dataset={dataset} | query={query}")
return "Metadata query not understood."
except Exception as e:
logger.error(f"Metadata agent failed | {e}")
return "Metadata agent error." |