Microsoft Agent Framework İş Akışları Düzenlemeleri - Sıralı

Sıralı düzenlemede aracılar, her aracının sırayla görevi işlediği bir işlem hattında düzenlenir ve çıktıyı dizideki bir sonraki aracıya geçirir.

Azure OpenAI İstemcisini Ayarlama

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);

Sırayla çalışacak özel aracılar oluşturun:

// 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));

Sıralı Orkestrasyonu Kurma

kullanarak AgentWorkflowBuilderiş akışını oluşturun:

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

Sıralı İş Akışını Çalıştırma

İş akışını uygulayın ve olayları işleyin.

// 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}");
}

Örnek Çıkış

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!

İnsan Katılımıyla Sıralı Orkestrasyon

Sıralı düzenleme, araç onayı aracılığıyla döngüdeki insan etkileşimlerini destekler. Aracılar, ApprovalRequiredAIFunction ile sarmalanmış araçları kullandığında, iş akışı durdurulur ve bir RequestInfoEvent içeren ToolApprovalRequestContent çıkış sağlar. Dış sistemler (insan operatörü gibi) araç çağrısını inceleyebilir, onaylayabilir veya reddedebilir ve iş akışı buna göre devam eder.

Onay Gerektiren Araçlarla Ajanları Tanımlama

Hassas araçların ApprovalRequiredAIFunction ile sarmalandığı aracılar oluşturun.

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");

Onay İşleme ile Derleme ve Çalıştırma

Sıralı iş akışını normal şekilde oluşturun. Onay akışı olay akışı üzerinden işlenir:

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)));
    }
}

Önemli Kavramlar

• Sıralı İşleme: Her bir temsilci, önceki temsilcinin çıkışını sırayla işler
• AgentWorkflowBuilder.BuildSequential(): Bir dizi aracı kullanarak bir iş akışı oluşturur
• ChatClientAgent: Belirli yönergelerle sohbet istemcisi tarafından desteklenen bir aracıyı temsil eder
• InProcessExecution.RunStreamingAsync(): İş akışını çalıştırır ve gerçek zamanlı olay akışı için bir StreamingRun döndürür
• Olay İşleme: Aracının ilerlemesini AgentResponseUpdateEvent, tamamlanmasını WorkflowOutputEvent izleme
• Araç Onayı: Yürütmeden önce insan onayı istemek için hassas araçları ApprovalRequiredAIFunction ile sarmak
• RequestInfoEvent: Bir araç onay gerektirdiğinde ortaya çıkar; araç arama ayrıntılarıyla birlikte içerir ToolApprovalRequestContent

Sıralı orkestrasyonda, her aracı görevi sırayla yönetir ve çıkış, birinden diğerine akarak devam eder. İki aşamalı bir işlem için aracıları tanımlayarak başlayın:

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",
)

Sıralı Orkestrasyonu Kurma

Bu SequentialBuilder sınıfı, etmenlerin görevleri sırayla işlediği bir işlem hattı oluşturur. Her temsilci konuşma geçmişinin tamamını görür ve yanıtını ekler.

from agent_framework.orchestrations import SequentialBuilder

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

Sıralı İş Akışını Çalıştırma

İş akışını yürüt ve son çıktıyı topla. Terminal çıkışı, son aracının yanıt iletilerini içeren bir AgentResponse çıktıdır:

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}")

Örnek Çıkış

===== 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!

Gelişmiş: Özel Yürütücülerle Ajanların Birleştirilmesi

Sıralı orkestrasyon, özel işlem gerektiren durumlar için aracıların özel yürütücülerle karıştırılmasını destekler. LLM gerektirmeyen özel bir mantığa ihtiyacınız olduğunda bu yararlı olur:

Özel Yürütücü Tanımlama

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]))

Karma Sıralı İş Akışı Oluşturma

# 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()

Özel Yürütücü ile Örnek Çıktı

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

Ajanlar Arasındaki Bağlamı Denetleme

Varsayılan olarak, bir SequentialBuilder iş akışındaki her bir ajan, önceki ajanın tam konuşmasını (giriş + yanıt iletileri) kullanır. Ayarı chain_only_agent_responses=True , dizideki tüm aracıları yalnızca önceki aracının yanıt iletilerini kullanacak şekilde yapılandırıyor:

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

Bu, çeviri işlem hatları, aşamalı iyileştirme ve her bir aracın, öncesindeki aracın çıktılarını dönüştürmeye odaklanırken, önceki konuşma döngülerinden etkilenmeden çalışması gereken diğer senaryolar için faydalıdır.

Tam bir örnek için bkz. Agent Framework deposundaki sequential_chain_only_agent_responses.py .

Ara Çıkışlar

Varsayılan olarak, SequentialBuilderson katılımcıyı terminal çıkış kaynağı (output_from ) olarak seçer. Yalnızca o katılımcının çıktısı, bir "output" olayı olarak görünür.

Önceki katılımcıların çıktılarını da görüntülemek için, ara kaynak olarak belirlemek istediğiniz katılımcılarla birlikte intermediate_output_from iletin. Bu, bu katılımcıları varsayılan final kümesinden örtük olarak çıkarır — "intermediate" olayları yerine "output" olayları üretirler:

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

Hem "intermediate" hem de "output" olaylarını akış modunda gerçek zamanlı işleyebilirsiniz:

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)

İnsan Katılımıyla Sıralı Orkestrasyon

Sıralı orkestrasyonlar, insan girdisi olan etkileşimleri iki şekilde destekler: hassas araç çağrılarını denetlemek için araç onayı ve her bir aracı yanıtından sonra duraklatıp geri bildirim toplamak için bilgi isteme.

Sıralı İş Akışlarında Araç Onayı

Yürütmeden önce insan onayı gerektiren araçları işaretlemek için kullanın @tool(approval_mode="always_require") . Ajan aracı çağırdığında iş akışı durur ve bir request_info olayı yayar.

@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()

Olay akışını işleme ve onay isteklerini işleme:

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)

Aracı Geri Bildirimi için Bilgi İste

Belirli aracılar yanıt verdikten sonra duraklatmak için .with_request_info() kullanarak bir sonraki aracı başlamadan önce dış girdiye (insan incelemesi gibi) izin verin.

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)

Önemli Kavramlar

• Paylaşılan Bağlam: Varsayılan olarak, her bir aracı, giriş ve yanıt iletileri dahil olmak üzere önceki aracın tam konuşmasını tüketir
• Bağlam Denetimi: Aracıları yalnızca önceki aracının yanıt iletilerini kullanacak şekilde yapılandırmak için kullanın chain_only_agent_responses=True
• AgentResponse Çıktısı: İş akışının terminal çıkışı, son aracının yanıtını içeren bir AgentResponse çıktıdır (konuşmanın tamamını değil)
• Order Matters: Ajanlar kesinlikle listede participants belirtilen sırayla yürütülür
• Esnek Katılımcılar: Aracıları ve özel yürütücüleri istediğiniz sırada karıştırabilirsiniz
• Özel Sonlandırıcı Sözleşmesi: Son katılımcı olarak kullanılan özel yürütücü, terminal çıkışını üretmek için ctx.yield_output(AgentResponse(...)) çağırmalıdır.
• Ara Çıktılar: Katılımcıların ilerlemesini yalnızca son katılımcının nihai çıktısı olarak değil, ara iş akışı olayları olarak görüntülemek için intermediate_output_from=[...] veya intermediate_output_from="all_other" kullanın
• Araç Onayı: İnsan incelemesi gerektiren hassas işlemler için kullanın @tool(approval_mode="always_require")
• Bilgi Talebi: Dış geri bildirim almak için belirli aracıların ardından .with_request_info(agents=[...]) ile duraklatma yapın

Go, ile workflow/agentworkflowsıralı aracı iş akışları oluşturabilir. NewSequentialWorkflowBuilder her aracıyı iş akışı yürütücüsü olarak barındırır, sırasıyla bağlar ve son ileti toplu işlemini iş akışı çıktısı olarak verir.

Dökümhane Yapılandırmasını Ayarlama

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 aracılarınızı tanımlama

Sırayla çalışacak özel aracılar oluşturun:

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")

Sıralı Orkestrasyonu Kurma

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

Sıralı İş Akışını Çalıştırma

İş akışını yürütür ve çıkış olaylarını işler:

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)
    }
}

Örnek Çıkış

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

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

İnsan Katılımıyla Sıralı Orkestrasyon

Barındırılan aracı onay gerektiren bir araç kullandığında, sıralı iş akışları araç onayı beklenirken duraklatılabilir. Aracı tool.ApprovalRequiredFunc ile sarın, ardından workflow.RequestInfoEvent bekleyin ve ToolApprovalResponseContent ile yanıt verin.

Onay Gerektiren Araçlarla Ajanları Tanımlama

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
}

Onay İşleme ile Derleme ve Çalıştırma

Olay akışında onay isteklerini işleme:

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
    }
}

Gelişmiş: Özel Yürütücülerle Ajanların Birleştirilmesi

Karma işlem hatlarında, aracıları agentworkflow.New ile barındırın ve bunları workflow.NewBuilder ile özel yürütücülere bağlayın:

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
}

Ajanlar Arasındaki Bağlamı Denetleme

NewSequentialWorkflowBuilder varsayılan barındırılan aracı yapılandırmasını kullanır ve burada her aşağı akış aracısı önceki aracının gelen iletilerini ve yanıt iletilerini alır. Yalnızca önceki ajan yanıtlarını zincirlemek için WithChainOnlyAgentResponses(true) değerini ayarlayın:

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

Ara Çıkışlar

Varsayılan olarak, NewSequentialWorkflowBuilder her katılımcının çıktısını ara iş akışı çıktısı olarak üretir ve nihai ileti grubunu nihai çıktı olarak üretir. İstediğiniz katılımcı çıkışlarını açıkça seçmek için WithIntermediateOutputFrom ve WithOutputFrom öğelerini birleştirin:

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

Ara katılımcı çıkışlarını terminal çıkışlarından ayırmak için kullanın OutputEvent.IsIntermediate() .

Önemli Kavramlar

• Sıralı İşleme: Her aracı veya yürütücü, önceki adımın çıkışını sırayla işler.
• agentworkflow. NewSequentialWorkflowBuilder(): Aracı koleksiyonundan bir işlem hattı iş akışı oluşturur.
• Barındırılan Aracılar: agentworkflow.New İleti iletme, rol yeniden atama, güncelleştirme olayları ve istek kesme işlemleri için aracı yapılandırma seçeneklerini kullanıma sunar.
• Özel Yürütücüler: Manuel workflow.NewBuilder işlem hatları, barındırılan ajanlar ile deterministik yürütücüleri bir arada kullanabilir.
• Araç Onayı: Onay gerektiren araçlar iş akışını duraklatır ve ToolApprovalRequestContent içeren RequestInfoEvent değerleri üretir.
• Ara Çıkışlar: WithIntermediateOutputFrom, seçili katılımcı çıkışlarını workflow.OutputTagIntermediate ile işaretler.