adispatch_custom_event#

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

向处理程序分派一个临时事件。

Parameters:

  • name (str) – 临时事件的名称。

  • data (Any) – 用于临时事件的数据。自由格式的数据。理想情况下应该是可JSON序列化的，以避免下游的序列化问题，但这不是强制要求的。

  • config (可选[RunnableConfig]) – 可选的配置对象。与异步API镜像，但并非严格必需。

Return type:

无

示例

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

示例：与astream事件一起使用

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

警告

如果使用 python <= 3.10 和异步，您必须指定 config 参数，否则函数将引发错误。这是由于 python <= 3.10 中的 asyncio 限制，导致 LangChain 无法自动为用户传播配置对象。

在版本0.2.15中添加。

使用 adispatch_custom_event 的示例