Aracılığıyla paylaş

Microsoft Agent Framework İş Akışları - Durum

Bu belge, Microsoft Agent Framework İş Akışı sisteminde Durum'a genel bir bakış sağlar.

Genel Bakış

Durum, bir iş akışı içindeki birden çok yürütücünin ortak verilere erişmesine ve bunları değiştirmesine olanak tanır. Bu özellik, doğrudan ileti geçirmenin uygun veya verimli olmadığı durumlarda iş akışının farklı bölümlerinin bilgi paylaşması gereken senaryolar için gereklidir.

Eyalete Yazma

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)

Duruma Erişme

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

Durum Yalıtımı

Gerçek dünyadaki uygulamalarda, birden çok görev veya isteği işlerken durumun düzgün yönetilmesi kritik önem taşır. Düzgün yalıtım olmadan, farklı iş akışı yürütmeleri arasındaki paylaşılan durum beklenmeyen davranışlara, veri bozulmasına ve yarış koşullarına yol açabilir. Bu bölümde, en iyi yöntemler ve yaygın tuzaklar hakkında içgörüler sağlayarak Microsoft Agent Framework İş Akışları içinde durum yalıtımının nasıl sağlanmaları açıklanır.

Değişken İş Akışı Oluşturucuları ile Sabit İş Akışları Karşılaştırması

İş akışları iş akışı oluşturucuları tarafından oluşturulur. İş akışı oluşturucuları, genellikle değişebilir olarak kabul edilir; yani, oluşturucu veya iş akışı oluşturulduktan sonra başlangıç yürütücüsü veya diğer yapılandırmalar eklenebilir veya değiştirilebilir. Öte yandan, iş akışları bir iş akışı oluşturulduktan sonra değiştirilemez (bir iş akışını değiştirmek için genel API yoktur).

Farklı iş akışı yürütmelerinde durumun nasıl yönetildiğini etkilediği için bu ayrım önemlidir. İstenmeyen durum paylaşımına neden olabileceğinden, birden çok görev veya istek için tek bir iş akışı örneğinin yeniden kullanılması önerilmez. Bunun yerine, uygun durum yalıtımı ve iş parçacığı güvenliği sağlamak üzere her görev veya istek için oluşturucudan yeni bir iş akışı örneği oluşturmanız önerilir.

Yardımcı Yöntemlerle Durum Yalıtımını Sağlama

Yürütücü örnekleri bir kez oluşturulduğunda ve birden çok iş akışı derlemesinde paylaşıldığında, iç durumları tüm iş akışı yürütmelerinde paylaşılır. Yürütücü iş akışı başına yalıtılması gereken değiştirilebilir durum içeriyorsa bu sorunlara yol açabilir. Uygun durum yalıtımı ve iş parçacığı güvenliği sağlamak için, yürütücü nesne yaratımı ve iş akışı yapısını oluşturma süreçlerini, her çağrıda yeni ve bağımsız örnekler üreten bir yardımcı yöntemle sarmalayın.

Çok yakında...

Yalıtılmamış örnek (paylaşılan durum):

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

Yalıtılmış örnek (yardımcı yöntem):

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

Tip

Düzgün durum yalıtımı ve iş parçacığı güvenliği sağlamak için, yardımcı yönteminde oluşturulan yürütücü örneklerinin dış değiştirilebilir durumu paylaşmadığından da emin olun.

Ajan Durum Yönetimi

Aracı bağlamı iş parçacıkları aracılığıyla yönetilir. Varsayılan olarak, bir iş akışındaki her ajan, özel bir yürütücü tarafından yönetilmediği sürece kendi iş parçacığını alır. Daha fazla bilgi için Bkz. Aracılarla Çalışma.

Aracı iş parçacıkları iş akışı çalıştırmaları arasında kalıcıdır. Bu, bir iş akışının ilk çalıştırmasında bir aracı çağrılırsa, aracı tarafından oluşturulan içeriğin aynı iş akışı örneğinin sonraki çalıştırmalarında kullanılabilir olacağı anlamına gelir. Bu, tek bir görev içinde sürekliliği korumak için yararlı olsa da, aynı iş akışı örneği farklı görevler veya istekler için yeniden kullanılırsa istenmeyen durum paylaşımına da yol açabilir. Her görevin izole bir aracı durumuna sahip olduğundan emin olmak için, her çağrının kendi iş parçacıklarıyla yeni aracı örnekleri üretmesi için aracı ve iş akışı oluşturma işlemlerini bir yardımcı yöntem içerisine yerleştirin.

Çok yakında...

İzole edilmemiş örnek (paylaşılan ajan durumu):

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

Yalıtılmış örnek (yardımcı yöntem):

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

Özet

Microsoft Agent Framework İş Akışlarında durum yalıtımı, yürütücü ve aracı örneklemesiyle iş akışının oluşturulmasıyla birlikte yardımcı yöntemlerle sarılarak etkili bir şekilde yönetilebilir. Her yeni iş akışına ihtiyaç duyduğunuzda yardımcı yöntemini çağırarak her örneğin yeni ve bağımsız bir duruma sahip olduğundan emin olur ve farklı iş akışı yürütmeleri arasında istenmeyen durum paylaşımından kaçınabilirsiniz.

Sonraki Adımlar

  • Last updated on