مهام سير عمل إطار عمل العامل Microsoft - Human-in-the-loop (HITL)

توفر هذه الصفحة نظرة عامة على تفاعلات Human-in-the-loop (HITL) في نظام سير عمل إطار عمل عامل Microsoft. يتم تحقيق HITL من خلال آلية معالجة الطلب والاستجابة في مهام سير العمل، والتي تسمح للمنفذين بإرسال الطلبات إلى الأنظمة الخارجية (مثل المشغلين البشريين) وانتظار استجاباتهم قبل المتابعة في تنفيذ سير العمل.

نظرة عامة

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

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

تمكين معالجة الطلب والاستجابة في سير العمل

تتم معالجة الطلبات والاستجابات عبر نوع خاص يسمى RequestPort.

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

// Create a request port that receives requests of type NumberSignal and responses of type int.
var numberRequestPort = RequestPort.Create<NumberSignal, int>("GuessNumber");

إضافة منفذ الإدخال إلى سير عمل.

JudgeExecutor judgeExecutor = new(42);
var workflow = new WorkflowBuilder(numberRequestPort)
    .AddEdge(numberRequestPort, judgeExecutor)
    .AddEdge(judgeExecutor, numberRequestPort)
    .WithOutputFrom(judgeExecutor)
    .Build();

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

internal enum NumberSignal
{
    Init,
    Above,
    Below,
}

internal sealed class JudgeExecutor() : Executor<int>("Judge")
{
    private readonly int _targetNumber;
    private int _tries;

    public JudgeExecutor(int targetNumber) : this()
    {
        this._targetNumber = targetNumber;
    }

    public override async ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        this._tries++;
        if (message == this._targetNumber)
        {
            await context.YieldOutputAsync($"{this._targetNumber} found in {this._tries} tries!", cancellationToken);
        }
        else if (message < this._targetNumber)
        {
            await context.SendMessageAsync(NumberSignal.Below, cancellationToken: cancellationToken);
        }
        else
        {
            await context.SendMessageAsync(NumberSignal.Above, cancellationToken: cancellationToken);
        }
    }
}

في Python، يرسل المنفذون طلبات باستخدام ctx.request_info() الاستجابات والتعامل معها باستخدام @response_handler مصمم الديكور.

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

تمكين معالجة الطلب والاستجابة في سير العمل

from dataclasses import dataclass

from agent_framework import (
    Executor,
    WorkflowBuilder,
    WorkflowContext,
    handler,
    response_handler,
)


@dataclass
class NumberSignal:
    hint: str  # "init", "above", or "below"


class JudgeExecutor(Executor):
    def __init__(self, target_number: int):
        super().__init__(id="judge")
        self._target_number = target_number
        self._tries = 0

    @handler
    async def handle_guess(self, guess: int, ctx: WorkflowContext[int, str]) -> None:
        self._tries += 1
        if guess == self._target_number:
            await ctx.yield_output(f"{self._target_number} found in {self._tries} tries!")
        elif guess < self._target_number:
            await ctx.request_info(request_data=NumberSignal(hint="below"), response_type=int)
        else:
            await ctx.request_info(request_data=NumberSignal(hint="above"), response_type=int)

    @response_handler
    async def on_human_response(
        self,
        original_request: NumberSignal,
        response: int,
        ctx: WorkflowContext[int, str],
    ) -> None:
        await self.handle_guess(response, ctx)


judge = JudgeExecutor(target_number=42)
workflow = WorkflowBuilder(start_executor=judge).build()

يقوم @response_handler مصمم الديكور تلقائيا بتسجيل الأسلوب لمعالجة الاستجابات لنوعي الطلب والاستجابة المحددين. يطابق إطار العمل الاستجابات الواردة للمعالج الصحيح استنادا إلى نوع التعليقات التوضيحية original_request للمعلمات و response .

تدعم مهام سير العمل أنماطا بشرية في التكرار الحلقي من خلال RequestPort، والتي توقف التنفيذ مؤقتا وتنتظر الإدخال الخارجي.

approvalPort := workflow.RequestPort{
    ID:       "ApprovalPort",
    Request:  reflect.TypeFor[string](),
    Response: reflect.TypeFor[bool](),
}

approval := approvalPort.Bind()
finalize := workflow.NewExecutor("FinalizeExecutor", func(approved bool) string {
    if approved {
        return "Request approved by the human reviewer"
    }
    return "Request rejected by the human reviewer"
}).Bind()

wf, err := workflow.NewBuilder(approval).
    AddEdge(approval, finalize).
    WithOutputFrom(finalize).
    Build()

RequestPort يحدد قناة طلب/استجابة مكتوبة بين سير العمل والعالم الخارجي. عندما يصل المنفذ إلى منفذ طلب، يتوقف سير العمل مؤقتا ويبعث حدث طلب خارجي. يستأنف سير العمل عند توفير استجابة خارجية.

معالجة الطلبات والاستجابات

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

await using StreamingRun handle = await InProcessExecution.RunStreamingAsync(workflow, NumberSignal.Init);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    switch (evt)
    {
        case RequestInfoEvent requestInputEvt:
            // Handle `RequestInfoEvent` from the workflow
            int guess = ...; // Get the guess from the human operator or any external system
            await handle.SendResponseAsync(requestInputEvt.Request.CreateResponse(guess));
            break;

        case WorkflowOutputEvent outputEvt:
            // The workflow has yielded output
            Console.WriteLine($"Workflow completed with result: {outputEvt.Data}");
            return;
    }
}

Tip

راجع العينة الكاملة للمشروع الكامل القابل للتشغيل.

يمكن للمنفذين إرسال الطلبات مباشرة دون الحاجة إلى مكون منفصل. عندما يستدعي ctx.request_info()المنفذ ، يصدر سير العمل WorkflowEvent مع type == "request_info". يمكنك الاشتراك في هذه الأحداث لمعالجة الطلبات الواردة من سير العمل. عند تلقي استجابة من نظام خارجي، أرسلها مرة أخرى إلى سير العمل باستخدام آلية الاستجابة. يوجه إطار العمل الاستجابة تلقائيا إلى أسلوب المنفذ @response_handler .

from collections.abc import AsyncIterable

from agent_framework import WorkflowEvent


async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str, int] | None:
    """Process events from the workflow stream to capture requests."""
    requests: list[tuple[str, NumberSignal]] = []
    async for event in stream:
        if event.type == "request_info":
            requests.append((event.request_id, event.data))

    # Handle any pending human feedback requests.
    if requests:
        responses: dict[str, int] = {}
        for request_id, request in requests:
            guess = ...  # Get the guess from the human operator or any external system.
            responses[request_id] = guess
        return responses

    return None

# Initiate the first run of the workflow with an initial guess.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(25, stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    # Run the workflow until there is no more human feedback to provide,
    # in which case this workflow completes.
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Tip

راجع هذا النموذج الكامل لملف كامل قابل للتشغيل.

استمع إلى workflow.RequestInfoEvent، وأنشئ استجابة من الطلب، واستأنف التشغيل بهذه الاستجابة:

run, err := inproc.Default.Run(ctx, wf, "Approve deployment to production?")
if err != nil {
    return err
}

var request *workflow.ExternalRequest
for evt := range run.NewEvents() {
    if requestEvent, ok := evt.(workflow.RequestInfoEvent); ok {
        request = requestEvent.Request
        break
    }
}

response, err := request.CreateResponse(true)
if err != nil {
    return err
}

if _, err := run.Resume(ctx, response); err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Println(output.Output)
    }
}

Tip

راجع عينة human-in-the-loop للحصول على ملف كامل قابل للتشغيل.

Human-in-the-Loop مع تنسيقات الوكيل

يعمل النمط الموضح RequestPort أعلاه مع المنفذين المخصصين و WorkflowBuilder. عند استخدام تنسيقات الوكيل (مثل مهام سير عمل الدردشة التسلسلية أو المتزامنة أو الجماعية)، يتم تحقيق الموافقة على الأدوات من خلال آلية طلب/استجابة الإنسان في التكرار الحلقي.

يمكن للوكلاء استخدام الأدوات التي تتطلب موافقة بشرية قبل التنفيذ. عندما يحاول العامل استدعاء أداة مطلوبة للموافقة، يتوقف سير العمل مؤقتا ويبعث RequestInfoEvent مثل RequestPort النمط تماما، ولكن حمولة الحدث تحتوي على ToolApprovalRequestContent (C# وGo) أو Content مع type == "function_approval_request" (Python) بدلا من نوع طلب مخصص.

للسيناريوهات التفاعلية حيث يحتاج العامل إلى جمع مزيد من المعلومات من المستخدم والتكرار قبل المتابعة؛ بدلا من الموافقة على استدعاء أداة أو رفضه فقط؛ استخدم تنسيق التسليم. التسليم تفاعلي بشكل افتراضي: عندما يستجيب عامل دون تسليمه إلى عامل آخر، يعود عنصر التحكم إلى المستخدم للإدخال التالي، والذي يمكن متعدد الأدوار ذهابا وإيابا داخل التنسيق. لا تتوقف تنسيقات الدردشة المتتالية والمتزامنة والمجموعة مؤقتا لإدخال مستخدم النموذج الحر من تلقاء نفسه؛ قم بإقرانها ب RequestPort في سير عمل مخصص WorkflowBuilder عندما تحتاج إلى عنصر التحكم هذا بين الخطوات.

Tip

للحصول على أمثلة كاملة مع التعليمات البرمجية، راجع:

• التزامن التسلسلي مع HITL
• عينة GroupChatToolApproval (C#)
• نموذج الموافقة على الأدوات المتتالية (Python)
• نموذج معلومات الطلب التسلسلي (Python)

نقاط التحقق والطلبات

لمعرفة المزيد حول نقاط التحقق، راجع نقاط التحقق.

عند إنشاء نقطة تحقق، يتم أيضا حفظ الطلبات المعلقة كجزء من حالة نقطة التحقق. عند الاستعادة من نقطة تحقق، ستتم إعادة إصدار أي طلبات معلقة ككائنات RequestInfoEvent ، مما يسمح لك بالتقاطها والاستجابة لها. يمكنك أيضا الاستئناف من نقطة تحقق وتقديم استجابات في نفس الاستدعاء عن طريق تمرير كل checkpoint_id من و responses إلى workflow.run(...).

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

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

• تعرف على التنسيق التسلسلي باستخدام HITL.
• تعرف على كيفية إدارة الحالة في مهام سير العمل.
• تعرف على كيفية إنشاء نقاط التحقق والاستئناف منها.
• تعرف على كيفية مراقبة مهام سير العمل.
• تعرف على كيفية تصور مهام سير العمل.