adispatch_custom_event

Example:

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

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)

async for event in foo_.ainvoke_stream(
    {"a": "1"},
    version="v2",
    config={"callbacks": [CustomCallbackManager()]}
):
    print(event)