adispatch_custom_event() — langchain Function Reference

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

Function python CoreAbstractions Serialization calls 1

Entity Profile

CoreAbstractions→ Serialization→ adispatch_custom_event() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/langchain_core/callbacks/manager.py lines 2475–2597

async def adispatch_custom_event(
    name: str, data: Any, *, config: RunnableConfig | None = None
) -> None:
    """Dispatch an adhoc event to the handlers.

    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 (
            AsyncCallbackHandler,
            adispatch_custom_event
        )
        from langchain_core.runnable import RunnableLambda

        class CustomCallbackManager(AsyncCallbackHandler):
            async 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}")

        callback = CustomCallbackManager()

        async def foo(inputs):
            await adispatch_custom_event("my_event", {"bar": "buzz})
            return inputs

        foo_ = RunnableLambda(foo)
        await foo_.ainvoke({"a": "1"}, {"callbacks": [CustomCallbackManager()]})
        ```

    Example: Use with astream events

        ```python
        from langchain_core.callbacks import (
            AsyncCallbackHandler,
            adispatch_custom_event
        )
        from langchain_core.runnable import RunnableLambda

        class CustomCallbackManager(AsyncCallbackHandler):
            async 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}")

        callback = CustomCallbackManager()

        async def foo(inputs):
            await adispatch_custom_event("event_type_1", {"bar": "buzz})
            await adispatch_custom_event("event_type_2", 5)
            return inputs

        foo_ = RunnableLambda(foo)