處理工作流程中的請求和回應

本教學課程示範如何使用 Agent Framework 工作流程處理工作流程中的請求和回應。您將瞭解如何建立互動式工作流程，以暫停執行以要求外部來源（例如人類或其他系統）的輸入，然後在提供回應後繼續。

涵蓋概念

請求與回應

在 .NET 中，人機迴圈工作流程會使用 RequestPort 和外部要求處理來暫停執行並收集使用者輸入。此模式可啟用互動式工作流程，讓系統在執行期間可從外部來源請求資訊。

先決條件

.NET 8.0 SDK 或更新版本。
已設定 Azure OpenAI 服務端點和部署。
已安裝並驗證 Azure CLI （針對 Azure 認證驗證）。
對 C# 和非同步程式設計有基本的了解。
新的主控台應用程式。

安裝 NuGet 套件

首先，安裝 .NET 專案所需的套件：

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

關鍵組件

RequestPort 和外部請求

A RequestPort 可作為工作流程與外部輸入來源之間的橋樑。當工作流程需要輸入時，它會產生一個由您的應用程式處理的RequestInfoEvent。

// Create a RequestPort for handling human input requests
RequestPort numberRequestPort = RequestPort.Create<NumberSignal, int>("GuessNumber");

訊號類型

定義訊號類型以傳達不同的請求類型：

/// <summary>
/// Signals used for communication between guesses and the JudgeExecutor.
/// </summary>
internal enum NumberSignal
{
    Init,     // Initial guess request
    Above,    // Previous guess was too high
    Below,    // Previous guess was too low
}

工作流程執行器

建立處理使用者輸入並提供意見反應的執行器：

/// <summary>
/// Executor that judges the guess and provides feedback.
/// </summary>
internal sealed class JudgeExecutor : Executor<int>("Judge")
{
    private readonly int _targetNumber;
    private int _tries;

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

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

建立工作流程

在回饋迴路中連接 RequestPort 和執行單元：

internal static class WorkflowHelper
{
    internal static ValueTask<Workflow<NumberSignal>> GetWorkflowAsync()
    {
        // Create the executors
        RequestPort numberRequestPort = RequestPort.Create<NumberSignal, int>("GuessNumber");
        JudgeExecutor judgeExecutor = new(42);

        // Build the workflow by connecting executors in a loop
        return new WorkflowBuilder(numberRequestPort)
            .AddEdge(numberRequestPort, judgeExecutor)
            .AddEdge(judgeExecutor, numberRequestPort)
            .WithOutputFrom(judgeExecutor)
            .BuildAsync<NumberSignal>();
    }
}

執行互動式工作流程

在工作流程執行期間處理外部請求：

private static async Task Main()
{
    // Create the workflow
    var workflow = await WorkflowHelper.GetWorkflowAsync().ConfigureAwait(false);

    // Execute the workflow
    await using StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init).ConfigureAwait(false);
    await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false))
    {
        switch (evt)
        {
            case RequestInfoEvent requestInputEvt:
                // Handle human input request from the workflow
                ExternalResponse response = HandleExternalRequest(requestInputEvt.Request);
                await handle.SendResponseAsync(response).ConfigureAwait(false);
                break;

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

要求處理

處理不同類型的輸入請求：

private static ExternalResponse HandleExternalRequest(ExternalRequest request)
{
    switch (request.DataAs<NumberSignal?>())
    {
        case NumberSignal.Init:
            int initialGuess = ReadIntegerFromConsole("Please provide your initial guess: ");
            return request.CreateResponse(initialGuess);
        case NumberSignal.Above:
            int lowerGuess = ReadIntegerFromConsole("You previously guessed too large. Please provide a new guess: ");
            return request.CreateResponse(lowerGuess);
        case NumberSignal.Below:
            int higherGuess = ReadIntegerFromConsole("You previously guessed too small. Please provide a new guess: ");
            return request.CreateResponse(higherGuess);
        default:
            throw new ArgumentException("Unexpected request type.");
    }
}

private static int ReadIntegerFromConsole(string prompt)
{
    while (true)
    {
        Console.Write(prompt);
        string? input = Console.ReadLine();
        if (int.TryParse(input, out int value))
        {
            return value;
        }
        Console.WriteLine("Invalid input. Please enter a valid integer.");
    }
}

實作概念

RequestInfoEvent 流程

工作流程執行：工作流程運行直到需要外部輸入為止
請求生成：RequestPort 生成包含 RequestInfoEvent 請求的詳細信息
外部處理：您的應用程式會擷取事件並收集使用者輸入
回應提交：回傳 ExternalResponse 以繼續工作流程
工作流程恢復：工作流程繼續使用提供的輸入進行處理

工作流程生命週期

串流執行：用於 StreamAsync 即時監控事件
事件處理：處理輸入請求的流程RequestInfoEvent和處理完成事項的流程WorkflowOutputEvent
回應協調：使用工作流程的回應處理機制將回應與請求匹配

實施流程

工作流程初始化：工作流程首先將 a NumberSignal.Init 傳送到 RequestPort。
請求生成：RequestPort 會產生 RequestInfoEvent 從使用者取得的初始猜測請求。
工作流程暫停：工作流程會暫停並等待外部輸入，同時應用程式處理請求。
人工回應：外部應用程式會收集使用者輸入，並將訊息 ExternalResponse 傳回工作流程。
處理和回饋：處理 JudgeExecutor 猜測並完成工作流程或發送新訊號（上方/下方）以請求另一次猜測。
循環延續：重複該過程，直到猜出正確的數字。

框架優勢

類型安全：強類型確保維護請求-響應合約
事件驅動：豐富的事件系統提供對工作流程執行的可見性
可暫停執行：工作流程可以在等待外部輸入時無限暫停
狀態管理：工作流程狀態在暫停-恢復週期中保留
靈活整合： RequestPorts 可以與任何外部輸入源（UI、API、控制台等）集成。

完整樣品

如需完整實作，請參閱人機互動基礎範例。

這種模式使得構建複雜的交互式應用程序成為可能，用戶可以在自動化工作流程中的關鍵決策點提供輸入。

您將構建什麼

您將建立互動式數字猜測遊戲工作流程，以示範要求-回應模式：

進行智能猜測的 AI 代理
可以直接使用 request_info API 傳送請求的執行器
代理和人類互動之間協調的輪次管理員，使用 @response_handler
互動式控制台輸入/輸出，提供即時回饋

先決條件

Python 3.10 或更新版本
已設定 Azure OpenAI 部署
已設定 Azure CLI 驗證（az login）
對 Python 非同步程式設計有基本的了解

重要概念

請求和回應功能

執行程式具有內建的請求和回應功能，可實現人機迴圈互動：

呼叫 ctx.request_info(request_data=request_data, response_type=response_type) 以傳送請求
使用 @response_handler 裝飾器來處理回應
定義自訂請求/回應類型，無需繼承要求

Request-Response 流程

執行器可以直接使用 ctx.request_info() 發送請求，並使用 @response_handler 裝飾器處理回應：

執行器呼叫 ctx.request_info(request_data=request_data, response_type=response_type)
工作流程會發出包含請求數據的 RequestInfoEvent
外部系統（人工、API 等）處理請求
回應會透過 send_responses_streaming() 發送
工作流程會繼續，並將回應傳遞至執行程式的方法@response_handler

設定環境

首先，安裝所需的套件：

pip install agent-framework-core --pre
pip install azure-identity

定義請求和回應模型

首先定義請求-回應通訊的資料結構：

import asyncio
from dataclasses import dataclass
from pydantic import BaseModel

from agent_framework import (
    AgentExecutor,
    AgentExecutorRequest,
    AgentExecutorResponse,
    ChatMessage,
    Executor,
    RequestInfoEvent,
    Role,
    WorkflowBuilder,
    WorkflowContext,
    WorkflowOutputEvent,
    WorkflowRunState,
    WorkflowStatusEvent,
    handler,
    response_handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

@dataclass
class HumanFeedbackRequest:
    """Request message for human feedback in the guessing game."""
    prompt: str = ""
    guess: int | None = None

class GuessOutput(BaseModel):
    """Structured output from the AI agent with response_format enforcement."""
    guess: int

這是 HumanFeedbackRequest 結構化請求承載的簡單資料類別：

請求載荷的強型別化
前向相容驗證
與回應的明確關聯語意
用於豐富使用者介面提示的情境欄位（如先前的猜測）

建立回合管理員

回合管理器協調 AI 代理和人類之間的流程：

class TurnManager(Executor):
    """Coordinates turns between the AI agent and human player.

    Responsibilities:
    - Start the game by requesting the agent's first guess
    - Process agent responses and request human feedback
    - Handle human feedback and continue the game or finish
    """

    def __init__(self, id: str | None = None):
        super().__init__(id=id or "turn_manager")

    @handler
    async def start(self, _: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
        """Start the game by asking the agent for an initial guess."""
        user = ChatMessage(Role.USER, text="Start by making your first guess.")
        await ctx.send_message(AgentExecutorRequest(messages=[user], should_respond=True))

    @handler
    async def on_agent_response(
        self,
        result: AgentExecutorResponse,
        ctx: WorkflowContext,
    ) -> None:
        """Handle the agent's guess and request human guidance."""
        # Parse structured model output (defensive default if agent didn't reply)
        text = result.agent_run_response.text or ""
        last_guess = GuessOutput.model_validate_json(text).guess if text else None

        # Craft a clear human prompt that defines higher/lower relative to agent's guess
        prompt = (
            f"The agent guessed: {last_guess if last_guess is not None else text}. "
            "Type one of: higher (your number is higher than this guess), "
            "lower (your number is lower than this guess), correct, or exit."
        )
        # Send a request using the request_info API
        await ctx.request_info(
            request_data=HumanFeedbackRequest(prompt=prompt, guess=last_guess),
            response_type=str
        )

    @response_handler
    async def on_human_feedback(
        self,
        original_request: HumanFeedbackRequest,
        feedback: str,
        ctx: WorkflowContext[AgentExecutorRequest, str],
    ) -> None:
        """Continue the game or finish based on human feedback."""
        reply = feedback.strip().lower()
        # Use the correlated request's guess to avoid extra state reads
        last_guess = original_request.guess

        if reply == "correct":
            await ctx.yield_output(f"Guessed correctly: {last_guess}")
            return

        # Provide feedback to the agent for the next guess
        user_msg = ChatMessage(
            Role.USER,
            text=f'Feedback: {reply}. Return ONLY a JSON object matching the schema {{"guess": <int 1..10>}}.',
        )
        await ctx.send_message(AgentExecutorRequest(messages=[user_msg], should_respond=True))

建立工作流程

建立連接所有元件的主要工作流程：

async def main() -> None:
    # Create the chat agent with structured output enforcement
    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
    agent = chat_client.create_agent(
        instructions=(
            "You guess a number between 1 and 10. "
            "If the user says 'higher' or 'lower', adjust your next guess. "
            'You MUST return ONLY a JSON object exactly matching this schema: {"guess": <integer 1..10>}. '
            "No explanations or additional text."
        ),
        response_format=GuessOutput,
    )

    # Create workflow components
    turn_manager = TurnManager(id="turn_manager")
    agent_exec = AgentExecutor(agent=agent, id="agent")

    # Build the workflow graph
    workflow = (
        WorkflowBuilder()
        .set_start_executor(turn_manager)
        .add_edge(turn_manager, agent_exec)  # Ask agent to make/adjust a guess
        .add_edge(agent_exec, turn_manager)  # Agent's response goes back to coordinator
        .build()
    )

    # Execute the interactive workflow
    await run_interactive_workflow(workflow)

async def run_interactive_workflow(workflow):
    """Run the workflow with human-in-the-loop interaction."""
    pending_responses: dict[str, str] | None = None
    completed = False
    workflow_output: str | None = None

    print("🎯 Number Guessing Game")
    print("Think of a number between 1 and 10, and I'll try to guess it!")
    print("-" * 50)

    while not completed:
        # First iteration uses run_stream("start")
        # Subsequent iterations use send_responses_streaming with pending responses
        stream = (
            workflow.send_responses_streaming(pending_responses)
            if pending_responses
            else workflow.run_stream("start")
        )

        # Collect events for this turn
        events = [event async for event in stream]
        pending_responses = None

        # Process events to collect requests and detect completion
        requests: list[tuple[str, str]] = []  # (request_id, prompt)
        for event in events:
            if isinstance(event, RequestInfoEvent) and isinstance(event.data, HumanFeedbackRequest):
                # RequestInfoEvent for our HumanFeedbackRequest
                requests.append((event.request_id, event.data.prompt))
            elif isinstance(event, WorkflowOutputEvent):
                # Capture workflow output when yielded
                workflow_output = str(event.data)
                completed = True

        # Check workflow status
        pending_status = any(
            isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS
            for e in events
        )
        idle_with_requests = any(
            isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
            for e in events
        )

        if pending_status:
            print("🔄 State: IN_PROGRESS_PENDING_REQUESTS (requests outstanding)")
        if idle_with_requests:
            print("⏸️  State: IDLE_WITH_PENDING_REQUESTS (awaiting human input)")

        # Handle human requests if any
        if requests and not completed:
            responses: dict[str, str] = {}
            for req_id, prompt in requests:
                print(f"\n🤖 {prompt}")
                answer = input("👤 Enter higher/lower/correct/exit: ").lower()

                if answer == "exit":
                    print("👋 Exiting...")
                    return
                responses[req_id] = answer
            pending_responses = responses

    # Show final result
    print(f"\n🎉 {workflow_output}")

執行範例

如需完整的工作實作，請參閱 Human-in-the-Loop 猜測遊戲範例。

運作方式

工作流程初始化：工作流程從請求 AI 代理的初始猜測開始 TurnManager 。
代理回應：AI 代理進行猜測並傳回結構化 JSON，該 JSON 流回 TurnManager.
人工請求：TurnManager 處理客服專員的猜測後，呼叫 ctx.request_info() 並附帶 HumanFeedbackRequest。
工作流程暫停：工作流程會發出 RequestInfoEvent 並繼續執行，直到無法再執行任何動作為止，然後等待人工輸入。
人工回應：外部應用程式收集人工輸入並使用 send_responses_streaming()將回應傳回。
恢復及繼續：工作流程恢復，TurnManager的@response_handler方法處理人工反饋，然後結束遊戲或向代理發送另一個請求。

主要優點

結構化通訊：類型安全的請求和回應模型可防止運行時錯誤
關聯性：請求 ID 確保響應與正確的請求相符
可暫停執行：工作流程可以在等待外部輸入時無限暫停
狀態保留：工作流程狀態會在暫停-繼續週期中維護
事件驅動：豐富的事件系統提供對工作流程狀態和過渡的可見性

這種模式使得建立複雜的互動式應用程式成為可能，其中人工智慧代理和人類可以在結構化工作流程中無縫協作。

後續步驟

瞭解檢查點處理和恢復工作流程

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-11-05

共用方式為

處理工作流程中的請求和回應

涵蓋概念

先決條件

安裝 NuGet 套件

關鍵組件

RequestPort 和外部請求

訊號類型

工作流程執行器

建立工作流程

執行互動式工作流程

要求處理

實作概念

RequestInfoEvent 流程

工作流程生命週期

實施流程

框架優勢

完整樣品

您將構建什麼

先決條件

重要概念

請求和回應功能

Request-Response 流程

設定環境

定義請求和回應模型

建立回合管理員

建立工作流程

執行範例

運作方式

主要優點

後續步驟

意見反應

其他資源