主题
注意：instrumentation 模块（llama-index v0.10.20 及之后版本可用）是为了取代旧的 callbacks 模块。在弃用过渡期内，llama-index 库同时支持这两个模块来为你的 LLM 应用做 instrumentation。但等所有现有集成都迁到新的 instrumentation 模块之后，我们将不再支持 callbacks 模块。
新的 instrumentation 模块可对 llama-index 应用做 instrumentation。具体来说，你可以用自定义逻辑，也可以用该模块自带的机制来处理事件、跟踪 span。用户还能定义自己的事件，并指定在代码逻辑的哪个位置、何时发出它们。下面列出该 instrumentation 模块的核心类及其简要说明：
Event—— 表示应用代码执行过程中某个事件发生的时间点。EventHandler—— 监听Event的发生，并在这些时刻执行代码逻辑。Span—— 表示应用中某段代码的执行流程，因而包含若干Event。SpanHandler—— 负责Span的进入、退出与丢弃（例如因错误提前退出）。Dispatcher—— 发出Event，并向相应的处理器传递进入/退出/丢弃Span的信号。
用 instrumentation 模块做可观测性
instrumentation 的一个核心用途就是可观测性。我们与第三方合作方间的原生 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 类似，二者都是结构化的数据类。不同之处在于，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 的层级结构
dispatcher 也有与 Python 标准 logging 库及其 Logger 类类似的层级结构。具体说，除根 dispatcher 外，每个 dispatcher 都有一个父级；在处理事件或 span 时可以向父级传播（这是默认行为）。这种层级化的处理方式让你既能定义“全局”的事件处理器，也能定义“局部”的。
看下 project 结构里定义的 3 个 dispatcher：一个在 project 顶层，另两个分别在其子模块 llama1 与 llama2 中。在这样的布置下，挂在 project 根 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