순차 오케스트레이션에서 에이전트들이 파이프라인 형태로 구성됩니다. 각 에이전트는 차례로 작업을 처리하여 순서대로 다음 에이전트에 출력을 전달합니다. 이는 문서 검토, 데이터 처리 파이프라인 또는 다단계 추론과 같이 각 단계가 이전 단계를 기반으로 빌드되는 워크플로에 적합합니다.
학습 내용
- 에이전트의 순차 파이프라인을 만드는 방법
- 각 에이전트가 이전 에이전트의 출력을 기반으로 작업을 수행하도록 연결하는 방법
- 특수 작업에 대한 사용자 지정 실행기와 에이전트를 혼합하는 방법
- 파이프라인을 통해 대화 흐름을 추적하는 방법
에이전트 설정
순차 오케스트레이션에서 에이전트는 각 에이전트가 차례로 작업을 처리하고 순서대로 다음 에이전트에 출력을 전달하는 파이프라인으로 구성됩니다.
Azure OpenAI 클라이언트 설정
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.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 AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
.GetChatClient(deploymentName)
.AsIChatClient();
순서대로 작동하는 특수 에이전트를 만듭니다.
// 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!") };
StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
List<ChatMessage> result = new();
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
if (evt is AgentRunUpdateEvent e)
{
Console.WriteLine($"{e.ExecutorId}: {e.Data}");
}
else if (evt is WorkflowOutputEvent outputEvt)
{
result = (List<ChatMessage>)outputEvt.Data!;
break;
}
}
// Display final result
foreach (var message in result)
{
Console.WriteLine($"{message.Role}: {message.Content}");
}
샘플 출력
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!
주요 개념
- 순차적 처리: 각 에이전트는 이전 에이전트의 출력을 순서대로 처리합니다.
- AgentWorkflowBuilder.BuildSequential(): 에이전트 컬렉션에서 파이프라인 워크플로를 만듭니다.
- ChatClientAgent: 특정 지침이 포함된 채팅 클라이언트에서 백업하는 에이전트를 나타냅니다.
- StreamingRun: 이벤트 스트리밍 기능을 사용하여 실시간 실행 제공
-
이벤트 처리:
AgentRunUpdateEvent을 통해 에이전트 진행을 모니터링하고,WorkflowOutputEvent을 통해 완료 확인
순차 오케스트레이션에서 각 에이전트는 차례로 작업을 처리하고 출력은 한 쪽에서 다음으로 흐릅니다. 먼저 2단계 프로세스에 대한 에이전트를 정의해 보겠습니다.
from agent_framework.azure import AzureChatClient
from azure.identity import AzureCliCredential
# 1) Create agents using AzureChatClient
chat_client = AzureChatClient(credential=AzureCliCredential())
writer = chat_client.create_agent(
instructions=(
"You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."
),
name="writer",
)
reviewer = chat_client.create_agent(
instructions=(
"You are a thoughtful reviewer. Give brief feedback on the previous assistant message."
),
name="reviewer",
)
순차 오케스트레이션 설정
이 클래스는 SequentialBuilder 에이전트가 작업을 순서대로 처리하는 파이프라인을 만듭니다. 각 에이전트는 전체 대화 기록을 보고 응답을 추가합니다.
from agent_framework import SequentialBuilder
# 2) Build sequential workflow: writer -> reviewer
workflow = SequentialBuilder().participants([writer, reviewer]).build()
순차 워크플로 실행
워크플로를 실행하고 각 에이전트의 기여를 보여 주는 최종 대화를 수집합니다.
from agent_framework import ChatMessage, WorkflowOutputEvent
# 3) Run and print final conversation
output_evt: WorkflowOutputEvent | None = None
async for event in workflow.run_stream("Write a tagline for a budget-friendly eBike."):
if isinstance(event, WorkflowOutputEvent):
output_evt = event
if output_evt:
print("===== Final Conversation =====")
messages: list[ChatMessage] | Any = output_evt.data
for i, msg in enumerate(messages, start=1):
name = msg.author_name or ("assistant" if msg.role == Role.ASSISTANT else "user")
print(f"{'-' * 60}\n{i:02d} [{name}]\n{msg.text}")
샘플 출력
===== Final Conversation =====
------------------------------------------------------------
01 [user]
Write a tagline for a budget-friendly eBike.
------------------------------------------------------------
02 [writer]
Ride farther, spend less—your affordable eBike adventure starts here.
------------------------------------------------------------
03 [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 Executor, WorkflowContext, handler
from agent_framework import ChatMessage, Role
class Summarizer(Executor):
"""Simple summarizer: consumes full conversation and appends an assistant summary."""
@handler
async def summarize(
self,
conversation: list[ChatMessage],
ctx: WorkflowContext[list[ChatMessage]]
) -> None:
users = sum(1 for m in conversation if m.role == Role.USER)
assistants = sum(1 for m in conversation if m.role == Role.ASSISTANT)
summary = ChatMessage(
role=Role.ASSISTANT,
text=f"Summary -> users:{users} assistants:{assistants}"
)
await ctx.send_message(list(conversation) + [summary])
혼합 순차 워크플로 빌드
# Create a content agent
content = chat_client.create_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()
사용자 지정 실행기를 사용한 샘플 출력
------------------------------------------------------------
01 [user]
Explain the benefits of budget eBikes for commuters.
------------------------------------------------------------
02 [content]
Budget eBikes offer commuters an affordable, eco-friendly alternative to cars and public transport.
Their electric assistance reduces physical strain and allows riders to cover longer distances quickly,
minimizing travel time and fatigue. Budget models are low-cost to maintain and operate, making them accessible
for a wider range of people. Additionally, eBikes help reduce traffic congestion and carbon emissions,
supporting greener urban environments. Overall, budget eBikes provide cost-effective, efficient, and
sustainable transportation for daily commuting needs.
------------------------------------------------------------
03 [assistant]
Summary -> users:1 assistants:1
주요 개념
- 공유 컨텍스트: 각 참가자는 모든 이전 메시지를 포함하여 전체 대화 기록을 받습니다.
-
주문 문제: 에이전트는 목록에 지정된
participants()순서로 엄격하게 실행됩니다. - 유연한 참가자: 에이전트와 사용자 지정 실행기를 순서대로 혼합할 수 있습니다.
- 대화 흐름: 각 에이전트/실행기가 대화에 추가하여 완전한 대화를 구축합니다.