Upravitelní agenti

Microsoft Agent Framework podporuje vytváření vlastních agentů tím, že dědí z AIAgent třídy a implementuje požadované metody.

Tento článek ukazuje, jak vytvořit jednoduchého vlastního agenta, který zopakuje vstup uživatele velkými písmeny. Ve většině případů bude sestavování vlastního agenta zahrnovat složitější logiku a integraci se službou AI.

Začínáme

Přidejte do projektu požadované balíčky NuGet.

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

Vytvoření vlastního agenta

Sezení agenta

Pokud chcete vytvořit vlastního agenta, potřebujete také relaci, která slouží ke sledování stavu jedné konverzace, včetně historie zpráv a jakéhokoli jiného stavu, který musí agent udržovat.

Abyste mohli snadno začít, můžete dědit z různých základních tříd, které implementují běžné mechanismy úložiště relací.

1. InMemoryAgentSession – uloží historii chatu do paměti a dá se serializovat do formátu JSON.
2. ServiceIdAgentSession - neukládá žádnou historii chatu, ale umožňuje přidružit ID k relaci, ve které se dá historie chatu uložit externě.

V tomto příkladu použijete InMemoryAgentSession jako základní třídu pro přizpůsobenou relaci.

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

Třída Agent

Dále vytvořte samotnou třídu agenta tím, že zdědíte třídu AIAgent.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Vytváření relací

Sezení jsou vždy vytvářena dvěma továrními metodami v třídě agenta. To umožňuje agentu řídit, jak se relace vytvářejí a deserializují. Agenti mohou proto při vytváření relace připojit jakýkoli další stav nebo chování, které je potřeba.

Je potřeba implementovat dvě 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));

Logika centrálního agenta

Základní logikou agenta je vzít všechny vstupní zprávy, převést jejich text na velká a velká písmena a vrátit je jako zprávy odpovědi.

Přidejte následující metodu, která bude obsahovat tuto logiku. Vstupní zprávy se naklonují, protože různé aspekty vstupních zpráv musí být změněny tak, aby byly platné zprávy odpovědí. Například role musí být změněna 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 spuštění agenta

Nakonec je potřeba implementovat dvě základní metody, které se používají ke spuštění agenta: jednu pro jiné než streamování a jednu pro streamování.

U obou metod musíte zajistit, aby byla relace poskytnuta, a pokud ne, vytvořte novou relaci. Zprávy lze načíst a předat do relace s ChatHistoryProvider. Pokud to neuděláte, uživatel nebude moct mít s agentem vícenásobnou konverzaci a každé spuštění bude mít novou interakci.

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

Návod

Najdete kompletní spustitelné příklady v ukázkách .NET.

Použití agenta

AIAgent Pokud jsou všechny metody implementované správně, agent by byl standardní AIAgent a podporuje standardní operace agenta.

Další informace o tom, jak spouštět agenty a pracovat s nimi, najdete v úvodních kurzech agenta.

Microsoft Agent Framework podporuje vytváření vlastních agentů tím, že dědí z BaseAgent třídy a implementuje požadované metody.

Tento dokument ukazuje, jak vytvořit zjednodušeného vlastního agenta, který vrací vstup uživatele s danou předponou. Ve většině případů bude sestavování vlastního agenta zahrnovat složitější logiku a integraci se službou AI.

Začínáme

Přidejte do projektu požadované balíčky Pythonu.

pip install agent-framework-core --pre

Vytvoření vlastního agenta

Protokol Agenta

Architektura poskytuje SupportsAgentRun protokol, který definuje rozhraní, které musí implementovat všichni agenti. Vlastní agenti mohou buď implementovat tento protokol přímo, nebo pro snadnější práci rozšířit třídu BaseAgent.

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

Návod

Přidejte @overload podpisy do run(), aby IDE a statické kontrolory typů mohly odvozovat návratový typ na základě stream (Awaitable[AgentResponse] pro stream=False a ResponseStream[AgentResponseUpdate, AgentResponse] pro stream=True).

Použití baseagentu

Doporučeným přístupem je rozšířit BaseAgent třídu, která poskytuje běžné funkce a zjednodušuje implementaci:

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)

Použití agenta

Pokud jsou všechny metody agenta implementované správně, agent podporuje standardní operace, včetně streamování prostřednictvím 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()

Další informace o tom, jak spouštět agenty a pracovat s nimi, najdete v úvodních kurzech agenta.

Další kroky

Spouštění agentů