عوامل مخصصة

يدعم Microsoft Agent Framework إنشاء عوامل مخصصة عن طريق الوراثة AIAgent من الفئة وتنفيذ الأساليب المطلوبة.

توضح هذه المقالة كيفية إنشاء عامل مخصص بسيط يقوم بنسخ إدخال المستخدم مرة أخرى في الأحرف الكبيرة. في معظم الحالات، سيتضمن بناء وكيلك منطقا أكثر تعقيدا والتكامل مع خدمة الذكاء الاصطناعي.

الشروع في العمل

أضف حزم NuGet المطلوبة إلى مشروعك.

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

إنشاء عامل مخصص

جلسة عمل العامل

لإنشاء عامل مخصص، تحتاج أيضا إلى جلسة عمل، والتي يتم استخدامها لتتبع حالة محادثة واحدة، بما في ذلك محفوظات الرسائل، وأي حالة أخرى يحتاج العامل إلى الاحتفاظ بها.

لتسهيل البدء، يمكنك الوراثة من الفئات الأساسية المختلفة التي تنفذ آليات تخزين الجلسة الشائعة.

1. InMemoryAgentSession - يخزن محفوظات الدردشة في الذاكرة ويمكن تسلسلها إلى JSON.
2. ServiceIdAgentSession - لا يخزن أي محفوظات دردشة، ولكنه يسمح لك بربط معرف بجلسة العمل، والتي يمكن بموجبها تخزين محفوظات الدردشة خارجيا.

على سبيل المثال، ستستخدم InMemoryAgentSession كفئة أساسية لجلسة العمل المخصصة.

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

فئة العامل

بعد ذلك، قم بإنشاء فئة العامل نفسها عن طريق الوراثة AIAgent من الفئة .

internal sealed class UpperCaseParrotAgent : AIAgent
{
}

إنشاء جلسات العمل

يتم إنشاء جلسات العمل دائما عبر طريقتين للمصنع في فئة العامل. يسمح هذا للعامل بالتحكم في كيفية إنشاء جلسات العمل وإلغاء تسلسلها. لذلك يمكن للوكلاء إرفاق أي حالة أو سلوكيات إضافية مطلوبة لجلسة العمل عند إنشائها.

هناك طريقتان مطلوبتان لتنفيذهما:

    protected override ValueTask<AgentSession> CreateSessionCoreAsync(CancellationToken cancellationToken = default) 
        => new(new CustomAgentSession());

    protected override ValueTask<AgentSession> DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
        => new(new CustomAgentSession(serializedState, jsonSerializerOptions));

منطق العامل الأساسي

المنطق الأساسي للعامل هو أخذ أي رسائل إدخال وتحويل نصها إلى أحرف كبيرة وإعادتها كرسائل استجابة.

أضف الأسلوب التالي لاحتواء هذا المنطق. يتم نسخ رسائل الإدخال، حيث يجب تعديل جوانب مختلفة من رسائل الإدخال لتكون رسائل استجابة صالحة. على سبيل المثال، يجب تغيير الدور إلى 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;
        });

أساليب تشغيل العامل

وأخيرا، تحتاج إلى تنفيذ طريقتين أساسيتين تستخدمان لتشغيل العامل: واحدة لعدم الدفق والأخرى للبث.

بالنسبة لكلتا الطريقتين، تحتاج إلى التأكد من توفير جلسة عمل، وإذا لم يكن الأمر كما هو الحال، قم بإنشاء جلسة عمل جديدة. يمكن استرداد الرسائل وتم تمريرها إلى في ChatHistoryProvider جلسة العمل. إذا لم تقم بذلك، فلن يتمكن المستخدم من إجراء محادثة متعددة الأدوار مع العامل وسيكون كل تشغيل تفاعلا جديدا.

    protected override async Task<AgentResponse> RunCoreAsync(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
        };
    }

    protected override async IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(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

راجع نماذج .NET للحصول على أمثلة كاملة قابلة للتشغيل.

ادوات

يحتوي العرف AIAgent على أي سطح أداة تقرر منحه. إذا قمت بتضمين ملف موجود IChatClient وتمريرهtools، فإنك ترث دعم أداة هذا العميل — راجع، على سبيل المثال، OpenAI أو Azure OpenAI أو Microsoft صفحات موفر Foundry لما يدعمه العملاء الأساسيون. إذا لم يتصل الوكيل المخصص بعميل الدردشة (على سبيل المثال، عامل echo أعلاه)، فلا توجد أدوات لاستدعاء.

استخدام العامل

إذا تم تنفيذ جميع الأساليب AIAgent بشكل صحيح، فسيكون العامل قياسيا AIAgent ويدعم عمليات الوكيل القياسية.

لمزيد من المعلومات حول كيفية التشغيل والتفاعل مع الوكلاء، راجع البرامج التعليمية لبدء تشغيل العامل.

يدعم Microsoft Agent Framework إنشاء عوامل مخصصة عن طريق الوراثة BaseAgent من الفئة وتنفيذ الأساليب المطلوبة.

يوضح هذا المستند كيفية إنشاء عامل مخصص بسيط يردد إدخال المستخدم مرة أخرى ببادئة. في معظم الحالات، سيتضمن بناء وكيلك منطقا أكثر تعقيدا والتكامل مع خدمة الذكاء الاصطناعي.

الشروع في العمل

أضف حزم Python المطلوبة إلى مشروعك.

pip install agent-framework-core

إنشاء عامل مخصص

بروتوكول العامل

يوفر SupportsAgentRun إطار العمل البروتوكول الذي يحدد الواجهة التي يجب على جميع العوامل تنفيذها. يمكن للوكلاء المخصصين إما تنفيذ هذا البروتوكول مباشرة أو توسيع 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."""
        ...

Tip

أضف @overload التواقيع إلى run() حتى تستنتج IDEs ومدققات النوع الثابت نوع الإرجاع استنادا stream إلى (Awaitable[AgentResponse] ل stream=False وResponseStream[AgentResponseUpdate, AgentResponse]).stream=True

استخدام BaseAgent

النهج الموصى به هو توسيع 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)

ادوات

يحتوي العرف BaseAgent على أي سطح أداة تقرر منحه. إذا قمت بتضمين عميل دردشة موجود وتمريرهtools، فإنك ترث دعم أدوات هذا العميل — راجع، على سبيل المثال، OpenAI أو Microsoft Foundry أو صفحات موفر Anthropic لما يدعمه العملاء الأساسيون. إذا لم يتصل الوكيل المخصص بعميل الدردشة (على سبيل المثال، عامل echo أعلاه)، فلا توجد أدوات لاستدعاء.

استخدام العامل

إذا تم تنفيذ جميع أساليب العامل بشكل صحيح، فإن العامل يدعم العمليات القياسية، بما في ذلك الدفق عبر 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()

لمزيد من المعلومات حول كيفية التشغيل والتفاعل مع الوكلاء، راجع البرامج التعليمية لبدء تشغيل العامل.

الموفرون المخصصون

يمكنك إنشاء موفر مخصص عن طريق تنفيذه agent.ProviderConfig وتمريره إلى agent.New:

import (
    "context"
    "iter"

    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/message"
)

a := agent.New(agent.ProviderConfig{
    ProviderName: "my-custom-provider",
    Run: func(ctx context.Context, messages []*message.Message,
        options ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
        // Your custom LLM logic here
        return func(yield func(*agent.ResponseUpdate, error) bool) {
            yield(&agent.ResponseUpdate{
                Role: message.RoleAssistant,
                Contents: []message.Content{
                    &message.TextContent{Text: "Hello from custom provider!"},
                },
            }, nil)
        }
    },
}, agent.Config{
    Name: "CustomAgent",
})

حقول ProviderConfig

الحقل	الغرض
CreateSession	إنشاء جلسة عمل جديدة للموفر
Run	تنفيذ طلب وتيار تحديثات الاستجابة
Middlewares	إضافة البرامج الوسيطة ذات نطاق الموفر التي تعمل بعد موفري المحفوظات والسياق
Format	إنشاء واصف تنسيق استجابة للإخراج المنظم
Unmarshal	فك ترميز الإخراج المنظم إلى نوع هدف

الخطوات التالية

عوامل التشغيل