仪器设备#

注意：instrumentation 模块（在 llama-index v0.10.20 及更高版本中可用）旨在取代传统的 callbacks 模块。在弃用期间，llama-index 库将同时支持这两个模块，作为仪器化 LLM 应用程序的手段。然而，在所有现有集成迁移到新的 instrumentation 模块之后的某个时刻，我们将不再支持 callbacks 模块。

新的 instrumentation 模块允许仪器化 llama-index 应用程序。特别是，用户可以使用自定义逻辑以及模块中提供的功能来处理事件和跟踪跨度。用户还可以定义自己的事件，并指定它们在代码逻辑中的何时何地应该被触发。以下是 instrumentation 模块的核心类及其简要描述：

Event — 表示应用程序代码执行过程中某个特定事件发生的时刻。
EventHandler — 监听 Event 的发生，并在这些时刻执行代码逻辑。
Span — 表示应用程序代码中特定部分的执行流程，因此包含 Event。
SpanHandler — 负责进入、退出和丢弃（即因错误而提前退出）Span。
Dispatcher — 发出 Event，以及信号以进入/退出/丢弃 Span 到适当的处理程序。

用法#

使用新的 instrumentation 模块涉及 3 个高级步骤。

定义一个 dispatcher
（可选）定义并将您的 EventHandler 附加到 dispatcher
（可选）将您的 SpanHandler 定义并附加到 dispatcher

这样做将使您能够处理事件并获取在 llama-index 库和扩展包中传输的跨度。

例如，如果我想要跟踪库中进行的每个 LLM 调用：

from typing import Dict, List

from llama_index.core.instrumentation.events.llm import (
    LLMChatEndEvent,
    LLMChatStartEvent,
    LLMChatInProgressEvent,
)


class ExampleEventHandler(BaseEventHandler):
    events: List[BaseEvent] = []

    @classmethod
    def class_name(cls) -> str:
        """类名。"""
        return "ExampleEventHandler"

    def handle(self, event: BaseEvent) -> None:
        """处理事件的逻辑。"""
        print("-----------------------")
        # 所有事件都具有这些属性
        print(event.id_)
        print(event.timestamp)
        print(event.span_id)

        # 特定事件属性
        if isinstance(event, LLMChatStartEvent):
            # 初始
            print(event.messages)
            print(event.additional_kwargs)
            print(event.model_dict)
        elif isinstance(event, LLMChatInProgressEvent):
            # 流式传输
            print(event.response.delta)
        elif isinstance(event, LLMChatEndEvent):
            # 最终响应
            print(event.response)

        self.events.append(event)
        print("-----------------------")

请参阅完整指南以了解 LlamaIndex 中记录的所有事件，或访问 API 参考以获取更多详细信息。

定义自定义 `EventHandler`#

用户可以通过子类化 BaseEventHandler 并为抽象方法 handle() 提供逻辑来创建自己的自定义处理程序。

from llama_index.core.instrumentation.event_handlers.base import (
    BaseEventHandler,
)


class MyEventHandler(BaseEventHandler):
    """我的自定义 EventHandler。"""

    @classmethod
    def class_name(cls) -> str:
        """类名。"""
        return "MyEventHandler"

    def handle(self, event: BaseEvent, **kwargs) -> Any:
        """处理事件的逻辑。"""
        print(event.class_name())


my_event_handler = MyEventHandler()

定义完处理程序后，您可以将其附加到所需的调度程序：

import llama_index.core.instrumentation as instrument

dispatcher = instrument.get_dispatcher(__name__)
dispatcher.add_event_handler(my_event_handler)

定义自定义 `Event`#

用户可以通过子类化 BaseEvent 来创建自己的自定义事件。BaseEvent 类带有 timestamp 和 id_ 字段。要向此事件负载添加更多项目，只需将它们添加为新的 Fields（因为它们是 pydantic.BaseModel 的子类）。

from llama_index.core.instrumentation.event.base import BaseEvent


class MyEvent(BaseEvent):
    """我的自定义 Event。"""

    new_field_1 = Field(...)
    new_field_2 = Field(...)

定义完自定义事件后，您可以使用调度程序在应用程序代码的所需实例中触发事件。

import llama_index.core.instrumentation as instrument

dispatcher = instrument.get_dispatcher(__name__)
dispatcher.event(MyEvent(new_field_1=..., new_field_2=...))

定义自定义 `Span`#

Span 和 Event 一样，都是结构化数据类。不过，与 Event 不同的是，Span 顾名思义，跨越程序执行流程的一段时间。你可以定义一个自定义的 Span 来存储任何你想要的信息。

from typing import Any
from llama_index.core.bridge.pydantic import Field


class MyCustomSpan(BaseSpan):
    custom_field_1: Any = Field(...)
    custom_field_2: Any = Field(...)

为了处理你的新的 Span 类型，你还需要定义一个自定义的 SpanHandler，通过继承 BaseSpanHandler 类。在子类化这个基类时，需要定义三个抽象方法，分别是：new_span()、prepare_to_exit_span() 和 prepare_to_drop_span()。

from typing import Any, Optional
from llama_index.core.instrumentation.span.base import BaseSpan
from llama_index.core.instrumentation.span_handlers import BaseSpanHandler


class MyCustomSpanHandler(BaseSpanHandler[MyCustomSpan]):
    @classmethod
    def class_name(cls) -> str:
        """类名。"""
        return "MyCustomSpanHandler"

    def new_span(
        self, id: str, parent_span_id: Optional[str], **kwargs
    ) -> Optional[MyCustomSpan]:
        """创建一个 span。"""
        # 创建一个新的 MyCustomSpan 的逻辑
        pass

    def prepare_to_exit_span(
        self, id: str, result: Optional[Any] = None, **kwargs
    ) -> Any:
        """准备退出一个 span 的逻辑。"""
        pass

    def prepare_to_drop_span(
        self, id: str, err: Optional[Exception], **kwargs
    ) -> Any:
        """准备放弃一个 span 的逻辑。"""
        pass

要使用你的新的 SpanHandler（以及相关的 Span 类型），你只需要将它添加到你想要的调度程序中。

import llama_index.core.instrumentation as instrument
from llama_index.core.instrumentation.span_handler import SimpleSpanHandler

dispatcher = (
    instrument.get_dispatcher()
)  # 如果没有指定名称参数，默认为根

my_span_handler = MyCustomSpanHandler()
dispatcher.add_span_handler(my_span_handler)

进入/退出一个 `Span`#

要向 SpanHandler 发送信号以进入/退出一个 Span，我们使用 span_enter()、span_exit() 方法。还有一个 span_drop() 方法，可以用来处理由于覆盖代码执行中的错误而导致 Span 被缩短的情况。

import llama_index.core.instrumentation as instrument

dispatcher = instrument.get_dispatcher(__name__)


def func():
    dispatcher.span_enter(...)
    try:
        val = ...
    except:
        ...
        dispatcher.span_drop(...)
    else:
        dispatcher.span_exit(...)
        return val


# 或者，通过装饰器进行语法糖化


@dispatcher.span
def func():
    ...

利用 `dispatcher` 层次结构#

与标准的 Python logging 库及其 Logger 类类似，dispatcher 也存在类似的层次结构。具体来说，除了根 dispatcher 外，所有的 dispatcher 都有一个父级，当处理事件或 span 时，可以将它们传播到其父级（这是默认行为）。这种分层处理事件和 span 的方法允许定义“全局”事件处理程序以及“局部”事件处理程序。

考虑下面定义的项目结构。有 3 个 dispatcher：一个位于 project 的顶层，另外两个位于各自的子模块 llama1 和 llama2 中。通过这种设置，附加到项目根 dispatcher 的任何 EventHandler 都将订阅在 llama1 和 llama2 中执行的所有 Event。另一方面，定义在各自的 llama<x> 子模块中的 EventHandler 只会订阅发生在各自子模块内部执行的 Event。

project
├── __init__.py  # 有一个 dispatcher=instrument.get_dispatcher(__name__)
├── llama1
│   ├── __init__.py  # 有一个 dispatcher=instrument.get_dispatcher(__name__)
│   └── app_query_engine.py
└── llama2
    ├── __init__.py  # 有一个 dispatcher=instrument.get_dispatcher(__name__)
    └── app_query_engine.py

笔记指南：#

API 参考#

仪器化 API 参考

仪器设备#

用法#

定义自定义 EventHandler#

定义自定义 Event#

定义自定义 Span#

进入/退出一个 Span#

利用 dispatcher 层次结构#