Delen via

Gepersonaliseerde agents

Microsoft Agent Framework ondersteunt het bouwen van aangepaste agents door te erven van de AIAgent klasse en door de vereiste methoden te implementeren.

In dit artikel wordt beschreven hoe u een eenvoudige aangepaste agent bouwt die gebruikersinvoer in hoofdletters weergeeft. In de meeste gevallen omvat het bouwen van uw eigen agent complexere logica en integratie met een AI-service.

Aan de slag komen

Voeg de vereiste NuGet-pakketten toe aan uw project.

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

Een aangepaste agent maken

De agentthread

Als u een aangepaste agent wilt maken, hebt u ook een thread nodig, die wordt gebruikt om de status van één gesprek bij te houden, inclusief berichtgeschiedenis en elke andere status die de agent moet onderhouden.

Om het eenvoudig te maken om aan de slag te gaan, kunt u overnemen van verschillende basisklassen die algemene threadopslagmechanismen implementeren.

  1. InMemoryAgentThread - slaat de chatgeschiedenis op in het geheugen en kan worden geserialiseerd naar JSON.
  2. ServiceIdAgentThread - slaat geen chatgeschiedenis op, maar hiermee kunt u een id koppelen aan de thread, waaronder de chatgeschiedenis extern kan worden opgeslagen.

In dit voorbeeld gebruikt u de InMemoryAgentThread basisklasse voor de aangepaste thread.

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

De agentklasse

Maak vervolgens de agentklasse zelf door deze over te nemen van de AIAgent klasse.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Threads samenstellen

Threads worden altijd gemaakt via twee factory-methoden in de agentklasse. Hierdoor kan de agent bepalen hoe threads worden gemaakt en gedeserialiseerd. Agenten kunnen daarom alle aanvullende staten of gedragingen die nodig zijn aan de thread koppelen tijdens de constructie ervan.

Er moeten twee methoden worden geïmplementeerd:

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

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

Kernagentlogica

De kernlogica van de agent is het nemen van invoerberichten, het converteren van de tekst naar hoofdletters en het retourneren als antwoordberichten.

Voeg de volgende methode toe om deze logica te bevatten. De invoerberichten worden gekloond, omdat verschillende aspecten van de invoerberichten moeten worden gewijzigd om geldige antwoordberichten te zijn. De rol moet bijvoorbeeld worden gewijzigd in 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;
        });

Methoden voor agentuitvoering

Ten slotte moet u de twee kernmethoden implementeren die worden gebruikt om de agent uit te voeren: één voor niet-streaming en één voor streaming.

Voor beide methoden moet u ervoor zorgen dat er een thread wordt opgegeven en zo niet, dan moet u een nieuwe thread creëren. De thread kan vervolgens worden bijgewerkt met de nieuwe berichten door aan te roepen NotifyThreadOfNewMessagesAsync. Als u dit niet doet, kan de gebruiker geen meerfasen gesprek met de agent voeren en zal elke sessie een nieuwe interactie zijn.

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

De agent gebruiken

Als de AIAgent methoden allemaal correct zijn geïmplementeerd, zou de agent een standaardagent AIAgent zijn en standaardagentbewerkingen ondersteunen.

Voor meer informatie over hoe je agents uitvoert en ermee werkt, raadpleeg de Aan de slag met agent-tutorials.

Microsoft Agent Framework ondersteunt het bouwen van aangepaste agents door middel van overerving van de BaseAgent klasse en het implementeren van de vereiste methoden.

Dit document laat zien hoe u een eenvoudige aangepaste agent bouwt die gebruikersinvoer weergeeft met een voorvoegsel. In de meeste gevallen omvat het bouwen van uw eigen agent complexere logica en integratie met een AI-service.

Aan de slag komen

Voeg de vereiste Python-pakketten toe aan uw project.

pip install agent-framework-core --pre

Een aangepaste agent maken

Het agentprotocol

Het framework biedt het AgentProtocol protocol waarmee de interface wordt gedefinieerd die alle agents moeten implementeren. Aangepaste agents kunnen dit protocol rechtstreeks implementeren of de BaseAgent klasse uitbreiden voor het gemak.

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

BaseAgent gebruiken

De aanbevolen aanpak is het uitbreiden van de BaseAgent klasse, die algemene functionaliteit biedt en de implementatie vereenvoudigt:

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)

De agent gebruiken

Als agentmethoden correct zijn geïmplementeerd, ondersteunt de agent alle standaardagentbewerkingen.

Voor meer informatie over hoe je agents uitvoert en ermee werkt, raadpleeg de Aan de slag met agent-tutorials.

Volgende stappen

Uitvoerende agents

  • Last updated on