Agenci dostosowani

Program Microsoft Agent Framework obsługuje tworzenie agentów niestandardowych przez dziedziczenie po AIAgent klasie i implementowanie wymaganych metod.

W tym artykule pokazano, jak utworzyć prostego agenta niestandardowego, który powtarza dane wejściowe użytkownika wielkimi literami. W większości przypadków tworzenie własnego agenta wymaga bardziej złożonej logiki i integracji z usługą sztucznej inteligencji.

Wprowadzenie

Dodaj wymagane pakiety NuGet do projektu.

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

Tworzenie agenta niestandardowego

Sesja agenta

Aby utworzyć agenta niestandardowego, potrzebujesz również sesji, która służy do śledzenia stanu pojedynczej konwersacji, w tym historii komunikatów i dowolnego innego stanu, który musi obsługiwać agent.

Aby ułatwić rozpoczęcie pracy, można dziedziczyć z różnych klas bazowych, które implementują typowe mechanizmy przechowywania sesji.

1. InMemoryAgentSession — przechowuje historię czatów w pamięci i może być serializowany do formatu JSON.
2. ServiceIdAgentSession — nie przechowuje żadnej historii czatów, ale umożliwia skojarzenie identyfikatora z sesją, w ramach której historia czatu może być przechowywana zewnętrznie.

W tym przykładzie użyjesz InMemoryAgentSession jako klasy podstawowej dla sesji niestandardowej.

internal sealed class CustomAgentSession : InMemoryAgentSession
{
    internal CustomAgentSession() : base() { }
    internal CustomAgentSession(JsonElement serializedSessionState, JsonSerializerOptions? jsonSerializerOptions = null)
        : base(serializedSessionState, jsonSerializerOptions) { }
}

Klasa agenta

Następnie utwórz samą klasę agenta, dziedzicząc po AIAgent klasie.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Tworzenie sesji

Sesje są zawsze tworzone za pomocą dwóch metod fabrycznych w klasie agenta. Dzięki temu agent może kontrolować sposób tworzenia i deserializacji sesji. W związku z tym agenci mogą dołączać wszelkie dodatkowe stany lub zachowania wymagane do sesji w momencie konstruowania.

Do zaimplementowania są dwie metody:

    public override Task<AgentSession> CreateSessionAsync(CancellationToken cancellationToken = default) 
        => Task.FromResult<AgentSession>(new CustomAgentSession());

    public override Task<AgentSession> DeserializeSessionAsync(JsonElement serializedSession, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
        => Task.FromResult<AgentSession>(new CustomAgentSession(serializedSession, jsonSerializerOptions));

Podstawowa logika agenta

** Podstawową logiką agenta jest przyjęcie dowolnych komunikatów wejściowych, przekształcenie ich tekstu na wielkie litery i zwrócenie ich jako komunikaty zwrotne.

Dodaj następującą metodę, aby zawierała tę logikę. Komunikaty wejściowe są klonowane, ponieważ różne aspekty komunikatów wejściowych muszą zostać zmodyfikowane w celu wprowadzania prawidłowych komunikatów odpowiedzi. Na przykład rola musi zostać zmieniona na 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;
        });

Metody działania agenta

Na koniec należy zaimplementować dwie podstawowe metody używane do uruchamiania agenta: jedną bez przesyłania strumieniowego i jedną z przesyłaniem strumieniowym.

W przypadku obu metod należy upewnić się, że sesja jest udostępniana, a jeśli nie, utwórz nową sesję. Komunikaty można pobrać i przekazać do ChatHistoryProvider tej sesji. Jeśli tego nie zrobisz, użytkownik nie będzie mógł przeprowadzić wielokrotnej rozmowy z agentem, a każda sesja będzie nową interakcją.

    public override async Task<AgentResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
    {
        session ??= await this.CreateSessionAsync(cancellationToken);

        // Get existing messages from the store
        var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
        var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);

        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();

        // Notify the session of the input and output messages.
        var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
        {
            ResponseMessages = responseMessages
        };
        await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);

        return new AgentResponse
        {
            AgentId = this.Id,
            ResponseId = Guid.NewGuid().ToString(),
            Messages = responseMessages
        };
    }

    public override async IAsyncEnumerable<AgentResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        session ??= await this.CreateSessionAsync(cancellationToken);

        // Get existing messages from the store
        var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
        var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);

        List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();

        // Notify the session of the input and output messages.
        var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
        {
            ResponseMessages = responseMessages
        };
        await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);

        foreach (var message in responseMessages)
        {
            yield return new AgentResponseUpdate
            {
                AgentId = this.Id,
                AuthorName = this.DisplayName,
                Role = ChatRole.Assistant,
                Contents = message.Contents,
                ResponseId = Guid.NewGuid().ToString(),
                MessageId = Guid.NewGuid().ToString()
            };
        }
    }

Tip

Zobacz przykłady dla platformy .NET , aby uzyskać pełne przykłady możliwych do uruchomienia.

Korzystanie z agenta

AIAgent Jeśli wszystkie metody są prawidłowo zaimplementowane, agent będzie standardowym AIAgent i obsługuje standardowe operacje agenta.

Aby uzyskać więcej informacji na temat uruchamiania agentów i interakcji z nimi, zobacz samouczki wprowadzające do agenta.

Program Microsoft Agent Framework obsługuje tworzenie agentów niestandardowych przez dziedziczenie po BaseAgent klasie i implementowanie wymaganych metod.

W tym dokumencie pokazano, jak utworzyć prostego agenta niestandardowego, który odtwarza dane wejściowe użytkownika, dodając prefiks. W większości przypadków tworzenie własnego agenta wymaga bardziej złożonej logiki i integracji z usługą sztucznej inteligencji.

Wprowadzenie

Dodaj wymagane pakiety języka Python do projektu.

pip install agent-framework-core --pre

Tworzenie agenta niestandardowego

Protokół agenta

Platforma udostępnia SupportsAgentRun protokół definiujący interfejs, który każdy agent musi implementować. Agenci niestandardowi mogą zaimplementować ten protokół bezpośrednio lub rozszerzyć klasę BaseAgent dla wygody.

from typing import Any, Literal, overload
from collections.abc import Awaitable, Sequence
from agent_framework import (
    AgentResponse,
    AgentResponseUpdate,
    AgentSession,
    Message,
    ResponseStream,
    SupportsAgentRun,
)

class MyCustomAgent(SupportsAgentRun):
    """A custom agent that implements the SupportsAgentRun directly."""

    @property
    def id(self) -> str:
        """Returns the ID of the agent."""
        ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        """Execute the agent and return either an awaitable response or a ResponseStream."""
        ...

Tip

Dodaj @overload podpisy do run(), aby IDEs i kontrolery typów statycznych wnioskowały typ zwracany na podstawie stream (Awaitable[AgentResponse] dla stream=False i ResponseStream[AgentResponseUpdate, AgentResponse] dla stream=True).

Korzystanie z agenta BaseAgent

Zalecaną metodą jest rozszerzenie BaseAgent klasy, która zapewnia typowe funkcje i upraszcza implementację:

import asyncio
from collections.abc import AsyncIterable, Awaitable, Sequence
from typing import Any, Literal, overload

from agent_framework import (
    AgentResponse,
    AgentResponseUpdate,
    AgentSession,
    BaseAgent,
    Content,
    Message,
    ResponseStream,
    normalize_messages,
)


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:
        super().__init__(
            name=name,
            description=description,
            echo_prefix=echo_prefix,
            **kwargs,
        )

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        """Execute the agent.

        Args:
            messages: The message(s) to process.
            stream: If True, return a ResponseStream of updates.
            session: The conversation session (optional).

        Returns:
            When stream=False: An awaitable AgentResponse.
            When stream=True: A ResponseStream with AgentResponseUpdate items and final response support.
        """
        if stream:
            return ResponseStream(
                self._run_stream(messages=messages, session=session, **kwargs),
                finalizer=AgentResponse.from_updates,
            )
        return self._run(messages=messages, session=session, **kwargs)

    async def _run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> AgentResponse:
        normalized_messages = normalize_messages(messages)

        if not normalized_messages:
            response_message = Message(
                role="assistant",
                contents=[Content.from_text("Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")],
            )
        else:
            last_message = normalized_messages[-1]
            echo_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"
            response_message = Message(role="assistant", contents=[Content.from_text(echo_text)])

        if session is not None:
            stored = session.state.setdefault("memory", {}).setdefault("messages", [])
            stored.extend(normalized_messages)
            stored.append(response_message)

        return AgentResponse(messages=[response_message])

    async def _run_stream(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentResponseUpdate]:
        normalized_messages = 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:
            last_message = normalized_messages[-1]
            response_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"

        words = response_text.split()
        for i, word in enumerate(words):
            chunk_text = f" {word}" if i > 0 else word
            yield AgentResponseUpdate(
                contents=[Content.from_text(chunk_text)],
                role="assistant",
            )
            await asyncio.sleep(0.1)

        if session is not None:
            complete_response = Message(role="assistant", contents=[Content.from_text(response_text)])
            stored = session.state.setdefault("memory", {}).setdefault("messages", [])
            stored.extend(normalized_messages)
            stored.append(complete_response)

Korzystanie z agenta

Jeśli wszystkie metody agenta są prawidłowo implementowane, agent obsługuje standardowe działania, w tym przesyłanie strumieniowe przez ResponseStream:

stream = echo_agent.run("Stream this response", stream=True, session=echo_agent.create_session())
async for update in stream:
    print(update.text or "", end="", flush=True)
final_response = await stream.get_final_response()

Aby uzyskać więcej informacji na temat uruchamiania agentów i interakcji z nimi, zobacz samouczki wprowadzające do agenta.

Dalsze kroki

Uruchamianie agentów