دليل الترقية: واجهات برمجة تطبيقات سير العمل ونظام Request-Response

يساعدك هذا الدليل على ترقية مهام سير العمل Python إلى أحدث تغييرات واجهة برمجة التطبيقات المقدمة في الإصدار 1.0.0b251104.

نظرة عامة على التغييرات

يتضمن هذا الإصدار تحسينين رئيسيين لنظام سير العمل:

1. واجهات برمجة تطبيقات تنفيذ سير العمل الموحدة

تم توحيد أساليب تنفيذ سير العمل للتبسيط:

• الأساليب الموحدة run(..., stream=True)run(): استبدال أساليب منفصلة خاصة بنقاط التفتيش (run_stream_from_checkpoint()، run_from_checkpoint())
• واجهة واحدة: استخدام checkpoint_id المعلمة لاستئناف من نقاط التحقق بدلا من أساليب منفصلة
• نقاط التفتيش المرنة: تكوين تخزين نقطة التحقق في وقت الإنشاء أو التجاوز في وقت التشغيل
• دلالات أكثر وضوحا: معلمات حصرية message متبادلة (تشغيل جديد) و checkpoint_id (استئناف)

2. نظام Request-Response المبسط

تم تبسيط نظام الاستجابة للطلبات:

• لا مزيد من RequestInfoExecutor: يمكن للمنفذين الآن إرسال الطلبات مباشرة
• مصمم الديكور الجديد@response_handler: استبدال RequestResponse معالجات الرسائل
• أنواع الطلبات المبسطة: لا يوجد توريث من RequestInfoMessage مطلوب
• القدرات المضمنة: تدعم جميع المنفذين تلقائيا وظيفة استجابة الطلب
• الرسوم البيانية لسير العمل الأنظف: إزالة RequestInfoExecutor العقد من مهام سير العمل

الجزء الأول: واجهات برمجة تطبيقات تنفيذ سير العمل الموحدة

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

استئناف من نقاط التفتيش

قبل (واجهة برمجة التطبيقات القديمة):

# OLD: Separate method for checkpoint resume
async for event in workflow.run_stream_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage
):
    print(f"Event: {event}")

بعد (واجهة برمجة تطبيقات جديدة):

# NEW: Unified method with checkpoint_id parameter
async for event in workflow.run(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage,  # Optional if configured at build time
    stream=True,
):
    print(f"Event: {event}")

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

• استخدام checkpoint_id المعلمة بدلا من أسلوب منفصل
• لا يمكن توفير كل من message و checkpoint_id (حصري بشكل متبادل)
• يجب توفير إما message (تشغيل جديد) أو checkpoint_id (استئناف)
• checkpoint_storage اختياري إذا تم تكوين نقاط التفتيش في وقت الإنشاء

واجهة برمجة التطبيقات غير المتدفقة

يتبع الأسلوب غير المتدفق run() نفس النمط:

القديمة:

result = await workflow.run_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage
)

الجديد:

result = await workflow.run(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage  # Optional if configured at build time
)

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

عند استئناف من نقطة تحقق تحتوي على أحداث معلومات الطلب المعلقة، تعيد واجهة برمجة التطبيقات إصدار هذه الأحداث تلقائيا. يمكنك التقاطها والرد عليها، أو توفيرها responsescheckpoint_id في نفس المكالمة.

قبل (السلوك القديم):

# OLD: Could provide responses directly during resume
responses = {
    "request-id-1": "user response data",
    "request-id-2": "another response"
}

async for event in workflow.run_stream_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage,
    responses=responses  # No longer supported
):
    print(f"Event: {event}")

بعد (سلوك جديد):

# Capture re-emitted pending requests
requests: dict[str, Any] = {}

async for event in workflow.run(checkpoint_id="checkpoint-id", stream=True):
    if event.type == "request_info":
        # Pending requests are automatically re-emitted
        print(f"Pending request re-emitted: {event.request_id}")
        requests[event.request_id] = event.data

# Collect user responses
responses: dict[str, Any] = {}
for request_id, request_data in requests.items():
    response = handle_request(request_data)  # Your logic here
    responses[request_id] = response

# Send responses back to workflow
async for event in workflow.run(responses=responses, stream=True):
    if event.type == "output":
        print(f"Workflow output: {event.data}")

مثال كامل على Human-in-the-Loop

فيما يلي مثال كامل يوضح استئناف نقطة التحقق مع موافقة بشرية معلقة:

from agent_framework import (
    Executor,
    FileCheckpointStorage,
    WorkflowBuilder,
    handler,
    response_handler,
)

# ... (Executor definitions omitted for brevity)

async def run_interactive_session(
    workflow: Workflow,
    initial_message: str | None = None,
    checkpoint_id: str | None = None,
) -> str:
    """Run workflow until completion, handling human input interactively."""

    requests: dict[str, HumanApprovalRequest] = {}
    responses: dict[str, str] | None = None
    completed_output: str | None = None

    while True:
        # Determine which API to call
        if responses:
            # Send responses from previous iteration
            event_stream = workflow.run(responses=responses, stream=True)
            requests.clear()
            responses = None
        else:
            # Start new run or resume from checkpoint
            if initial_message:
                event_stream = workflow.run(initial_message, stream=True)
            elif checkpoint_id:
                event_stream = workflow.run(checkpoint_id=checkpoint_id, stream=True)
            else:
                raise ValueError("Either initial_message or checkpoint_id required")

        # Process events
        async for event in event_stream:
            if event.type == "status":
                print(event)
            if event.type == "output":
                completed_output = event.data
            if event.type == "request_info":
                if isinstance(event.data, HumanApprovalRequest):
                    requests[event.request_id] = event.data

        # Check completion
        if completed_output:
            break

        # Prompt for user input if we have pending requests
        if requests:
            responses = prompt_for_responses(requests)
            continue

        raise RuntimeError("Workflow stopped without completing or requesting input")

    return completed_output

الجزء الثاني: نظام Request-Response المبسط

بعد الترحيل إلى واجهات برمجة تطبيقات سير العمل الموحدة، قم بتحديث أنماط استجابة الطلب لاستخدام النظام المتكامل الجديد.

1. تحديث عمليات الاستيراد

قبل:

from agent_framework import (
    RequestInfoExecutor,
    RequestInfoMessage,
    RequestResponse,
    # ... other imports
)

بعد:

from agent_framework import (
    response_handler,
    # ... other imports
    # Remove: RequestInfoExecutor, RequestInfoMessage, RequestResponse
)

2. تحديث أنواع الطلبات

قبل:

from dataclasses import dataclass
from agent_framework import RequestInfoMessage

@dataclass
class UserApprovalRequest(RequestInfoMessage):
    """Request for user approval."""
    prompt: str = ""
    context: str = ""

بعد:

from dataclasses import dataclass

@dataclass
class UserApprovalRequest:
    """Request for user approval."""
    prompt: str = ""
    context: str = ""

3. تحديث الرسم البياني لسير العمل

قبل:

# Old pattern: Required RequestInfoExecutor in workflow
approval_executor = ApprovalRequiredExecutor(id="approval")
request_info_executor = RequestInfoExecutor(id="request_info")

workflow = (
    WorkflowBuilder(start_executor=approval_executor)
    .add_edge(approval_executor, request_info_executor)
    .add_edge(request_info_executor, approval_executor)
    .build()
)

بعد:

# New pattern: Direct request-response capabilities
approval_executor = ApprovalRequiredExecutor(id="approval")

workflow = (
    WorkflowBuilder(start_executor=approval_executor)
    .build()
)

4. تحديث إرسال الطلب

قبل:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def process(self, message: str, ctx: WorkflowContext[UserApprovalRequest]) -> None:
        request = UserApprovalRequest(
            prompt=f"Please approve: {message}",
            context="Important operation"
        )
        await ctx.send_message(request)

بعد:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def process(self, message: str, ctx: WorkflowContext) -> None:
        request = UserApprovalRequest(
            prompt=f"Please approve: {message}",
            context="Important operation"
        )
        await ctx.request_info(request_data=request, response_type=bool)

5. تحديث معالجة الاستجابة

قبل:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def handle_approval(
        self,
        response: RequestResponse[UserApprovalRequest, bool],
        ctx: WorkflowContext[Never, str]
    ) -> None:
        if response.data:
            await ctx.yield_output("Approved!")
        else:
            await ctx.yield_output("Rejected!")

بعد:

class ApprovalRequiredExecutor(Executor):
    @response_handler
    async def handle_approval(
        self,
        original_request: UserApprovalRequest,
        approved: bool,
        ctx: WorkflowContext
    ) -> None:
        if approved:
            await ctx.yield_output("Approved!")
        else:
            await ctx.yield_output("Rejected!")

ملخص الفوائد

واجهات برمجة تطبيقات سير العمل الموحدة

1. واجهة مبسطة: أسلوب واحد للشواط الأولية واستئناف نقطة التحقق
2. دلالات أكثر وضوحا: تجعل المعلمات الحصرية المتبادلة الهدف صريحا
3. نقاط التحقق المرنة: التكوين في وقت الإنشاء أو التجاوز في وقت التشغيل
4. تقليل الحمل المعرفي: أساليب أقل لتذكرها وصيانتها

نظام Request-Response

1. البنية المبسطة: لا حاجة لمكونات منفصلة RequestInfoExecutor
2. أمان النوع: مواصفات النوع المباشر في request_info() المكالمات
3. التعليمات البرمجية الأنظف: عمليات استيراد أقل ورسومات بيانية أبسط لسير العمل
4. أداء أفضل: تقليل حمل توجيه الرسائل
5. تصحيح الأخطاء المحسن: تدفق تنفيذ أكثر وضوحا ومعالجة الأخطاء

اختبار الترحيل الخاص بك

قائمة التحقق من الجزء 1: واجهات برمجة تطبيقات سير العمل

1. تحديث استدعاءات واجهة برمجة التطبيقات: استبدال run_stream_from_checkpoint() ب run(checkpoint_id=..., stream=True)
2. تحديث استدعاءات واجهة برمجة التطبيقات: استبدال run_from_checkpoint() ب run(checkpoint_id=...)
3. استخدام شكل السيرة الذاتية الحالي: تمرير الاستجابات مع workflow.run(responses=..., stream=True) أو مع checkpoint_id عند الاستئناف والاستجابة في مكالمة واحدة
4. إضافة التقاط الحدث: تنفيذ المنطق لالتقاط أحداث request_info المعاد إصدارها (event.type == "request_info")
5. استئناف اختبار نقطة التحقق: تحقق من إعادة إصدار الطلبات المعلقة ومعالجتها بشكل صحيح

قائمة اختيار الجزء 2: نظام Request-Response

1. التحقق من الواردات: تأكد من عدم بقاء أي عمليات استيراد قديمة (RequestInfoExecutor، ، RequestInfoMessageRequestResponse)
2. التحقق من أنواع الطلبات: تأكيد إزالة التوريث RequestInfoMessage
3. اختبار الرسم البياني لسير العمل: التحقق من RequestInfoExecutor إزالة العقد
4. التحقق من صحة المعالجات: تأكد من @response_handler تطبيق المحسنات
5. اختبار من طرف إلى طرف: تشغيل سيناريوهات سير العمل الكاملة

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

بعد إكمال الترحيل:

1. مراجعة البرنامج التعليمي للطلبات والاستجابات المحدثة
2. استكشاف الأنماط المتقدمة في دليل المستخدم
3. تحقق من العينات المحدثة في المستودع

للحصول على مساعدة إضافية، راجع وثائق إطار عمل العامل أو تواصل مع الفريق والمجتمع.