codekingpro/portable-devtools

1"""Graph lifecycle callback interfaces and event payloads.2 3This module defines the public callback surface for observing LangGraph-specific4lifecycle transitions such as interrupt and resume.5"""6 7from __future__ import annotations8 9from collections.abc import Sequence10from dataclasses import dataclass11from typing import Any, Literal, TypeAlias, TypeVar12from uuid import UUID13 14from langchain_core.callbacks import BaseCallbackHandler, BaseCallbackManager15from langchain_core.callbacks.manager import ahandle_event, handle_event16from langchain_core.runnables import RunnableConfig17 18from langgraph.types import Interrupt19 20__all__ = (21    "GraphCallbackHandler",22    "GraphInterruptEvent",23    "GraphLifecycleEvent",24    "GraphLifecycleStatus",25    "GraphResumeEvent",26    "get_async_graph_callback_manager_for_config",27    "get_sync_graph_callback_manager_for_config",28)29 30 31GraphLifecycleStatus: TypeAlias = Literal[32    "input",33    "pending",34    "done",35    "interrupt_before",36    "interrupt_after",37    "out_of_steps",38]39"""Allowed lifecycle statuses reported in graph lifecycle callback events."""40 41 42@dataclass(frozen=True)43class GraphInterruptEvent:44    """Graph lifecycle event emitted when execution pauses for interrupts."""45 46    run_id: UUID | None47    """Run id for the current graph execution, if available."""48 49    status: GraphLifecycleStatus50    """Loop status when the interrupt was captured."""51 52    checkpoint_id: str53    """Checkpoint id associated with the interrupted execution."""54 55    checkpoint_ns: tuple[str, ...]56    """Checkpoint namespace path for the current graph or subgraph."""57 58    interrupts: tuple[Interrupt, ...]59    """Interrupt payloads that caused the graph to pause."""60 61 62@dataclass(frozen=True)63class GraphResumeEvent:64    """Graph lifecycle event emitted when execution resumes from a checkpoint."""65 66    run_id: UUID | None67    """Run id for the current graph execution, if available."""68 69    status: GraphLifecycleStatus70    """Loop status when the resume was captured."""71 72    checkpoint_id: str73    """Checkpoint id the graph resumed from."""74 75    checkpoint_ns: tuple[str, ...]76    """Checkpoint namespace path for the current graph or subgraph."""77 78 79GraphLifecycleEvent: TypeAlias = GraphInterruptEvent | GraphResumeEvent80"""Union of all public graph lifecycle callback event payloads.81 82Use this alias when a callback or helper can receive either interrupt or resume83lifecycle events.84"""85 86 87class GraphCallbackHandler(BaseCallbackHandler):88    """Base class for graph-level lifecycle callbacks.89 90    Subclass this handler to observe graph lifecycle transitions that are91    specific to LangGraph execution, rather than generic LangChain runnable92    callbacks.93 94    Instances can be passed through `config["callbacks"]` when invoking a95    graph. Only handlers that inherit from `GraphCallbackHandler` receive these96    lifecycle events.97    """98 99    def on_interrupt(self, event: GraphInterruptEvent) -> Any:100        """Run when graph execution pauses due to one or more interrupts.101 102        Args:103            event: Interrupt lifecycle event payload.104        """105 106    def on_resume(self, event: GraphResumeEvent) -> Any:107        """Run when graph execution resumes from a persisted checkpoint.108 109        Args:110            event: Resume lifecycle event payload.111        """112 113 114_MISSING = object()115 116 117def _filter_graph_handlers(118    handlers: list[BaseCallbackHandler],119) -> list[GraphCallbackHandler]:120    return [h for h in handlers if isinstance(h, GraphCallbackHandler)]121 122 123def _init_base_manager(124    manager: BaseCallbackManager,125    handlers: Sequence[GraphCallbackHandler] | None,126    inheritable_handlers: Sequence[GraphCallbackHandler] | None,127    parent_run_id: UUID | None,128    *,129    tags: list[str] | None,130    inheritable_tags: list[str] | None,131    metadata: dict[str, Any] | None,132    inheritable_metadata: dict[str, Any] | None,133    run_id: UUID | None,134) -> None:135    base_handlers: list[BaseCallbackHandler] = []136    base_inheritable_handlers: list[BaseCallbackHandler] = []137    if handlers is not None:138        base_handlers.extend(handlers)139    if inheritable_handlers is not None:140        base_inheritable_handlers.extend(inheritable_handlers)141    BaseCallbackManager.__init__(142        manager,143        handlers=base_handlers,144        inheritable_handlers=base_inheritable_handlers,145        parent_run_id=parent_run_id,146        tags=tags,147        inheritable_tags=inheritable_tags,148        metadata=metadata,149        inheritable_metadata=inheritable_metadata,150    )151    manager.run_id = run_id  # type: ignore[attr-defined]152 153 154def _configure_graph_callbacks(155    cls: type[_GraphManagerT],156    callbacks: object | None,157    *,158    run_id: UUID | None,159) -> _GraphManagerT:160    if callbacks is None:161        return cls(run_id=run_id)162    if isinstance(callbacks, cls):163        return callbacks.copy(run_id=run_id)164    if isinstance(callbacks, (_GraphCallbackManager, _AsyncGraphCallbackManager)):165        # Cross-type: extract handlers into the requested cls.166        return cls(167            handlers=_filter_graph_handlers(callbacks.handlers),168            inheritable_handlers=_filter_graph_handlers(callbacks.inheritable_handlers),169            parent_run_id=callbacks.parent_run_id,170            tags=callbacks.tags.copy(),171            inheritable_tags=callbacks.inheritable_tags.copy(),172            metadata=callbacks.metadata.copy(),173            inheritable_metadata=callbacks.inheritable_metadata.copy(),174            run_id=run_id,175        )176    if isinstance(callbacks, BaseCallbackManager):177        return cls(178            handlers=_filter_graph_handlers(callbacks.handlers),179            inheritable_handlers=_filter_graph_handlers(callbacks.inheritable_handlers),180            parent_run_id=callbacks.parent_run_id,181            tags=callbacks.tags.copy(),182            inheritable_tags=callbacks.inheritable_tags.copy(),183            metadata=callbacks.metadata.copy(),184            inheritable_metadata=callbacks.inheritable_metadata.copy(),185            run_id=run_id,186        )187    if isinstance(callbacks, GraphCallbackHandler):188        return cls((callbacks,), run_id=run_id)189    if isinstance(callbacks, (str, bytes)) or not isinstance(callbacks, Sequence):190        raise TypeError("callbacks must be a handler, sequence, or manager")191    return cls(_filter_graph_handlers(list(callbacks)), run_id=run_id)192 193 194def _copy_graph_manager(195    manager: _GraphCallbackManager | _AsyncGraphCallbackManager,196    cls: type[_GraphManagerT],197    run_id: UUID | None | object,198) -> _GraphManagerT:199    resolved_run_id: UUID | None200    if run_id is _MISSING:201        resolved_run_id = manager.run_id202    else:203        if run_id is not None and not isinstance(run_id, UUID):204            raise TypeError("run_id must be a UUID or None")205        resolved_run_id = run_id206 207    return cls(208        handlers=_filter_graph_handlers(manager.handlers),209        inheritable_handlers=_filter_graph_handlers(manager.inheritable_handlers),210        parent_run_id=manager.parent_run_id,211        tags=manager.tags.copy(),212        inheritable_tags=manager.inheritable_tags.copy(),213        metadata=manager.metadata.copy(),214        inheritable_metadata=manager.inheritable_metadata.copy(),215        run_id=resolved_run_id,216    )217 218 219class _GraphCallbackManager(BaseCallbackManager):220    """Sync dispatcher for graph lifecycle events."""221 222    run_id: UUID | None223 224    def __init__(225        self,226        handlers: Sequence[GraphCallbackHandler] | None = None,227        inheritable_handlers: Sequence[GraphCallbackHandler] | None = None,228        parent_run_id: UUID | None = None,229        *,230        tags: list[str] | None = None,231        inheritable_tags: list[str] | None = None,232        metadata: dict[str, Any] | None = None,233        inheritable_metadata: dict[str, Any] | None = None,234        run_id: UUID | None = None,235    ) -> None:236        _init_base_manager(237            self,238            handlers,239            inheritable_handlers,240            parent_run_id,241            tags=tags,242            inheritable_tags=inheritable_tags,243            metadata=metadata,244            inheritable_metadata=inheritable_metadata,245            run_id=run_id,246        )247 248    def copy(249        self,250        *,251        run_id: UUID | None | object = _MISSING,252    ) -> _GraphCallbackManager:253        return _copy_graph_manager(self, _GraphCallbackManager, run_id)254 255    @classmethod256    def configure(257        cls,258        callbacks: object | None = None,259        *,260        run_id: UUID | None = None,261    ) -> _GraphCallbackManager:262        return _configure_graph_callbacks(cls, callbacks, run_id=run_id)263 264    def on_interrupt(self, event: GraphInterruptEvent) -> None:265        handle_event(266            self.handlers,267            "on_interrupt",268            None,269            event,270        )271 272    def on_resume(self, event: GraphResumeEvent) -> None:273        handle_event(274            self.handlers,275            "on_resume",276            None,277            event,278        )279 280 281class _AsyncGraphCallbackManager(BaseCallbackManager):282    """Async dispatcher for graph lifecycle events."""283 284    run_id: UUID | None285 286    @property287    def is_async(self) -> bool:288        """Return whether the manager is async."""289        return True290 291    def __init__(292        self,293        handlers: Sequence[GraphCallbackHandler] | None = None,294        inheritable_handlers: Sequence[GraphCallbackHandler] | None = None,295        parent_run_id: UUID | None = None,296        *,297        tags: list[str] | None = None,298        inheritable_tags: list[str] | None = None,299        metadata: dict[str, Any] | None = None,300        inheritable_metadata: dict[str, Any] | None = None,301        run_id: UUID | None = None,302    ) -> None:303        _init_base_manager(304            self,305            handlers,306            inheritable_handlers,307            parent_run_id,308            tags=tags,309            inheritable_tags=inheritable_tags,310            metadata=metadata,311            inheritable_metadata=inheritable_metadata,312            run_id=run_id,313        )314 315    def copy(316        self,317        *,318        run_id: UUID | None | object = _MISSING,319    ) -> _AsyncGraphCallbackManager:320        return _copy_graph_manager(self, _AsyncGraphCallbackManager, run_id)321 322    @classmethod323    def configure(324        cls,325        callbacks: object | None = None,326        *,327        run_id: UUID | None = None,328    ) -> _AsyncGraphCallbackManager:329        return _configure_graph_callbacks(cls, callbacks, run_id=run_id)330 331    async def on_interrupt(self, event: GraphInterruptEvent) -> None:332        await ahandle_event(333            self.handlers,334            "on_interrupt",335            None,336            event,337        )338 339    async def on_resume(self, event: GraphResumeEvent) -> None:340        await ahandle_event(341            self.handlers,342            "on_resume",343            None,344            event,345        )346 347 348_GraphManagerT = TypeVar(349    "_GraphManagerT", _GraphCallbackManager, _AsyncGraphCallbackManager350)351 352GraphCallbacks: TypeAlias = (353    _GraphCallbackManager354    | _AsyncGraphCallbackManager355    | BaseCallbackManager356    | GraphCallbackHandler357    | Sequence[BaseCallbackHandler]358    | Sequence[GraphCallbackHandler]359    | None360)361 362 363def get_sync_graph_callback_manager_for_config(364    config: RunnableConfig,365    *,366    run_id: UUID | None = None,367) -> _GraphCallbackManager:368    """Build a sync graph lifecycle callback manager from a runnable config.369 370    This helper filters `config["callbacks"]` down to handlers that inherit371    from `GraphCallbackHandler` and binds the provided `run_id` onto the372    returned manager.373    """374    return _GraphCallbackManager.configure(375        config.get("callbacks"),376        run_id=run_id,377    )378 379 380def get_async_graph_callback_manager_for_config(381    config: RunnableConfig,382    *,383    run_id: UUID | None = None,384) -> _AsyncGraphCallbackManager:385    """Build an async graph lifecycle callback manager from a runnable config.386 387    This helper filters `config["callbacks"]` down to handlers that inherit388    from `GraphCallbackHandler` and binds the provided `run_id` onto the389    returned manager.390    """391    return _AsyncGraphCallbackManager.configure(392        config.get("callbacks"),393        run_id=run_id,394    )395