|
1 | 1 | from __future__ import annotations
|
2 | 2 |
|
3 | 3 | import asyncio
|
| 4 | +import importlib.util |
| 5 | +import inspect |
4 | 6 | import os
|
| 7 | +from glob import glob |
5 | 8 | from logging import Logger
|
| 9 | +from pathlib import Path |
6 | 10 | from time import time_ns
|
7 | 11 | from typing import TYPE_CHECKING, ClassVar
|
8 | 12 |
|
@@ -135,6 +139,10 @@ def _init_persona_classes(self) -> None:
|
135 | 139 | "ERROR: Jupyter AI has no AI personas available. "
|
136 | 140 | + "Please verify your server configuration and open a new issue on our GitHub repo if this warning persists."
|
137 | 141 | )
|
| 142 | + |
| 143 | + # TODO: check whether in the end we need a full class for this, or if we can just use a simple function that returns a list of persona classes from the local filesystem. |
| 144 | + persona_classes.extend(LocalPersonaLoader(self.root_dir).load_persona_classes()) |
| 145 | + |
138 | 146 | PersonaManager._persona_classes = persona_classes
|
139 | 147 |
|
140 | 148 | def _init_personas(self) -> dict[str, BasePersona]:
|
@@ -260,3 +268,107 @@ def get_chat_dir(self) -> str:
|
260 | 268 | """
|
261 | 269 | abspath = self.get_chat_path(absolute=True)
|
262 | 270 | return os.path.dirname(abspath)
|
| 271 | + |
| 272 | + |
| 273 | +class LocalPersonaLoader(LoggingConfigurable): |
| 274 | + """ |
| 275 | + Load _persona class declarations_ from the local filesystem. |
| 276 | +
|
| 277 | + Those class declarations are then used to instantiate personas by the `PersonaManager`. |
| 278 | +
|
| 279 | + TODO: wire to local .jupyter directory system, for now just use the current |
| 280 | + directory. |
| 281 | + """ |
| 282 | + |
| 283 | + def __init__( |
| 284 | + self, |
| 285 | + *args, |
| 286 | + root_dir: str | None = None, |
| 287 | + **kwargs, |
| 288 | + ): |
| 289 | + # Forward other arguments to parent class |
| 290 | + super().__init__(*args, **kwargs) |
| 291 | + |
| 292 | + # Set default root directory to current working directory if not provided |
| 293 | + self.root_dir = root_dir or os.getcwd() |
| 294 | + |
| 295 | + self.log.info(f"LocalPersonaLoader initialized with root directory: {self.root_dir}") |
| 296 | + |
| 297 | + def load_persona_classes(self) -> list[type[BasePersona]]: |
| 298 | + """ |
| 299 | + Loads persona classes from Python files in the local filesystem. |
| 300 | + |
| 301 | + Scans the root_dir for .py files, dynamically imports them, and extracts |
| 302 | + any class declarations that are subclasses of BasePersona. |
| 303 | + """ |
| 304 | + persona_classes: list[type[BasePersona]] = [] |
| 305 | + |
| 306 | + # Check if root directory exists |
| 307 | + if not os.path.exists(self.root_dir): |
| 308 | + self.log.info(f"Root directory does not exist: {self.root_dir}") |
| 309 | + return persona_classes |
| 310 | + |
| 311 | + # Find all .py files in the root directory |
| 312 | + py_files = glob(os.path.join(self.root_dir, "*.py")) |
| 313 | + |
| 314 | + if not py_files: |
| 315 | + self.log.info(f"No Python files found in directory: {self.root_dir}") |
| 316 | + return persona_classes |
| 317 | + |
| 318 | + self.log.info(f"Found {len(py_files)} Python files in {self.root_dir}") |
| 319 | + self.log.info("PENDING: Loading persona classes from local Python files...") |
| 320 | + start_time_ns = time_ns() |
| 321 | + |
| 322 | + for py_file in py_files: |
| 323 | + try: |
| 324 | + # Get module name from file path |
| 325 | + module_name = Path(py_file).stem |
| 326 | + |
| 327 | + # Skip if module name starts with underscore (private modules) |
| 328 | + if module_name.startswith('_'): |
| 329 | + continue |
| 330 | + |
| 331 | + # Create module spec and load the module |
| 332 | + spec = importlib.util.spec_from_file_location(module_name, py_file) |
| 333 | + if spec is None or spec.loader is None: |
| 334 | + self.log.warning(f" - Unable to create module spec for {py_file}") |
| 335 | + continue |
| 336 | + |
| 337 | + module = importlib.util.module_from_spec(spec) |
| 338 | + spec.loader.exec_module(module) |
| 339 | + |
| 340 | + # Find all classes in the module that are BasePersona subclasses |
| 341 | + module_persona_classes = [] |
| 342 | + for name, obj in inspect.getmembers(module, inspect.isclass): |
| 343 | + # Check if it's a subclass of BasePersona but not BasePersona itself |
| 344 | + if (issubclass(obj, BasePersona) and |
| 345 | + obj is not BasePersona and |
| 346 | + obj.__module__ == module_name): |
| 347 | + module_persona_classes.append(obj) |
| 348 | + |
| 349 | + if module_persona_classes: |
| 350 | + persona_classes.extend(module_persona_classes) |
| 351 | + class_names = [cls.__name__ for cls in module_persona_classes] |
| 352 | + self.log.info( |
| 353 | + f" - Loaded {len(module_persona_classes)} persona class(es) from '{py_file}': {class_names}" |
| 354 | + ) |
| 355 | + else: |
| 356 | + self.log.debug(f" - No persona classes found in '{py_file}'") |
| 357 | + |
| 358 | + except Exception: |
| 359 | + # On exception, log an error and continue |
| 360 | + # This mirrors the error handling pattern from entry point loading |
| 361 | + self.log.exception( |
| 362 | + f" - Unable to load persona classes from '{py_file}' due to an exception printed below." |
| 363 | + ) |
| 364 | + continue |
| 365 | + |
| 366 | + if len(persona_classes) > 0: |
| 367 | + elapsed_time_ms = (time_ns() - start_time_ns) // 1_000_000 |
| 368 | + self.log.info( |
| 369 | + f"SUCCESS: Loaded {len(persona_classes)} persona classes from local filesystem. Time elapsed: {elapsed_time_ms}ms." |
| 370 | + ) |
| 371 | + else: |
| 372 | + self.log.info("No persona classes found in local filesystem.") |
| 373 | + |
| 374 | + return persona_classes |
0 commit comments