Agentes Personalizados

O Microsoft Agent Framework dá suporte à criação de agentes personalizados herdando da AIAgent classe e implementando os métodos necessários.

Este artigo mostra como criar um agente personalizado simples que repete a entrada do usuário em maiúsculas. Na maioria dos casos, a criação de seu próprio agente envolverá uma lógica e integração mais complexas com um serviço de IA.

Introdução

Adicione os pacotes NuGet necessários ao seu projeto.

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

Criar um agente personalizado

A Sessão do Agente

Para criar um agente personalizado, você também precisa de uma sessão, que é usada para acompanhar o estado de uma única conversa, incluindo o histórico de mensagens e qualquer outro estado que o agente precise manter.

Para facilitar a introdução, você pode herdar de várias classes base que implementam mecanismos comuns de armazenamento de sessão.

1. InMemoryAgentSession – armazena o histórico de chat na memória e pode ser serializado em JSON.
2. ServiceIdAgentSession - não armazena nenhum histórico de chat, mas permite associar uma ID à sessão, na qual o histórico de chat pode ser armazenado externamente.

Para este exemplo, você usará a InMemoryAgentSession classe como base para a sessão personalizada.

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

A classe Agent

Em seguida, crie a própria classe de agente herdando da AIAgent classe.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Criação de sessões

As sessões são sempre criadas por meio de dois métodos de fábrica na classe de agente. Isso permite que o agente controle como as sessões são criadas e desserializadas. Os agentes podem, portanto, anexar qualquer estado ou comportamento adicional necessários à sessão quando construídos.

Dois métodos são necessários para serem implementados:

    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));

Lógica do agente principal

A lógica principal do agente é pegar qualquer mensagem de entrada, converter seu texto em maiúsculas e devolvê-las como mensagens de resposta.

Adicione o método a seguir para conter essa lógica. As mensagens de entrada são clonadas, pois vários aspectos das mensagens de entrada precisam ser modificados para serem mensagens de resposta válidas. Por exemplo, a função deve ser alterada para 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étodos de execução do agente

Por fim, você precisa implementar os dois métodos principais que são usados para executar o agente: um para não streaming e outro para streaming.

Para ambos os métodos, você precisa garantir que uma sessão seja fornecida e, caso contrário, crie uma nova sessão. As mensagens podem ser recuperadas e enviadas para a ChatHistoryProvider na sessão. Se você não fizer isso, o usuário não poderá ter uma conversa em múltiplos turnos com o agente, e cada execução será uma interação nova.

    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()
            };
        }
    }

Dica

Consulte os exemplos do .NET para obter exemplos executáveis completos.

Usando o agente

Se todos os AIAgent métodos forem implementados corretamente, o agente será um padrão AIAgent e oferecerá suporte a operações de agente padrão.

Para obter mais informações sobre como executar e interagir com agentes, consulte os tutoriais de introdução do Agente.

O Microsoft Agent Framework dá suporte à criação de agentes personalizados herdando da BaseAgent classe e implementando os métodos necessários.

Este documento mostra como criar um agente personalizado simples que ecoa de volta a entrada do usuário com um prefixo. Na maioria dos casos, a criação de seu próprio agente envolverá uma lógica e integração mais complexas com um serviço de IA.

Introdução

Adicione os pacotes do Python necessários ao seu projeto.

pip install agent-framework-core --pre

Criar um agente personalizado

O Protocolo do Agente

A estrutura fornece o SupportsAgentRun protocolo que define a interface que todos os agentes devem implementar. Os agentes personalizados podem implementar esse protocolo diretamente ou estender a BaseAgent classe para conveniência.

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."""
        ...

Dica

Adicione @overload assinaturas a run() de modo que IDEs e verificadores de tipo estático infiram o tipo de retorno com base em stream (Awaitable[AgentResponse] para stream=False e ResponseStream[AgentResponseUpdate, AgentResponse] para stream=True).

Usando BaseAgent

A abordagem recomendada é estender a BaseAgent classe, que fornece funcionalidade comum e simplifica a implementação:

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)

Usando o agente

Se todos os métodos de agente forem implementados corretamente, o agente oferecerá suporte a operações padrão, incluindo streaming por meio de 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()

Para obter mais informações sobre como executar e interagir com agentes, consulte os tutoriais de introdução do Agente.

Próximas etapas

Agentes em execução