如何用 LlamaIndex instrumentation 模块构建自定义 LLM 调用的事件与 span 跟踪【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index如果你在基于 LlamaIndex 开发 LLM 应用想在自己的代码里捕获每一次 LLM 调用的输入/输出、记录执行流程中每个阶段的耗时官方提供的路径是instrumentation模块。该模块从 llama-index v0.10.20 开始提供用于取代旧的callbacks模块——过渡期内两个模块可以同时使用但官方文档明确说明等所有集成迁移完成后将不再支持callbacks。本文基于 instrumentation 官方文档和两个示例 notebookbasic_usage.ipynb、observe_api_calls.ipynb完成一条连续任务定义自定义EventHandler捕获 LLM 事件附加SpanHandler跟踪 span运行一次查询验证并了解如何扩展出自定义Event、Span和给自有函数打 span。适用前提版本要求llama-index v0.10.20 或更高版本instrumentation模块才可用。示例依赖下面的验证示例通过VectorStoreIndex触发真实的 OpenAI 调用。observe_api_calls.ipynb中的前置代码是os.environ[OPENAI_API_KEY] sk-...即运行前需要设置你自己的OPENAI_API_KEY环境变量。可选依赖使用SimpleSpanHandler的print_trace_trees()输出 trace 树时需要treelib包缺失时源码中给出的提示是pip install treelib。五个核心抽象在使用前需要区分五个概念引自官方文档类职责Event表示执行过程中某一时刻发生的一次行为EventHandler监听Event的发生并在这些时刻执行你的代码逻辑Span表示代码中某部分执行的完整流程跨一段时间内部包含EventSpanHandler负责Span的进入、退出和丢弃因错误提前退出Dispatcher把Event以及进入/退出/丢弃Span的信号分发给对应的 handler官方文档把使用流程概括为 3 步定义一个dispatcher可选定义并挂上EventHandler可选定义并挂上SpanHandler。第一步定义捕获 LLM 调用的 EventHandler自定义EventHandler的方式是继承BaseEventHandler并实现抽象方法handle()。针对跟踪每次 LLM 调用这个目标observe_api_calls.ipynb给出的示例按事件类型分支from llama_index.core.instrumentation.event_handlers import BaseEventHandler from llama_index.core.instrumentation.events.llm import ( LLMCompletionEndEvent, LLMChatEndEvent, ) class ModelEventHandler(BaseEventHandler): classmethod def class_name(cls) - str: Class name. return ModelEventHandler def handle(self, event) - None: Logic for handling event. if isinstance(event, LLMCompletionEndEvent): print(fLLM Prompt length: {len(event.prompt)}) print(fLLM Completion: {str(event.response.text)}) elif isinstance(event, LLMChatEndEvent): messages_str \n.join([str(x) for x in event.messages]) print(fLLM Input Messages length: {len(messages_str)}) print(fLLM Response: {str(event.response.message)})LLM 事件类的可用字段以 events/llm.py 为准LLMChatStartEventmessages输入消息列表、additional_kwargs、model_dictLLMChatInProgressEvent流式场景下携带当前response官方文档示例中读取event.response.deltaLLMChatEndEventmessages和最终的responseOptional[ChatResponse]LLMCompletionStartEvent/LLMCompletionEndEvent对应complete()调用带prompt与responseCompletionResponse。除了 LLM 事件observe_api_calls.ipynb 还演示了EmbeddingEndEvent可读取event.chunks如果你也要跟踪 embedding 调用可以在handle()中加一个isinstance分支。所有事件都带有id_、timestamp、span_id属性可以直接打印用于排查。第二步把 handler 挂到 dispatcher 上get_dispatcher()不带参数时默认返回root dispatcher带模块名如__name__时返回该模块的 dispatcherimport llama_index.core.instrumentation as instrument # root dispatcher不带 name 参数默认是 root root_dispatcher instrument.get_dispatcher() root_dispatcher.add_event_handler(ModelEventHandler())dispatcher 存在类似 Pythonlogging.Logger的层级结构除 root 外的每个 dispatcher 都有 parent处理事件和 span 时默认会向上传播propagate默认开启。这意味着挂在root dispatcher上的EventHandler会订阅llama-index库内部和你所有子模块中产生的事件——做跟踪所有 LLM 调用这类全局目标时挂 root 是正确选择挂在某个模块自己的 dispatcherinstrument.get_dispatcher(__name__)上的 handler 只订阅该子模块内的执行事件——适合做局部调试。basic_usage.ipynb中还演示了按名字取特定内部 dispatcher 的方式例如instrument.get_dispatcher(llama_index.core.base.base_query_engine)一般用于验证层级关系不必默认使用。第三步附加 SimpleSpanHandler跑一次查询验证SpanHandler与特定的Span类型配对工作。示例 notebook 直接使用现成的SimpleSpanHandler处理SimpleSpanfrom llama_index.core.instrumentation.span_handlers import SimpleSpanHandler span_handler SimpleSpanHandler() root_dispatcher.add_span_handler(span_handler)然后触发会调用 LLM 的查询from llama_index.core import Document, VectorStoreIndex index VectorStoreIndex.from_documents([Document.example()]) query_engine index.as_query_engine() response query_engine.query(Tell me about LLMs?)验证方式一handler 的打印输出。basic_usage.ipynb中handle()里对每个事件执行print(event.dict())同步query()后控制台会按序输出QueryStartEvent、RetrievalStartEvent、RetrievalEndEvent、SynthesizeStartEvent、GetResponseStartEvent、LLMPredictStartEvent、LLMPredictEndEvent等事件的timestamp/id_/class_name以上事件序列是 notebook 的文档示例输出实际id_和时间戳每次运行都不同对LLMChatEndEvent等 LLM 事件前面定义的ModelEventHandler会打印 prompt 长度与响应文本。验证方式二打印 span trace 树。SimpleSpanHandler记录了每个 span 的开始/结束时间、耗时和父子关系span_handler.print_trace_trees() # 需要已安装 treelibbasic_usage.ipynb中的文档示例输出形如BaseQueryEngine.query-bda10f51-... (1.762367) └── RetrieverQueryEngine._query-35da82df-... (1.760649) ├── BaseRetriever.retrieve-237d1b8d-... (0.19558) │ └── VectorIndexRetriever._retrieve-af6479b8-... (0.194024) └── BaseSynthesizer.synthesize-bf923672-... (1.564853) └── ... └── LLM.predict-a5ab2252-... (1.552019)每个节点是方法名-span_id (秒)缩进体现调用层级。上面的具体数值同样是文档示例只用于说明输出结构。Dispatcher对异步方法和流式方法同样生效basic_usage.ipynb用await query_engine.aquery(...)和chat_engine.stream_chat(...)复测事件含流式增量StreamChatDeltaReceivedEvent照常到达 handler。可选定义自定义 Event 与 Span如果你想在库内置事件之外记录自己的业务数据instrumentation模块允许扩展自定义Event继承BaseEvent自带timestamp和id_用 pydanticField加字段然后通过任意 dispatcher 触发from llama_index.core.instrumentation.event.base import BaseEvent from llama_index.core.bridge.pydantic import Field import llama_index.core.instrumentation as instrument class MyEvent(BaseEvent): My custom Event. new_field_1 Field(...) new_field_2 Field(...) dispatcher instrument.get_dispatcher(__name__) dispatcher.event(MyEvent(new_field_1..., new_field_2...))最后两行中...是文档中的占位替换成你两个字段的具体值即可。自定义Span及其SpanHandlerSpan与Event一样是结构化数据类但跨一段执行时间。定义BaseSpan子类保存你想记录的信息from typing import Any from llama_index.core.bridge.pydantic import Field from llama_index.core.instrumentation.span.base import BaseSpan class MyCustomSpan(BaseSpan): custom_field_1: Any Field(...) custom_field_2: Any Field(...)要处理这个新 Span 类型需继承BaseSpanHandler并实现三个抽象方法new_span()创建 span、prepare_to_exit_span()span 正常退出前、prepare_to_drop_span()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: 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. # 在此创建 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上面的三个方法是文档中的骨架pass需要填入你处理MyCustomSpan的逻辑后再挂到 dispatcher 上my_span_handler MyCustomSpanHandler() dispatcher.add_span_handler(my_span_handler)可选给自有函数打 span对库内部函数你无需手动打点但要跟踪自己代码里的一段执行官方文档提供两种写法装饰器推荐dispatcher.span会自动向SpanHandler发送进入/退出信号函数抛异常时发送 drop 信号import llama_index.core.instrumentation as instrument dispatcher instrument.get_dispatcher(__name__) dispatcher.span def my_pipeline_step(): # 你的业务逻辑 return done手动信号dispatcher.span_enter(...)、dispatcher.span_exit(...)、dispatcher.span_drop(...)分别用于进入、正常退出、以及span 因错误被提前截断的场景。文档示例的模式是进入后try业务逻辑异常分支调用span_drop(...)正常分支调用span_exit(...)并返回结果。从 Dispatcher 实现 可以看到装饰器的实际行为正常返回时执行span_exit捕获到异常时先发出携带err_str的SpanDropEvent再执行span_drop并继续抛出原异常——所以你的 handler 里可以通过SpanDropEvent识别失败的调用。注意事项与限制handler 抛异常不会中断应用dispatcher 分发事件和 span 信号时把 handler 的异常吞掉dispatcher.py中try/except BaseException: pass。如果handler 没反应先检查自己的handle()逻辑是否抛了异常而不是假设事件没产生。get_dispatch_event()已弃用自 llama-index-core v0.10.41 起弃用请直接用dispatcher.event(...)发事件如果你的第三方集成还在触发弃用告警升级对应集成即可。与callbacks的关系instrumentation是callbacks的替代者新代码应直接使用instrumentation避免踩到后续移除旧模块的坑。事件与 span 会沿 dispatcher 层级向 root 传播propagate默认开启。root dispatcher 的propagate为False见basic_usage.ipynb中对 root dispatcher 的打印示例即链路到 root 为止。完成上述步骤后你的应用就同时具备了LLM 调用的事件级可见性每次调用的输入长度、响应内容、流式增量、以及方法级调用树每个阶段的耗时与父子关系。如果需要把这些 span 接入第三方可观测平台官方文档指向 observability 指南 中提到的原生集成以及 Instrumentation API Reference 查阅全部内置事件类型。【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何用 LlamaIndex instrumentation 模块构建自定义 LLM 调用的事件与 span 跟踪
如何用 LlamaIndex instrumentation 模块构建自定义 LLM 调用的事件与 span 跟踪【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index如果你在基于 LlamaIndex 开发 LLM 应用想在自己的代码里捕获每一次 LLM 调用的输入/输出、记录执行流程中每个阶段的耗时官方提供的路径是instrumentation模块。该模块从 llama-index v0.10.20 开始提供用于取代旧的callbacks模块——过渡期内两个模块可以同时使用但官方文档明确说明等所有集成迁移完成后将不再支持callbacks。本文基于 instrumentation 官方文档和两个示例 notebookbasic_usage.ipynb、observe_api_calls.ipynb完成一条连续任务定义自定义EventHandler捕获 LLM 事件附加SpanHandler跟踪 span运行一次查询验证并了解如何扩展出自定义Event、Span和给自有函数打 span。适用前提版本要求llama-index v0.10.20 或更高版本instrumentation模块才可用。示例依赖下面的验证示例通过VectorStoreIndex触发真实的 OpenAI 调用。observe_api_calls.ipynb中的前置代码是os.environ[OPENAI_API_KEY] sk-...即运行前需要设置你自己的OPENAI_API_KEY环境变量。可选依赖使用SimpleSpanHandler的print_trace_trees()输出 trace 树时需要treelib包缺失时源码中给出的提示是pip install treelib。五个核心抽象在使用前需要区分五个概念引自官方文档类职责Event表示执行过程中某一时刻发生的一次行为EventHandler监听Event的发生并在这些时刻执行你的代码逻辑Span表示代码中某部分执行的完整流程跨一段时间内部包含EventSpanHandler负责Span的进入、退出和丢弃因错误提前退出Dispatcher把Event以及进入/退出/丢弃Span的信号分发给对应的 handler官方文档把使用流程概括为 3 步定义一个dispatcher可选定义并挂上EventHandler可选定义并挂上SpanHandler。第一步定义捕获 LLM 调用的 EventHandler自定义EventHandler的方式是继承BaseEventHandler并实现抽象方法handle()。针对跟踪每次 LLM 调用这个目标observe_api_calls.ipynb给出的示例按事件类型分支from llama_index.core.instrumentation.event_handlers import BaseEventHandler from llama_index.core.instrumentation.events.llm import ( LLMCompletionEndEvent, LLMChatEndEvent, ) class ModelEventHandler(BaseEventHandler): classmethod def class_name(cls) - str: Class name. return ModelEventHandler def handle(self, event) - None: Logic for handling event. if isinstance(event, LLMCompletionEndEvent): print(fLLM Prompt length: {len(event.prompt)}) print(fLLM Completion: {str(event.response.text)}) elif isinstance(event, LLMChatEndEvent): messages_str \n.join([str(x) for x in event.messages]) print(fLLM Input Messages length: {len(messages_str)}) print(fLLM Response: {str(event.response.message)})LLM 事件类的可用字段以 events/llm.py 为准LLMChatStartEventmessages输入消息列表、additional_kwargs、model_dictLLMChatInProgressEvent流式场景下携带当前response官方文档示例中读取event.response.deltaLLMChatEndEventmessages和最终的responseOptional[ChatResponse]LLMCompletionStartEvent/LLMCompletionEndEvent对应complete()调用带prompt与responseCompletionResponse。除了 LLM 事件observe_api_calls.ipynb 还演示了EmbeddingEndEvent可读取event.chunks如果你也要跟踪 embedding 调用可以在handle()中加一个isinstance分支。所有事件都带有id_、timestamp、span_id属性可以直接打印用于排查。第二步把 handler 挂到 dispatcher 上get_dispatcher()不带参数时默认返回root dispatcher带模块名如__name__时返回该模块的 dispatcherimport llama_index.core.instrumentation as instrument # root dispatcher不带 name 参数默认是 root root_dispatcher instrument.get_dispatcher() root_dispatcher.add_event_handler(ModelEventHandler())dispatcher 存在类似 Pythonlogging.Logger的层级结构除 root 外的每个 dispatcher 都有 parent处理事件和 span 时默认会向上传播propagate默认开启。这意味着挂在root dispatcher上的EventHandler会订阅llama-index库内部和你所有子模块中产生的事件——做跟踪所有 LLM 调用这类全局目标时挂 root 是正确选择挂在某个模块自己的 dispatcherinstrument.get_dispatcher(__name__)上的 handler 只订阅该子模块内的执行事件——适合做局部调试。basic_usage.ipynb中还演示了按名字取特定内部 dispatcher 的方式例如instrument.get_dispatcher(llama_index.core.base.base_query_engine)一般用于验证层级关系不必默认使用。第三步附加 SimpleSpanHandler跑一次查询验证SpanHandler与特定的Span类型配对工作。示例 notebook 直接使用现成的SimpleSpanHandler处理SimpleSpanfrom llama_index.core.instrumentation.span_handlers import SimpleSpanHandler span_handler SimpleSpanHandler() root_dispatcher.add_span_handler(span_handler)然后触发会调用 LLM 的查询from llama_index.core import Document, VectorStoreIndex index VectorStoreIndex.from_documents([Document.example()]) query_engine index.as_query_engine() response query_engine.query(Tell me about LLMs?)验证方式一handler 的打印输出。basic_usage.ipynb中handle()里对每个事件执行print(event.dict())同步query()后控制台会按序输出QueryStartEvent、RetrievalStartEvent、RetrievalEndEvent、SynthesizeStartEvent、GetResponseStartEvent、LLMPredictStartEvent、LLMPredictEndEvent等事件的timestamp/id_/class_name以上事件序列是 notebook 的文档示例输出实际id_和时间戳每次运行都不同对LLMChatEndEvent等 LLM 事件前面定义的ModelEventHandler会打印 prompt 长度与响应文本。验证方式二打印 span trace 树。SimpleSpanHandler记录了每个 span 的开始/结束时间、耗时和父子关系span_handler.print_trace_trees() # 需要已安装 treelibbasic_usage.ipynb中的文档示例输出形如BaseQueryEngine.query-bda10f51-... (1.762367) └── RetrieverQueryEngine._query-35da82df-... (1.760649) ├── BaseRetriever.retrieve-237d1b8d-... (0.19558) │ └── VectorIndexRetriever._retrieve-af6479b8-... (0.194024) └── BaseSynthesizer.synthesize-bf923672-... (1.564853) └── ... └── LLM.predict-a5ab2252-... (1.552019)每个节点是方法名-span_id (秒)缩进体现调用层级。上面的具体数值同样是文档示例只用于说明输出结构。Dispatcher对异步方法和流式方法同样生效basic_usage.ipynb用await query_engine.aquery(...)和chat_engine.stream_chat(...)复测事件含流式增量StreamChatDeltaReceivedEvent照常到达 handler。可选定义自定义 Event 与 Span如果你想在库内置事件之外记录自己的业务数据instrumentation模块允许扩展自定义Event继承BaseEvent自带timestamp和id_用 pydanticField加字段然后通过任意 dispatcher 触发from llama_index.core.instrumentation.event.base import BaseEvent from llama_index.core.bridge.pydantic import Field import llama_index.core.instrumentation as instrument class MyEvent(BaseEvent): My custom Event. new_field_1 Field(...) new_field_2 Field(...) dispatcher instrument.get_dispatcher(__name__) dispatcher.event(MyEvent(new_field_1..., new_field_2...))最后两行中...是文档中的占位替换成你两个字段的具体值即可。自定义Span及其SpanHandlerSpan与Event一样是结构化数据类但跨一段执行时间。定义BaseSpan子类保存你想记录的信息from typing import Any from llama_index.core.bridge.pydantic import Field from llama_index.core.instrumentation.span.base import BaseSpan class MyCustomSpan(BaseSpan): custom_field_1: Any Field(...) custom_field_2: Any Field(...)要处理这个新 Span 类型需继承BaseSpanHandler并实现三个抽象方法new_span()创建 span、prepare_to_exit_span()span 正常退出前、prepare_to_drop_span()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: 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. # 在此创建 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上面的三个方法是文档中的骨架pass需要填入你处理MyCustomSpan的逻辑后再挂到 dispatcher 上my_span_handler MyCustomSpanHandler() dispatcher.add_span_handler(my_span_handler)可选给自有函数打 span对库内部函数你无需手动打点但要跟踪自己代码里的一段执行官方文档提供两种写法装饰器推荐dispatcher.span会自动向SpanHandler发送进入/退出信号函数抛异常时发送 drop 信号import llama_index.core.instrumentation as instrument dispatcher instrument.get_dispatcher(__name__) dispatcher.span def my_pipeline_step(): # 你的业务逻辑 return done手动信号dispatcher.span_enter(...)、dispatcher.span_exit(...)、dispatcher.span_drop(...)分别用于进入、正常退出、以及span 因错误被提前截断的场景。文档示例的模式是进入后try业务逻辑异常分支调用span_drop(...)正常分支调用span_exit(...)并返回结果。从 Dispatcher 实现 可以看到装饰器的实际行为正常返回时执行span_exit捕获到异常时先发出携带err_str的SpanDropEvent再执行span_drop并继续抛出原异常——所以你的 handler 里可以通过SpanDropEvent识别失败的调用。注意事项与限制handler 抛异常不会中断应用dispatcher 分发事件和 span 信号时把 handler 的异常吞掉dispatcher.py中try/except BaseException: pass。如果handler 没反应先检查自己的handle()逻辑是否抛了异常而不是假设事件没产生。get_dispatch_event()已弃用自 llama-index-core v0.10.41 起弃用请直接用dispatcher.event(...)发事件如果你的第三方集成还在触发弃用告警升级对应集成即可。与callbacks的关系instrumentation是callbacks的替代者新代码应直接使用instrumentation避免踩到后续移除旧模块的坑。事件与 span 会沿 dispatcher 层级向 root 传播propagate默认开启。root dispatcher 的propagate为False见basic_usage.ipynb中对 root dispatcher 的打印示例即链路到 root 为止。完成上述步骤后你的应用就同时具备了LLM 调用的事件级可见性每次调用的输入长度、响应内容、流式增量、以及方法级调用树每个阶段的耗时与父子关系。如果需要把这些 span 接入第三方可观测平台官方文档指向 observability 指南 中提到的原生集成以及 Instrumentation API Reference 查阅全部内置事件类型。【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考