Microsoft Agent Framework 工作流程編排 - 循序

在循序編排中，代理程式會組織在管道中，每個代理程式依序處理任務，並將輸出傳遞給序列中的下一個代理程式。

設定 Azure OpenAI 用戶端

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// 1) Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ??
    throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var client = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetProjectOpenAIClient()
    .GetProjectResponsesClient()
    .AsIChatClient(deploymentName);

建立將依序運作的專用代理程式：

// 2) Helper method to create translation agents
static ChatClientAgent GetTranslationAgent(string targetLanguage, IChatClient chatClient) =>
    new(chatClient,
        $"You are a translation assistant who only responds in {targetLanguage}. Respond to any " +
        $"input by outputting the name of the input language and then translating the input to {targetLanguage}.");

// Create translation agents for sequential processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

設定循序協調流程

使用下列方式 AgentWorkflowBuilder建置工作流程：

// 3) Build sequential workflow
var workflow = AgentWorkflowBuilder.BuildSequential(translationAgents);

執行循序工作流程

執行工作流程並處理事件：

// 4) Run the workflow
var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, world!") };

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

string? lastExecutorId = null;
List<ChatMessage> result = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
    {
        if (e.ExecutorId != lastExecutorId)
        {
            lastExecutorId = e.ExecutorId;
            Console.WriteLine();
            Console.Write($"{e.ExecutorId}: ");
        }

        Console.Write(e.Update.Text);
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = outputEvt.As<List<ChatMessage>>()!;
        break;
    }
}

// Display final result
Console.WriteLine();
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

範例輸出

French_Translation: User: Hello, world!
French_Translation: Assistant: English detected. Bonjour, le monde !
Spanish_Translation: Assistant: French detected. ¡Hola, mundo!
English_Translation: Assistant: Spanish detected. Hello, world!

有人在迴路中的序列協作

序列編排透過工具審核支持人機互動。 當代理使用以 ApprovalRequiredAIFunction 包裹的工具時，工作流程會暫停並輸出包含 RequestInfoEvent 的 ToolApprovalRequestContent。 外部系統（如人工操作員）可以檢查工具呼叫，批准或拒絕，工作流程隨即恢復。

使用需核准的工具來定義代理

建立使用敏感工具的代理程式，並將其包裹於 ApprovalRequiredAIFunction：

ChatClientAgent deployAgent = new(
    client,
    "You are a DevOps engineer. Check staging status first, then deploy to production.",
    "DeployAgent",
    "Handles deployments",
    [
        AIFunctionFactory.Create(CheckStagingStatus),
        new ApprovalRequiredAIFunction(AIFunctionFactory.Create(DeployToProduction))
    ]);

ChatClientAgent verifyAgent = new(
    client,
    "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
    "VerifyAgent",
    "Verifies deployments");

建置與執行並進行核准處理

正常建立連續工作流程。 審核流程透過事件流程處理：

var workflow = AgentWorkflowBuilder.BuildSequential([deployAgent, verifyAgent]);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is RequestInfoEvent e &&
        e.Request.TryGetDataAs(out ToolApprovalRequestContent? approvalRequest))
    {
        await run.SendResponseAsync(
            e.Request.CreateResponse(approvalRequest.CreateResponse(approved: true)));
    }
}

超越工具審核：互動回饋

工具審核允許人工接受或拒絕特定工具呼叫，但序列協調不包含內建的步驟讓代理間自由回應使用者回饋，且無法將控制權還給先前代理。 當代理需要互動式詢問使用者更多資訊並反覆進行後再繼續時;例如，在呼叫訂位工具前收集訂位細節;改用以下其中一種方法：

• 交接編排 預設為互動式：當代理在未進行交接的情況下回應時，控制權會返回給使用者，由其提供下一次輸入，從而可在編排流程中進行多輪來回互動。 限制每個代理只能使用單一的交接目標，以模擬一個仍會暫停使用者輸入的連續流程。
• 以 WorkflowBuilder 和 RequestPort 建立的 自訂工作流程 可讓你在任何時點向使用者傳送具型別的請求，並將回應路由回傳給執行器；你可以將該執行器放在管線中代理程式的前面或後面。

關鍵概念

• 順序處理：每個代理依序處理前一個代理的輸出
• AgentWorkflowBuilder.BuildSequential（）：從代理程式集合建立管線工作流程
• ChatClientAgent：代表由聊天客戶端支援的代理，並提供特定說明
• InProcessExecution.RunStreamingAsync（）：執行工作流程並回傳即時 StreamingRun 事件串流
• 事件處理：透過 AgentResponseUpdateEvent 監控代理的進度，透過 WorkflowOutputEvent 監控完成情況
• 工具審核：將敏感工具 ApprovalRequiredAIFunction 包裹起來，要求人工批准後才能執行
• RequestInfoEvent：當工具需要核准時會發出;包含 ToolApprovalRequestContent 工具呼叫詳情
• 互動式 HITL：順序協調涵蓋工具審核;若能進行互動式互動，讓代理從使用者那裡收集更多資訊，請使用 交接協調 或自訂 RequestPort 工作流程

在循序協調流程中，每個代理程式依序處理任務，輸出從一個代理程式流向下一個代理程式。 首先定義兩個階段流程中的代理人：

import os
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

# 1) Create agents using FoundryChatClient
chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

writer = chat_client.as_agent(
    instructions=(
        "You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."
    ),
    name="writer",
)

reviewer = chat_client.as_agent(
    instructions=(
        "You are a thoughtful reviewer. Give brief feedback on the previous assistant message."
    ),
    name="reviewer",
)

設定循序協調流程

該 SequentialBuilder 類創建了一個管道，代理程式在其中按順序處理任務。 每個客服專員都會看到完整的對話歷史記錄並添加他們的回應：

from agent_framework.orchestrations import SequentialBuilder

# 2) Build sequential workflow: writer -> reviewer
workflow = SequentialBuilder(participants=[writer, reviewer]).build()

執行循序工作流程

執行工作流程並收集最終輸出。 終端輸出為包含最後代理回應訊息的AgentResponse:

from agent_framework import AgentResponse

# 3) Run and print the last agent's response
events = await workflow.run("Write a tagline for a budget-friendly eBike.")
outputs = events.get_outputs()

if outputs:
    print("===== Final Response =====")
    final: AgentResponse = outputs[0]
    for msg in final.messages:
        name = msg.author_name or "assistant"
        print(f"[{name}]\n{msg.text}")

範例輸出

===== Final Response =====
[reviewer]
This tagline clearly communicates affordability and the benefit of extended travel, making it
appealing to budget-conscious consumers. It has a friendly and motivating tone, though it could
be slightly shorter for more punch. Overall, a strong and effective suggestion!

進階：將代理程式與自訂執行器混合

循序協調流程支援將代理程式與自訂執行程式混合，以進行專門處理。 當您需要不需要 LLM 的自訂邏輯時，這很有用：

定義自訂執行程式

from agent_framework import AgentExecutorResponse, AgentResponse, Executor, WorkflowContext, handler
from agent_framework import Message
from typing_extensions import Never

class Summarizer(Executor):
    """Terminator custom executor: consumes full conversation and yields a summary as the workflow's final answer."""

    @handler
    async def summarize(
        self,
        agent_response: AgentExecutorResponse,
        ctx: WorkflowContext[Never, AgentResponse]
    ) -> None:
        if not agent_response.full_conversation:
            await ctx.yield_output(AgentResponse(messages=[Message("assistant", ["No conversation to summarize."])]))
            return

        users = sum(1 for m in agent_response.full_conversation if m.role == "user")
        assistants = sum(1 for m in agent_response.full_conversation if m.role == "assistant")
        summary = Message("assistant", [f"Summary -> users:{users} assistants:{assistants}"])
        await ctx.yield_output(AgentResponse(messages=[summary]))

建立混合循序工作流程

# Create a content agent
content = chat_client.as_agent(
    instructions="Produce a concise paragraph answering the user's request.",
    name="content",
)

# Build sequential workflow: content -> summarizer
summarizer = Summarizer(id="summarizer")
workflow = SequentialBuilder(participants=[content, summarizer]).build()

使用自訂執行程式的範例輸出

===== Final Summary =====
Summary -> users:1 assistants:1

控制代理間的上下文

預設情況下，工作流程中的 SequentialBuilder 每個代理會消耗前一位代理的完整對話（輸入+回應訊息）。 設定 chain_only_agent_responses=True 會將序列中的所有代理設定為只使用前一個代理的回應訊息：

workflow = SequentialBuilder(
    participants=[writer, translator, reviewer],
    chain_only_agent_responses=True,
).build()

這對於翻譯流程、漸進式精煉及其他情境非常有用，讓每個代理專注於轉換前一位代理的輸出，不受先前對話回合影響。

完整範例請參見代理框架倉庫中的 sequential_chain_only_agent_responses.py 。

中間輸出

預設情況下， SequentialBuilder 指定 最後一位參與者 為終端輸出來源（output_from）。 只有該參與者的產出會以 "output" 事件的形式出現。

若也要顯示較早的參與者輸出，請傳遞 intermediate_output_from，並附上你想指定為中間來源的參與者。 這隱含地將參與者從預設最終集合降級——他們發出的是 "intermediate" 事件而非 "output" 事件：

workflow = SequentialBuilder(
    participants=[writer, reviewer, editor],
    intermediate_output_from=[writer, reviewer],
).build()

你可以在串流模式下即時處理這兩個"intermediate""output"事件：

from agent_framework import AgentResponseUpdate

# Track the last author to format streaming output.
last_author: str | None = None

async for event in workflow.run("Write a tagline for a budget-friendly eBike.", stream=True):
    if event.type in ("output", "intermediate") and isinstance(event.data, AgentResponseUpdate):
        update = event.data
        author = update.author_name
        if author != last_author:
            if last_author is not None:
                print()  # Newline between different authors
            label = "FINAL" if event.type == "output" else "intermediate"
            print(f"[{label}] {author}: {update.text}", end="", flush=True)
            last_author = author
        else:
            print(update.text, end="", flush=True)

有人在迴路中的序列協作

序列協調支持人機互動的兩種方式為：用於控制敏感工具調用的工具審核，以及在每次代理程式回應後暫停以收集反饋的資訊請求。

序列工作流程中的工具核准

用於 @tool(approval_mode="always_require") 標記需要人工核准才能執行的工具。 當代理嘗試呼叫該工具時，工作流程會暫停並發出 request_info 事件。

@tool(approval_mode="always_require")
def execute_database_query(query: str) -> str:
    return f"Query executed successfully: {query}"


database_agent = Agent(
    client=chat_client,
    name="DatabaseAgent",
    instructions="You are a database assistant.",
    tools=[execute_database_query],
)

workflow = SequentialBuilder(participants=[database_agent]).build()

處理事件串流並處理核准請求：

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info" and event.data.type == "function_approval_request":
            responses[event.request_id] = event.data.to_function_approval_response(approved=True)
    return responses if responses else None

stream = workflow.run("Check the schema and update all pending orders", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

請求資訊以獲取客服人員回饋

在特定代理人回應之後，使用 .with_request_info() 暫停，以允許外部輸入（如人工審查），然後再開始下一位代理人的回應：

drafter = Agent(
    client=chat_client,
    name="drafter",
    instructions="You are a document drafter. Create a brief draft on the given topic.",
)

editor = Agent(
    client=chat_client,
    name="editor",
    instructions="You are an editor. Review and improve the draft. Incorporate any human feedback.",
)

finalizer = Agent(
    client=chat_client,
    name="finalizer",
    instructions="You are a finalizer. Create a polished final version.",
)

# Enable request info for the editor agent only
workflow = (
    SequentialBuilder(participants=[drafter, editor, finalizer])
    .with_request_info(agents=["editor"])
    .build()
)

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info":
            responses[event.request_id] = AgentRequestInfoResponse.approve()
    return responses if responses else None

stream = workflow.run("Write a brief introduction to artificial intelligence.", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

關鍵概念

• 共享上下文：預設情況下，每個代理會接收前一位代理的完整對話內容，包括輸入與回應訊息
• 情境控制：用於 chain_only_agent_responses=True 配置代理只接收前一個代理的回應訊息
• AgentResponse 輸出：工作流程的終端輸出是一個 AgentResponse，其中包含最後一個代理的回應（而非完整對話）
• 訂單事項：代理嚴格按照清單中 participants 指定的順序執行
• 靈活的參與者：您可以按任何順序混合代理和自訂執行者
• 自訂終止者合約：作為最後參與者的自訂執行者必須呼叫 ctx.yield_output(AgentResponse(...)) 以產生終端輸出
• 中間輸出：使用 intermediate_output_from=[...] 或 intermediate_output_from="all_other" 呈現參與者的進度作為中間工作流程事件，而非僅僅是最後一位參與者的終端輸出
• 工具核准：用於 @tool(approval_mode="always_require") 需要人工審查的敏感作業
• 請求資訊：在 .with_request_info(agents=[...]) 特定客服人員發言後暫停，尋求外部回饋

Go 可以使用 workflow/agentworkflow 建立序列式代理工作流程。 NewSequentialWorkflowBuilder 將每個代理作為工作流程執行器來承載，依序連接，並輸出最終的訊息批次作為工作流程輸出。

設定鑄造廠配置

endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

定義你的 Go 代理人

建立將依序運作的專用代理程式：

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(
        endpoint,
        token,
        foundryprovider.ModelDeployment(model),
        foundryprovider.AgentConfig{
            Instructions: fmt.Sprintf(
                "You are a translation assistant who only responds in %s. Respond to any input by outputting the name of the input language and then translating the input to %s.",
                language,
                language,
            ),
            Config: agent.Config{Name: language},
        },
    )
}

frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")

設定循序協調流程

wf, err := agentworkflow.NewSequentialWorkflowBuilder(
    frenchAgent,
    spanishAgent,
    englishAgent,
).
    WithName("translation-pipeline").
    Build()
if err != nil {
    return err
}

執行循序工作流程

執行工作流程並處理輸出事件：

run, err := inproc.Default.RunStreaming(ctx, wf, []*message.Message{message.NewText("Hello, world!")})
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

lastExecutorID := ""
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch e := evt.(type) {
    case workflow.OutputEvent:
        switch value := e.Output.(type) {
        case *agent.ResponseUpdate:
            if e.ExecutorID != lastExecutorID {
                lastExecutorID = e.ExecutorID
                fmt.Printf("\n%s: ", e.ExecutorID)
            }
            fmt.Print(value.String())
        case []*message.Message:
            fmt.Println("\n===== Final Response =====")
            for _, msg := range value {
                fmt.Printf("%s: %s\n", msg.Role, msg.String())
            }
        }
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

範例輸出

French: English detected. Bonjour, le monde !
Spanish: French detected. ¡Hola, mundo!
English: Spanish detected. Hello, world!

===== Final Response =====
assistant: Spanish detected. Hello, world!

有人在迴路中的序列協作

當託管代理使用需核准的工具時，序列工作流程可能會暫停以進行工具核准。 用 tool.ApprovalRequiredFunc 包裝工具，然後監聽 workflow.RequestInfoEvent，並以 ToolApprovalResponseContent 回應。

使用需核准的工具來定義代理

deployAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a DevOps engineer. Check staging status first, then deploy to production.",
        Config: agent.Config{
            Name:  "DeployAgent",
            Tools: []tool.Tool{tool.ApprovalRequiredFunc(deployTool)},
        },
    },
)

verifyAgent := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
        Config:      agent.Config{Name: "VerifyAgent"},
    },
)

wf, err := agentworkflow.NewSequentialWorkflowBuilder(deployAgent, verifyAgent).
    WithName("deployment-pipeline").
    Build()
if err != nil {
    return err
}

建置與執行並進行核准處理

在事件串流中處理批准請求：

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    requestEvent, ok := evt.(workflow.RequestInfoEvent)
    if !ok {
        continue
    }

    requestContent, ok := requestEvent.Request.Data.As(reflect.TypeFor[*message.ToolApprovalRequestContent]())
    if !ok {
        continue
    }

    approvalRequest := requestContent.(*message.ToolApprovalRequestContent)
    response, err := requestEvent.Request.CreateResponse(approvalRequest.CreateResponse(true, "approved"))
    if err != nil {
        return err
    }

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

進階：將代理程式與自訂執行器混合

對於混合式管線，請使用 agentworkflow.New 託管代理程式，並將其連線至使用 workflow.NewBuilder 的自訂執行器：

writer := agentworkflow.New(writerAgent, agentworkflow.Config{})

summarizer := workflow.NewExecutor("Summarizer", func(messages []*message.Message) string {
    return summarizeMessages(messages)
}).Bind()

wf, err := workflow.NewBuilder(writer).
    AddEdge(writer, summarizer).
    WithOutputFrom(summarizer).
    Build()
if err != nil {
    return err
}

控制代理間的上下文

NewSequentialWorkflowBuilder 使用預設的託管代理設定，每個下游代理接收前一個代理的訊息與回應訊息。 若要僅串接先前的代理回應，請設 WithChainOnlyAgentResponses(true)：

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithChainOnlyAgentResponses(true).
    Build()
if err != nil {
    return err
}

中間輸出

預設情況下， NewSequentialWorkflowBuilder 會將每位參與者的輸出作為中介工作流程輸出，並以最終訊息批次作為終端輸出。 要明確選擇你想要的參與者輸出，請結合 WithIntermediateOutputFrom 和 WithOutputFrom：

wf, err := agentworkflow.NewSequentialWorkflowBuilder(frenchAgent, spanishAgent, englishAgent).
    WithIntermediateOutputFrom(frenchAgent, spanishAgent).
    WithOutputFrom(englishAgent).
    Build()
if err != nil {
    return err
}

用 OutputEvent.IsIntermediate() 來區分中間參與者輸出與終端輸出。

關鍵概念

• 順序處理：每個代理或執行者依序處理前一步的輸出。
• agentworkflow.NewSequentialWorkflowBuilder()：從一組代理程式建立管線工作流程。
• 託管代理：agentworkflow.New揭露代理設定選項，包括訊息轉發、角色重新指派、更新事件及請求攔截。
• 自訂執行者：手動 workflow.NewBuilder 管線可混合託管代理與確定性執行者。
• 工具核准：核准所需的工具會暫停工作流程並輸出 RequestInfoEvent 包含 ToolApprovalRequestContent的值。
• 中間輸出：WithIntermediateOutputFrom 會以 workflow.OutputTagIntermediate 標記所選參與者的輸出。