检测#
注意：`instrumentation` 模块（在 llama-index v0.10.20 及更高版本中可用）旨在取代旧的 `callbacks` 模块。在弃用期间，llama-index 库同时支持这两个模块，作为检测您的 LLM 应用程序的一种方式。然而，在所有现有集成迁移到新的 `instrumentation` 模块后，我们将在某个时候停止支持 `callbacks` 模块。
新的 `instrumentation` 模块可以对 `llama-index` 应用程序进行检测。特别地，可以使用自定义逻辑以及模块中提供的功能来处理事件和跟踪 span。用户还可以定义自己的事件，并指定在代码逻辑中的何处以及何时应该触发它们。下面列出了 `instrumentation` 模块的核心类及其简要说明
- `Event` — 表示应用程序代码执行过程中某个特定时刻发生的事件。
- `EventHandler` — 监听 `Event` 的发生，并在这些时刻执行代码逻辑。
- `Span` — 表示应用程序代码特定部分的执行流程，因此包含 `Event`。
- `SpanHandler` — 负责 `Span` 的进入、退出和丢弃（即因错误而提前退出）。
- `Dispatcher` — 发出 `Event`，并向适当的处理程序发送进入/退出/丢弃 `Span` 的信号。
使用 Instrumentation 模块进行可观测性#
检测的一个核心用例是可观测性。我们与第三方合作伙伴的原生检测集成允许您获得跨越整个调用堆栈的详细跟踪。
查看我们的可观测性指南，了解有关支持的合作伙伴的更多详细信息。
用法#
使用新的 `instrumentation` 模块涉及 3 个高层步骤。
- 定义一个 `dispatcher`
- （可选）定义并将您的 `EventHandler` 附加到 `dispatcher`
- （可选）定义并将您的 `SpanHandler` 附加到 `dispatcher`
这样做将使您能够处理在 `llama-index` 库和扩展包中传输的事件并获取 span。
例如，如果我想跟踪库中进行的每一个 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("-----------------------")
查看关于 LlamaIndex 中记录的所有事件的完整指南，或访问API 参考了解更多详情。
定义自定义 `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()
定义处理程序后，您可以将其附加到所需的 `dispatcher`。
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):
"""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=...))
定义自定义 `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()`。
import inspect
from typing import Any, Dict, 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,
bound_args: inspect.BoundArguments,
instance: Optional[Any] = None,
parent_span_id: Optional[str] = None,
tags: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> Optional[MyCustomSpan]:
"""Create a span."""
# logic for creating a new MyCustomSpan
pass
def prepare_to_exit_span(
self,
id_: str,
bound_args: inspect.BoundArguments,
instance: Optional[Any] = None,
result: Optional[Any] = None,
**kwargs: Any,
) -> Any:
"""Logic for preparing to exit a span."""
pass
def prepare_to_drop_span(
self,
id_: str,
bound_args: inspect.BoundArguments,
instance: Optional[Any] = None,
err: Optional[BaseException] = None,
**kwargs: Any,
) -> 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)
进入/退出 `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():
...
利用 `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 # 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
