Compartilhar via

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 documento 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

Criando um agente personalizado

O thread do agente

Para criar um agente personalizado, você também precisa de um thread, que é usado 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 threads.

  1. InMemoryAgentThread – armazena o histórico de chat na memória e pode ser serializado em JSON.
  2. ServiceIdAgentThread - não armazena nenhum histórico de chat, mas permite associar uma ID ao thread, no qual o histórico de chat pode ser armazenado externamente.

Para este exemplo, usaremos a InMemoryAgentThread classe como base para nosso thread personalizado.

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

A classe Agent

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

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Construindo threads de execução

Os threads são sempre criados por meio de dois métodos de fábrica na classe de agente. Isso permite que o agente controle como os threads são criados e desserializados. Os agentes podem, portanto, anexar qualquer estado ou comportamento adicional necessário ao thread quando construídos.

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

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

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

Lógica do agente principal

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

Queremos adicionar o método a seguir para conter essa lógica. Estamos clonando as mensagens de entrada, 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, precisamos implementar os dois métodos principais usados para executar o agente. Um para conteúdo não transmitido via streaming e outro para streaming.

Para ambos os métodos, precisamos garantir que um thread seja fornecido e, caso contrário, criaremos um novo thread. Em seguida, o thread pode ser atualizado com as novas mensagens chamando NotifyThreadOfNewMessagesAsync. Se não fizermos isso, o usuário não poderá ter um diálogo contínuo com o agente, e cada interação será uma nova experiência.

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

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.

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

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

Criando um agente personalizado

O Protocolo do Agente

A estrutura fornece o AgentProtocol 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 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."""
        ...

Usando BaseAgent

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

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)

Usando o agente

Se todos os métodos de agente forem implementados corretamente, o agente oferecerá suporte a todas as operações de agente padrão.

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

Próximas etapas

Agentes em execução

  • Last updated on