监控模块#
注意：instrumentation模块（在llama-index v0.10.20及更高版本中可用）旨在取代传统的callbacks模块。在弃用过渡期间，llama-index库同时支持这两个模块来实现LLM应用的监控功能。不过，在所有现有集成都迁移到新的instrumentation模块后，我们将不再支持callbacks模块。
新的instrumentation模块允许对llama-index应用进行监控。具体而言，用户既可以使用模块提供的功能，也可以通过自定义逻辑来处理事件和跟踪跨度。用户还可以定义自己的事件，并指定在代码逻辑的哪些位置触发这些事件。以下是instrumentation模块的核心类及其简要说明：
Event— 表示应用程序代码执行过程中某个特定时刻发生的事件EventHandler— 监听Event的发生，并在这些时刻执行代码逻辑Span— 表示应用程序代码中特定部分的执行流程，因此包含多个EventSpanHandler— 负责处理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:
"""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_字段。要添加更多字段到事件负载中，只需将它们添加为新的Field（因为它们继承自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层级结构#
dispatcher的层级结构与标准Pythonlogging库及其Logger类类似。具体来说，除根dispatcher外，所有dispatcher都有一个父级，并且在处理事件或跨度时可以将它们传播到其父级（这是默认行为）。这种处理事件和跨度的层级方法允许定义"全局"事件处理器和"局部"处理器。
考虑以下项目结构。有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