追踪

AgentScope实现了基于OpenTelemetry的追踪功能，用于监控和调试智能体应用程序的执行，其特点包括

• 为LLM、工具、智能体、格式化器等提供内置追踪功能。

• 支持错误和异常追踪

• 在 AgentScope Studio 中提供原生的追踪可视化功能

• 支持连接到 第三方平台，例如 Arize-Phoenix、Langfuse 等。

设置

注意

连接到AgentScope Studio或第三方追踪端点应在应用程序开始时通过agentscope.init函数完成。

智能体工作室

Agentscope Studio 中的追踪功能

当连接到AgentScope Studio时，只需在agentscope.init函数中提供studio_url参数。

import agentscope

agentscope.init(studio_url="http://xxx:port")

第三方平台

要连接到第三方追踪平台，请在agentscope.init函数中设置tracing_url参数。 tracing_url是您的OpenTelemetry收集器或任何支持OTLP（OpenTelemetry协议）的兼容后端的URL。

import agentscope

# Connect to OpenTelemetry-compatible backends
agentscope.init(tracing_url="https://your-tracing-backend:port/traces")

以Arize-Phoenix和Langfuse为例：

Arize-Phoenix: 您需要在环境变量中设置 PHOENIX_API_KEY。

Connect to Arize Phoenix

# Arize Phoenix Integration
import os

PHOENIX_API_KEY = os.environ.get("PHOENIX_API_KEY")
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"api_key={PHOENIX_API_KEY}"

agentscope.init(tracing_url="https://app.phoenix.arize.com/v1/traces")

LangFuse: 您需要在您的环境变量中设置 LANGFUSE_PUBLIC_KEY 和 LANGFUSE_SECRET_KEY。授权 标头是通过这些密钥构建的。

Connect to LangFuse

import os, base64

LANGFUSE_PUBLIC_KEY = os.environ["LANGFUSE_PUBLIC_KEY"]
LANGFUSE_SECRET_KEY = os.environ["LANGFUSE_SECRET_KEY"]
LANGFUSE_AUTH_STRING = f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}"

LANGFUSE_AUTH = base64.b64encode(LANGFUSE_AUTH_STRING.encode("utf-8")).decode("ascii")
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

# EU data region
agentscope.init(tracing_url="https://cloud.langfuse.com/api/public/otel/v1/traces")
# US data region
# agentscope.init(tracing_url="https://us.cloud.langfuse.com/api/public/otel/v1/traces")

自定义追踪

如上所述，AgentScope 中的追踪功能是基于 OpenTelemetry 实现的。 这意味着使用 OpenTelemetry SDK 自行实现的追踪代码与 AgentScope 原生兼容。

除了这些，AgentScope还内置了以下装饰器来追踪相应的模块：

• @trace_llm: 追踪继承自 ChatModelBase 类的 __call__ 函数

• @trace_reply: 跟踪继承自AgentBase的类的reply函数

• @trace_format: 追踪继承自FormatterBase类的format函数

• @trace: 追踪一般函数

追踪LLMs

@trace_llm 装饰器用于追踪 ChatModelBase 类的 __call__ 函数。

注意

您的LLM类必须继承自ChatModelBase

Tracing new ChatModel class

class ExampleChatModel(ChatModelBase):
    """An example Model"""

    ...

    @trace_llm
    async def __call__(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> AsyncGenerator[ChatResponse, None] | ChatResponse:
        """LLM call"""
        ...

追踪代理

@trace_reply 装饰器用于智能体实现并追踪 reply 函数。

注意

您的智能体类必须继承自 AgentBase

Tracing new Agent class

class ExampleAgent(AgentBase):
    """An example agent class"""

    @tracer_reply
    async def reply(self, *args: Any, **kwargs: Any) -> Msg:
        """Reply to the message."""
        ...

追踪格式化器

@trace_format 装饰器用于格式化器实现以及追踪 format 函数。

注意

你的格式化器类必须继承自 FormatterBase

Tracing new Formatter class

class ExampleFormatter(FormatterBase):
        """A simple example formatter class"""

        @trace_format
        async def format(self, *args: Any, **kwargs: Any) -> list[dict]:
            """Example formatting"""

通用追踪

@trace 装饰器与上述装饰器不同，它是一种通用追踪装饰器，可应用于任何函数。 它需要一个 name 参数来标识被追踪的函数，并且能够追踪多种类型的函数，包括：

• 同步函数

• 同步生成器函数

• 异步函数

• 异步生成器函数

General tracing example

# 1. Synchronous function
@trace(name='simple_function')
def simple_function(name: str, age: int) -> str:
    """A simple function with automatic tracing."""
    return f"Hello, {name}! You are {age} years old."

# 2. Synchronous generator function
@trace(name='number_generator')
def number_generator(n: int) -> Generator[int, None, None]:
    """Generate numbers from 0 to n-1."""
    for i in range(n):
        yield i

# 3. Asynchronous function
@trace(name='async_function')
async def async_function(data: dict) -> dict:
    """Process data asynchronously."""
    return {"processed": data}

# 4. Asynchronous generator function
@trace(name='async_stream')
async def async_stream(n: int) -> AsyncGenerator[str, None]:
    """Generate stream of data asynchronously."""
    for i in range(n):
        yield f"data_{i}"