Microsoft Agent Framework İş Akışları Düzenlemeleri - Eşzamanlı

Eşzamanlı düzenleme, birden çok aracının aynı görev üzerinde paralel olarak çalışmasını sağlar. Her aracı girişi bağımsız olarak işler ve sonuçları toplanır, ardından birleştirilir. Bu yaklaşım, beyin fırtınası, grup akıl yürütme veya oylama sistemleri gibi farklı perspektiflerin veya çözümlerin değerli olduğu senaryolar için uygundur.

Öğrenecekler

• Farklı uzmanlığa sahip birden çok aracı tanımlama
• Bu aracıları tek bir görevde eşzamanlı çalışacak şekilde düzenleme
• Sonuçları toplama ve işleme

Eşzamanlı düzenlemede, birden çok etmen aynı görev üzerinde aynı anda ve bağımsız olarak çalışır ve aynı veri üzerinde farklı perspektifler sağlar.

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

Uyarı

DefaultAzureCredential geliştirme için uygundur ancak üretimde dikkatli bir şekilde dikkate alınması gerekir. Üretimde gecikme sorunları, istenmeyen kimlik bilgisi yoklama ve geri dönüş mekanizmalarından kaynaklanan olası güvenlik risklerini önlemek için belirli bir kimlik bilgisi (ör ManagedIdentityCredential. ) kullanmayı göz önünde bulundurun.

Ajanlarınızı Tanımlayın

Aynı görevde aynı anda çalışacak birden çok özel aracı 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 concurrent processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Eşzamanlı Düzenlemeyi Ayarlama

Aracıları paralel olarak çalıştırmak için kullanarak AgentWorkflowBuilder iş akışını oluşturun:

// 3) Build concurrent workflow
var workflow = AgentWorkflowBuilder.BuildConcurrent(translationAgents);

Eşzamanlı İş Akışını Çalıştırma ve Sonuçları Toplama

Aynı anda çalışan tüm aracılardan iş akışını yürüt 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));

List<ChatMessage> result = new();
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
    {
        Console.WriteLine($"{e.ExecutorId}: {e.Update.Text}");
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = outputEvt.As<List<ChatMessage>>()!;
        break;
    }
}

// Display aggregated results from all agents
Console.WriteLine("===== Final Aggregated Results =====");
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

Örnek Çıkış

French_Agent: English detected. Bonjour, le monde !
Spanish_Agent: English detected. ¡Hola, mundo!
English_Agent: English detected. Hello, world!

===== Final Aggregated Results =====
User: Hello, world!
Assistant: English detected. Bonjour, le monde !
Assistant: English detected. ¡Hola, mundo!
Assistant: English detected. Hello, world!

Önemli Kavramlar

• Paralel Yürütme: Tüm aracılar girişi aynı anda ve bağımsız olarak işler
• AgentWorkflowBuilder.BuildConcurrent(): Aracı koleksiyonundan eşzamanlı iş akışı oluşturur
• Otomatik Toplama: Tüm aracılardan elde edilen sonuçlar otomatik olarak nihai sonucta toplanır
• Olay Akışı: Ajanın ilerleme durumunu gerçek zamanlı izleme AgentResponseUpdateEvent
• Farklı Bakış Açıları: Her temsilci benzersiz uzmanlığını aynı soruna getirir

Ajanlar, görevleri işleyebilen özel varlıklardır. Aşağıdaki kod üç aracıyı tanımlar: araştırma uzmanı, pazarlama uzmanı ve hukuk uzmanı.

import os

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

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

researcher = chat_client.as_agent(
    instructions=(
        "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
        " opportunities, and risks."
    ),
    name="researcher",
)

marketer = chat_client.as_agent(
    instructions=(
        "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
        " aligned to the prompt."
    ),
    name="marketer",
)

legal = chat_client.as_agent(
    instructions=(
        "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns"
        " based on the prompt."
    ),
    name="legal",
)

Eşzamanlı Düzenlemeyi Ayarlama

sınıfı, ConcurrentBuilder birden çok aracıyı paralel olarak çalıştırmak için bir iş akışı oluşturmanıza olanak tanır. Aracıların listesini katılımcı olarak geçirirsiniz.

from agent_framework.orchestrations import ConcurrentBuilder

# 2) Build a concurrent workflow
# Participants are either Agents (type of SupportsAgentRun) or Executors
workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

Eşzamanlı İş Akışını Çalıştırma ve Sonuçları Toplama

Varsayılan toplayıcı, katılımcı başına bir asistan iletisi içeren bir AgentResponse oluşturur.

from agent_framework import AgentResponse

# 3) Run with a single prompt and print the aggregated agent responses
events = await workflow.run("We are launching a new budget-friendly electric bike for urban commuters.")
outputs = events.get_outputs()

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

Örnek Çıkış

===== Final Aggregated Results =====
------------------------------------------------------------

[researcher]:
**Insights:**

- **Target Demographic:** Urban commuters seeking affordable, eco-friendly transport;
    likely to include students, young professionals, and price-sensitive urban residents.
- **Market Trends:** E-bike sales are growing globally, with increasing urbanization,
    higher fuel costs, and sustainability concerns driving adoption.
...
------------------------------------------------------------

[marketer]:
**Value Proposition:**
"Empowering your city commute: Our new electric bike combines affordability, reliability, and
    sustainable design—helping you conquer urban journeys without breaking the bank."
...
------------------------------------------------------------

[legal]:
**Constraints, Disclaimers, & Policy Concerns for Launching a Budget-Friendly Electric Bike for Urban Commuters:**

**1. Regulatory Compliance**
- Verify that the electric bike meets all applicable federal, state, and local regulations
    regarding e-bike classification, speed limits, power output, and safety features.

Gelişmiş: Özel Ajan Yöneticileri

Eşzamanlı düzenleme, aracıları ek mantıkla sarmalayan özel yürütücüleri destekler. Aracıların nasıl başlatıldığı ve istekleri nasıl işledikleri üzerinde daha fazla denetime ihtiyacınız olduğunda bu yararlı olur:

Özel Aracı Yürütücüleri Tanımlayın

from agent_framework import (
    AgentExecutorRequest,
    AgentExecutorResponse,
    Agent,
    Executor,
    WorkflowContext,
    handler,
)

class ResearcherExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "researcher"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
                " opportunities, and risks."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

class MarketerExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "marketer"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
                " aligned to the prompt."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

Özel Yürütücülerle İş Akışı Oluşturma

chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher = ResearcherExec(chat_client)
marketer = MarketerExec(chat_client)
legal = LegalExec(chat_client)

workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

Gelişmiş: Özel Toplayıcı

Varsayılan olarak, eşzamanlı orkestrasyon tüm ajan yanıtlarını her katılımcı için bir yardımcı mesajı ile tek bir AgentResponse içinde toplar. Bu davranışı, sonuçları belirli bir şekilde işleyen özel bir toplayıcıyla geçersiz kılabilirsiniz:

Özel Toplayıcı Tanımlama

from agent_framework import AgentExecutorResponse

# Create a summarizer agent for the aggregator
summarizer_agent = chat_client.as_agent(
    instructions=(
        "You are a helpful assistant that consolidates multiple domain expert outputs "
        "into one cohesive, concise summary with clear takeaways. Keep it under 200 words."
    ),
    name="summarizer",
)

# Define a custom aggregator callback
async def summarize_results(results: list[AgentExecutorResponse]) -> str:
    # Extract one final assistant message per agent
    expert_sections: list[str] = []
    for r in results:
        try:
            messages = getattr(r.agent_response, "messages", [])
            final_text = messages[-1].text if messages and hasattr(messages[-1], "text") else "(no content)"
            expert_sections.append(f"{r.executor_id}:\n{final_text}")
        except Exception as e:
            expert_sections.append(f"{r.executor_id}: (error: {type(e).__name__}: {e})")

    # Ask the model to synthesize a concise summary of the experts' outputs
    prompt = "\n\n".join(expert_sections)
    response = await summarizer_agent.run(prompt)
    # Return the model's final assistant text as the completion result
    return response.messages[-1].text if response.messages else ""

Özel Toplayıcı ile İş Akışı Oluşturma

workflow = (
    ConcurrentBuilder(participants=[researcher, marketer, legal])
    .with_aggregator(summarize_results)
    .build()
)

output = None
async for event in workflow.run("We are launching a new budget-friendly electric bike for urban commuters.", stream=True):
    if event.type == "output":
        output = event.data

if output:
    print("===== Final Consolidated Output =====")
    print(output)

Özel Toplayıcı ile Örnek Çıkış

===== Final Consolidated Output =====
Urban e-bike demand is rising rapidly due to eco-awareness, urban congestion, and high fuel costs,
with market growth projected at a ~10% CAGR through 2030. Key customer concerns are affordability,
easy maintenance, convenient charging, compact design, and theft protection. Differentiation opportunities
include integrating smart features (GPS, app connectivity), offering subscription or leasing options, and
developing portable, space-saving designs. Partnering with local governments and bike shops can boost visibility.

Risks include price wars eroding margins, regulatory hurdles, battery quality concerns, and heightened expectations
for after-sales support. Accurate, substantiated product claims and transparent marketing (with range disclaimers)
are essential. All e-bikes must comply with local and federal regulations on speed, wattage, safety certification,
and labeling. Clear warranty, safety instructions (especially regarding batteries), and inclusive, accessible
marketing are required. For connected features, data privacy policies and user consents are mandatory.

Effective messaging should target young professionals, students, eco-conscious commuters, and first-time buyers,
emphasizing affordability, convenience, and sustainability. Slogan suggestion: "Charge Ahead—City Commutes Made
Affordable." Legal review in each target market, compliance vetting, and robust customer support policies are
critical before launch.

Ara Çıkışlar

Varsayılan olarak, yalnızca toplayıcının çıkışı iş akışı output olayı olarak görünür. intermediate_outputs=True öğesini, her katılımcının bireysel çıktısını da gösterecek şekilde ayarlayın.

workflow = ConcurrentBuilder(
    participants=[researcher, marketer, legal],
    intermediate_outputs=True,
).build()

Bu olayları akış modunda gerçek zamanlı olarak 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("Analyze our new product launch strategy.", stream=True):
    if event.type == "output" 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
            print(f"{author}: {update.text}", end="", flush=True)
            last_author = author
        else:
            print(update.text, end="", flush=True)

Önemli Kavramlar

• Paralel Yürütme: Tüm aracılar görev üzerinde aynı anda ve bağımsız olarak çalışır
• AgentResponse Çıktısı: Varsayılan toplayıcı, katılımcı başına tek bir AgentResponse yardımcı iletisi verir (kullanıcı istemi dahil değildir)
• Farklı Bakış Açıları: Her temsilci benzersiz uzmanlığını aynı soruna getirir
• Esnek Katılımcılar: Ajanları doğrudan kullanabilir veya özel yürütücülerde sarmalayabilirsiniz
• Özel İşleme: Sonuçları etki alanına özgü yollarla sentezlemek için varsayılan toplayıcıyı geçersiz kılma
• Ara Çıkışlar: Toplayıcının son çıkışına ek olarak her katılımcının tek tek çıkışını ortaya çıkarmak için ayarlayın intermediate_outputs=True

Sonraki Adımlar

Sıralı Düzenleme