Microsoft 에이전트 프레임워크 워크플로 - 워크플로를 에이전트로 사용

요구 사항

워크플로를 에이전트로 사용하려면 워크플로의 시작 실행기가 입력으로 처리 IEnumerable<ChatMessage> 할 수 있어야 합니다. ChatClientAgent 또는 다른 에이전트 기반 실행기를 사용할 때 이 조건은 자동으로 충족됩니다.

워크플로 에이전트 만들기

확장 메서드를 AsAgent() 사용하여 호환되는 워크플로를 에이전트로 변환합니다.

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

// First, build your workflow
var workflow = AgentWorkflowBuilder
    .CreateSequentialPipeline(researchAgent, writerAgent, reviewerAgent)
    .Build();

// Convert the workflow to an agent
AIAgent workflowAgent = workflow.AsAgent(
    id: "content-pipeline",
    name: "Content Pipeline Agent",
    description: "A multi-agent workflow that researches, writes, and reviews content"
);

AsAgent 매개 변수

워크플로 에이전트 사용

스레드 만들기

워크플로 에이전트와의 각 대화에서 상태를 관리하려면 스레드가 필요합니다.

// Create a new thread for the conversation
AgentThread thread = workflowAgent.GetNewThread();

스트리밍이 아닌 실행

전체 응답을 원하는 간단한 사용 사례의 경우:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

AgentRunResponse response = await workflowAgent.RunAsync(messages, thread);

foreach (ChatMessage message in response.Messages)
{
    Console.WriteLine($"{message.AuthorName}: {message.Text}");
}

스트리밍 실행

워크플로가 실행됨에 따라 실시간 업데이트의 경우:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Process streaming updates from each agent in the workflow
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

외부 입력 요청 처리

워크플로에 외부 입력(사용 RequestInfoExecutor)을 요청하는 실행기가 포함된 경우 이러한 요청은 에이전트 응답에서 함수 호출로 표시됩니다.

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Check for function call requests
    foreach (AIContent content in update.Contents)
    {
        if (content is FunctionCallContent functionCall)
        {
            // Handle the external input request
            Console.WriteLine($"Workflow requests input: {functionCall.Name}");
            Console.WriteLine($"Request data: {functionCall.Arguments}");

            // Provide the response in the next message
        }
    }
}

스레드 직렬화 및 재개

워크플로 에이전트 스레드는 지속성을 위해 직렬화되고 나중에 다시 시작될 수 있습니다.

// Serialize the thread state
JsonElement serializedThread = thread.Serialize();

// Store serializedThread to your persistence layer...

// Later, resume the thread
AgentThread resumedThread = workflowAgent.DeserializeThread(serializedThread);

// Continue the conversation
await foreach (var update in workflowAgent.RunStreamingAsync(newMessages, resumedThread))
{
    Console.Write(update.Text);
}

워크플로 에이전트를 사용하여 검사점 지정

검사점을 사용하도록 설정하여 프로세스 다시 시작에서 워크플로 상태를 유지합니다.

// Create a checkpoint manager with your storage backend
var checkpointManager = new CheckpointManager(new FileCheckpointStorage("./checkpoints"));

// Create workflow agent with checkpointing enabled
AIAgent workflowAgent = workflow.AsAgent(
    id: "persistent-workflow",
    name: "Persistent Workflow Agent",
    checkpointManager: checkpointManager
);

요구 사항

워크플로를 에이전트로 사용하려면 워크플로의 시작 실행기가 입력으로 처리 list[ChatMessage] 할 수 있어야 합니다. 이를 ChatAgent나 AgentExecutor를 사용할 때 자동으로 충족합니다.

워크플로 에이전트 만들기

호환되는 워크플로를 호출 as_agent() 하여 에이전트로 변환합니다.

from agent_framework import WorkflowBuilder, ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Create your chat client and agents
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

researcher = ChatAgent(
    name="Researcher",
    instructions="Research and gather information on the given topic.",
    chat_client=chat_client,
)

writer = ChatAgent(
    name="Writer", 
    instructions="Write clear, engaging content based on research.",
    chat_client=chat_client,
)

# Build your workflow
workflow = (
    WorkflowBuilder()
    .set_start_executor(researcher)
    .add_edge(researcher, writer)
    .build()
)

# Convert the workflow to an agent
workflow_agent = workflow.as_agent(name="Content Pipeline Agent")

as_agent 매개 변수

워크플로 에이전트 사용

스레드 만들기

워크플로 에이전트와의 각 대화에서 상태를 관리하려면 스레드가 필요합니다.

# Create a new thread for the conversation
thread = workflow_agent.get_new_thread()

스트리밍이 아닌 실행

전체 응답을 원하는 간단한 사용 사례의 경우:

from agent_framework import ChatMessage, Role

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

response = await workflow_agent.run(messages, thread=thread)

for message in response.messages:
    print(f"{message.author_name}: {message.text}")

스트리밍 실행

워크플로가 실행됨에 따라 실시간 업데이트의 경우:

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

async for update in workflow_agent.run_stream(messages, thread=thread):
    # Process streaming updates from each agent in the workflow
    if update.text:
        print(update.text, end="", flush=True)

외부 입력 요청 처리

워크플로에 외부 입력(사용 RequestInfoExecutor)을 요청하는 실행기가 포함되어 있으면 이러한 요청이 함수 호출로 표시됩니다. 워크플로 에이전트는 보류 중인 요청을 추적하고 계속하기 전에 응답을 예상합니다.

from agent_framework import (
    FunctionCallContent,
    FunctionApprovalRequestContent,
    FunctionApprovalResponseContent,
)

async for update in workflow_agent.run_stream(messages, thread=thread):
    for content in update.contents:
        if isinstance(content, FunctionApprovalRequestContent):
            # The workflow is requesting external input
            request_id = content.id
            function_call = content.function_call

            print(f"Workflow requests input: {function_call.name}")
            print(f"Request data: {function_call.arguments}")

            # Store the request_id to provide a response later

# Check for pending requests
if workflow_agent.pending_requests:
    print(f"Pending requests: {list(workflow_agent.pending_requests.keys())}")

보류 중인 요청에 대한 응답 제공

외부 입력 요청 후 워크플로 실행을 계속하려면 다음을 수행합니다.

# Create a response for the pending request
response_content = FunctionApprovalResponseContent(
    id=request_id,
    function_call=function_call,
    approved=True,
)

response_message = ChatMessage(
    role=Role.USER,
    contents=[response_content],
)

# Continue the workflow with the response
async for update in workflow_agent.run_stream([response_message], thread=thread):
    if update.text:
        print(update.text, end="", flush=True)

완성된 예시

다음은 스트리밍 출력이 있는 워크플로 에이전트를 보여주는 전체 예제입니다.

import asyncio
from agent_framework import (
    ChatAgent,
    ChatMessage,
    Role,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework._workflows import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

    # Create specialized agents
    researcher = ChatAgent(
        name="Researcher",
        instructions="Research the given topic and provide key facts.",
        chat_client=chat_client,
    )

    writer = ChatAgent(
        name="Writer",
        instructions="Write engaging content based on the research provided.",
        chat_client=chat_client,
    )

    reviewer = ChatAgent(
        name="Reviewer",
        instructions="Review the content and provide a final polished version.",
        chat_client=chat_client,
    )

    # Build a sequential workflow
    workflow = (
        SequentialBuilder()
        .add_agents([researcher, writer, reviewer])
        .build()
    )

    # Convert to a workflow agent
    workflow_agent = workflow.as_agent(name="Content Creation Pipeline")

    # Create a thread and run the workflow
    thread = workflow_agent.get_new_thread()
    messages = [ChatMessage(role=Role.USER, content="Write about quantum computing")]

    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run_stream(messages, thread=thread):
        # Show when different agents are responding
        if update.author_name and update.author_name != current_author:
            if current_author:
                print("\n" + "-" * 40)
            print(f"\n[{update.author_name}]:")
            current_author = update.author_name

        if update.text:
            print(update.text, end="", flush=True)

    print("\n" + "=" * 60)
    print("Workflow completed!")


if __name__ == "__main__":
    asyncio.run(main())

이벤트 변환 이해

워크플로가 에이전트로 실행되면 워크플로 이벤트가 에이전트 응답으로 변환됩니다. 응답 유형은 사용하는 방법에 따라 달라집니다.

• run(): 워크플로가 완료된 후 전체 결과를 포함하는 AgentRunResponse을 반환
• run_stream() AgentRunResponseUpdate: 워크플로가 실행되면 개체를 생성하여 실시간 업데이트를 제공합니다.

실행하는 동안 내부 워크플로 이벤트는 다음과 같이 에이전트 응답에 매핑됩니다.

이 변환을 사용하면 필요할 때 자세한 워크플로 정보에 액세스할 수 있는 동시에 표준 에이전트 인터페이스를 사용할 수 있습니다.