为机器人实现自定义存储

适用于：SDK v4

机器人的交互划分为三个方面：与 Azure AI 机器人服务的活动交换，使用内存存储加载和保存机器人和对话状态，以及与后端服务的集成。

本文介绍如何扩展 Azure AI 机器人服务与机器人的内存状态和存储之间的语义。

先决条件

• 了解 Microsoft Bot Framework 的基础知识、使用活动处理程序的事件驱动对话以及管理状态。
• C#、Python 或 Java 中横向扩展示例的副本。

本文重点介绍示例的 C# 版本。

背景

Bot Framework SDK 包括机器人状态和内存存储的默认实现。 此实现适用于需要将这些片段与少量的初始化代码一起使用的应用程序，正如许多示例中所演示的那样。

SDK 是一个框架，而不是具有固定行为的应用程序。 换句话说，框架中许多机制的实现是默认实现，而不是唯一可能的实现。 该框架不会规定与 Azure AI 机器人服务的活动交换，以及加载和保存任何机器人状态这两者之间的关系。

本文介绍在不太适用于应用程序时，修改默认状态和存储实现的语义的一种方法。 横向扩展示例提供了状态和存储的一种备用实现，其语义与默认实现不同。 这一替代方案在该框架下同样适用。 根据你的应用场景，此备用解决方案可能更适合你正在开发的应用程序。

默认适配器和存储提供程序的行为

在默认实现中，机器人在收到活动时会加载与该对话对应的状态。 然后，它会基于此状态和传入活动执行对话逻辑。 在运行该对话的过程中，会创建并立即发送一个或多个出站活动。 对话处理完成后，机器人将保存更新的状态，并覆盖旧状态。

不过，这种情况也可能会出一些问题。

• 如果保存操作由于某种原因而失败，则状态与用户在频道上看到的内容隐式地失去同步。 用户已经看到机器人的响应，并认为状态已经向前移动，但并非如此。 此错误可能比状态更新成功但用户未收到响应消息的情形更糟。

  此类状态错误可能对对话设计产生影响。 例如，对话可能需要与用户进行额外的，或者冗余的确认交换。

• 如果该实现被部署为跨多个节点的横向扩展形式，则状态可能会被意外覆盖。 此错误可能会令人困惑，因为对话可能已将活动发送到包含确认消息的通道。

  考虑一个披萨订购机器人，机器人会询问用户的配料选择，而用户会发送两条快速消息：一条是添加蘑菇，一条是添加奶酪。 在横向扩展方案中，机器人的多个实例可能处于活动状态，并且两条用户消息可由不同的计算机上的两个单独的实例处理。 这种冲突称为竞争条件，其中一台计算机可能会覆盖另一台计算机写入的状态。 但是，由于回复已发送，用户收到了其订单中已添加蘑菇和奶酪的确认信息。 遗憾的是，送达的披萨仅包含蘑菇或奶酪，而不是包含两者。

乐观锁

横向扩展示例围绕状态管理引入了一些锁机制。 此示例实现“乐观锁定”，让每个实例都像自己是唯一运行实例一样运行，然后检查任何并发冲突。 此锁定听起来可能很复杂，但有已知解决方案存在，并且可以使用 Bot Framework 中的云存储技术和正确的扩展点。

示例使用基于实体标记标头 (ETag) 的标准 HTTP 机制。 了解此机制对于理解以下代码至关重要。 下图演示了序列。

该示意图有对某个资源执行更新的两个客户端。

1. 当客户端发出 GET 请求并从服务器返回资源时，服务器纳入一个 ETag 标头。

   ETag 标头是代表资源状态的不透明值。 如果资源已更改，服务器会更新资源的 ETag。

2. 当客户端想要保留状态更改时，它会向服务器发出 POST 请求，并在 If-Match 前置条件标头中包含 ETag 值。

3. 如果请求的 ETag 值与服务器的值不匹配，则前置条件检查失败并显示 412（前置条件失败）响应。

   此失败表示服务器上的当前值不再与客户端处理的原始值匹配。

4. 如果客户端收到前置条件失败响应，客户端通常会获取资源的新值，应用所需的更新，并尝试再次发布资源更新。

   如果没有其他客户端更新资源，则第二个 POST 请求会成功。 否则，客户端可以重试。

此过程之所以称为“乐观”，是因为获得了资源的客户端可继续执行处理，而资源本身并未“锁定”，因为其他客户端可以不受限制地访问它。 客户端之间对于资源应处于何种状态的任何争议，都要等到处理完成后才能确定。 在分布式系统中，此策略通常比相对的“悲观”方法更佳。

如上所述的乐观锁定机制假定可以安全地重试程序逻辑。 理想情况下，这些服务请求是幂等的。 在计算机科学中，幂等操作是指使用相同输入参数调用多次不会造成额外影响的操作。 实现了 GET、PUT 和 DELETE 请求方法的纯 HTTP REST 服务通常具有幂等性。 如果服务请求不会产生额外的效果，则作为重试策略的一部分，可以安全地重新执行请求。

横向扩展示例和本文的其余部分都假设你的机器人所使用的后端服务均为幂等的 HTTP REST 服务。

缓冲出站活动

发送活动不是幂等操作。 活动通常是将信息中继给用户的消息，两次或多次重复相同的消息可能会令人困惑或产生误导。

乐观锁定意味着机器人逻辑可能需要多次重新运行。 为了避免多次发送任何给定活动，请在向用户发送活动之前等待状态更新操作成功。 机器人逻辑应如下图所示。

在对话执行中生成重试循环后，在保存操作的前置条件失败时，会出现以下行为。

有了这种机制，前面示例中的披萨机器人就绝不会错误地确认已将某种披萨配料添加到订单中。 即使将机器人部署在多台计算机上，乐观锁方案也能有效地将状态更新序列化。 在披萨机器人中，添加商品后的确认信息现在甚至能够准确反映完整的状态信息。 例如，如果用户快速键入“奶酪”，然后键入“蘑菇”，并且这些消息由机器人的两个不同的实例处理，则后一个完成的实例可以包括“包含奶酪和蘑菇的披萨”作为响应的一部分。

这个新的自定义存储解决方案有三点是 SDK 中的默认实现无法做到的：

1. 它使用 ETag 来检测争用。

2. 检测到 ETag 失败时，它会重试处理。

3. 它等待发送出站活动，直到它已成功保存状态。

本文的余下内容将介绍这三个部分的实现。

实现 ETag 支持

首先，为我们的新存储定义一个支持 ETag 的接口。 该接口有助于在 ASP.NET 中使用依赖项注入机制。 从接口开始，你可以为单元测试和生产环境分别实现不同的版本。 例如，单元测试版本可能在内存中运行，不需要网络连接。

接口由 load 和 save 方法组成。 这两种方法都将使用键参数来标识要从存储中加载或保存到存储中的状态。

• Load 将返回状态值及其关联的 ETag。
• “保存”将具有状态值和关联的 ETag 的参数，并返回一个布尔值，该值指示操作是否成功。 返回值不会用作常规错误指示器，而是作为前置条件失败的特定指示器。 检查返回代码属于重试循环逻辑的一部分。

若要使存储实现广泛适用，请避免对存储实现提出序列化要求。 但是，许多新型存储服务支持 JSON 作为 content-type。 在 C# 中，可以使用 JObject 类型来表示 JSON 对象。 在 JavaScript 或 TypeScript 中，JSON 是一个常规本机对象。

以下是自定义接口的定义。

IStore.cs

public interface IStore
{
    Task<(JObject content, string etag)> LoadAsync(string key);

    Task<bool> SaveAsync(string key, JObject content, string etag);
}

下面是 Azure Blob 存储的实现。

BlobStore.cs

public class BlobStore : IStore
{
    private readonly CloudBlobContainer _container;

    public BlobStore(string accountName, string accountKey, string containerName)
    {
        if (string.IsNullOrWhiteSpace(accountName))
        {
            throw new ArgumentException(nameof(accountName));
        }

        if (string.IsNullOrWhiteSpace(accountKey))
        {
            throw new ArgumentException(nameof(accountKey));
        }

        if (string.IsNullOrWhiteSpace(containerName))
        {
            throw new ArgumentException(nameof(containerName));
        }

        var storageCredentials = new StorageCredentials(accountName, accountKey);
        var cloudStorageAccount = new CloudStorageAccount(storageCredentials, useHttps: true);
        var client = cloudStorageAccount.CreateCloudBlobClient();
        _container = client.GetContainerReference(containerName);
    }

    public async Task<(JObject content, string etag)> LoadAsync(string key)
    {
        if (string.IsNullOrWhiteSpace(key))
        {
            throw new ArgumentException(nameof(key));
        }

        var blob = _container.GetBlockBlobReference(key);
        try
        {
            var content = await blob.DownloadTextAsync();
            var obj = JObject.Parse(content);
            var etag = blob.Properties.ETag;
            return (obj, etag);
        }
        catch (StorageException e)
            when (e.RequestInformation.HttpStatusCode == (int)HttpStatusCode.NotFound)
        {
            return (new JObject(), null);
        }
    }

    public async Task<bool> SaveAsync(string key, JObject obj, string etag)
    {
        if (string.IsNullOrWhiteSpace(key))
        {
            throw new ArgumentException(nameof(key));
        }

        if (obj == null)
        {
            throw new ArgumentNullException(nameof(obj));
        }

        var blob = _container.GetBlockBlobReference(key);
        blob.Properties.ContentType = "application/json";
        var content = obj.ToString();
        if (etag != null)
        {
            try
            {
                await blob.UploadTextAsync(content, Encoding.UTF8, new AccessCondition { IfMatchETag = etag }, new BlobRequestOptions(), new OperationContext());
            }
            catch (StorageException e)
                when (e.RequestInformation.HttpStatusCode == (int)HttpStatusCode.PreconditionFailed)
            {
                return false;
            }
        }
        else
        {
            await blob.UploadTextAsync(content);
        }

        return true;
    }
}

Azure Blob 存储完成大部分工作。 每个方法会检查特定异常以满足调用代码的预期。

• 方法 LoadAsync 在响应带有“未找到”状态代码的存储异常时，返回一个 null 值。
• 方法 SaveAsync 在响应带有“前置条件失败”代码的存储异常时，返回 false。

实现重试循环

重试循环的设计实现了序列图中显示的行为。

1. 收到活动时，为对话状态创建一个键。

   对于自定义存储，活动与对话状态之间的关系与默认实现中的情况相同。 因此，可以采用与默认状态实现相同的方式构造键。

2. 尝试加载对话状态。

3. 运行机器人的对话流程，并捕获要发送的传出活动。

4. 尝试保存对话状态。

   • 成功后，发送出站活动并退出。

   • 失败时，从加载对话状态步骤重复此过程。

     新加载的对话状态会获得新的当前 ETag 和对话状态。 重新运行该对话框后，保存状态这一步就有机会成功。

下面是消息活动处理程序的实现。

ScaleoutBot.cs

protected override async Task OnMessageActivityAsync(ITurnContext<IMessageActivity> turnContext, CancellationToken cancellationToken)
{
    // Create the storage key for this conversation.
    var key = $"{turnContext.Activity.ChannelId}/conversations/{turnContext.Activity.Conversation?.Id}";

    // The execution sits in a loop because there might be a retry if the save operation fails.
    while (true)
    {
        // Load any existing state associated with this key
        var (oldState, etag) = await _store.LoadAsync(key);

        // Run the dialog system with the old state and inbound activity, the result is a new state and outbound activities.
        var (activities, newState) = await DialogHost.RunAsync(_dialog, turnContext.Activity, oldState, cancellationToken);

        // Save the updated state associated with this key.
        var success = await _store.SaveAsync(key, newState, etag);

        // Following a successful save, send any outbound Activities, otherwise retry everything.
        if (success)
        {
            if (activities.Any())
            {
                // This is an actual send on the TurnContext we were given and so will actual do a send this time.
                await turnContext.SendActivitiesAsync(activities, cancellationToken);
            }

            break;
        }
    }
}

实现出站活动缓冲机制

下一个要求是缓冲出站活动，直到成功保存操作发生后，这需要自定义适配器实现。 自定义 SendActivitiesAsync 方法不应将这些活动发送供使用，而应将其添加到列表中。 对话代码将不需要修改。

• 在此特定场景中，不支持 更新活动 和 删除活动 操作，相关方法将引发 未实现 异常。
• 某些通道使用发送活动操作的返回值，允许机器人修改或删除之前发送的信息，例如，禁用通道中显示的卡片上的按钮。 需要状态时，这些消息交换可能特别复杂，这不属于本文的讨论范畴。
• 对话会创建和使用此自定义适配器，以便可以缓冲活动。
• 机器人的回合处理程序将使用更标准的方式，通过 AdapterWithErrorHandler 向用户发送活动。

下面是自定义适配器的实现。

DialogHostAdapter.cs

public class DialogHostAdapter : BotAdapter
{
    private List<Activity> _response = new List<Activity>();

    public IEnumerable<Activity> Activities => _response;

    public override Task<ResourceResponse[]> SendActivitiesAsync(ITurnContext turnContext, Activity[] activities, CancellationToken cancellationToken)
    {
        foreach (var activity in activities)
        {
            _response.Add(activity);
        }

        return Task.FromResult(new ResourceResponse[0]);
    }

    #region Not Implemented
    public override Task DeleteActivityAsync(ITurnContext turnContext, ConversationReference reference, CancellationToken cancellationToken)
    {
        throw new NotImplementedException();
    }

    public override Task<ResourceResponse> UpdateActivityAsync(ITurnContext turnContext, Activity activity, CancellationToken cancellationToken)
    {
        throw new NotImplementedException();
    }
    #endregion
}

在机器人中使用自定义存储

最后一步是将这些自定义类和方法与现有框架类和方法一起使用。

• 主要重试循环会成为机器人ActivityHandler.OnMessageActivityAsync方法的一部分，并通过依赖注入纳入你的自定义存储。
• 对话框承载代码被添加到 DialogHost 类中，该类提供一个静态 RunAsync 方法。 对话主机：
  • 接收传入的活动和原有状态，然后返回生成的活动和新状态。
  • 创建自定义适配器，并在其他方面以与 SDK 相同的方式运行对话。
  • 创建一个自定义状态属性访问器，即一个将对话状态传递到对话系统的适配器。 访问器使用引用语义将访问器句柄传递给对话系统。

下面是对话主机的实现。

DialogHost.cs

public static class DialogHost
{
    // The serializer to use. Moving the serialization to this layer will make the storage layer more pluggable.
    private static readonly JsonSerializer StateJsonSerializer = new JsonSerializer() { TypeNameHandling = TypeNameHandling.All };

    /// <summary>
    /// A function to run a dialog while buffering the outbound Activities.
    /// </summary>
    /// <param name="dialog">THe dialog to run.</param>
    /// <param name="activity">The inbound Activity to run it with.</param>
    /// <param name="oldState">Th eexisting or old state.</param>
    /// <returns>An array of Activities 'sent' from the dialog as it executed. And the updated or new state.</returns>
    public static async Task<(Activity[], JObject)> RunAsync(Dialog dialog, IMessageActivity activity, JObject oldState, CancellationToken cancellationToken)
    {
        // A custom adapter and corresponding TurnContext that buffers any messages sent.
        var adapter = new DialogHostAdapter();
        var turnContext = new TurnContext(adapter, (Activity)activity);

        // Run the dialog using this TurnContext with the existing state.
        var newState = await RunTurnAsync(dialog, turnContext, oldState, cancellationToken);

        // The result is a set of activities to send and a replacement state.
        return (adapter.Activities.ToArray(), newState);
    }

    /// <summary>
    /// Execute the turn of the bot. The functionality here closely resembles that which is found in the
    /// IBot.OnTurnAsync method in an implementation that is using the regular BotFrameworkAdapter.
    /// Also here in this example the focus is explicitly on Dialogs but the pattern could be adapted
    /// to other conversation modeling abstractions.
    /// </summary>
    /// <param name="dialog">The dialog to be run.</param>
    /// <param name="turnContext">The ITurnContext instance to use. Note this is not the one passed into the IBot OnTurnAsync.</param>
    /// <param name="state">The existing or old state of the dialog.</param>
    /// <returns>The updated or new state of the dialog.</returns>
    private static async Task<JObject> RunTurnAsync(Dialog dialog, ITurnContext turnContext, JObject state, CancellationToken cancellationToken)
    {
        // If we have some state, deserialize it. (This mimics the shape produced by BotState.cs.)
        var dialogStateProperty = state?[nameof(DialogState)];
        var dialogState = dialogStateProperty?.ToObject<DialogState>(StateJsonSerializer);

        // A custom accessor is used to pass a handle on the state to the dialog system.
        var accessor = new RefAccessor<DialogState>(dialogState);

        // Run the dialog.
        await dialog.RunAsync(turnContext, accessor, cancellationToken);

        // Serialize the result (available as Value on the accessor), and put its value back into a new JObject.
        return new JObject { { nameof(DialogState), JObject.FromObject(accessor.Value, StateJsonSerializer) } };
    }
}

最后，下面是自定义状态属性访问器的实现。

RefAccessor.cs

public class RefAccessor<T> : IStatePropertyAccessor<T>
    where T : class
{
    public RefAccessor(T value)
    {
        Value = value;
    }

    public T Value { get; private set; }

    public string Name => nameof(T);

    public Task<T> GetAsync(ITurnContext turnContext, Func<T> defaultValueFactory = null, CancellationToken cancellationToken = default(CancellationToken))
    {
        if (Value == null)
        {
            if (defaultValueFactory == null)
            {
                throw new KeyNotFoundException();
            }

            Value = defaultValueFactory();
        }

        return Task.FromResult(Value);
    }

    #region Not Implemented
    public Task DeleteAsync(ITurnContext turnContext, CancellationToken cancellationToken = default(CancellationToken))
    {
        throw new NotImplementedException();
    }

    public Task SetAsync(ITurnContext turnContext, T value, CancellationToken cancellationToken = default(CancellationToken))
    {
        throw new NotImplementedException();
    }
    #endregion
}

其他信息

横向扩展示例可从 GitHub 上的 Bot Framework 示例存储库（C#、Python 和 Java）获取。