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.

Cet article explique comment créer un agent personnalisé simple qui répète l'entrée 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éer un agent personnalisé

Session de l’agent

Pour créer un agent personnalisé, vous avez également besoin d’une session, qui est utilisée pour suivre 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 session courants.

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

Pour cet exemple, vous allez utiliser la InMemoryAgentSession classe de base pour la session personnalisée.

internal sealed class CustomAgentSession : InMemoryAgentSession
{
    internal CustomAgentSession() : base() { }
    internal CustomAgentSession(JsonElement serializedSessionState, JsonSerializerOptions? jsonSerializerOptions = null)
        : base(serializedSessionState, jsonSerializerOptions) { }
}

Classe Agent

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

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Création de sessions

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

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

    public override Task<AgentSession> CreateSessionAsync(CancellationToken cancellationToken = default) 
        => Task.FromResult<AgentSession>(new CustomAgentSession());

    public override Task<AgentSession> DeserializeSessionAsync(JsonElement serializedSession, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
        => Task.FromResult<AgentSession>(new CustomAgentSession(serializedSession, 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.

Ajoutez la méthode suivante pour contenir cette logique. Les messages d’entrée sont clonés, 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 modifié en 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, vous devez implémenter les deux méthodes principales utilisées pour exécuter l’agent : une pour la non-diffusion en continu et une pour la diffusion en continu.

Pour les deux méthodes, vous devez vous assurer qu’une session est fournie et, si ce n’est pas le cas, créer une session. Les messages peuvent être récupérés et transmis dans la ChatHistoryProvider session. Si vous ne le faites pas, l’utilisateur ne pourra pas avoir de conversation à plusieurs tours avec l’agent, et chaque session sera une nouvelle interaction.

    public override async Task<AgentResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
    {
        session ??= await this.CreateSessionAsync(cancellationToken);

        // Get existing messages from the store
        var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
        var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);

        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();

        // Notify the session of the input and output messages.
        var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
        {
            ResponseMessages = responseMessages
        };
        await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);

        return new AgentResponse
        {
            AgentId = this.Id,
            ResponseId = Guid.NewGuid().ToString(),
            Messages = responseMessages
        };
    }

    public override async IAsyncEnumerable<AgentResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        session ??= await this.CreateSessionAsync(cancellationToken);

        // Get existing messages from the store
        var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
        var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);

        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();

        // Notify the session of the input and output messages.
        var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
        {
            ResponseMessages = responseMessages
        };
        await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);

        foreach (var message in responseMessages)
        {
            yield return new AgentResponseUpdate
            {
                AgentId = this.Id,
                AuthorName = this.DisplayName,
                Role = ChatRole.Assistant,
                Contents = message.Contents,
                ResponseId = Guid.NewGuid().ToString(),
                MessageId = Guid.NewGuid().ToString()
            };
        }
    }

Conseil / Astuce

Consultez les exemples .NET pour obtenir des exemples exécutables complets.

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éer un agent personnalisé

Le protocole agent

L’infrastructure fournit le SupportsAgentRun 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 typing import Any, Literal, overload
from collections.abc import Awaitable, Sequence
from agent_framework import (
    AgentResponse,
    AgentResponseUpdate,
    AgentSession,
    Message,
    ResponseStream,
    SupportsAgentRun,
)

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

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

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        """Execute the agent and return either an awaitable response or a ResponseStream."""
        ...

Conseil / Astuce

Ajoutez @overload des signatures à run() pour que les IDE et les vérificateurs de type statique déduisent le type de retour basé(e)s sur stream (Awaitable[AgentResponse] pour stream=False et ResponseStream[AgentResponseUpdate, AgentResponse] pour stream=True).

Utilisation de BaseAgent

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

import asyncio
from collections.abc import AsyncIterable, Awaitable, Sequence
from typing import Any, Literal, overload

from agent_framework import (
    AgentResponse,
    AgentResponseUpdate,
    AgentSession,
    BaseAgent,
    Content,
    Message,
    ResponseStream,
    normalize_messages,
)


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:
        super().__init__(
            name=name,
            description=description,
            echo_prefix=echo_prefix,
            **kwargs,
        )

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        """Execute the agent.

        Args:
            messages: The message(s) to process.
            stream: If True, return a ResponseStream of updates.
            session: The conversation session (optional).

        Returns:
            When stream=False: An awaitable AgentResponse.
            When stream=True: A ResponseStream with AgentResponseUpdate items and final response support.
        """
        if stream:
            return ResponseStream(
                self._run_stream(messages=messages, session=session, **kwargs),
                finalizer=AgentResponse.from_updates,
            )
        return self._run(messages=messages, session=session, **kwargs)

    async def _run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> AgentResponse:
        normalized_messages = normalize_messages(messages)

        if not normalized_messages:
            response_message = Message(
                role="assistant",
                contents=[Content.from_text("Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")],
            )
        else:
            last_message = normalized_messages[-1]
            echo_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"
            response_message = Message(role="assistant", contents=[Content.from_text(echo_text)])

        if session is not None:
            stored = session.state.setdefault("memory", {}).setdefault("messages", [])
            stored.extend(normalized_messages)
            stored.append(response_message)

        return AgentResponse(messages=[response_message])

    async def _run_stream(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentResponseUpdate]:
        normalized_messages = 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:
            last_message = normalized_messages[-1]
            response_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"

        words = response_text.split()
        for i, word in enumerate(words):
            chunk_text = f" {word}" if i > 0 else word
            yield AgentResponseUpdate(
                contents=[Content.from_text(chunk_text)],
                role="assistant",
            )
            await asyncio.sleep(0.1)

        if session is not None:
            complete_response = Message(role="assistant", contents=[Content.from_text(response_text)])
            stored = session.state.setdefault("memory", {}).setdefault("messages", [])
            stored.extend(normalized_messages)
            stored.append(complete_response)

Utilisation de l’agent

Si les méthodes de l’agent sont correctement implémentées, l’agent prend en charge les opérations standard, notamment la diffusion en continu via ResponseStream:

stream = echo_agent.run("Stream this response", stream=True, session=echo_agent.create_session())
async for update in stream:
    print(update.text or "", end="", flush=True)
final_response = await stream.get_final_response()

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

Prochaines étapes

Agents en cours d’exécution

  • Last updated on