Agentes personalizados

Microsoft Agent Framework admite la compilación de agentes personalizados heredando de la AIAgent clase e implementando los métodos necesarios.

En este documento se muestra cómo crear un agente personalizado simple que devuelve la entrada del usuario en mayúsculas. En la mayoría de los casos, la creación de su propio agente implicará una lógica y una integración más complejas con un servicio de inteligencia artificial.

Introducción

Agregue los paquetes NuGet necesarios al proyecto.

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

Creación de un agente personalizado

Subproceso del agente

Para crear un agente personalizado, también necesita un subproceso, que se usa para realizar un seguimiento del estado de una sola conversación, incluido el historial de mensajes, y cualquier otro estado que el agente necesite mantener.

Para facilitar la introducción, puede heredar de varias clases base que implementan mecanismos comunes de almacenamiento de hilos.

1. InMemoryAgentThread : almacena el historial de chats en la memoria y se puede serializar en JSON.
2. ServiceIdAgentThread : no almacena ningún historial de chat, pero le permite asociar un identificador al subproceso, en el que el historial de chat se puede almacenar externamente.

En este ejemplo, usaremos InMemoryAgentThread como clase base para nuestro subproceso personalizado.

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

La clase Agent

A continuación, queremos crear la propia clase del agente heredando de la AIAgent clase .

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Construcción de hilos

Los hilos siempre se crean a través de dos métodos de creación en la clase del agente. Esto permite al agente controlar cómo se crean y deserializan los subprocesos. Por lo tanto, los agentes pueden adjuntar cualquier estado o comportamiento adicional necesario al subproceso cuando se construye.

Es necesario implementar dos métodos:

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

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

Lógica del agente principal

La lógica principal del agente es tomar los mensajes de entrada, convertir su texto en mayúsculas y devolverlos como mensajes de respuesta.

Queremos agregar el siguiente método para contener esta lógica. Estamos clonando los mensajes de entrada, ya que es necesario modificar varios aspectos de los mensajes de entrada para que sean mensajes de respuesta válidos. Por ejemplo, el rol debe cambiarse a 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 ejecución del agente

Por último, es necesario implementar los dos métodos principales que se usan para ejecutar el agente. Uno para no streaming y otro para streaming.

Para ambos métodos, necesitamos asegurarnos de que se proporcione un hilo y, si no, creamos un nuevo hilo. A continuación, el hilo se puede actualizar con los nuevos mensajes llamando a NotifyThreadOfNewMessagesAsync. Si no lo hacemos, el usuario no podrá tener una conversación multiturno con el agente y cada ejecución será una interacción nueva.

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

Uso del agente

Si todos los AIAgent métodos se implementan correctamente, el agente sería un estándar AIAgent y admitiría operaciones de agente estándar.

Consulte los tutoriales de introducción del agente para obtener más información sobre cómo ejecutar e interactuar con agentes.

Microsoft Agent Framework admite la compilación de agentes personalizados heredando de la BaseAgent clase e implementando los métodos necesarios.

En este documento se muestra cómo crear un agente personalizado sencillo que devuelve la entrada del usuario con un prefijo. En la mayoría de los casos, la creación de su propio agente implicará una lógica y una integración más complejas con un servicio de inteligencia artificial.

Introducción

Agregue los paquetes de Python necesarios al proyecto.

pip install agent-framework-core --pre

Creación de un agente personalizado

El protocolo del agente

El marco proporciona el AgentProtocol protocolo que define la interfaz que todos los agentes deben implementar. Los agentes personalizados pueden implementar este protocolo directamente o ampliar la BaseAgent clase para mayor comodidad.

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

Uso de BaseAgent

El enfoque recomendado es ampliar la BaseAgent clase , que proporciona funcionalidad común y simplifica la implementación:

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)

Uso del agente

Si todos los métodos del agente se implementan correctamente, el agente admitiría todas las operaciones del agente estándar.

Consulte los tutoriales de introducción del agente para obtener más información sobre cómo ejecutar e interactuar con agentes.