Aracılığıyla paylaş

Microsoft Agent Framework İş Akışları - İstek ve Yanıt

Bu sayfa, Microsoft Agent Framework İş Akışı sisteminde İstek ve Yanıt işlemenin nasıl çalıştığına genel bir bakış sağlar.

Genel Bakış

bir iş akışındaki yürütücüler iş akışının dışına istek gönderebilir ve yanıtları bekleyebilir. Bu, yürütücünün döngüdeki insan etkileşimleri veya diğer zaman uyumsuz işlemler gibi dış sistemlerle etkileşim kurması gereken senaryolar için kullanışlıdır.

İş Akışında İstek ve Yanıt İşlemeyi Etkinleştirme

İstekler ve yanıtlar adlı InputPortözel bir tür aracılığıyla işlenir.

// Create an input port that receives requests of type CustomRequestType and responses of type CustomResponseType.
var inputPort = InputPort.Create<CustomRequestType, CustomResponseType>("input-port");

Giriş bağlantı noktasını bir iş akışına ekleyin.

var executorA = new SomeExecutor();
var workflow = new WorkflowBuilder(inputPort)
    .AddEdge(inputPort, executorA)
    .AddEdge(executorA, inputPort)
    .Build<CustomRequestType>();

Şimdi, iş akışında executorA her iki yönde de bağlı inputPort olduğundan, executorA üzerinden inputPortistek gönderebilmesi ve yanıt alabilmesi gerekir. İstek göndermek ve yanıt almak için içinde yapmanız SomeExecutor gerekenler aşağıdadır.

internal sealed class SomeExecutor() : Executor<CustomResponseType>("SomeExecutor")
{
    public async ValueTask HandleAsync(CustomResponseType message, IWorkflowContext context)
    {
        // Process the response...
        ...
        // Send a request
        await context.SendMessageAsync(new CustomRequestType(...)).ConfigureAwait(false);
    }
}

Alternatif olarak, SomeExecutor istek gönderme ve yanıt işlemeyi iki işleyiciye ayırabilir.

internal sealed class SomeExecutor() : Executor("SomeExecutor")
{
    protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
    {
        return routeBuilder
            .AddHandler<CustomResponseType>(this.HandleCustomResponseAsync)
            .AddHandler<OtherDataType>(this.HandleOtherDataAsync);
    }

    public async ValueTask HandleCustomResponseAsync(CustomResponseType message, IWorkflowContext context)
    {
        // Process the response...
        ...
    }

    public async ValueTask HandleOtherDataAsync(OtherDataType message, IWorkflowContext context)
    {
        // Process the message...
        ...
        // Send a request
        await context.SendMessageAsync(new CustomRequestType(...)).ConfigureAwait(false);
    }
}

ctx.request_info() kullanarak yürütücüler istek gönderebilir ve @response_handler ile yanıtları işleyebilir.

from agent_framework import response_handler, WorkflowBuilder

executor_a = SomeExecutor()
executor_b = SomeOtherExecutor()
workflow_builder = WorkflowBuilder()
workflow_builder.set_start_executor(executor_a)
workflow_builder.add_edge(executor_a, executor_b)
workflow = workflow_builder.build()

executor_a yerleşik özellikleri kullanarak doğrudan istek gönderebilir ve yanıt alabilir.

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

class SomeExecutor(Executor):

    @handler
    async def handle_data(
        self,
        data: OtherDataType,
        context: WorkflowContext,
    ):
        # Process the message...
        ...
        # Send a request using the API
        await context.request_info(
            request_data=CustomRequestType(...),
            response_type=CustomResponseType
        )

    @response_handler
    async def handle_response(
        self,
        original_request: CustomRequestType,
        response: CustomResponseType,
        context: WorkflowContext,
    ):
        # Process the response...
        ...

Dekoratör, @response_handler belirtilen istek ve yanıt türlerine yönelik yanıtları işlemek için yöntemi otomatik olarak kaydeder.

İstekleri ve Yanıtları İşleme

Bir InputPort, bir istek aldığında bir RequestInfoEvent yayar. İş akışından gelen istekleri işlemek için bu olaylara abone olabilirsiniz. Bir dış sistemden yanıt aldığınızda, yanıt mekanizmasını kullanarak bu yanıtı iş akışına geri gönderin. Çerçeve, yanıtı otomatik olarak özgün isteği gönderen yürütücüye yönlendirir.

StreamingRun handle = await InProcessExecution.StreamAsync(workflow, input).ConfigureAwait(false);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false))
{
    switch (evt)
    {
        case RequestInfoEvent requestInputEvt:
            // Handle `RequestInfoEvent` from the workflow
            ExternalResponse response = requestInputEvt.Request.CreateResponse<CustomResponseType>(...);
            await handle.SendResponseAsync(response).ConfigureAwait(false);
            break;

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

Yürütücüler ayrı bir bileşene gerek kalmadan doğrudan istek gönderebilir. Yürütücü ctx.request_info()'i çağırdığında, iş akışı RequestInfoEvent yayar. İş akışından gelen istekleri işlemek için bu olaylara abone olabilirsiniz. Bir dış sistemden yanıt aldığınızda, yanıt mekanizmasını kullanarak bu yanıtı iş akışına geri gönderin. Çerçeve, yanıtı yürütücüsü @response_handler yöntemine otomatik olarak yönlendirir.

from agent_framework import RequestInfoEvent

while True:
    request_info_events : list[RequestInfoEvent] = []
    pending_responses : dict[str, CustomResponseType] = {}

    stream = workflow.run_stream(input) if not pending_responses else workflow.send_responses_streaming(pending_responses)

    async for event in stream:
        if isinstance(event, RequestInfoEvent):
            # Handle `RequestInfoEvent` from the workflow
            request_info_events.append(event)

    if not request_info_events:
        break

    for request_info_event in request_info_events:
        # Handle `RequestInfoEvent` from the workflow
        response = CustomResponseType(...)
        pending_responses[request_info_event.request_id] = response

Denetim Noktaları ve İstekler

Denetim noktaları hakkında daha fazla bilgi edinmek için bkz. Denetim noktaları.

Bir denetim noktası oluşturulduğunda, bekleyen istekler de denetim noktası durumunun bir parçası olarak kaydedilir. Bir denetim noktasından geri yükleme yaptığınızda, bekleyen istekler yeniden nesne halinde RequestInfoEvent yayımlanarak bunları yakalayabilmenizi ve onlara yanıt verebilmenizi sağlar. Sürdürme işlemi sırasında yanıtları doğrudan sağlayamazsınız; bunun yerine, yeniden yayılan olayları dinlemeniz ve standart yanıt mekanizmasını kullanarak yanıt vermeniz gerekir.

Sonraki Adımlar

  • Last updated on