Freigeben über

Benutzerdefinierte Agenten

Das Microsoft Agent Framework unterstützt das Erstellen von benutzerdefinierten Agents, indem sie von der AIAgent Klasse erben und die erforderlichen Methoden implementieren.

In diesem Dokument wird gezeigt, wie Sie einen einfachen benutzerdefinierten Agent erstellen, der die Benutzereingabe in Großbuchstaben zurückgibt. In den meisten Fällen umfasst das Erstellen Ihres eigenen Agents komplexere Logik und Integration in einen KI-Dienst.

Erste Schritte

Fügen Sie dem Projekt die erforderlichen NuGet-Pakete hinzu.

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

Erstellen eines benutzerdefinierten Agents

Der Agentthread

Um einen benutzerdefinierten Agent zu erstellen, benötigen Sie auch einen Thread, der zum Nachverfolgen des Status einer einzelnen Unterhaltung verwendet wird, einschließlich des Nachrichtenverlaufs und eines anderen Status, den der Agent verwalten muss.

Um den Einstieg zu erleichtern, können Sie von verschiedenen Basisklassen erben, die allgemeine Threadspeichermechanismen implementieren.

  1. InMemoryAgentThread – speichert den Chatverlauf im Arbeitsspeicher und kann in JSON serialisiert werden.
  2. ServiceIdAgentThread – speichert keinen Chatverlauf, ermöglicht ihnen jedoch, eine ID mit dem Thread zu verknüpfen, unter der der Chatverlauf extern gespeichert werden kann.

In diesem Beispiel verwenden wir die InMemoryAgentThread Basisklasse für unseren benutzerdefinierten Thread.

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

Die Agent-Klasse

Als Nächstes möchten wir die Agentklasse selbst erstellen, indem wir von der AIAgent Klasse erben.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Erstellen von Threads

Threads werden immer über zwei Factorymethoden für die Agentklasse erstellt. Dadurch kann der Agent steuern, wie Threads erstellt und deserialisiert werden. Agents können daher beim Erstellen alle zusätzlichen Zustände oder Verhaltensweisen an den Thread anfügen, die erforderlich sind.

Es sind zwei Methoden erforderlich, die implementiert werden müssen:

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

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

Kern-Agent-Logik

Die Kernlogik des Agents besteht darin, alle Eingabenachrichten zu übernehmen, ihren Text in Großbuchstaben zu konvertieren und als Antwortnachrichten zurückzugeben.

Wir möchten die folgende Methode hinzufügen, um diese Logik zu enthalten. Wir klonen die Eingabemeldungen, da verschiedene Aspekte der Eingabenachrichten geändert werden müssen, um gültige Antwortnachrichten zu sein. Die Rolle muss auf Assistant geändert werden.

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

Agent-Ausführungsmethoden

Schließlich müssen wir die beiden Kernmethoden implementieren, die zum Ausführen des Agents verwendet werden. Eine für Nicht-Streaming und eine für Streaming.

Für beide Methoden müssen wir sicherstellen, dass ein Thread bereitgestellt wird, und wenn nicht, wird ein neuer Thread erstellt. Der Thread kann dann mit den neuen Nachrichten aktualisiert werden, indem NotifyThreadOfNewMessagesAsync aufgerufen wird. Wenn dies nicht der Fall ist, kann der Benutzer keine Multi-Turn-Gespräch mit dem Agent führen, und jeder Vorgang ist eine neue Interaktion.

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

Den Agent verwenden

Wenn die AIAgent-Methoden alle ordnungsgemäß implementiert sind, wäre der Agent ein Standardagent und würde Standardbetriebsfunktionen eines Agents unterstützen.

Weitere Informationen zum Ausführen und Interagieren mit Agenten finden Sie in den Einführungstutorials für Agenten.

Das Microsoft Agent Framework unterstützt das Erstellen von benutzerdefinierten Agents, indem sie von der BaseAgent Klasse erben und die erforderlichen Methoden implementieren.

Dieses Dokument zeigt, wie Sie einen einfachen benutzerdefinierten Agent erstellen, der die Benutzereingabe mit einem Präfix wiedergibt. In den meisten Fällen umfasst das Erstellen Ihres eigenen Agents komplexere Logik und Integration in einen KI-Dienst.

Erste Schritte

Fügen Sie ihrem Projekt die erforderlichen Python-Pakete hinzu.

pip install agent-framework-core --pre

Erstellen eines benutzerdefinierten Agents

Das Agent-Protokoll

Das Framework stellt das AgentProtocol Protokoll bereit, das die Schnittstelle definiert, die alle Agents implementieren müssen. Benutzerdefinierte Agents können dieses Protokoll entweder direkt implementieren oder die BaseAgent Klasse zur Vereinfachung erweitern.

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

Verwenden von BaseAgent

Der empfohlene Ansatz besteht darin, die BaseAgent Klasse zu erweitern, die allgemeine Funktionen bereitstellt und die Implementierung vereinfacht:

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)

Den Agent verwenden

Wenn agent-Methoden alle ordnungsgemäß implementiert sind, würde der Agent alle Standard-Agent-Vorgänge unterstützen.

Weitere Informationen zum Ausführen und Interagieren mit Agenten finden Sie in den Einführungstutorials für Agenten.

Nächste Schritte

Ausführen von Agenten

  • Last updated on