可以通过继承 BaseInstrumentor 类，编写 Instrumentor 以实现埋点插桩逻辑。
假设应用使用Dashscope SDK进行大模型调用，可以参考如下示例代码对Dashscope库中BaseApi类的call方法进行埋点插桩：
from typing import Any, Collection
from aliyun.instrumentation.dashscope._wrapper import DashscopeRequestWrapper
from aliyun.instrumentation.dashscope.package import _instruments
from aliyun.instrumentation.dashscope.version import __version__
from aliyun.semconv.logger import getLogger
from wrapt import wrap_function_wrapper
from opentelemetry import trace as trace_api
from opentelemetry.instrumentation.instrumentor import BaseInstrumentor # type: ignore
logger = getLogger(__name__)
_MODULE = "dashscope.client.base_api"
__all__ = ["AliyunDashScopeInstrumentor"]
class AliyunDashScopeInstrumentor(BaseInstrumentor): # type: ignore
__slots__ = (
"_original_call",
)
# 该函数定义了应该对哪个依赖里的代码进行插桩
# 使用wrapt对目标函数进行插桩增强
def instrumentation_dependencies(self) -> Collection[str]:
return ("dashscope >= 1.0.0",)
# 在_instrument函数中执行实际的插桩逻辑
def _instrument(self, **kwargs: Any) -> None:
if not (tracer_provider := kwargs.get("tracer_provider")):
tracer_provider = trace_api.get_tracer_provider()
tracer = trace_api.get_tracer(__name__, __version__, tracer_provider)
wrap_function_wrapper(
module=_MODULE,
name="BaseApi.call",
wrapper=DashscopeRequestWrapper(tracer=tracer),
)
# 该函数定义了取消插桩时应该执行的动作，如无需定义请忽略
def _uninstrument(self, **kwargs: Any) -> None:
pass
