Partager via

Agents personnalisés

Microsoft Agent Framework prend en charge la création d’agents personnalisés en hériter de la AIAgent classe et en implémentant les méthodes requises.

Ce document montre comment créer un agent personnalisé simple qui répète les données utilisateur en majuscules. Dans la plupart des cas, la création de votre propre agent implique une logique et une intégration plus complexes avec un service IA.

Getting Started

Ajoutez les packages NuGet requis à votre projet.

dotnet add package Microsoft.Agents.AI.Abstractions --prerelease

Création d’un agent personnalisé

Thread de l’agent

Pour créer un agent personnalisé, vous avez également besoin d’un thread, qui est utilisé pour effectuer le suivi de l’état d’une conversation unique, y compris l’historique des messages et tout autre état dont l’agent a besoin pour gérer.

Pour faciliter la prise en main, vous pouvez hériter de différentes classes de base qui implémentent des mécanismes de stockage de thread courants.

  1. InMemoryAgentThread - stocke l’historique des conversations en mémoire et peut être sérialisé en JSON.
  2. ServiceIdAgentThread - ne stocke pas d’historique de conversation, mais vous permet d’associer un ID au thread, sous lequel l’historique des conversations peut être stocké en externe.

Pour cet exemple, nous allons utiliser la InMemoryAgentThread classe de base pour notre thread personnalisé.

internal sealed class CustomAgentThread : InMemoryAgentThread
{
    internal CustomAgentThread() : base() { }
    internal CustomAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
        : base(serializedThreadState, jsonSerializerOptions) { }
}

Classe Agent

Ensuite, nous voulons créer la classe d’agent elle-même en hériter de la AIAgent classe.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Construction de threads

Les threads sont toujours créés via deux méthodes de fabrique sur la classe d’agent. Cela permet à l’agent de contrôler la façon dont les threads sont créés et désérialisés. Les agents peuvent donc attacher tout état ou comportement supplémentaire nécessaire au thread lors de la construction.

Deux méthodes doivent être implémentées :

    public override AgentThread GetNewThread() => new CustomAgentThread();

    public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null)
        => new CustomAgentThread(serializedThread, jsonSerializerOptions);

Logique de l’agent principal

La logique principale de l’agent consiste à prendre tous les messages d’entrée, à convertir leur texte en majuscules et à les renvoyer sous forme de messages de réponse.

Nous voulons ajouter la méthode suivante pour contenir cette logique. Nous clonez les messages d’entrée, car différents aspects des messages d’entrée doivent être modifiés pour être des messages de réponse valides. Par exemple, le rôle doit être remplacé par Assistant.

    private static IEnumerable<ChatMessage> CloneAndToUpperCase(IEnumerable<ChatMessage> messages, string agentName) => messages.Select(x =>
        {
            var messageClone = x.Clone();
            messageClone.Role = ChatRole.Assistant;
            messageClone.MessageId = Guid.NewGuid().ToString();
            messageClone.AuthorName = agentName;
            messageClone.Contents = x.Contents.Select(c => c is TextContent tc ? new TextContent(tc.Text.ToUpperInvariant())
            {
                AdditionalProperties = tc.AdditionalProperties,
                Annotations = tc.Annotations,
                RawRepresentation = tc.RawRepresentation
            } : c).ToList();
            return messageClone;
        });

Méthodes de fonctionnement de l’agent

Enfin, nous devons implémenter les deux méthodes principales utilisées pour exécuter l’agent. Un pour la non-diffusion en continu et un pour la diffusion en continu.

Pour les deux méthodes, nous devons vérifier qu'un thread est disponible, et si ce n'est pas le cas, nous devons en créer un nouveau. Le thread peut ensuite être mis à jour avec les nouveaux messages en appelant NotifyThreadOfNewMessagesAsync. Si ce n’est pas le cas, l’utilisateur ne pourra pas avoir une conversation à plusieurs tours avec l’agent, et chaque session commencera par une nouvelle interaction.

    public override async Task<AgentRunResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
    {
        thread ??= this.GetNewThread();
        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
        await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), cancellationToken);
        return new AgentRunResponse
        {
            AgentId = this.Id,
            ResponseId = Guid.NewGuid().ToString(),
            Messages = responseMessages
        };
    }

    public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        thread ??= this.GetNewThread();
        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
        await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), cancellationToken);
        foreach (var message in responseMessages)
        {
            yield return new AgentRunResponseUpdate
            {
                AgentId = this.Id,
                AuthorName = this.DisplayName,
                Role = ChatRole.Assistant,
                Contents = message.Contents,
                ResponseId = Guid.NewGuid().ToString(),
                MessageId = Guid.NewGuid().ToString()
            };
        }
    }

Utilisation de l’agent

Si les AIAgent méthodes sont correctement implémentées, l’agent serait un agent standard AIAgent et prendrait en charge les opérations d’agent standard.

Pour plus d’informations sur l’exécution et l’interaction avec les agents, consultez les didacticiels de prise en main de l’agent .

Microsoft Agent Framework prend en charge la création d’agents personnalisés en hériter de la BaseAgent classe et en implémentant les méthodes requises.

Ce document montre comment créer un agent personnalisé simple qui renvoie l’entrée utilisateur avec un préfixe. Dans la plupart des cas, la création de votre propre agent implique une logique et une intégration plus complexes avec un service IA.

Getting Started

Ajoutez les packages Python requis à votre projet.

pip install agent-framework-core --pre

Création d’un agent personnalisé

Le protocole agent

L’infrastructure fournit le AgentProtocol protocole qui définit l’interface que tous les agents doivent implémenter. Les agents personnalisés peuvent implémenter ce protocole directement ou étendre la BaseAgent classe pour des raisons pratiques.

from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, AgentThread, ChatMessage
from collections.abc import AsyncIterable
from typing import Any

class MyCustomAgent(AgentProtocol):
    """A custom agent that implements the AgentProtocol directly."""

    @property
    def id(self) -> str:
        """Returns the ID of the agent."""
        ...

    async def run(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AgentRunResponse:
        """Execute the agent and return a complete response."""
        ...

    def run_stream(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentRunResponseUpdate]:
        """Execute the agent and yield streaming response updates."""
        ...

Utilisation de BaseAgent

L’approche recommandée consiste à étendre la BaseAgent classe, qui fournit des fonctionnalités courantes et simplifie l’implémentation :

from agent_framework import (
    BaseAgent,
    AgentRunResponse,
    AgentRunResponseUpdate,
    AgentThread,
    ChatMessage,
    Role,
    TextContent,
)
from collections.abc import AsyncIterable
from typing import Any


class EchoAgent(BaseAgent):
    """A simple custom agent that echoes user messages with a prefix."""

    echo_prefix: str = "Echo: "

    def __init__(
        self,
        *,
        name: str | None = None,
        description: str | None = None,
        echo_prefix: str = "Echo: ",
        **kwargs: Any,
    ) -> None:
        """Initialize the EchoAgent.

        Args:
            name: The name of the agent.
            description: The description of the agent.
            echo_prefix: The prefix to add to echoed messages.
            **kwargs: Additional keyword arguments passed to BaseAgent.
        """
        super().__init__(
            name=name,
            description=description,
            echo_prefix=echo_prefix,
            **kwargs,
        )

    async def run(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AgentRunResponse:
        """Execute the agent and return a complete response.

        Args:
            messages: The message(s) to process.
            thread: The conversation thread (optional).
            **kwargs: Additional keyword arguments.

        Returns:
            An AgentRunResponse containing the agent's reply.
        """
        # Normalize input messages to a list
        normalized_messages = self._normalize_messages(messages)

        if not normalized_messages:
            response_message = ChatMessage(
                role=Role.ASSISTANT,
                contents=[TextContent(text="Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")],
            )
        else:
            # For simplicity, echo the last user message
            last_message = normalized_messages[-1]
            if last_message.text:
                echo_text = f"{self.echo_prefix}{last_message.text}"
            else:
                echo_text = f"{self.echo_prefix}[Non-text message received]"

            response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=echo_text)])

        # Notify the thread of new messages if provided
        if thread is not None:
            await self._notify_thread_of_new_messages(thread, normalized_messages, response_message)

        return AgentRunResponse(messages=[response_message])

    async def run_stream(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentRunResponseUpdate]:
        """Execute the agent and yield streaming response updates.

        Args:
            messages: The message(s) to process.
            thread: The conversation thread (optional).
            **kwargs: Additional keyword arguments.

        Yields:
            AgentRunResponseUpdate objects containing chunks of the response.
        """
        # Normalize input messages to a list
        normalized_messages = self._normalize_messages(messages)

        if not normalized_messages:
            response_text = "Hello! I'm a custom echo agent. Send me a message and I'll echo it back."
        else:
            # For simplicity, echo the last user message
            last_message = normalized_messages[-1]
            if last_message.text:
                response_text = f"{self.echo_prefix}{last_message.text}"
            else:
                response_text = f"{self.echo_prefix}[Non-text message received]"

        # Simulate streaming by yielding the response word by word
        words = response_text.split()
        for i, word in enumerate(words):
            # Add space before word except for the first one
            chunk_text = f" {word}" if i > 0 else word

            yield AgentRunResponseUpdate(
                contents=[TextContent(text=chunk_text)],
                role=Role.ASSISTANT,
            )

            # Small delay to simulate streaming
            await asyncio.sleep(0.1)

        # Notify the thread of the complete response if provided
        if thread is not None:
            complete_response = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)])
            await self._notify_thread_of_new_messages(thread, normalized_messages, complete_response)

Utilisation de l’agent

Si les méthodes de l’agent sont correctement implémentées, l’agent prend en charge toutes les opérations d’agent standard.

Pour plus d’informations sur l’exécution et l’interaction avec les agents, consultez les didacticiels de prise en main de l’agent .

Étapes suivantes

Agents en cours d’exécution

  • Last updated on