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

요구 사항

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

워크플로 에이전트 만들기

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

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

// Create agents
AIAgent researchAgent = chatClient.AsAIAgent("You are a researcher. Research and gather information on the given topic.");
AIAgent writerAgent = chatClient.AsAIAgent("You are a writer. Write clear, engaging content based on research.");
AIAgent reviewerAgent = chatClient.AsAIAgent("You are a reviewer. Review the content and provide a final polished version.");

// Build a sequential workflow
var workflow = new WorkflowBuilder(researchAgent)
    .AddEdge(researchAgent, writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

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

AsAIAgent 매개 변수

워크플로 에이전트 사용

세션 만들기

워크플로 에이전트와의 각 대화에는 상태를 관리하는 세션이 필요합니다.

// Create a new session for the conversation
AgentSession session = await workflowAgent.CreateSessionAsync();

스트리밍이 아닌 실행

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

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

AgentResponse response = await workflowAgent.RunAsync(messages, session);

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 (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // Process streaming updates from each agent in the workflow
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

외부 입력 요청 처리

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

await foreach (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // 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 session state
JsonElement serializedSession = await workflowAgent.SerializeSessionAsync(session);

// Store serializedSession to your persistence layer...

// Later, resume the session
AgentSession resumedSession = await workflowAgent.DeserializeSessionAsync(serializedSession);

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

요구 사항

워크플로를 에이전트로 사용하려면 워크플로의 시작 실행기가 메시지 입력을 처리할 수 있어야 합니다. Agent 또는 에이전트 기반 실행기를 사용할 때 자동으로 충족됩니다.

워크플로 에이전트 만들기

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

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

# Create your chat client and agents
client = FoundryChatClient(
    project_endpoint="<your-endpoint>",
    model="<your-deployment>",
    credential=AzureCliCredential(),
)

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

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

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

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

as_agent 매개 변수

워크플로 에이전트 사용

세션 만들기

필요에 따라 세션을 만들어 여러 번 대화 상태를 관리할 수 있습니다.

# Create a new session for the conversation
session = await workflow_agent.create_session()

스트리밍이 아닌 실행

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

# You can pass a plain string as input
response = await workflow_agent.run("Write an article about AI trends")

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

스트리밍 실행

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

async for update in workflow_agent.run(
    "Write an article about AI trends",
    stream=True,
):
    if update.text:
        print(update.text, end="", flush=True)

외부 입력 요청 처리

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

from agent_framework import Content, Message, WorkflowAgent

response = await workflow_agent.run("Process my request")

# Look for function calls in the response
human_review_function_call = None
for message in response.messages:
    for content in message.contents:
        if content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
            human_review_function_call = content

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

외부 입력 요청 후 워크플로 실행을 계속하려면 함수 결과를 만들고 다시 보냅니다.

if human_review_function_call:
    # Parse the request arguments
    request = WorkflowAgent.RequestInfoFunctionArgs.from_json(
        human_review_function_call.arguments
    )

    # Create a response (your custom response type)
    result_data = MyResponseType(approved=True, feedback="Looks good")

    # Create the function call result
    function_result = Content.from_function_result(
        call_id=human_review_function_call.call_id,
        result=result_data,
    )

    # Send the response back to continue the workflow
    response = await workflow_agent.run(Message("tool", [function_result]))

완성된 예시

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

import asyncio
import os

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


async def main():
    # Set up the chat client
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

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

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

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

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

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

    # Run the workflow
    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run(
        "Write about quantum computing",
        stream=True,
    ):
        # 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()하는 방법에 따라 달라집니다.

• run(): 워크플로가 완료된 후 전체 결과를 포함하는 AgentResponse을 반환
• run(..., stream=True): 워크플로가 실행되면 개체의 AgentResponseUpdate 비동기 반복 가능 항목을 반환하여 실시간 업데이트를 제공합니다.

as_agent()는 "output"(종결) 및 "intermediate" 이벤트를 호출자에게 모두 전달합니다. 전달된 이벤트 유형 집합은 .입니다 AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}. 다른 모든 워크플로 내부 이벤트가 삭제됩니다.

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

이 변환을 사용하면 필요할 때 자세한 워크플로 정보에 액세스할 수 있는 동시에 표준 에이전트 인터페이스를 사용할 수 있습니다. .text 및 AgentResponse의 AgentResponseUpdate 속성은 최종("output") 응답만 반환합니다. 중간 진행 상황에 액세스하려면 text_reasoning 콘텐츠 항목을 검사하세요.

Go는 워크플로를 workflow/agentworkflow 에이전트로 감쌉니다. 이렇게 하면 공급자가 백그라운드에서 워크플로를 실행하는 동안 호출자가 일반 에이전트 실행 API를 사용할 수 있습니다.

요구 사항

워크플로의 시작 실행기는 []*message.Message를 허용해야 합니다. 호스팅된 에이전트 실행기와 messageworkflow.Configure로 구성된 실행기는 이 요구 사항을 충족합니다.

워크플로 에이전트 만들기

호환되는 워크플로를 에이전트로 래핑하는 데 사용합니다 agentworkflow.New .

wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
    IncludeOutputsInResponse: true,
    Config: agent.Config{
        Name: "WorkflowAgent",
    },
})
if err != nil {
    return err
}

agentworkflow. AgentConfig 매개 변수

워크플로 에이전트 사용

세션 만들기

워크플로 상태가 차례로 유지되도록 하려면 에이전트 세션을 만듭니다.

session, err := wfAgent.CreateSession(ctx)
if err != nil {
    return err
}

스트리밍이 아닌 실행

비스트리밍 실행의 경우 RunText 또는 Run를 사용하고 응답을 수집합니다:

response, err := wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session)).Collect()
if err != nil {
    return err
}
fmt.Println(response.String())

스트리밍 실행

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

for update, err := range wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

외부 입력 요청 처리

워크플로의 외부 요청은 에이전트 응답에서 함수 호출 콘텐츠로 표시됩니다. 요청 콘텐츠에 대한 응답 메시지를 검사하고 나중에 실행에서 일치하는 응답을 보냅니다.

var requestCall *message.FunctionCallContent
for content := range response.Contents() {
    if call, ok := content.(*message.FunctionCallContent); ok {
        requestCall = call
        break
    }
}

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

워크플로 실행을 계속하려면 일치하는 응답 콘텐츠를 워크플로 에이전트에 반환합니다.

result := &message.FunctionResultContent{
    CallID: requestCall.CallID,
    Result: "approved",
}

response, err = wfAgent.Run(
    ctx,
    []*message.Message{{
        Role:     message.RoleTool,
        Contents: []message.Content{result},
    }},
    agent.WithSession(session),
).Collect()
if err != nil {
    return err
}

세션 직렬화 및 재개

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

// Serialize the session state.
serializedSession, err := json.Marshal(session)
if err != nil {
    return err
}

// Store serializedSession to your persistence layer...

// Later, resume the session.
var resumedSession agent.Session
if err := json.Unmarshal(serializedSession, &resumedSession); err != nil {
    return err
}

for update, err := range wfAgent.RunText(ctx, "Continue the article", agent.WithSession(&resumedSession), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

완성된 예시

다음 예제에서는 콘텐츠 파이프라인 워크플로를 빌드하고, 에이전트로 래핑하고, 일반 에이전트 API를 통해 응답을 스트리밍합니다.

package main

import (
    "cmp"
    "context"
    "fmt"
    "log"
    "os"

    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
)

func main() {
    ctx := context.Background()
    endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
    model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

    credential, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        log.Fatal(err)
    }

    researcher := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Research and gather information on the given topic.",
        Config:      agent.Config{Name: "Researcher"},
    })
    writer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Write clear, engaging content based on research.",
        Config:      agent.Config{Name: "Writer"},
    })
    reviewer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Review the content and provide a final polished version.",
        Config:      agent.Config{Name: "Reviewer"},
    })

    wf, err := agentworkflow.NewSequentialWorkflowBuilder(researcher, writer, reviewer).
        WithName("content-pipeline").
        Build()
    if err != nil {
        log.Fatal(err)
    }

    wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
        IncludeOutputsInResponse: true,
        Config: agent.Config{
            Name: "Content Pipeline Agent",
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    session, err := wfAgent.CreateSession(ctx)
    if err != nil {
        log.Fatal(err)
    }

    for update, err := range wfAgent.RunText(ctx, "Write about quantum computing", agent.WithSession(session), agent.Stream(true)) {
        if err != nil {
            log.Fatal(err)
        }
        if text := update.String(); text != "" {
            fmt.Print(text)
        }
    }
}