다음을 통해 공유

Microsoft 에이전트 프레임워크 워크플로 - 상태

이 문서에서는 Microsoft 에이전트 프레임워크 워크플로 시스템의 상태에 대한 개요를 제공합니다.

개요

상태를 사용하면 워크플로 내의 여러 실행기가 공통 데이터에 액세스하고 수정할 수 있습니다. 이 기능은 직접 메시지 전달이 가능하거나 효율적이지 않은 경우 워크플로의 여러 부분에서 정보를 공유해야 하는 시나리오에 필수적입니다.

상태에 쓰기

using Microsoft.Agents.AI.Workflows;

internal sealed partial class FileReadExecutor(): Executor("FileReadExecutor")
{
    /// <summary>
    /// Reads a file and stores its content in a shared state.
    /// </summary>
    /// <param name="message">The path to the embedded resource file.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The ID of the shared state where the file content is stored.</returns>
    [MessageHandler]
    private async ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        // Read file content from embedded resource
        string fileContent = File.ReadAllText(message);
        // Store file content in a shared state for access by other executors
        string fileID = Guid.NewGuid().ToString();
        await context.QueueStateUpdateAsync<string>(fileID, fileContent, scopeName: "FileContent");

        return fileID;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class FileReadExecutor(Executor):

    @handler
    async def handle(self, file_path: str, ctx: WorkflowContext[str]):
        # Read file content from embedded resource
        with open(file_path, 'r') as file:
            file_content = file.read()
        # Store file content in state for access by other executors
        file_id = str(uuid.uuid4())
        ctx.set_state(file_id, file_content)

        await ctx.send_message(file_id)

상태 접근

using Microsoft.Agents.AI.Workflows;

internal sealed partial class WordCountingExecutor() : Executor("WordCountingExecutor")
{
    /// <summary>
    /// Counts the number of words in the file content stored in a shared state.
    /// </summary>
    /// <param name="message">The ID of the shared state containing the file content.</param>
    /// <param name="context">The workflow context for accessing shared states.</param>
    /// <returns>The number of words in the file content.</returns>
    [MessageHandler]
    private async ValueTask<int> HandleAsync(string message, IWorkflowContext context)
    {
        // Retrieve the file content from the shared state
        var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent")
            ?? throw new InvalidOperationException("File content state not found");

        return fileContent.Split([' ', '\n', '\r'], StringSplitOptions.RemoveEmptyEntries).Length;
    }
}

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class WordCountingExecutor(Executor):

    @handler
    async def handle(self, file_id: str, ctx: WorkflowContext[int]):
        # Retrieve the file content from state
        file_content = ctx.get_state(file_id)
        if file_content is None:
            raise ValueError("File content state not found")

        await ctx.send_message(len(file_content.split()))

상태 격리

실제 애플리케이션에서는 여러 작업 또는 요청을 처리할 때 상태를 적절하게 관리하는 것이 중요합니다. 적절한 격리가 없으면 서로 다른 워크플로 실행 간의 공유 상태가 예기치 않은 동작, 데이터 손상 및 경합 상태로 이어질 수 있습니다. 이 섹션에서는 Microsoft 에이전트 프레임워크 워크플로 내에서 상태 격리를 보장하여 모범 사례 및 일반적인 함정에 대한 인사이트를 제공하는 방법을 설명합니다.

변경 가능한 워크플로 작성기 및 변경할 수 없는 워크플로

워크플로 작성자는 워크플로를 만듭니다. 워크플로 작성기는 일반적으로 변경 가능한 것으로 간주됩니다. 여기서는 작성기를 만든 후 또는 워크플로가 빌드된 후에도 시작 실행기 또는 다른 구성을 추가, 수정할 수 있습니다. 반면 워크플로는 워크플로를 빌드한 후에는 수정할 수 없다는 측면에서 변경할 수 없습니다(워크플로를 수정할 공용 API 없음).

이 구분은 여러 워크플로 실행에서 상태를 관리하는 방법에 영향을 주므로 중요합니다. 의도하지 않은 상태 공유로 이어질 수 있으므로 여러 작업 또는 요청에 단일 워크플로 인스턴스를 다시 사용하지 않는 것이 좋습니다. 대신, 적절한 상태 격리 및 스레드 안전을 보장하기 위해 각 작업 또는 요청에 대해 작성기에서 새 워크플로 인스턴스를 만드는 것이 좋습니다.

도우미 메서드를 사용하여 상태 격리 보장

실행기 인스턴스가 한 번 만들어지고 여러 워크플로 빌드에서 공유되는 경우 내부 상태는 모든 워크플로 실행에서 공유됩니다. 이로 인해 실행기에서 워크플로별로 격리해야 하는 변경 가능한 상태가 포함된 경우 문제가 발생할 수 있습니다. 적절한 상태 격리 및 스레드 안전을 보장하려면 각 호출이 새로운 독립 인스턴스를 생성하도록 도우미 메서드 내에 실행기 인스턴스화 및 워크플로 빌드를 래핑합니다.

곧 출시 예정...

격리되지 않은 예제(공유 상태):

executor_a = CustomExecutorA()
executor_b = CustomExecutorB()

# executor_a and executor_b are shared across all workflows built from this builder
workflow_builder = WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b)

workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()
# workflow_a and workflow_b share the same executor instances and their mutable state

격리된 예제(도우미 메서드):

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated state.

    Each call produces independent executor instances, ensuring no state
    leaks between workflow runs.
    """
    executor_a = CustomExecutorA()
    executor_b = CustomExecutorB()

    return WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b).build()

# Each workflow has its own executor instances with independent state
workflow_a = create_workflow()
workflow_b = create_workflow()

팁 (조언)

적절한 상태 격리 및 스레드 보안을 보장하려면 도우미 메서드 내에서 만든 실행기 인스턴스가 외부 변경 가능한 상태를 공유하지 않는지 확인합니다.

에이전트 상태 관리

에이전트 컨텍스트는 에이전트 스레드를 통해 관리됩니다. 기본적으로 사용자 지정 실행기에서 에이전트를 관리하지 않는 한 워크플로의 각 에이전트는 자체 스레드를 가져옵니다. 자세한 내용은 에이전트 작업을 참조하세요.

에이전트 스레드는 워크플로 실행에서 유지됩니다. 즉, 에이전트가 워크플로의 첫 번째 실행에서 호출되는 경우 에이전트에서 생성된 콘텐츠는 동일한 워크플로 인스턴스의 후속 실행에서 사용할 수 있습니다. 이는 단일 작업 내에서 연속성을 유지하는 데 유용할 수 있지만 동일한 워크플로 인스턴스를 다른 작업 또는 요청에 다시 사용하는 경우 의도하지 않은 상태 공유로 이어질 수도 있습니다. 각 태스크에 격리된 에이전트 상태가 있는지 확인하려면 각 호출이 자체 스레드를 사용하여 새 에이전트 인스턴스를 생성하도록 도우미 메서드 내에서 에이전트 및 워크플로 만들기를 래핑합니다.

곧 출시 예정...

격리되지 않은 예제(공유 에이전트 상태):

writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

# writer_agent and reviewer_agent are shared across all workflows
workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

격리된 예제(도우미 메서드):

def create_workflow() -> Workflow:
    """Create a fresh workflow with isolated agent state.

    Each call produces new agent instances with their own threads,
    ensuring no conversation history leaks between workflow runs.
    """
    writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
        name="writer_agent",
    )
    reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).as_agent(
        instructions=(
            "You are an excellent content reviewer."
            "Provide actionable feedback to the writer about the provided content."
            "Provide the feedback in the most concise manner possible."
        ),
        name="reviewer_agent",
    )

    return WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()

# Each workflow has its own agent instances and threads
workflow_a = create_workflow()
workflow_b = create_workflow()

요약

Microsoft Agent Framework 워크플로의 상태 격리는 도우미 메서드 내의 워크플로 빌드와 함께 실행기 및 에이전트 인스턴스화를 래핑하여 효과적으로 관리할 수 있습니다. 새 워크플로가 필요할 때마다 도우미 메서드를 호출하여 각 인스턴스에 새로운 독립 상태가 있는지 확인하고 다른 워크플로 실행 간에 의도하지 않은 상태 공유를 방지합니다.

다음 단계

  • Last updated on