مهام سير عمل إطار عمل عامل Microsoft - الحالة

يوفر هذا المستند نظرة عامة على الحالة في نظام سير عمل إطار عمل عامل Microsoft.

نظرة عامة

تسمح الحالة للعديد من المنفذين داخل سير العمل بالوصول إلى البيانات الشائعة وتعديلها. هذه الميزة ضرورية للسيناريوهات التي تحتاج فيها أجزاء مختلفة من سير العمل إلى مشاركة المعلومات حيث لا يكون تمرير الرسائل المباشرة ممكنا أو فعالا.

رؤية الحالة وسلوك النطاق

QueueStateUpdateAsync و ReadStateAsync كلاهما على دراية بالنطاق:

• إذا كان scopeName هو null، يتم استخدام النطاق الافتراضي الخاص للمنفذ.
• إذا scopeName تم تعيين (على سبيل المثال، "SharedResponse")، تتم كتابة القيمة إلى نطاق مشترك يمكن لأي منفذ قراءته عند استخدام نفس اسم النطاق.

يتبع توقيت الرؤية قواعد الفوقية:

• يمكن للمنفذ الذي يستدعي QueueStateUpdateAsync قراءة القيمة المحدثة مباشرة في نفس المعالج.
• يرى المنفذون الآخرون أن التحديث يبدأ في الخطوة فائقة التالية.

لمشاركة الحالة عبر المنفذين، استخدم نفس اسم النطاق غير الفارغ في كل من استدعاءات الكتابة والقراءة:

private const string SharedScope = "SharedResponse";

await context.QueueStateUpdateAsync("Response", blanketResponse, scopeName: SharedScope, cancellationToken);

var finalResponse = await context.ReadStateAsync<string>("Response", scopeName: SharedScope, cancellationToken);

WorkflowContext.set_state() وتعمل WorkflowContext.get_state() على حالة سير العمل المتوفرة لمنفذي انتقال البيانات من الخادم أثناء تنفيذ سير العمل.

استخدم مفاتيح متناسقة عبر المنفذين لكتابة القيمة نفسها وقراءتها:

ctx.set_state("response", blanket_response)
final_response = ctx.get_state("response")

الكتابة إلى الحالة

using Microsoft.Agents.AI.Workflows;

internal sealed class FileReadExecutor() : Executor<string, string>("FileReadExecutor")
{
    public override async ValueTask<string> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // 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("N");
        await context.QueueStateUpdateAsync(fileID, fileContent, scopeName: "FileContent", cancellationToken);

        return fileID;
    }
}

import uuid

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)

fileRead := workflow.NewExecutor("FileReadExecutor", func(ctx *workflow.Context, path string) (string, error) {
    fileContent, err := os.ReadFile(path)
    if err != nil {
        return "", err
    }

    fileID := uuid.NewString()
    if err := ctx.QueueStateUpdate(fileID, "FileContent", string(fileContent)); err != nil {
        return "", err
    }

    return fileID, nil
}).Bind()

حالة الوصول

using Microsoft.Agents.AI.Workflows;

internal sealed class WordCountingExecutor() : Executor<string, int>("WordCountingExecutor")
{
    public override async ValueTask<int> HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // Retrieve the file content from the shared state
        var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent", cancellationToken)
            ?? 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()))

fileProcess := workflow.NewExecutor("FileProcessExecutor", func(ctx *workflow.Context, fileID string) (FileSummary, error) {
    value, err := ctx.ReadState(fileID, "FileContent")
    if err != nil {
        return FileSummary{}, err
    }

    fileContent, ok := value.(string)
    if !ok {
        return FileSummary{}, fmt.Errorf("file content %q was not found", fileID)
    }

    return FileSummary{
        FileID:  fileID,
        Summary: summarize(fileContent),
    }, nil
}).Bind()

kwargs وقت التشغيل على نطاق سير العمل

بالنسبة للقيم التي يجب أن تتدفق إلى العوامل والأدوات دون أن تصبح حالة سير عمل مشتركة، قم بتمريرها ك workflow.run()function_invocation_kwargs= أو client_kwargs=.

• يعد التعيين العادي بدون مفاتيح معرف المنفذ عموميا، ويتلقىه كل منفذ عامل مطابق.
• يتم استهداف تعيين عادي مع مفاتيح معرف المنفذ، ويتلقى كل منفذ إدخاله الخاص فقط.
• يستخدم WorkflowInvocationKwargs لدمج القيم المشتركة مع التجاوزات الخاصة بالمنفذ. يمكن لمفاتيحه executor_kwargs استخدام معرفات المنفذ أو أسماء عوامل ملفوفة فريدة. تفوز التجاوزات عندما يظهر نفس المفتاح في كلا التعيينين.
• تنطبق نفس القواعد على function_invocation_kwargs و client_kwargs. في مهام سير العمل المتداخلة، تظل المفاتيح المتطابقة مع المنفذ الأصل محددة النطاق للرسم البياني الأصلي، بينما تستمر القيم والإدخالات العمومية التي تستهدف المنفذين الفرعيين فقط في سير العمل التابع.

from agent_framework import WorkflowInvocationKwargs

await workflow.run(
    "Create the report",
    function_invocation_kwargs=WorkflowInvocationKwargs(
        global_kwargs={
            "tenant": "contoso",
            "request_id": "req-42",
        },
        executor_kwargs={
            "researcher": {
                "request_id": "research-42",
                "db_config": {"connection_string": "..."},
            },
            "writer": {
                "user_preferences": {"format": "markdown"},
            },
        },
    ),
)

Tip

تستخدم kwargs المستهدفة من قبل المنفذ معرفات منفذ سير العمل. بالنسبة للوكلاء الملتفين، هذا المعرف هو اسم العامل بشكل افتراضي. إذا قمت بتمرير صريح id مختلف إلى AgentExecutor(...)، WorkflowInvocationKwargs.executor_kwargs فلا يزال بإمكانك استخدام اسم العامل عندما يكون فريدا. استخدم معرفات المنفذ عند تكرار أسماء الوكلاء.

عزل الحالة

في تطبيقات العالم الحقيقي، تعد إدارة الحالة بشكل صحيح أمرا بالغ الأهمية عند معالجة مهام أو طلبات متعددة. بدون العزل المناسب، يمكن أن تؤدي الحالة المشتركة بين عمليات تنفيذ سير العمل المختلفة إلى سلوك غير متوقع، وفساد في البيانات، وظروف تعارض. يشرح هذا القسم كيفية ضمان عزل الحالة داخل مهام سير عمل إطار عمل العامل Microsoft، وتوفير رؤى حول أفضل الممارسات والمزالق الشائعة.

منشئو سير العمل القابل للتغيير مقابل مهام سير العمل غير القابلة للتغيير

يتم إنشاء مهام سير العمل بواسطة منشئي سير العمل. عادة ما يعتبر منشئو سير العمل قابلين للتغيير، حيث يمكن للمرء إضافة منفذ البدء أو التكوينات الأخرى أو تعديلها بعد إنشاء المنشئ أو حتى بعد إنشاء سير العمل. من ناحية أخرى، تكون مهام سير العمل غير قابلة للتغيير من حيث أنه بمجرد إنشاء سير عمل، لا يمكن تعديله (لا توجد واجهة برمجة تطبيقات عامة لتعديل سير عمل).

هذا التمييز مهم لأنه يؤثر على كيفية إدارة الحالة عبر عمليات تنفيذ سير العمل المختلفة. لا يوصى بإعادة استخدام مثيل سير عمل واحد لمهام أو طلبات متعددة، حيث يمكن أن يؤدي ذلك إلى مشاركة حالة غير مقصودة. بدلا من ذلك، يوصى بإنشاء مثيل سير عمل جديد من المنشئ لكل مهمة أو طلب لضمان عزل الحالة المناسب وسلامة مؤشر الترابط.

ضمان عزل الحالة باستخدام أساليب المساعد

عند إنشاء مثيلات المنفذ مرة واحدة ومشاركتها عبر بنيات سير عمل متعددة، تتم مشاركة حالتها الداخلية عبر جميع عمليات تنفيذ سير العمل. يمكن أن يؤدي هذا إلى مشكلات إذا كان المنفذ يحتوي على حالة قابلة للتغيير يجب عزلها لكل سير عمل. لضمان عزل الحالة المناسب وسلامة مؤشر الترابط، قم بتضمين إنشاء مثيل للمنفذ وبناء سير العمل داخل أسلوب مساعد بحيث ينتج عن كل استدعاء مثيلات جديدة ومستقلة.

قريبا...

مثال غير معزول (حالة مشتركة):

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

مثال غير معزول (حالة مشتركة):

executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
    return input, nil
}).Bind()
executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
    return input, nil
}).Bind()

builder := workflow.NewBuilder(executorA).AddEdge(executorA, executorB)

workflowA, err := builder.Build()
if err != nil {
    return err
}
workflowB, err := builder.Build()
if err != nil {
    return err
}

مثال معزول (أسلوب المساعد):

func createWorkflow() (*workflow.Workflow, error) {
    executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
        return input, nil
    }).Bind()
    executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
        return input, nil
    }).Bind()

    return workflow.NewBuilder(executorA).AddEdge(executorA, executorB).Build()
}

workflowA, err := createWorkflow()
if err != nil {
    return err
}
workflowB, err := createWorkflow()
if err != nil {
    return err
}

Tip

لضمان عزل الحالة المناسبة وسلامة مؤشر الترابط، تأكد أيضا من أن مثيلات المنفذ التي تم إنشاؤها داخل أسلوب المساعد لا تشترك في حالة قابلة للتغيير الخارجية.

إعادة تعيين المنفذين المشتركين

إذا كنت بحاجة إلى مشاركة مثيلات المنفذ عبر عمليات تشغيل سير العمل — على سبيل المثال، عندما يكون إنشاء المنفذ مكلفا أو عندما يتم كشف سير العمل كعامل — يجب على المنفذين المناسبين تنفيذ IResettableExecutor. توفر هذه الواجهة أسلوبا ResetAsync() يستدعيه وقت تشغيل سير العمل تلقائيا بين عمليات التشغيل لمسح الحالة القديمة.

للحصول على تفاصيل حول وقت وكيفية تنفيذ IResettableExecutor، راجع المنفذون القابلون لإعادة التعيين.

إعادة تعيين المنفذين المشتركين

يمكن لروابط المنفذ Go إعادة تعيين حالة المنفذ المشترك باستخدام ResetFunc. الارتباطات التي تم إنشاؤها باستخدام BindNewExecutorFunc إنشاء منفذ جديد لكل جلسة سير عمل ولا تحتاج إلى ربط إعادة تعيين.

للحصول على التفاصيل، راجع المنفذون القابلون لإعادة التعيين.

إدارة حالة العامل

تتم إدارة سياق العامل عبر مؤشرات ترابط العامل. بشكل افتراضي، سيحصل كل عامل في سير العمل على مؤشر ترابط خاص به ما لم تتم إدارة العامل بواسطة منفذ مخصص. لمزيد من المعلومات، راجع العمل مع الوكلاء.

تستمر مؤشرات ترابط العامل عبر عمليات تشغيل سير العمل. وهذا يعني أنه إذا تم استدعاء عامل في التشغيل الأول لسير عمل، فسيتوفر المحتوى الذي تم إنشاؤه بواسطة العامل في عمليات التشغيل اللاحقة لنفس مثيل سير العمل. على الرغم من أن هذا يمكن أن يكون مفيدا للحفاظ على الاستمرارية داخل مهمة واحدة، فإنه يمكن أن يؤدي أيضا إلى مشاركة الحالة غير المقصودة إذا تمت إعادة استخدام مثيل سير العمل نفسه لمهام أو طلبات مختلفة. للتأكد من أن كل مهمة تحتوي على حالة عامل معزولة، قم بتضمين العامل وإنشاء سير العمل داخل أسلوب مساعد بحيث ينتج عن كل استدعاء مثيلات عامل جديدة مع مؤشرات الترابط الخاصة بها.

قريبا...

مثال غير معزول (حالة العامل المشترك):

writer_agent = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    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 = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    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 = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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 = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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()

تتم إدارة حالة عامل Go من خلال agent.Session. يحتفظ الوكلاء في مهام سير العمل بجلسة العمل الخاصة بهم عبر المنعطفات ما لم يتم إنشاء عامل جديد أو سير عمل أو جلسة عمل جديدة.

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

_, err = writerAgent.RunText(ctx, "first request", agent.WithSession(session)).Collect()
if err != nil {
    return err
}

_, err = writerAgent.RunText(ctx, "follow-up request", agent.WithSession(session)).Collect()
if err != nil {
    return err
}

يمكن لمنفذي الوكيل المستضافين الذين تم إنشاؤهم باستخدام agentworkflow.New بدء جلسة عامل جديدة عن طريق إرسال agentworkflow.ResetSignal{}.

المُلَخَّص

يمكن إدارة عزل الحالة في Microsoft سير عمل إطار عمل العامل بشكل فعال عن طريق التفاف المنفذ وإنشاء مثيل للعامل جنبا إلى جنب مع بناء سير العمل داخل أساليب المساعد. من خلال استدعاء أسلوب المساعد في كل مرة تحتاج فيها إلى سير عمل جديد، تأكد من أن كل مثيل له حالة جديدة ومستقلة وتجنب مشاركة الحالة غير المقصودة بين عمليات تنفيذ سير العمل المختلفة.

الخطوات التالية

• تعرف على كيفية إنشاء نقاط التحقق والاستئناف منها.
• تعرف على كيفية مراقبة مهام سير العمل.
• تعرف على كيفية تصور مهام سير العمل.