如何分发自定义回调事件

前置条件

本指南假设您熟悉以下概念：

回调
自定义回调处理器
Astream Events API 该 astream_events 方法将展示自定义回调事件。

在某些情况下，您可能希望从 Runnable 内部分发自定义回调事件，以便在自定义回调处理程序中或通过 Astream Events API 显示它。

例如，如果您有一个包含多个步骤的长时间运行工具，您可以在步骤之间分派自定义事件，并使用这些自定义事件来监控进度。您还可以将这些自定义事件展示给应用程序的最终用户，以向他们显示当前任务的进展情况。

要分发自定义事件，您需要为事件确定两个属性：name 和 data。

属性	类型	描述
name	str	A user defined name for the event.
data	Any	The data associated with the event. This can be anything, though we suggest making it JSON serializable.

重要

调度自定义回调事件需要 langchain-core>=0.2.15。
自定义回调事件只能从现有的 Runnable 内部分发。
如果使用 astream_events，则必须使用 version='v2' 才能查看自定义事件。
在 LangSmith 中发送或渲染自定义回调事件目前尚不支持。

兼容性

LangChain 无法自动传播配置（包括用于 astream_events() 的回调）到子运行器，如果您在 Python <=3.10 中运行异步代码。这是您可能无法看到自定义运行器或工具发出事件的常见原因。

如果您运行的是 python<=3.10，则需要在异步环境中手动将 RunnableConfig 对象传播到子可运行对象。关于如何手动传播配置的示例，请参阅下方 bar RunnableLambda 的实现。

如果您运行的是 python>=3.11，RunnableConfig 将在异步环境中自动传播到子可运行对象。然而，如果您的代码可能在其他 Python 版本中运行，手动传播 RunnableConfig 仍然是一个好主意。

Astream Events API

使用自定义事件最有用的方式是通过 Astream Events API。

我们可以使用 async adispatch_custom_event API 在异步环境中发射自定义事件。

重要

要通过 astream events API 查看自定义事件，您需要使用 astream_events 的较新 v2 API。

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


@RunnableLambda
async def foo(x: str) -> str:
    await adispatch_custom_event("event1", {"x": x})
    await adispatch_custom_event("event2", 5)
    return x


async for event in foo.astream_events("hello world", version="v2"):
    print(event)

API 参考:分发自定义事件 |RunnableLambda |可运行配置

{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'foo', 'tags': [], 'run_id': 'f354ffe8-4c22-4881-890a-c1cad038a9a6', 'metadata': {}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': 'f354ffe8-4c22-4881-890a-c1cad038a9a6', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': 'f354ffe8-4c22-4881-890a-c1cad038a9a6', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': 'f354ffe8-4c22-4881-890a-c1cad038a9a6', 'name': 'foo', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}
{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': 'f354ffe8-4c22-4881-890a-c1cad038a9a6', 'name': 'foo', 'tags': [], 'metadata': {}, 'parent_ids': []}

在 Python <= 3.10 中，您必须手动传播配置！

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


@RunnableLambda
async def bar(x: str, config: RunnableConfig) -> str:
    """An example that shows how to manually propagate config.

    You must do this if you're running python<=3.10.
    """
    await adispatch_custom_event("event1", {"x": x}, config=config)
    await adispatch_custom_event("event2", 5, config=config)
    return x


async for event in bar.astream_events("hello world", version="v2"):
    print(event)

API 参考:分发自定义事件 |RunnableLambda |可运行配置

{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'bar', 'tags': [], 'run_id': 'c787b09d-698a-41b9-8290-92aaa656f3e7', 'metadata': {}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': 'c787b09d-698a-41b9-8290-92aaa656f3e7', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': 'c787b09d-698a-41b9-8290-92aaa656f3e7', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': 'c787b09d-698a-41b9-8290-92aaa656f3e7', 'name': 'bar', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}
{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': 'c787b09d-698a-41b9-8290-92aaa656f3e7', 'name': 'bar', 'tags': [], 'metadata': {}, 'parent_ids': []}

异步回调处理器

您也可以通过异步回调处理程序消费已分派的事件。

from typing import Any, Dict, List, Optional
from uuid import UUID

from langchain_core.callbacks import AsyncCallbackHandler
from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


class AsyncCustomCallbackHandler(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 event {name} with data: {data}, with tags: {tags}, with metadata: {metadata} and run_id: {run_id}"
        )


@RunnableLambda
async def bar(x: str, config: RunnableConfig) -> str:
    """An example that shows how to manually propagate config.

    You must do this if you're running python<=3.10.
    """
    await adispatch_custom_event("event1", {"x": x}, config=config)
    await adispatch_custom_event("event2", 5, config=config)
    return x


async_handler = AsyncCustomCallbackHandler()
await foo.ainvoke(1, {"callbacks": [async_handler], "tags": ["foo", "bar"]})

API 参考:异步回调处理器 |分发自定义事件 |RunnableLambda |可运行配置

Received event event1 with data: {'x': 1}, with tags: ['foo', 'bar'], with metadata: {} and run_id: a62b84be-7afd-4829-9947-7165df1f37d9
Received event event2 with data: 5, with tags: ['foo', 'bar'], with metadata: {} and run_id: a62b84be-7afd-4829-9947-7165df1f37d9

同步回调处理器

让我们看看如何在同步环境中使用 dispatch_custom_event 来发射自定义事件。

您必须在现有的 Runnable 内调用 dispatch_custom_event。

from typing import Any, Dict, List, Optional
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.callbacks.manager import (
    dispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


class CustomHandler(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 event {name} with data: {data}, with tags: {tags}, with metadata: {metadata} and run_id: {run_id}"
        )


@RunnableLambda
def foo(x: int, config: RunnableConfig) -> int:
    dispatch_custom_event("event1", {"x": x})
    dispatch_custom_event("event2", {"x": x})
    return x


handler = CustomHandler()
foo.invoke(1, {"callbacks": [handler], "tags": ["foo", "bar"]})

API 参考:基础回调处理器 |分派自定义事件 |RunnableLambda |可运行配置

Received event event1 with data: {'x': 1}, with tags: ['foo', 'bar'], with metadata: {} and run_id: 27b5ce33-dc26-4b34-92dd-08a89cb22268
Received event event2 with data: {'x': 1}, with tags: ['foo', 'bar'], with metadata: {} and run_id: 27b5ce33-dc26-4b34-92dd-08a89cb22268

下一步

您已经了解了如何发射自定义事件，您可以查看更深入的指南，了解流式事件，这是利用自定义事件最简单的方法。

Astream Events API​

异步回调处理器​

同步回调处理器​

下一步​

Astream Events API

异步回调处理器

同步回调处理器

下一步