dispatch_custom_event() — langchain Function Reference

Architecture documentation for the dispatch_custom_event() function in manager.py from the langchain codebase.

Function python CoreAbstractions Serialization calls 1

Entity Profile

CoreAbstractions→ Serialization→ dispatch_custom_event() — langchain Function Reference

Dependency Diagram

graph TD
  457a6193_09b7_2858_bcc2_df476047b6b9["dispatch_custom_event()"]
  35cf5db6_bcb1_b854_6ebb_5e0368e51b58["manager.py"]
  457a6193_09b7_2858_bcc2_df476047b6b9 -->|defined in| 35cf5db6_bcb1_b854_6ebb_5e0368e51b58
  2e5dbbf3_7483_d224_7983_94f06da7a574["on_custom_event()"]
  457a6193_09b7_2858_bcc2_df476047b6b9 -->|calls| 2e5dbbf3_7483_d224_7983_94f06da7a574
  style 457a6193_09b7_2858_bcc2_df476047b6b9 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/callbacks/manager.py lines 2600–2671

def dispatch_custom_event(
    name: str, data: Any, *, config: RunnableConfig | None = None
) -> None:
    """Dispatch an adhoc event.

    Args:
        name: The name of the adhoc event.
        data: The data for the adhoc event.

            Free form data. Ideally should be JSON serializable to avoid serialization
            issues downstream, but this is not enforced.
        config: Optional config object.

            Mirrors the async API but not strictly needed.

    Raises:
        RuntimeError: If there is no parent run ID available to associate the event
            with.

    Example:
        ```python
        from langchain_core.callbacks import BaseCallbackHandler
        from langchain_core.callbacks import dispatch_custom_event
        from langchain_core.runnable import RunnableLambda

        class CustomCallbackManager(BaseCallbackHandler):
            def on_custom_event(
                self,
                name: str,
                data: Any,
                *,
                run_id: UUID,
                tags: list[str] | None = None,
                metadata: dict[str, Any] | None = None,
                **kwargs: Any,
            ) -> None:
                print(f"Received custom event: {name} with data: {data}")

        def foo(inputs):
            dispatch_custom_event("my_event", {"bar": "buzz})
            return inputs

        foo_ = RunnableLambda(foo)
        foo_.invoke({"a": "1"}, {"callbacks": [CustomCallbackManager()]})
        ```
    """
    # Import locally to prevent circular imports.
    from langchain_core.runnables.config import (  # noqa: PLC0415
        ensure_config,
        get_callback_manager_for_config,
    )

    config = ensure_config(config)
    callback_manager = get_callback_manager_for_config(config)
    # We want to get the callback manager for the parent run.
    # This is a work-around for now to be able to dispatch adhoc events from
    # within a tool or a lambda and have the metadata events associated
    # with the parent run rather than have a new run id generated for each.
    if callback_manager.parent_run_id is None:
        msg = (
            "Unable to dispatch an adhoc event without a parent run id."
            "This function can only be called from within an existing run (e.g.,"
            "inside a tool or a RunnableLambda or a RunnableGenerator.)"
            "If you are doing that and still seeing this error, try explicitly"
            "passing the config parameter to this function."
        )
        raise RuntimeError(msg)
    callback_manager.on_custom_event(
        name,
        data,
        run_id=callback_manager.parent_run_id,
    )