مهام سير العمل مع AG-UI

يمكن أن يعرض .NET MAF سير عمل من خلال AG-UI عن طريق تحويل سير العمل إلى AIAgent وتعيينه مثل أي عامل آخر:

AIAgent workflowAgent = AgentWorkflowBuilder
    .BuildSequential(researcher, reporter)
    .AsAIAgent();

app.MapAGUIServer("/", workflowAgent);

تقوم نقطة النهاية ببث النص القياسي للوكلاء المكونين وإخراج استدعاء الأدوات. AuthorName يحدد العامل الذي أنتج كل تحديث.

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

للحصول على حالة تعقب .NET الحالية، راجع microsoft/agent-framework#2494. لإنشاء سير العمل وتنفيذه بشكل مستقل عن AG-UI، راجع مفاهيم سير عمل MAF.

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

مراجعة اعتبارات الإنتاج والأمان

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

المتطلبات الأساسية

قبل أن تبدأ، تأكد من أنك تملك:

Python 3.10 أو أحدث
agent-framework-ag-uiوثبت agent-framework-foundry
الإلمام بالبرنامج التعليمي للبدء
الفهم الأساسي لمفاهيم سير عمل Agent Framework

متى تستخدم مهام سير العمل مع AG-UI

استخدم سير عمل بدلا من عامل واحد عند الحاجة:

التنسيق متعدد العوامل: توجيه المهام بين الوكلاء المتخصصين (على سبيل المثال، الفرز → استرداد المبلغ → الطلب)
خطوات التنفيذ المنظمة: تتبع التقدم من خلال مراحل محددة باستخدام STEP_STARTED / STEP_FINISHED الأحداث
المقاطعة / استئناف التدفقات: إيقاف التنفيذ مؤقتا لجمع المدخلات أو الموافقات البشرية، ثم استئناف
تدفق الأحداث المخصصة: إرسال أحداث خاصة بالمجال (request_info، ، statusworkflow_output) إلى العميل

التفاف سير عمل باستخدام AgentFrameworkWorkflow

AgentFrameworkWorkflow هو برنامج تضمين خفيف الوزن يقوم بتكييف أصلي Workflow مع بروتوكول AG-UI. يمكنك توفير مثيل سير عمل تم إنشاؤه مسبقا أو مصنع يقوم بإنشاء سير عمل جديد لكل مؤشر ترابط.

مثيل مباشر

استخدم مثيلا مباشرا عندما يمكن لعنصر سير عمل واحد خدمة جميع الطلبات بأمان (على سبيل المثال، البنية الأساسية لبرنامج ربط العمليات التجارية عديمة الحالة):

from agent_framework import Workflow
from agent_framework.ag_ui import AgentFrameworkWorkflow

workflow = build_my_workflow()  # returns a Workflow

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    name="my-workflow",
    description="Single-instance workflow.",
)

مصنع ذي نطاق مؤشر ترابط

استخدم workflow_factory عندما يحتاج كل مؤشر ترابط محادثة إلى حالة سير العمل الخاصة به. يتلقى thread_id المصنع ويعيد جديد Workflow:

from agent_framework.ag_ui import AgentFrameworkWorkflow

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda thread_id: build_my_workflow(),
    name="my-workflow",
    description="Thread-scoped workflow.",
)

Important

يجب أن تمر إماworkflowأوworkflow_factory، وليس كليهما. يرفع برنامج التضمين ValueError إذا تم توفير كليهما.

تسجيل نقطة النهاية

سجل سير العمل add_agent_framework_fastapi_endpoint بنفس الطريقة التي تسجل بها وكيلا واحدا:

from fastapi import FastAPI
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)

app = FastAPI(title="Workflow AG-UI Server")

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda thread_id: build_my_workflow(),
    name="handoff-demo",
    description="Multi-agent handoff workflow.",
)

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=ag_ui_workflow,
    path="/workflow",
)

يمكنك أيضا تمرير عارية Workflow مباشرة — تقوم نقطة النهاية بتضمينها تلقائيا في AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI الأحداث المنبعثة من مهام سير العمل

تصدر عمليات تشغيل سير العمل مجموعة أكثر ثراء من أحداث AG-UI مقارنة بتشغيل عامل واحد:

Event	عند إصداره	Description
`RUN_STARTED`	يبدأ التشغيل	وضع علامة على بداية تنفيذ سير العمل
`STEP_STARTED`	يبدأ المنفذ أو الفوقية	`step_name` يحدد العامل أو الخطوة (على سبيل المثال، `"triage_agent"`)
`TEXT_MESSAGE_*`	ينتج العامل نصا	أحداث نص الدفق القياسي
`TOOL_CALL_*`	يستدعي العامل أداة	أحداث استدعاء الأداة القياسية
`REASONING_*`	يصدر سير العمل نصا من منفذ تم تكوينه في `intermediate_output_from`	دفق النص المتوسط ككتلة تحليل. يتبع الاسم المستعار للحدث المهمل `"data"` نفس المسار.
`STEP_FINISHED`	اكتمال المنفذ أو التمام	إغلاق خطوة تعقب تقدم واجهة المستخدم
`CUSTOM` (`status`)	تغييرات حالة سير العمل	يحتوي على `{"state": "<value>"}` في قيمة الحدث
`CUSTOM` (`request_info`)	يطلب سير العمل إدخالا بشريا	يحتوي على حمولة الطلب للعميل لتقديم مطالبة
`CUSTOM` (`workflow_output`)	لا يمكن تحويل إخراج سير العمل إلى محتوى رسالة	يحتوي على الإخراج المتسلسل لعرض العميل المخصص.
`RUN_FINISHED`	اكتمال التشغيل	يتضمن `outcome.type == "interrupt"` و `outcome.interrupts` عندما ينتظر سير العمل الإدخال

يمكن للعملاء استخدام STEP_STARTED / STEP_FINISHED الأحداث لعرض مؤشرات التقدم التي توضح العامل النشط حاليا. يغلق التكامل المنطق المفتوح وكتل النص قبل حدث طرفي أو طلب إدخال بشري، بحيث يتلقى العملاء تسلسل حدث كامل.

عند فشل سير عمل Python، RUN_ERROR يستخدم الرسالة Workflow execution failed. العامة العامة بالإضافة إلى رمز خطأ. يعرض executor_failed الحدث بالمثل الرسالة العامة ونوع الخطأ. تظل تفاصيل الاستثناء الداخلي والتتبعات في سجلات الخادم.

المقاطعة والاستئناف

يمكن أن توقف مهام سير العمل التنفيذ مؤقتا لجمع موافقات المدخلات أو الأدوات البشرية. يعالج تكامل AG-UI هذا من خلال بروتوكول المقاطعة/الاستئناف.

كيفية عمل المقاطعة

أثناء التنفيذ، يرفع سير العمل طلبا معلقا (على سبيل المثال، HandoffAgentUserRequest طلب المزيد من التفاصيل، أو أداة مع approval_mode="always_require").
يصدر جسر AG-UI حدثا CUSTOM يحتوي name="request_info" على بيانات الطلب.

ينتهي التشغيل بحدث RUN_FINISHED يحتوي حقله outcome.interrupts على الطلبات المعلقة:

{
  "type": "RUN_FINISHED",
  "threadId": "abc123",
  "runId": "run_xyz",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "request-id-1",
        "reason": "input_required",
        "message": "Provide the requested information.",
        "responseSchema": { "type": "string" },
        "metadata": {
          "agent_framework": {
            "request_type": "HandoffAgentUserRequest"
          }
        }
      }
    ]
  }
}

يعرض العميل واجهة المستخدم للمستخدم للاستجابة (إدخال نص، وزر موافقة، وما إلى ذلك).

كيفية عمل السيرة الذاتية

يرسل العميل طلبا جديدا مع صفيف متعارف عليه resume . يحدد كل إدخال المقاطعة ويزود استجابة المستخدم:

{
  "threadId": "abc123",
  "messages": [],
  "resume": [
    {
      "interruptId": "request-id-1",
      "status": "resolved",
      "payload": "User's response text or approval decision"
    }
  ]
}

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

الاحتفاظ بنقاط التحقق من سير العمل واستئنافها

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

يستخدم المثال التالي التخزين في الذاكرة لسير عمل قصير الأجل:

from agent_framework import InMemoryCheckpointStorage
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)
from fastapi import FastAPI

app = FastAPI()
checkpoint_storage = InMemoryCheckpointStorage()
workflow = build_my_workflow()

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    checkpoint_storage=checkpoint_storage,
)
add_agent_framework_fastapi_endpoint(
    app,
    ag_ui_workflow,
    "/workflow",
)

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

AgentFrameworkWorkflow.run()يتلقى حمولة طلب AG-UI، لذلك يوفر العميل معرف نقطة التحقق من خلال الخصائص التي تمت إعادة توجيهها بدلا من وسيطة Pythoncheckpoint_id. لا يتضمن استئناف نقطة التحقق فقط رسالة مستخدم جديدة:

{
  "threadId": "abc123",
  "messages": [],
  "forwardedProps": {
    "checkpointId": "checkpoint-id-from-interrupt-metadata"
  }
}

يستعيد المحول حالة سير العمل المحفوظ ويستمر في التنفيذ. إذا كانت نقطة التحقق تحتوي على مقاطعة معلقة، قم بتضمين كل من معرف نقطة التحقق والحمولة المتعارف resume عليه في نفس الطلب. يستعيد المحول نقطة التحقق قبل أن يسلم استجابة المقاطعة. استخدم القيمة من metadata.agent_framework.checkpoint_id ك forwardedProps.checkpointId.

يربط المحول كل نقطة تحقق جديدة بنطاق اللقطة للطلب والمزود من threadIdالعميل . يرفض طلب استئناف عندما لا تتطابق أي من القيمتين. تظل نقاط التحقق المكتوبة قبل تقديم بيانات تعريف الملكية قابلة للاستئناس للتوافق.

لا يحل هذا الفحص محل تخويل نقطة النهاية أو تخزين نقطة التحقق المحمية. لمزيد من المعلومات، راجع اعتبارات الأمان.

InMemoryCheckpointStorage لا ينجو من إعادة تشغيل العملية. للحصول على خيارات التخزين الدائم وتحديد نقطة التحقق، راجع نقاط التحقق.

نقاط التحقق من سير العمل ولقطات مؤشر ترابط AG-UI

تستمر نقاط التحقق من سير العمل ولقطات مؤشر ترابط AG-UI في بيانات مختلفة:

آلية الثبات	المخازن	الغرض
نقطة التحقق من سير عمل إطار عمل العامل	حالة المنفذ ووقت التشغيل، بما في ذلك الطلبات المعلقة	استئناف تنفيذ سير العمل من حالة وقت التشغيل المحفوظة
لقطة مؤشر ترابط AG-UI	إخراج البروتوكول القابل لإعادة التشغيل، مثل الرسائل والحالة المشتركة وأحدث مقاطعة	إعادة ترطيب مؤشر الترابط المرئي للعميل

يمكنك تكوين كلتا الآليتين. لا تحل نقطة فحص سير العمل محل لقطة مؤشر ترابط AG-UI، ولا تحتوي لقطة مؤشر ترابط AG-UI على حالة المنفذ المطلوبة لاستئناف تنفيذ سير العمل.

مثال كامل: سير عمل التسليم متعدد العوامل

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

تحديد العوامل والأدوات

"""AG-UI workflow server with multi-agent handoff."""

import os

from agent_framework import Agent, Message, Workflow, tool
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)
from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import HandoffBuilder
from azure.identity import AzureCliCredential
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware


@tool(approval_mode="always_require")
def submit_refund(refund_description: str, amount: str, order_id: str) -> str:
    """Capture a refund request for manual review before processing."""
    return f"Refund recorded for order {order_id} (amount: {amount}): {refund_description}"


@tool(approval_mode="always_require")
def submit_replacement(order_id: str, shipping_preference: str, replacement_note: str) -> str:
    """Capture a replacement request for manual review before processing."""
    return f"Replacement recorded for order {order_id} (shipping: {shipping_preference}): {replacement_note}"


@tool(approval_mode="never_require")
def lookup_order_details(order_id: str) -> dict[str, str]:
    """Return order details for a given order ID."""
    return {
        "order_id": order_id,
        "item_name": "Wireless Headphones",
        "amount": "$129.99",
        "status": "delivered",
    }

إنشاء سير العمل

def create_handoff_workflow() -> Workflow:
    """Build a handoff workflow with triage, refund, and order agents."""
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

    triage = Agent(id="triage_agent", name="triage_agent", instructions="...", client=client)
    refund = Agent(id="refund_agent", name="refund_agent", instructions="...", client=client,
                   tools=[lookup_order_details, submit_refund])
    order = Agent(id="order_agent", name="order_agent", instructions="...", client=client,
                  tools=[lookup_order_details, submit_replacement])

    def termination_condition(conversation: list[Message]) -> bool:
        for msg in reversed(conversation):
            if msg.role == "assistant" and (msg.text or "").strip().lower().endswith("case complete."):
                return True
        return False

    builder = HandoffBuilder(
        name="support_workflow",
        participants=[triage, refund, order],
        termination_condition=termination_condition,
    )
    builder.add_handoff(triage, [refund], description="Route refund requests.")
    builder.add_handoff(triage, [order], description="Route replacement requests.")
    builder.add_handoff(refund, [order], description="Route to order after refund.")
    builder.add_handoff(order, [triage], description="Route back after completion.")

    return builder.with_start_agent(triage).build()

إنشاء تطبيق FastAPI

app = FastAPI(title="Workflow AG-UI Demo")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda _thread_id: create_handoff_workflow(),
    name="support_workflow",
    description="Customer support handoff workflow.",
)

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=ag_ui_workflow,
    path="/support",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8888)

تسلسل الحدث

ينتج عن التفاعل النموذجي متعدد الأدوار أحداثا مثل:

RUN_STARTED           threadId=abc123
STEP_STARTED          stepName=triage_agent
TEXT_MESSAGE_START     role=assistant
TEXT_MESSAGE_CONTENT   delta="I'll look into your refund..."
TEXT_MESSAGE_END
STEP_FINISHED         stepName=triage_agent
STEP_STARTED          stepName=refund_agent
TOOL_CALL_START       toolCallName=lookup_order_details
TOOL_CALL_ARGS        delta='{"order_id":"12345"}'
TOOL_CALL_END
TOOL_CALL_START       toolCallName=submit_refund
TOOL_CALL_ARGS        delta='{"order_id":"12345","amount":"$129.99",...}'
TOOL_CALL_END
RUN_FINISHED          outcome={type: "interrupt", interrupts: [{id: "...", reason: "tool_call"}]}

يمكن للعميل بعد ذلك عرض مربع حوار الموافقة والاستئناف مع قرار المستخدم.

تلقي الخصائص التي تمت إعادة توجيهها

يمكن للعملاء AG-UI (مثل CopilotKit) تضمين forwarded_props حقل (أو forwardedProps) في حمولة الإدخال. يمرر تكامل AG-UI تلقائيا هذه الخصائص إلى أسلوب سير العمل run عبر وسيطة function_invocation_kwargs الكلمة الأساسية:

class MyWorkflow(Workflow):
    async def run(
        self,
        *,
        message=None,
        responses=None,
        stream: bool = False,
        function_invocation_kwargs: dict | None = None,
    ):
        forwarded_props = (function_invocation_kwargs or {}).get("forwarded_props", {})
        # Use forwarded_props for custom routing, feature flags, etc.
        ...

التفاصيل الرئيسية:

يتم قبول كل من forwarded_props و forwardedProps في حمولة الإدخال؛ داخليا يتم تسويتها إلى forwarded_props.
ضمن الخصائص التي تمت إعادة توجيهها، checkpoint_id و checkpointId محجوزة لاستئناف نقطة التحقق لسير العمل.
إذا workflow.run() لم يقبل function_invocation_kwargs (أو **kwargs)، يتم إسقاط الخصائص بصمت — لا تتأثر مهام سير العمل الحالية.
يتم أيضا تخزين الخصائص التي تمت إعادة توجيهها في بيانات تعريف الجلسة ولكن تتم تصفيتها من بيانات التعريف المرتبطة ب LLM، لذلك لا تتسرب إلى طلبات عميل الدردشة.

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

Human-in-the-Loop

موارد إضافية

يمكن أن يعرض Go مهام سير العمل AG-UI عن طريق التفاف workflow.Workflow كعامل مع workflow/agentworkflow، ثم استضافة هذا العامل باستخدام provider/aguiprovider.

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

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(workflowAgent, aguiprovider.HandlerConfig{}))

Tip

راجع سير العمل كعينة عامل وعينة خادمAG-UI للحصول على أمثلة كاملة قابلة للتشغيل.

الملاحظات

هل كانت هذه الصفحة مفيدة؟

Last updated on 2026-09-18

مهام سير العمل مع AG-UI

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

المتطلبات الأساسية

متى تستخدم مهام سير العمل مع AG-UI

التفاف سير عمل باستخدام AgentFrameworkWorkflow

مثيل مباشر

مصنع ذي نطاق مؤشر ترابط

تسجيل نقطة النهاية

AG-UI الأحداث المنبعثة من مهام سير العمل

المقاطعة والاستئناف

كيفية عمل المقاطعة

كيفية عمل السيرة الذاتية

الاحتفاظ بنقاط التحقق من سير العمل واستئنافها

نقاط التحقق من سير العمل ولقطات مؤشر ترابط AG-UI

مثال كامل: سير عمل التسليم متعدد العوامل

تحديد العوامل والأدوات

إنشاء سير العمل

إنشاء تطبيق FastAPI

تسلسل الحدث

تلقي الخصائص التي تمت إعادة توجيهها

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

موارد إضافية

الملاحظات

الموارد الإضافية