遥测的更高级方案

项目
11/03/2024

注意

本文将使用 Aspire 仪表板进行演示。如果希望使用其他工具，请参阅你在安装说明中使用的工具的文档。

自动函数调用

自动函数调用是一种语义内核功能，允许内核在模型响应函数调用时自动执行函数，并将结果提供回模型。此功能适用于查询需要多次迭代函数调用才能获取最终自然语言响应的方案。有关更多详细信息，请参阅这些 GitHub 示例。

注意

所有模型都不支持函数调用。

提示

你将听到术语“工具”和“工具调用”有时与“函数”和“函数调用”互换使用。

先决条件

支持函数调用的 Azure OpenAI 聊天完成部署。
Docker
适用于操作系统的最新 .Net SDK 。

支持函数调用的 Azure OpenAI 聊天完成部署。
Docker
计算机上安装了 Python 3.10、3.11 或 3.12 。

注意

语义内核可观测性尚不适用于 Java。

安装

创建新的控制台应用程序

在终端中运行以下命令，在 C# 中创建新的控制台应用程序：

dotnet new console -n TelemetryAutoFunctionCallingQuickstart

命令完成后，导航到新创建的项目目录。

安装所需程序包

语义内核

dotnet add package Microsoft.SemanticKernel

OpenTelemetry 控制台导出程序

dotnet add package OpenTelemetry.Exporter.OpenTelemetryProtocol

使用语义内核创建简单应用程序

从项目目录中，使用你喜欢的编辑器打开 Program.cs 该文件。我们将创建一个简单的应用程序，该应用程序使用语义内核向聊天完成模型发送提示。将现有内容替换为以下代码，并填写所需的值 deploymentName， endpoint以及 apiKey：

using System.ComponentModel;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Connectors.OpenAI;
using OpenTelemetry;
using OpenTelemetry.Logs;
using OpenTelemetry.Metrics;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;

namespace TelemetryAutoFunctionCallingQuickstart
{
    class BookingPlugin
    {
        [KernelFunction("FindAvailableRooms")]
        [Description("Finds available conference rooms for today.")]
        public async Task<List<string>> FindAvailableRoomsAsync()
        {
            // Simulate a remote call to a booking system.
            await Task.Delay(1000);
            return ["Room 101", "Room 201", "Room 301"];
        }

        [KernelFunction("BookRoom")]
        [Description("Books a conference room.")]
        public async Task<string> BookRoomAsync(string room)
        {
            // Simulate a remote call to a booking system.
            await Task.Delay(1000);
            return $"Room {room} booked.";
        }
    }

    class Program
    {
        static async Task Main(string[] args)
        {
            // Endpoint to the Aspire Dashboard
            var endpoint = "http://localhost:4317";

            var resourceBuilder = ResourceBuilder
                .CreateDefault()
                .AddService("TelemetryAspireDashboardQuickstart");

            // Enable model diagnostics with sensitive data.
            AppContext.SetSwitch("Microsoft.SemanticKernel.Experimental.GenAI.EnableOTelDiagnosticsSensitive", true);

            using var traceProvider = Sdk.CreateTracerProviderBuilder()
                .SetResourceBuilder(resourceBuilder)
                .AddSource("Microsoft.SemanticKernel*")
                .AddOtlpExporter(options => options.Endpoint = new Uri(endpoint))
                .Build();

            using var meterProvider = Sdk.CreateMeterProviderBuilder()
                .SetResourceBuilder(resourceBuilder)
                .AddMeter("Microsoft.SemanticKernel*")
                .AddOtlpExporter(options => options.Endpoint = new Uri(endpoint))
                .Build();

            using var loggerFactory = LoggerFactory.Create(builder =>
            {
                // Add OpenTelemetry as a logging provider
                builder.AddOpenTelemetry(options =>
                {
                    options.SetResourceBuilder(resourceBuilder);
                    options.AddOtlpExporter(options => options.Endpoint = new Uri(endpoint));
                    // Format log messages. This is default to false.
                    options.IncludeFormattedMessage = true;
                    options.IncludeScopes = true;
                });
                builder.SetMinimumLevel(LogLevel.Information);
            });

            IKernelBuilder builder = Kernel.CreateBuilder();
            builder.Services.AddSingleton(loggerFactory);
            builder.AddAzureOpenAIChatCompletion(
                deploymentName: "your-deployment-name",
                endpoint: "your-azure-openai-endpoint",
                apiKey: "your-azure-openai-api-key"
            );
            builder.Plugins.AddFromType<BookingPlugin>();

            Kernel kernel = builder.Build();

            var answer = await kernel.InvokePromptAsync(
                "Reserve a conference room for me today.",
                new KernelArguments(
                    new OpenAIPromptExecutionSettings {
                        ToolCallBehavior = ToolCallBehavior.AutoInvokeKernelFunctions
                    }
                )
            );

            Console.WriteLine(answer);
        }
    }
}

在上面的代码中，我们首先定义了一个具有两个函数的模拟会议室预订插件： FindAvailableRoomsAsync 和 BookRoomAsync。然后创建一个简单的控制台应用程序，该应用程序将插件注册到内核，并要求内核在需要时自动调用函数。

创建新的 Python 虚拟环境

python -m venv telemetry-auto-function-calling-quickstart

激活虚拟环境。

telemetry-auto-function-calling-quickstart\Scripts\activate

安装所需程序包

pip install semantic-kernel opentelemetry-exporter-otlp-proto-grpc

使用语义内核创建简单的 Python 脚本

创建新的 Python 脚本，并使用你喜欢的编辑器打开它。

PowerShell
Bash

New-Item -Path telemetry_auto_function_calling_quickstart.py -ItemType file

touch telemetry_auto_function_calling_quickstart.py

我们将创建一个简单的 Python 脚本，该脚本使用语义内核向聊天完成模型发送提示。将现有内容替换为以下代码，并填写所需的值 deployment_name， endpoint以及 api_key：

import asyncio
import logging
from typing import Annotated

from opentelemetry._logs import set_logger_provider
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.metrics import set_meter_provider
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.metrics.view import DropAggregation, View
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.trace import set_tracer_provider

from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings
from semantic_kernel.functions.kernel_arguments import KernelArguments
from semantic_kernel.functions.kernel_function_decorator import kernel_function


class BookingPlugin:
    @kernel_function(
        name="find_available_rooms",
        description="Find available conference rooms for today.",
    )
    def find_available_rooms(self,) -> Annotated[list[str], "A list of available rooms."]:
        return ["Room 101", "Room 201", "Room 301"]

    @kernel_function(
        name="book_room",
        description="Book a conference room.",
    )
    def book_room(self, room: str) -> Annotated[str, "A confirmation message."]:
        return f"Room {room} booked."


# Endpoint to the Aspire Dashboard
endpoint = "http://localhost:4317"

# Create a resource to represent the service/sample
resource = Resource.create({ResourceAttributes.SERVICE_NAME: "telemetry-aspire-dashboard-quickstart"})


def set_up_logging():
    exporter = OTLPLogExporter(endpoint=endpoint)

    # Create and set a global logger provider for the application.
    logger_provider = LoggerProvider(resource=resource)
    # Log processors are initialized with an exporter which is responsible
    # for sending the telemetry data to a particular backend.
    logger_provider.add_log_record_processor(BatchLogRecordProcessor(exporter))
    # Sets the global default logger provider
    set_logger_provider(logger_provider)

    # Create a logging handler to write logging records, in OTLP format, to the exporter.
    handler = LoggingHandler()
    # Add filters to the handler to only process records from semantic_kernel.
    handler.addFilter(logging.Filter("semantic_kernel"))
    # Attach the handler to the root logger. `getLogger()` with no arguments returns the root logger.
    # Events from all child loggers will be processed by this handler.
    logger = logging.getLogger()
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)


def set_up_tracing():
    exporter = OTLPSpanExporter(endpoint=endpoint)

    # Initialize a trace provider for the application. This is a factory for creating tracers.
    tracer_provider = TracerProvider(resource=resource)
    # Span processors are initialized with an exporter which is responsible
    # for sending the telemetry data to a particular backend.
    tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
    # Sets the global default tracer provider
    set_tracer_provider(tracer_provider)


def set_up_metrics():
    exporter = OTLPMetricExporter(endpoint=endpoint)

    # Initialize a metric provider for the application. This is a factory for creating meters.
    meter_provider = MeterProvider(
        metric_readers=[PeriodicExportingMetricReader(exporter, export_interval_millis=5000)],
        resource=resource,
        views=[
            # Dropping all instrument names except for those starting with "semantic_kernel"
            View(instrument_name="*", aggregation=DropAggregation()),
            View(instrument_name="semantic_kernel*"),
        ],
    )
    # Sets the global default meter provider
    set_meter_provider(meter_provider)


# This must be done before any other telemetry calls
set_up_logging()
set_up_tracing()
set_up_metrics()


async def main():
    # Create a kernel and add a service
    kernel = Kernel()
    kernel.add_service(AzureChatCompletion(
        api_key="your-azure-openai-api-key",
        endpoint="your-azure-openai-endpoint",
        deployment_name="your-deployment-name"
    ))
    kernel.add_plugin(BookingPlugin(), "BookingPlugin")

    answer = await kernel.invoke_prompt(
        "Reserve a conference room for me today.",
        arguments=KernelArguments(
            settings=PromptExecutionSettings(
                function_choice_behavior=FunctionChoiceBehavior.Auto(),
            ),
        ),
    )
    print(answer)


if __name__ == "__main__":
    asyncio.run(main())

在上面的代码中，我们首先定义了一个具有两个函数的模拟会议室预订插件： find_available_rooms 和 book_room。然后创建一个简单的 Python 脚本，用于将插件注册到内核，并要求内核在需要时自动调用函数。

环境变量

有关设置所需环境变量以使内核能够发出 AI 连接器跨度的详细信息，请参阅本文。

注意

语义内核可观测性尚不适用于 Java。

启动 Aspire 仪表板

按照此处的说明启动仪表板。仪表板运行后，打开浏览器并导航到 http://localhost:18888 访问仪表板。

运行

使用以下命令运行控制台应用程序：

dotnet run

使用以下命令运行 Python 脚本：

python telemetry_auto_function_calling_quickstart.py

注意

语义内核可观测性尚不适用于 Java。

应该会看到与下面类似的输出：

Room 101 has been successfully booked for you today.

检查遥测数据

运行应用程序后，转到仪表板以检查遥测数据。

在 “跟踪 ”选项卡中查找应用程序的跟踪。跟踪中应有五个范围：

TracesAdvancedScenarioDotNet

这 5 个范围表示启用了自动函数调用的内核的内部操作。它首先调用请求函数调用的模型。然后内核会自动执行函数 FindAvailableRoomsAsync 并将结果返回到模型。然后，模型请求另一个函数调用进行预留，内核会自动执行该函数 BookRoomAsync 并将结果返回到模型。最后，模型将向用户返回自然语言响应。

如果单击最后一个范围，并在事件中 gen_ai.content.prompt 查找提示，则应看到类似于以下内容的内容：

[
  { "role": "user", "content": "Reserve a conference room for me today." },
  {
    "role": "Assistant",
    "content": null,
    "tool_calls": [
      {
        "id": "call_NtKi0OgOllJj1StLkOmJU8cP",
        "function": { "arguments": {}, "name": "FindAvailableRooms" },
        "type": "function"
      }
    ]
  },
  {
    "role": "tool",
    "content": "[\u0022Room 101\u0022,\u0022Room 201\u0022,\u0022Room 301\u0022]"
  },
  {
    "role": "Assistant",
    "content": null,
    "tool_calls": [
      {
        "id": "call_mjQfnZXLbqp4Wb3F2xySds7q",
        "function": { "arguments": { "room": "Room 101" }, "name": "BookRoom" },
        "type": "function"
      }
    ]
  },
  { "role": "tool", "content": "Room Room 101 booked." }
]

这是在模型和内核相互交互时生成的聊天历史记录。这在上次迭代中发送到模型以获取自然语言响应。

在 “跟踪 ”选项卡中查找应用程序的跟踪。应在范围下分组的跟踪中包括五个 AutoFunctionInvocationLoop 跨度：

TracesAdvancedScenarioPython

这 5 个范围表示启用了自动函数调用的内核的内部操作。它首先调用请求函数调用的模型。然后内核会自动执行函数 find_available_rooms 并将结果返回到模型。然后，模型请求另一个函数调用进行预留，内核会自动执行该函数 book_room 并将结果返回到模型。最后，模型将向用户返回自然语言响应。

如果单击最后一个范围，并在事件中 gen_ai.content.prompt 查找提示，则应看到类似于以下内容的内容：

[
  { "role": "user", "content": "Reserve a conference room for me today." },
  {
    "role": "assistant",
    "tool_calls": [
      {
        "id": "call_ypqO5v6uTRlYH9sPTjvkGec8",
        "type": "function",
        "function": {
          "name": "BookingPlugin-find_available_rooms",
          "arguments": "{}"
        }
      }
    ]
  },
  {
    "role": "tool",
    "content": "['Room 101', 'Room 201', 'Room 301']",
    "tool_call_id": "call_ypqO5v6uTRlYH9sPTjvkGec8"
  },
  {
    "role": "assistant",
    "tool_calls": [
      {
        "id": "call_XDZGeTfNiWRpYKoHoH9TZRoX",
        "type": "function",
        "function": {
          "name": "BookingPlugin-book_room",
          "arguments": "{\"room\":\"Room 101\"}"
        }
      }
    ]
  },
  {
    "role": "tool",
    "content": "Room Room 101 booked.",
    "tool_call_id": "call_XDZGeTfNiWRpYKoHoH9TZRoX"
  }
]

这是在模型和内核相互交互时生成的聊天历史记录。这在上次迭代中发送到模型以获取自然语言响应。

注意

语义内核可观测性尚不适用于 Java。

错误处理

如果在执行函数期间发生错误，内核将自动捕获错误并将错误消息返回到模型。然后，模型可以使用此错误消息向用户提供自然语言响应。

BookRoomAsync修改 C# 代码中的函数以模拟错误：

[KernelFunction("BookRoom")]
[Description("Books a conference room.")]
public async Task<string> BookRoomAsync(string room)
{
    // Simulate a remote call to a booking system.
    await Task.Delay(1000);

    throw new Exception("Room is not available.");
}

再次运行应用程序并观察仪表板中的跟踪。应会看到表示内核函数调用的跨度，并出现错误：

TracesAdvancedScenarioErrorDotNet

注意

模型对错误的响应很可能在每次运行应用程序时都有所不同，因为模型是随机的。你可能会看到模型同时保留所有三个房间，或者第一次保留一个房间，然后第二次保留其他两个房间，等等。

book_room修改 Python 代码中的函数以模拟错误：

@kernel_function(
    name="book_room",
    description="Book a conference room.",
)
async def book_room(self, room: str) -> Annotated[str, "A confirmation message."]:
    # Simulate a remote call to a booking system
    await asyncio.sleep(1)

    raise Exception("Room is not available.")

再次运行应用程序并观察仪表板中的跟踪。应会看到表示内核函数调用的跨度，并显示错误和堆栈跟踪：

TracesAdvancedScenarioErrorPython

注意

语义内核可观测性尚不适用于 Java。

后续步骤和进一步阅读

在生产环境中，服务可能会收到大量请求。语义内核将生成大量的遥测数据。其中一些可能对用例没有用处，并且会引入不必要的成本来存储数据。可以使用采样功能减少收集的遥测数据量。

语义内核中的可观测性不断改进。可以在 GitHub 存储库中找到最新的更新和新功能。

通过

遥测的更高级方案

自动函数调用

先决条件

安装

创建新的控制台应用程序

安装所需程序包

使用语义内核创建简单应用程序

创建新的 Python 虚拟环境

安装所需程序包

使用语义内核创建简单的 Python 脚本

环境变量

启动 Aspire 仪表板

运行

检查遥测数据

错误处理

后续步骤和进一步阅读

其他资源