diff --git a/mellea/plugins/base.py b/mellea/plugins/base.py index a68244835..f77bd8743 100644 --- a/mellea/plugins/base.py +++ b/mellea/plugins/base.py @@ -120,6 +120,13 @@ async def __aexit__( _HAS_PLUGIN_FRAMEWORK = False if TYPE_CHECKING: + from cpex.framework.base import Plugin as _CpexPlugin + from cpex.framework.models import ( + PluginContext, + PluginPayload, + PluginResult as _CFPluginResult, + ) + from mellea.core.backend import Backend from mellea.core.base import Context from mellea.stdlib.session import MelleaSession @@ -146,119 +153,146 @@ def __init__( # noqa: D107 super().__init__(f"Plugin blocked {hook_type}: {detail}{reason}") -if _HAS_PLUGIN_FRAMEWORK: +# --------------------------------------------------------------------------- +# Dynamic base classes — a single ClassDef node per class ensures Griffe's +# static AST parser always picks up the authoritative docstring. See #667. +# --------------------------------------------------------------------------- - class MelleaBasePayload(PluginPayload): - """Frozen base — all payloads are immutable by design. +if _HAS_PLUGIN_FRAMEWORK or TYPE_CHECKING: + _PayloadBase = PluginPayload + _PluginBase = _CpexPlugin +else: + _PayloadBase = object + _PluginBase = object - Plugins must use ``model_copy(update={...})`` to propose modifications - and return the copy via ``PluginResult.modified_payload``. The plugin - manager applies the hook's ``HookPayloadPolicy`` to filter changes to - writable fields only. - """ - session_id: str | None = None - request_id: str = "" - timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) - hook: str = "" - user_metadata: dict[str, Any] = Field(default_factory=dict) +class MelleaBasePayload(_PayloadBase): # type: ignore[misc,valid-type] + """Frozen base — all payloads are immutable by design. - class MelleaPlugin(_CpexPlugin): - """Base class for Mellea plugins with lifecycle hooks and typed accessors. + Plugins must use ``model_copy(update={...})`` to propose modifications + and return the copy via ``PluginResult.modified_payload``. The plugin + manager applies the hook's ``HookPayloadPolicy`` to filter changes to + writable fields only. + """ - Use this when you need lifecycle hooks (``initialize``/``shutdown``) - or typed context accessors. For simpler plugins, prefer ``@hook`` - on standalone functions or ``@plugin`` on plain classes. + session_id: str | None = None + request_id: str = "" + timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) + hook: str = "" + user_metadata: dict[str, Any] = Field(default_factory=dict) - Instances support the context manager protocol for temporary activation:: - class MyPlugin(MelleaPlugin): - def __init__(self): - super().__init__(PluginConfig(name="my-plugin", hooks=[...])) +class MelleaPlugin(_PluginBase): # type: ignore[misc,valid-type] + """Base class for Mellea plugins with lifecycle hooks and typed accessors. - async def some_hook(self, payload, ctx): - ... + Use this when you need lifecycle hooks (``initialize``/``shutdown``) + or typed context accessors. For simpler plugins, prefer ``@hook`` + on standalone functions or ``@plugin`` on plain classes. - with MyPlugin() as p: - result, ctx = instruct("...", ctx, backend) + Instances support the context manager protocol for temporary activation:: - # or async - async with MyPlugin() as p: - result, ctx = await ainstruct("...", ctx, backend) - """ + class MyPlugin(MelleaPlugin): + def __init__(self): + super().__init__(PluginConfig(name="my-plugin", hooks=[...])) - def get_backend(self, context: PluginContext) -> Backend | None: - """Get the Backend from the plugin context.""" - return context.global_context.state.get("backend") + async def some_hook(self, payload, ctx): + ... - def get_mellea_context(self, context: PluginContext) -> Context | None: - """Get the Mellea Context from the plugin context.""" - return context.global_context.state.get("context") + with MyPlugin() as p: + result, ctx = instruct("...", ctx, backend) - def get_session(self, context: PluginContext) -> MelleaSession | None: - """Get the MelleaSession from the plugin context.""" - return context.global_context.state.get("session") + # or async + async with MyPlugin() as p: + result, ctx = await ainstruct("...", ctx, backend) + """ - @property - def plugin_config(self) -> dict[str, Any]: - """Plugin-specific configuration from PluginConfig.config.""" - return self._config.config or {} + def get_backend(self, context: PluginContext) -> Backend | None: + """Get the Backend from the plugin context. - def __enter__(self) -> MelleaPlugin: - """Register this plugin for the duration of a ``with`` block.""" - if getattr(self, "_scope_id", None) is not None: - raise RuntimeError( - f"MelleaPlugin {self.name!r} is already active as a context manager. " - "Concurrent or nested reuse of the same instance is not supported; " - "create a new instance instead." - ) - import uuid + Args: + context: The plugin context provided by the hook framework. - from mellea.plugins.registry import register + Returns: + The active Backend, or ``None`` if unavailable. + """ + return context.global_context.state.get("backend") - self._scope_id = str(uuid.uuid4()) - register(self, session_id=self._scope_id) - return self + def get_mellea_context(self, context: PluginContext) -> Context | None: + """Get the Mellea Context from the plugin context. - def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: - """Deregister this plugin on exit.""" - scope_id = getattr(self, "_scope_id", None) - if scope_id is not None: - from mellea.plugins.manager import deregister_session_plugins + Args: + context: The plugin context provided by the hook framework. - deregister_session_plugins(scope_id) - self._scope_id = None # type: ignore[assignment] + Returns: + The active Mellea Context, or ``None`` if unavailable. + """ + return context.global_context.state.get("context") - async def __aenter__(self) -> MelleaPlugin: - """Async variant — delegates to the synchronous ``__enter__``.""" - return self.__enter__() + def get_session(self, context: PluginContext) -> MelleaSession | None: + """Get the MelleaSession from the plugin context. - async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: - """Async variant — delegates to the synchronous ``__exit__``.""" - self.__exit__(exc_type, exc_val, exc_tb) + Args: + context: The plugin context provided by the hook framework. - PluginResult: TypeAlias = _CFPluginResult # type: ignore[misc] + Returns: + The active MelleaSession, or ``None`` if unavailable. + """ + return context.global_context.state.get("session") -else: - # Provide a stub when the plugin framework is not installed. - class MelleaBasePayload: # type: ignore[no-redef] - """Stub — install ``"mellea[hooks]"`` for full plugin support.""" - - def __init__(self, *args: Any, **kwargs: Any) -> None: # noqa: D107 - raise ImportError( - "MelleaPlugin requires the ContextForge plugin framework. " - "Install it with: pip install 'mellea[hooks]'" + @property + def plugin_config(self) -> dict[str, Any]: + """Plugin-specific configuration from PluginConfig.config.""" + return self._config.config or {} + + def __enter__(self) -> MelleaPlugin: + """Register this plugin for the duration of a ``with`` block.""" + if getattr(self, "_scope_id", None) is not None: + raise RuntimeError( + f"MelleaPlugin {self.name!r} is already active as a context manager. " + "Concurrent or nested reuse of the same instance is not supported; " + "create a new instance instead." ) + import uuid - # Provide a stub when the plugin framework is not installed. - class MelleaPlugin: # type: ignore[no-redef] - """Stub — install ``"mellea[hooks]"`` for full plugin support.""" + from mellea.plugins.registry import register - def __init__(self, *args: Any, **kwargs: Any) -> None: # noqa: D107 - raise ImportError( - "MelleaPlugin requires the ContextForge plugin framework. " - "Install it with: pip install 'mellea[hooks]'" - ) + self._scope_id = str(uuid.uuid4()) + register(self, session_id=self._scope_id) + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + """Deregister this plugin on exit.""" + scope_id = getattr(self, "_scope_id", None) + if scope_id is not None: + from mellea.plugins.manager import deregister_session_plugins + + deregister_session_plugins(scope_id) + self._scope_id = None # type: ignore[assignment] - # Provide an alias when the plugin framework is not installed. - PluginResult: TypeAlias = Any # type: ignore[no-redef, misc] + async def __aenter__(self) -> MelleaPlugin: + """Async variant — delegates to the synchronous ``__enter__``.""" + return self.__enter__() + + async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + """Async variant — delegates to the synchronous ``__exit__``.""" + self.__exit__(exc_type, exc_val, exc_tb) + + +# When cpex is not installed, override __init__ to provide a clear error +# message. This is done post-hoc so there is only one ClassDef node in the +# AST for each class (the docstring-carrying definition above). +if not _HAS_PLUGIN_FRAMEWORK: + + def _stub_init(_self: Any, *_args: Any, **_kwargs: Any) -> None: + raise ImportError( + "This class requires the ContextForge plugin framework. " + "Install it with: pip install 'mellea[hooks]'" + ) + + MelleaBasePayload.__init__ = _stub_init # type: ignore[assignment,misc] + MelleaPlugin.__init__ = _stub_init # type: ignore[assignment,misc] + +if _HAS_PLUGIN_FRAMEWORK: + PluginResult: TypeAlias = _CFPluginResult # type: ignore[misc] +else: + PluginResult: TypeAlias = Any # type: ignore[no-redef,misc] diff --git a/mellea/stdlib/components/intrinsic/core.py b/mellea/stdlib/components/intrinsic/core.py index 7f6c52622..55f79b48b 100644 --- a/mellea/stdlib/components/intrinsic/core.py +++ b/mellea/stdlib/components/intrinsic/core.py @@ -13,10 +13,12 @@ def check_certainty(context: ChatContext, backend: AdapterMixin) -> float: assistant's response to a user's question. The context should end with a user question followed by an assistant answer. - :param context: Chat context containing user question and assistant answer. - :param backend: Backend instance that supports LoRA/aLoRA adapters. + Args: + context: Chat context containing user question and assistant answer. + backend: Backend instance that supports LoRA/aLoRA adapters. - :return: Certainty score as a float (higher = more certain). + Returns: + Certainty score as a float (higher = more certain). """ result_json = call_intrinsic("uncertainty", context, backend) return result_json["certainty"] @@ -40,11 +42,13 @@ def requirement_check( requirements. Appends an evaluation prompt to the context following the format specified by the Granite Guardian requirement checker model card. - :param context: Chat context containing user question and assistant answer. - :param backend: Backend instance that supports LoRA/aLoRA adapters. - :param requirement: set of requirements to satisfy + Args: + context: Chat context containing user question and assistant answer. + backend: Backend instance that supports LoRA/aLoRA adapters. + requirement: Set of requirements to satisfy. - :return: Score as a float between 0.0 and 1.0 (higher = more likely satisfied). + Returns: + Score as a float between 0.0 and 1.0 (higher = more likely satisfied). """ eval_message = f": {requirement}\n{_EVALUATION_PROMPT}" context = context.add(Message("user", eval_message))