LlamaIndex(十)——LlamaIndex Observability

一、Instrumentation简介

instrumentation模块（可在 llama-index v0.10.20 及以后的版本中使用）旨在取代旧的回调模块。在弃用期间，LlamaIndex 库支持两种模块，以便为您的 LLM 应用进行监控。然而，在所有现有集成迁移到新的监控模块后，LlamaIndex将不再支持回调模块。因此callback的文档就不看了，关注instrumentation模块更新即可。

新的instrumentation模块允许对LlamaIndex应用进行监控。特别地，用户可以使用模块提供的逻辑以及自定义逻辑来处理事件和跟踪跨度。用户还可以定义自己的事件，并指定在代码逻辑中的何时何地发出它们。

下面列出了监控模块的核心类以及它们各自的简要描述：

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

1.1 使用

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

1. 定义一个dispatcher
2. （可选）定义并附加EventHandler到dispatcher
3. （可选）定义并附加SpanHandler到dispatcher

这样做，将能够处理事件并获取在整个 LlamaIndex 库和扩展包中传输的跨度。
例如，如果想跟踪库中进行的每个 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:
        """Class name."""
        return "ExampleEventHandler"

    def handle(self, event: BaseEvent) -> None:
        """Logic for handling event."""
        print("-----------------------")
        # all events have these attributes
        print(event.id_)
        print(event.timestamp)
        print(event.span_id)

        # event specific attributes
        if isinstance(event, LLMChatStartEvent):
            # initial
            print(event.messages)
            print(event.additional_kwargs)
            print(event.model_dict)
        elif isinstance(event, LLMChatInProgressEvent):
            # streaming
            print(event.response.delta)
        elif isinstance(event, LLMChatEndEvent):
            # final response
            print(event.response)

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

完整例子、API Reference

1.1.1 自定义EventHandler

用户可以通过继承 BaseEventHandler 并为抽象方法 handle() 提供逻辑来创建自己的自定义处理器。

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


class MyEventHandler(BaseEventHandler):
    """My custom EventHandler."""

    @classmethod
    def class_name(cls) -> str:
        """Class name."""
        return "MyEventHandler"

    def handle(self, event: BaseEvent, **kwargs) -> Any:
        """Logic for handling event."""
        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)

1.1.2 自定义Event

用户可以通过继承 BaseEvent 类来创建自己的自定义事件。BaseEvent 类自带了一个时间戳(timestamp)以及一个 id_ 字段。要向这个事件负载中添加更多项，只需将它们作为新的 Fields 添加进去（因为它们是 pydantic.BaseModel 的子类）。

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


class MyEvent(BaseEvent):
    """My custom Event."""

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

一旦定义了自定义事件，就可以使用 Dispatcher 在应用程序代码的所需位置触发事件。

import llama_index.core.instrumentation as instrument

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

1.1.3 自定义 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 类型，还需要通过继承 BaseSpanHandler 类来定义自定义 SpanHandler。当子类化这个基类时，需要定义三个抽象方法，分别是：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:
        """Class name."""
        return "MyCustomSpanHandler"

    def new_span(
        self, id: str, parent_span_id: Optional[str], **kwargs
    ) -> Optional[MyCustomSpan]:
        """Create a span."""
        # logic for creating a new MyCustomSpan
        pass

    def prepare_to_exit_span(
        self, id: str, result: Optional[Any] = None, **kwargs
    ) -> Any:
        """Logic for preparing to exit a span."""
        pass

    def prepare_to_drop_span(
        self, id: str, err: Optional[Exception], **kwargs
    ) -> Any:
        """Logic for preparing to drop a span."""
        pass

要使用新 SpanHandler（和相关的 Span 类型），只需将其添加到想要使用的 Dispatcher 中。

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

dispatcher = (
    instrument.get_dispatcher()
)  # with no name argument, defaults to root

my_span_handler = MyCustomSpanHandler()
dispatcher.add_span_handler(my_span_handler)

1.1.4 进入/退出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


# or, syntactic sugar via decorators


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

1.1.5 利用 Dispatcher 层级结构

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

考虑下面定义的项目结构。有三个 Dispatcher：一个位于项目的最高级别，另外两个位于各个子模块 llama1 和 llama2 中。通过这种设置，附加到项目根 Dispatcher 的任何 EventHandler 将订阅在 llama1 和 llama2 中代码执行过程中发生的所有 Event。另一方面，在各自的 llama<x> 子模块中定义的 EventHandler 将只订阅在其各自子模块执行过程中发生的 Event。

project
├── __init__.py  # has a dispatcher=instrument.get_dispatcher(__name__)
├── llama1
│   ├── __init__.py  # has a dispatcher=instrument.get_dispatcher(__name__)
│   └── app_query_engine.py
└── llama2
    ├── __init__.py  # has a dispatcher=instrument.get_dispatcher(__name__)
    └── app_query_engine.py

1.2 NoteBooks

• Basic Usage
• Observing Model Calls
• Observing All Events

Instrumentation API Reference

官方资源

• 官方文档
• 官方博客
• 官方全部例子