dispatch_custom_event#

langchain_core.callbacks.manager.dispatch_custom_event(name: str, data: Any, *, config: RunnableConfig | None = None) → None[source]#

Dispatch an adhoc event.

Parameters:

  • name (str) – The name of the adhoc event.

  • data (Any) – 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[RunnableConfig]) – Optional config object. Mirrors the async API but not strictly needed.

Return type:

None

Example

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: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = 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()]})

New in version 0.2.15.

Examples using dispatch_custom_event

On this page