Aracılığıyla paylaş

Özel Ajanlar

Microsoft Agent Framework, sınıfından AIAgent devralarak ve gerekli yöntemleri uygulayarak özel aracılar oluşturulmasını destekler.

Bu makale, kullanıcı girişini büyük harfle tekrar eden basit bir özelleştirilmiş ajan nasıl oluşturulacağını gösterir. Çoğu durumda kendi aracınızı oluşturmak için yapay zeka hizmetiyle daha karmaşık mantık ve tümleştirme gerekir.

Başlangıç Yapmak

Projenize gerekli NuGet paketlerini ekleyin.

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

Özel Aracı Oluşturma

Aracı Oturumu

Özel bir aracı oluşturmak için, ileti geçmişi ve aracının koruması gereken diğer tüm durumları içeren tek bir konuşmanın durumunu izlemek için kullanılan bir oturuma da ihtiyacınız vardır.

Başlamayı kolaylaştırmak için, ortak oturum depolama mekanizmaları uygulayan çeşitli temel sınıflardan devralabilirsiniz.

  1. InMemoryAgentSession - sohbet geçmişini bellekte depolar ve JSON'a seri hale getirilebilir.
  2. ServiceIdAgentSession - herhangi bir sohbet geçmişi depolamaz, ancak bir kimliği sohbet geçmişinin harici olarak depolanabileceği oturumla ilişkilendirmenize olanak tanır.

Bu örnekte, özel oturum için temel sınıf olarak kullanacaksınız InMemoryAgentSession .

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

Agent sınıfı

Ardından, AIAgent sınıfından türeterek aracı sınıfının kendisini oluşturun.

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

Oturumların oluşturulması

Oturumlar her zaman aracı sınıfındaki iki fabrika yöntemiyle oluşturulur. Bu, aracının oturumların nasıl oluşturulduğunu ve seri durumdan çıkarıldığını denetlemesine olanak tanır. Bu nedenle, gerektiğinde aracılar oluşturulurken oturuma ek durumlar veya davranışlar ekleyebilir.

İki yöntemin uygulanması gerekir:

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

Temel ajan mantığı

Aracının temel mantığı, tüm giriş iletilerini almak, metinlerini büyük harfe dönüştürmek ve bunları yanıt iletileri olarak döndürmektir.

Bu mantığı içerecek şekilde aşağıdaki yöntemi ekleyin. Giriş iletilerinin çeşitli yönlerinin geçerli yanıt iletileri olması için değiştirilmesi gerektiğinden, giriş iletileri kopyalanır. Örneğin, rolün Assistant olarak değiştirilmesi gerekir.

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

Aracı çalıştırma yöntemleri

Son olarak, aracıyı çalıştırmak için kullanılan iki temel yöntemi uygulamanız gerekir: biri akış dışı ve biri akış için.

Her iki yöntem için de bir oturumun sağlandığından emin olmanız ve sağlanmadıysa yeni bir oturum oluşturmanız gerekir. İletiler alınabilir ve oturumda ChatHistoryProvider öğesine aktarılabilir. Bunu yapmazsanız, kullanıcı aracıyla çoklu tur konuşması yapamaz ve her oturum yeni bir etkileşim olacaktır.

    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

Çalıştırılabilir örneklerin tamamı için .NET örneklerine bakın.

Ajanı Kullanma

Yöntemlerin AIAgent tümü doğru uygulanırsa, aracı standart bir AIAgent olur ve standart aracı işlemlerini destekler.

Aracıları çalıştırma ve aracılarla etkileşim kurma hakkında daha fazla bilgi için bkz. Aracı kullanmaya başlama öğreticileri.

Microsoft Agent Framework, sınıfından BaseAgent devralarak ve gerekli yöntemleri uygulayarak özel aracılar oluşturulmasını destekler.

Bu belgede, ön ek ile kullanıcı girişini geri döndüren basit bir özel aracı oluşturma gösterilmektedir. Çoğu durumda kendi aracınızı oluşturmak için yapay zeka hizmetiyle daha karmaşık mantık ve tümleştirme gerekir.

Başlangıç Yapmak

Projenize gerekli Python paketlerini ekleyin.

pip install agent-framework-core --pre

Özel Aracı Oluşturma

Aracı Protokolü

Çerçeve, tüm aracıların SupportsAgentRun uygulaması gereken arabirimi tanımlayan protokolü sağlar. Özel aracılar bu protokolü doğrudan uygulayabilir veya kolaylık sağlamak için sınıfını BaseAgent genişletebilir.

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

IDE'lerin ve statik tür denetleyicilerinin, stream temel alınarak Awaitable[AgentResponse] için stream=False ve ResponseStream[AgentResponseUpdate, AgentResponse] için stream=True dönüş türünü çıkarması amacıyla @overload öğesine imzalar ekleyin.

BaseAgent kullanma

Önerilen yaklaşım, ortak işlevsellik sağlayan ve uygulamayı basitleştiren sınıfını genişletmektir BaseAgent :

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)

Ajanı Kullanma

Aracı yöntemlerinin tümü doğru uygulanırsa, aracı ResponseStream üzerinden akış da dahil olmak üzere standart işlemleri destekler.

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

Aracıları çalıştırma ve aracılarla etkileşim kurma hakkında daha fazla bilgi için bkz. Aracı kullanmaya başlama öğreticileri.

Sonraki Adımlar

Çalışan Ajanlar

  • Last updated on