Partilhar via

Agentes Personalizados

O Microsoft Agent Framework oferece 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 utilizador em maiúsculas. Na maioria dos casos, construir seu próprio agente envolverá lógica mais complexa e integração 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

Fio 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 os primeiros passos, você pode herdar de várias classes base que implementam mecanismos de armazenamento de threads comuns.

  1. InMemoryAgentThread - armazena o histórico de bate-papo na memória e pode ser serializado para JSON.
  2. ServiceIdAgentThread - Não armazena nenhum histórico de bate-papo, mas permite que você associe um ID ao tópico, sob o qual o histórico de bate-papo pode ser armazenado externamente.

Para este exemplo, usaremos o InMemoryAgentThread como a classe 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

Os threads são sempre criados por meio de dois métodos de fábrica na classe do 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ído.

São necessários dois métodos 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 central do agente é pegar todas as mensagens de entrada, converter seu texto em maiúsculas e devolvê-las como mensagens de resposta.

Queremos adicionar o seguinte método para conter essa lógica. Estamos clonando as mensagens de entrada, uma vez que vários aspetos das mensagens de entrada têm que ser modificados para serem mensagens de resposta válidas. Por exemplo, a função tem de 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 operação do agente

Finalmente, precisamos implementar os dois métodos principais que são usados para executar o agente. Um para atividades que não envolvem streaming e outro para funções de streaming.

Para ambos os métodos, precisamos garantir que um thread seja fornecido e, se não, criamos um novo thread. O thread pode então ser atualizado com as novas mensagens chamando NotifyThreadOfNewMessagesAsync. Se não fizermos isso, o usuário não poderá ter uma conversa de vários turnos com o agente e cada execução será uma nova interação.

    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 métodos AIAgent forem implementados corretamente, o agente será um agente padrão AIAgent e suportará operações padrão do agente.

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

O Microsoft Agent Framework oferece 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 a entrada do usuário com um prefixo. Na maioria dos casos, construir seu próprio agente envolverá lógica mais complexa e integração com um serviço de IA.

Introdução

Adicione os pacotes 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 por 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 o 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 os métodos do agente forem todos implementados corretamente, o agente suportará todas as operações padrão do agente.

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

Próximos passos

Agentes em execução

  • Last updated on