Partager via

Stockage de l’historique des conversations dans le stockage tiers

Ce tutoriel montre comment stocker l’historique des conversations de l’agent dans le stockage externe en implémentant un fichier personnalisé ChatMessageStore et en l’utilisant avec un ChatClientAgent.

Par défaut, lors de l’utilisation ChatClientAgent, l’historique des conversations est stocké en mémoire dans l’objet ou dans le AgentThread service d’inférence sous-jacent, si le service le prend en charge.

Lorsque les services n’ont pas besoin de stocker l’historique des conversations dans le service, il est possible de fournir un magasin personnalisé pour conserver l’historique des conversations au lieu de s’appuyer sur le comportement en mémoire par défaut.

Prerequisites

Pour connaître les prérequis, consultez l’étape Créer et exécuter un agent simple dans ce tutoriel.

Installer les packages NuGet

Pour utiliser Microsoft Agent Framework avec Azure OpenAI, vous devez installer les packages NuGet suivants :

dotnet add package Azure.AI.OpenAI --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.OpenAI --prerelease

En outre, vous allez utiliser le magasin vectoriel en mémoire pour stocker les messages de conversation.

dotnet add package Microsoft.SemanticKernel.Connectors.InMemory --prerelease

Créer un magasin ChatMessage personnalisé

Pour créer une classe personnalisée ChatMessageStore, vous devez implémenter la classe abstraite ChatMessageStore et fournir des implémentations pour les méthodes requises.

Méthodes de stockage et de récupération des messages

Les méthodes les plus importantes à implémenter sont les suivantes :

  • AddMessagesAsync - appelé pour ajouter de nouveaux messages au magasin.
  • GetMessagesAsync - appelé pour récupérer les messages à partir du magasin.

GetMessagesAsync doit retourner les messages dans l’ordre chronologique croissant. Tous les messages retournés par celui-ci seront utilisés par le ChatClientAgent lors des appels effectués vers le système sous-jacent IChatClient. Il est donc important que cette méthode considère les limites du modèle sous-jacent et retourne uniquement autant de messages que possible par le modèle.

Toute logique de réduction de l’historique des conversations, telle que la synthèse ou la suppression, doit être effectuée avant de retourner des messages à partir de GetMessagesAsync.

Sérialisation

ChatMessageStore les instances sont créées et attachées à un AgentThread lorsque le thread est créé et lorsqu'un thread est repris d'un état sérialisé.

Bien que les messages réels qui composent l’historique des conversations soient stockés en externe, l’instance ChatMessageStore peut avoir besoin de stocker des clés ou d’un autre état pour identifier l’historique des conversations dans le magasin externe.

Pour autoriser la persistance des threads, vous devez implémenter la SerializeStateAsync méthode de la ChatMessageStore classe. Vous devez également fournir un constructeur qui accepte un JsonElement paramètre, qui peut être utilisé pour désérialiser l’état lors de la reprise d’un thread.

Exemple d’implémentation de ChatMessageStore

L’exemple d’implémentation suivant stocke les messages de conversation dans un magasin vectoriel.

AddMessagesAsync insère ou met à jour des messages dans le stockage vectoriel, à l’aide d’une clé unique pour chaque message.

GetMessagesAsync récupère les messages du thread actuel à partir du magasin vectoriel, les commande par horodatage et les retourne dans l’ordre croissant.

Lorsque le premier message est reçu, le magasin génère une clé unique pour le thread, qui est ensuite utilisée pour identifier l’historique des conversations dans le magasin vectoriel pour les appels suivants.

La clé unique est stockée dans la propriété ThreadDbKey, qui est sérialisée et désérialisée à l'aide de la méthode SerializeStateAsync et du constructeur qui utilise un JsonElement. Cette clé est donc conservée dans le cadre de l’état AgentThread , ce qui permet au thread d’être repris ultérieurement et de continuer à utiliser le même historique de conversation.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.VectorData;
using Microsoft.SemanticKernel.Connectors.InMemory;

internal sealed class VectorChatMessageStore : ChatMessageStore
{
    private readonly VectorStore _vectorStore;

    public VectorChatMessageStore(
        VectorStore vectorStore,
        JsonElement serializedStoreState,
        JsonSerializerOptions? jsonSerializerOptions = null)
    {
        this._vectorStore = vectorStore ?? throw new ArgumentNullException(nameof(vectorStore));
        if (serializedStoreState.ValueKind is JsonValueKind.String)
        {
            this.ThreadDbKey = serializedStoreState.Deserialize<string>();
        }
    }

    public string? ThreadDbKey { get; private set; }

    public override async Task AddMessagesAsync(
        IEnumerable<ChatMessage> messages,
        CancellationToken cancellationToken)
    {
        this.ThreadDbKey ??= Guid.NewGuid().ToString("N");
        var collection = this._vectorStore.GetCollection<string, ChatHistoryItem>("ChatHistory");
        await collection.EnsureCollectionExistsAsync(cancellationToken);
        await collection.UpsertAsync(messages.Select(x => new ChatHistoryItem()
        {
            Key = this.ThreadDbKey + x.MessageId,
            Timestamp = DateTimeOffset.UtcNow,
            ThreadId = this.ThreadDbKey,
            SerializedMessage = JsonSerializer.Serialize(x),
            MessageText = x.Text
        }), cancellationToken);
    }

    public override async Task<IEnumerable<ChatMessage>> GetMessagesAsync(
        CancellationToken cancellationToken)
    {
        var collection = this._vectorStore.GetCollection<string, ChatHistoryItem>("ChatHistory");
        await collection.EnsureCollectionExistsAsync(cancellationToken);
        var records = collection
            .GetAsync(
                x => x.ThreadId == this.ThreadDbKey, 10,
                new() { OrderBy = x => x.Descending(y => y.Timestamp) },
                cancellationToken);

        List<ChatMessage> messages = [];
        await foreach (var record in records)
        {
            messages.Add(JsonSerializer.Deserialize<ChatMessage>(record.SerializedMessage!)!);
        }

        messages.Reverse();
        return messages;
    }

    public override JsonElement Serialize(JsonSerializerOptions? jsonSerializerOptions = null) =>
        // We have to serialize the thread id, so that on deserialization you can retrieve the messages using the same thread id.
        JsonSerializer.SerializeToElement(this.ThreadDbKey);

    private sealed class ChatHistoryItem
    {
        [VectorStoreKey]
        public string? Key { get; set; }
        [VectorStoreData]
        public string? ThreadId { get; set; }
        [VectorStoreData]
        public DateTimeOffset? Timestamp { get; set; }
        [VectorStoreData]
        public string? SerializedMessage { get; set; }
        [VectorStoreData]
        public string? MessageText { get; set; }
    }
}

Utilisation du ChatMessageStore personnalisé avec un ChatClientAgent

Pour utiliser le ChatMessageStore personnalisé, vous devez fournir une valeur ChatMessageStoreFactory lors de la création de l’agent. Cette fabrique permet à l’agent de créer une nouvelle instance souhaitée ChatMessageStore pour chaque thread.

Lors de la création d’un ChatClientAgent, il est possible de fournir un ChatClientAgentOptions qui permet d’ajouter le ChatMessageStoreFactory en plus de toutes les autres options d’agent.

using Azure.AI.OpenAI;
using Azure.Identity;
using OpenAI;

AIAgent agent = new AzureOpenAIClient(
    new Uri("https://<myresource>.openai.azure.com"),
    new AzureCliCredential())
     .GetChatClient("gpt-4o-mini")
     .CreateAIAgent(new ChatClientAgentOptions
     {
         Name = "Joker",
         Instructions = "You are good at telling jokes.",
         ChatMessageStoreFactory = ctx =>
         {
             // Create a new chat message store for this agent that stores the messages in a vector store.
             return new VectorChatMessageStore(
                new InMemoryVectorStore(),
                ctx.SerializedState,
                ctx.JsonSerializerOptions);
         }
     });

Ce tutoriel montre comment stocker l’historique des conversations de l’agent dans le stockage externe en implémentant un fichier personnalisé ChatMessageStore et en l’utilisant avec un ChatAgent.

Par défaut, lors de l’utilisation ChatAgent, l’historique des conversations est stocké en mémoire dans l’objet ou dans le AgentThread service d’inférence sous-jacent, si le service le prend en charge.

Lorsque les services ne nécessitent pas ou ne sont pas capables de stocker l’historique des conversations dans le service, il est possible de fournir un magasin personnalisé pour conserver l’historique des conversations au lieu de s’appuyer sur le comportement en mémoire par défaut.

Prerequisites

Pour connaître les prérequis, consultez l’étape Créer et exécuter un agent simple dans ce tutoriel.

Créer un magasin ChatMessage personnalisé

Pour créer un protocole personnalisé ChatMessageStore, vous devez implémenter le ChatMessageStore protocole et fournir des implémentations pour les méthodes requises.

Méthodes de stockage et de récupération des messages

Les méthodes les plus importantes à implémenter sont les suivantes :

  • add_messages - appelé pour ajouter de nouveaux messages au magasin.
  • list_messages - appelé pour récupérer les messages à partir du magasin.

list_messages doit retourner les messages dans l’ordre chronologique croissant. Tous les messages retournés par celui-ci seront utilisés par ChatAgent lors des appels au client de messagerie sous-jacent. Il est donc important que cette méthode considère les limites du modèle sous-jacent et retourne uniquement autant de messages que possible par le modèle.

Toute logique de réduction de l’historique des conversations, telle que la synthèse ou la suppression, doit être effectuée avant de retourner des messages à partir de list_messages.

Sérialisation

ChatMessageStore les instances sont créées et attachées à un AgentThread lorsque le thread est créé et lorsqu'un thread est repris d'un état sérialisé.

Bien que les messages réels qui composent l’historique des conversations soient stockés en externe, l’instance ChatMessageStore peut avoir besoin de stocker des clés ou d’un autre état pour identifier l’historique des conversations dans le magasin externe.

Pour autoriser la persistance des threads, vous devez implémenter les méthodes serialize_state et deserialize_state du protocole ChatMessageStore. Ces méthodes permettent à l’état du magasin d’être conservé et restauré lors de la reprise d’un thread.

Exemple d’implémentation de ChatMessageStore

L’exemple d’implémentation suivant stocke les messages de conversation dans Redis à l’aide de la structure de données Listes Redis.

Dans add_messages, il stocke les messages dans Redis à l’aide de RPUSH pour les ajouter à la fin de la liste dans l’ordre chronologique.

list_messages récupère les messages du thread actuel de Redis à l’aide de LRANGE et les retourne dans l’ordre chronologique croissant.

Lorsque le premier message est reçu, le magasin génère une clé unique pour le thread, qui est ensuite utilisé pour identifier l’historique des conversations dans Redis pour les appels suivants.

La clé unique et d'autres configurations sont stockées et peuvent être sérialisées et désérialisées en utilisant les méthodes serialize_state et deserialize_state. Cet état est donc conservé dans le cadre de l’état AgentThread , ce qui permet au thread d’être repris ultérieurement et de continuer à utiliser le même historique de conversation.

from collections.abc import Sequence
from typing import Any
from uuid import uuid4
from pydantic import BaseModel
import json
import redis.asyncio as redis
from agent_framework import ChatMessage


class RedisStoreState(BaseModel):
    """State model for serializing and deserializing Redis chat message store data."""

    thread_id: str
    redis_url: str | None = None
    key_prefix: str = "chat_messages"
    max_messages: int | None = None


class RedisChatMessageStore:
    """Redis-backed implementation of ChatMessageStore using Redis Lists."""

    def __init__(
        self,
        redis_url: str | None = None,
        thread_id: str | None = None,
        key_prefix: str = "chat_messages",
        max_messages: int | None = None,
    ) -> None:
        """Initialize the Redis chat message store.

        Args:
            redis_url: Redis connection URL (for example, "redis://localhost:6379").
            thread_id: Unique identifier for this conversation thread.
                      If not provided, a UUID will be auto-generated.
            key_prefix: Prefix for Redis keys to namespace different applications.
            max_messages: Maximum number of messages to retain in Redis.
                         When exceeded, oldest messages are automatically trimmed.
        """
        if redis_url is None:
            raise ValueError("redis_url is required for Redis connection")

        self.redis_url = redis_url
        self.thread_id = thread_id or f"thread_{uuid4()}"
        self.key_prefix = key_prefix
        self.max_messages = max_messages

        # Initialize Redis client
        self._redis_client = redis.from_url(redis_url, decode_responses=True)

    @property
    def redis_key(self) -> str:
        """Get the Redis key for this thread's messages."""
        return f"{self.key_prefix}:{self.thread_id}"

    async def add_messages(self, messages: Sequence[ChatMessage]) -> None:
        """Add messages to the Redis store.

        Args:
            messages: Sequence of ChatMessage objects to add to the store.
        """
        if not messages:
            return

        # Serialize messages and add to Redis list
        serialized_messages = [self._serialize_message(msg) for msg in messages]
        await self._redis_client.rpush(self.redis_key, *serialized_messages)

        # Apply message limit if configured
        if self.max_messages is not None:
            current_count = await self._redis_client.llen(self.redis_key)
            if current_count > self.max_messages:
                # Keep only the most recent max_messages using LTRIM
                await self._redis_client.ltrim(self.redis_key, -self.max_messages, -1)

    async def list_messages(self) -> list[ChatMessage]:
        """Get all messages from the store in chronological order.

        Returns:
            List of ChatMessage objects in chronological order (oldest first).
        """
        # Retrieve all messages from Redis list (oldest to newest)
        redis_messages = await self._redis_client.lrange(self.redis_key, 0, -1)

        messages = []
        for serialized_message in redis_messages:
            message = self._deserialize_message(serialized_message)
            messages.append(message)

        return messages

    async def serialize_state(self, **kwargs: Any) -> Any:
        """Serialize the current store state for persistence.

        Returns:
            Dictionary containing serialized store configuration.
        """
        state = RedisStoreState(
            thread_id=self.thread_id,
            redis_url=self.redis_url,
            key_prefix=self.key_prefix,
            max_messages=self.max_messages,
        )
        return state.model_dump(**kwargs)

    async def deserialize_state(self, serialized_store_state: Any, **kwargs: Any) -> None:
        """Deserialize state data into this store instance.

        Args:
            serialized_store_state: Previously serialized state data.
            **kwargs: Additional arguments for deserialization.
        """
        if serialized_store_state:
            state = RedisStoreState.model_validate(serialized_store_state, **kwargs)
            self.thread_id = state.thread_id
            self.key_prefix = state.key_prefix
            self.max_messages = state.max_messages

            # Recreate Redis client if the URL changed
            if state.redis_url and state.redis_url != self.redis_url:
                self.redis_url = state.redis_url
                self._redis_client = redis.from_url(self.redis_url, decode_responses=True)

    def _serialize_message(self, message: ChatMessage) -> str:
        """Serialize a ChatMessage to JSON string."""
        message_dict = message.model_dump()
        return json.dumps(message_dict, separators=(",", ":"))

    def _deserialize_message(self, serialized_message: str) -> ChatMessage:
        """Deserialize a JSON string to ChatMessage."""
        message_dict = json.loads(serialized_message)
        return ChatMessage.model_validate(message_dict)

    async def clear(self) -> None:
        """Remove all messages from the store."""
        await self._redis_client.delete(self.redis_key)

    async def aclose(self) -> None:
        """Close the Redis connection."""
        await self._redis_client.aclose()

Utilisation du ChatMessageStore personnalisé avec un ChatAgent

Pour utiliser le ChatMessageStore personnalisé, vous devez fournir une valeur chat_message_store_factory lors de la création de l’agent. Cette fabrique permet à l’agent de créer une nouvelle instance souhaitée ChatMessageStore pour chaque thread.

Lors de la création d’un ChatAgent, vous pouvez fournir le chat_message_store_factory paramètre en plus de toutes les autres options d’agent.

from azure.identity import AzureCliCredential
from agent_framework import ChatAgent
from agent_framework.openai import AzureOpenAIChatClient

# Create the chat agent with custom message store factory
agent = ChatAgent(
    chat_client=AzureOpenAIChatClient(
        endpoint="https://<myresource>.openai.azure.com",
        credential=AzureCliCredential(),
        ai_model_id="gpt-4o-mini"
    ),
    name="Joker",
    instructions="You are good at telling jokes.",
    chat_message_store_factory=lambda: RedisChatMessageStore(
        redis_url="redis://localhost:6379"
    )
)

# Use the agent with persistent chat history
thread = agent.get_new_thread()
response = await agent.run("Tell me a joke about pirates", thread=thread)
print(response.text)

Étapes suivantes

Ajout de la mémoire à un agent

  • Last updated on