واجهة برمجة تطبيقات سير العمل الوظيفي

تحذير

واجهة برمجة تطبيقات سير العمل الوظيفي تجريبية وتخضع للتغيير أو الإزالة في الإصدارات المستقبلية دون إشعار.

تتيح لك واجهة برمجة تطبيقات سير العمل الوظيفي كتابة مهام سير العمل كوظائف عادية Python غير متزامنة. بدلا من تحديد فئات المنفذ، وأسلاك الحواف، واستخدام WorkflowBuilder، يمكنك تزيين دالة async باستخدام @workflow واستخدام تدفق التحكم Python الأصلي — ifelse/، التكرارات for الحلقية، asyncio.gather - للتعبير عن المنطق الخاص بك.

للمقارنة جنبا إلى جنب مع واجهة برمجة تطبيقات الرسم البياني، راجع واجهات برمجة تطبيقات سير العمل في نظرة عامة على مهام سير العمل.

@workflow المصمم

تطبيق @workflow على دالة async لإنشاء عديم الحالة FunctionalWorkflowDefinition:

from agent_framework import workflow

@workflow
async def text_pipeline(text: str) -> str:
    upper = await to_upper_case(text)
    return await reverse_text(upper)

@workflow يدعم مصمم الديكور نموذج ذو معلمات مع وسيطات اختيارية:

from agent_framework import workflow

@workflow(name="my_pipeline", description="Uppercase then reverse")
async def text_pipeline(text: str) -> str:
    ...

@workflow البارامترات

المعلمه Type Description
name str | None اسم العرض لسير العمل. الإعدادات الافتراضية للدالة __name__.
description str | None وصف اختياري يمكن للبشر قراءته.

إنشاء سير عمل

قم باستدعاء .build() التعريف لإنشاء حالة .FunctionalWorkflow يمثل كل سير عمل مضمن مستدعي منطقي أو جلسة عمل واحدة. إنشاء مثيلات منفصلة للمتصلين المستقلين لعزل حالة التشغيل وإعادة التشغيل الخاصة بهم.

تمرير تخزين نقطة التحقق إلى .build() عندما يحتاج سير العمل إلى استمرار نتائج الخطوة والحالة:

from agent_framework import InMemoryCheckpointStorage

storage = InMemoryCheckpointStorage()
workflow_instance = text_pipeline.build(checkpoint_storage=storage)
المعلمه Type Description
checkpoint_storage CheckpointStorage | None التخزين الافتراضي لنتائج الخطوة الدائمة والحالة بين عمليات التشغيل.

توقيع دالة سير العمل

تتلقى المعلمة الأولى لوظيفة سير العمل الإدخال الذي تم تمريره إلى .run() على سير العمل المضمن. أضف معلمة ctx: RunContext فقط عندما تحتاج إلى HITL أو حالة المفتاح/القيمة أو الأحداث المخصصة — فهي اختيارية بخلاف ذلك:

# No ctx needed — just a plain pipeline
@workflow
async def simple_pipeline(data: str) -> str:
    result = await process(data)
    return result

# ctx needed for HITL, state, or custom events
@workflow
async def hitl_pipeline(data: str, ctx: RunContext) -> str:
    feedback = await ctx.request_info({"draft": data}, response_type=str)
    return feedback

RunContext يتم الكشف عن طريق كتابة التعليق التوضيحي أولا، ثم بواسطة اسم ctxالمعلمة ، لذلك يعمل كل من ctx: RunContext و معلمة عارية ctx .

تشغيل سير عمل

استدعاء .run() على الكائن الذي FunctionalWorkflow تم إرجاعه بواسطة .build():

workflow_instance = text_pipeline.build()

# .run() wraps the result in a WorkflowRunResult with events and state
result = await workflow_instance.run("hello world")
print(result.text)                        # first output as a string
print(result.get_outputs())               # list of terminal outputs
print(result.get_intermediate_outputs())  # list of intermediate outputs
print(result.get_final_state())           # WorkflowRunState.IDLE

run() البارامترات

المعلمه Type Description
message Any | None تم تمرير الإدخال إلى دالة سير العمل كوسيطة أولى لها.
stream bool إذا True، ترجع هذه الدالة ResponseStreamWorkflowEvent العناصر. تتغير افتراضيا إلى False.
responses dict[str, Any] | None استجابات HITL التي تم مفتاحها بواسطة request_id. يستخدم لاستئناف سير عمل معلق.
checkpoint_id str | None نقطة تحقق للاستعادة منها. checkpoint_storage يتطلب تعيين أثناء الإنشاء أو لهذا التشغيل.
checkpoint_storage CheckpointStorage | None يتجاوز تخزين نقطة التحقق وقت الإنشاء لهذا التشغيل.
include_status_events bool قم بتضمين أحداث تغيير الحالة في النتيجة غير المتدفقة.

توفير وضع إدخال واحد لكل استدعاء: messageأو responsesأو checkpoint_id. الاستثناء هو استئناف نقطة التحقق مع إدخال خارجي، حيث checkpoint_idresponses ويمكن تمريرها معا.

WorkflowRunResult

run()ترجع (غير المتدفقة) .WorkflowRunResult الأساليب الرئيسية:

الأسلوب / الخاصية المرتجعات Description
.text str الإخراج الأول كسلسلة. سلسلة فارغة إذا لم يتم إخراج سلسلة.
.get_outputs() list[Any] جميع مخرجات المحطة الطرفية المنبعثة من سير العمل (الأحداث مع type == "output").
.get_intermediate_outputs() list[Any] جميع المخرجات الوسيطة المنبعثة من سير العمل (الأحداث مع type == "intermediate").
.get_final_state() WorkflowRunState حالة التشغيل النهائية (IDLE، IDLE_WITH_PENDING_REQUESTS، FAILED، ...).
.get_request_info_events() list[WorkflowEvent] طلبات HITL المعلقة عندما تكون الحالة .IDLE_WITH_PENDING_REQUESTS

البث المباشر

تمرير stream=True لتلقي الأحداث كما يتم إنتاجها:

from agent_framework import workflow

@workflow
async def data_pipeline(url: str) -> str:
    raw = await fetch_data(url)
    return await transform_data(raw)

workflow_instance = data_pipeline.build()

# stream=True returns a ResponseStream you iterate with async for
stream = workflow_instance.run("https://example.com/api/data", stream=True)
async for event in stream:
    if event.type == "output":
        print(f"Output: {event.data}")

# After iteration, get_final_response() returns the WorkflowRunResult
result = await stream.get_final_response()
print(f"Final state: {result.get_final_state()}")

راجع python/samples/03-workflows/functional/basic_streaming_pipeline.py للحصول على مثال كامل.

@step المصمم

@step هو مصمم اختيار يضيف التخزين المؤقت للنتيجة وانبعاثات الحدث ونقاط التفتيش لكل خطوة إلى الوظائف غير المتزامنة الفردية:

from agent_framework import step, workflow

@step
async def fetch_data(url: str) -> dict:
    # expensive — hits a real API
    return await http_get(url)

@workflow
async def pipeline(url: str) -> str:
    raw = await fetch_data(url)
    return process(raw)

ماذا @step يفعل داخل سير العمل

  • تخزين النتائج مؤقتا — يتم تخزين النتيجة مع هوية إعادة التشغيل المشتقة من تعريف الخطوة والوسيطات المرتبطة بها. في استئناف HITL أو استعادة نقطة التحقق، ترجع الخطوة المكتملة نتيجتها المحفوظة على الفور بدلا من إعادة التنفيذ.
  • إصدار الأحداث — executor_invoked / / executor_completedexecutor_failed يتم إصدارها من أجل إمكانية الملاحظة. في ذاكرة التخزين المؤقت، executor_bypassed يتم إصدار بدلا من ذلك.
  • يحفظ نقاط التحقق — إذا كان سير العمل المضمن يحتوي checkpoint_storageعلى ، يتم حفظ نقطة تحقق بعد اكتمال كل خطوة.
  • حقن RunContext — إذا أعلنت دالة الخطوة عن معلمة ctx: RunContext ، يتم إدخال السياق النشط تلقائيا.

خارج سير العمل قيد التشغيل، @step يكون شفافا - تعمل الدالة بشكل مماثل لإصدارها غير المهين، ما يجعلها قابلة للاختبار بشكل كامل في عزلة.

متى تستخدم @step

استخدم @step على الدالات المكلفة لإعادة التشغيل: استدعاءات العامل أو طلبات واجهة برمجة التطبيقات الخارجية أو أي عملية تكون فيها إعادة التنفيذ على السيرة الذاتية مكلفة أو لها آثار جانبية. لا تزال الدالات العادية (بدون @step) تعمل داخل @workflow؛ فهي ببساطة تعيد التنفيذ عند استئناف سير العمل.

from agent_framework import InMemoryCheckpointStorage, step, workflow

@step  # cached — won't re-run on resume
async def call_llm(prompt: str) -> str:
    return (await agent.run(prompt)).text

# No @step — cheap, fine to re-run
async def validate(text: str) -> bool:
    return len(text) > 0

@workflow
async def pipeline(topic: str) -> str:
    draft = await call_llm(f"Write about: {topic}")
    ok = await validate(draft)
    return draft if ok else ""

storage = InMemoryCheckpointStorage()
workflow_instance = pipeline.build(checkpoint_storage=storage)

@step يقبل أيضا معلمة name :

@step(name="transform")
async def transform_data(raw: dict) -> str:
    ...

التحكم في هوية إعادة التشغيل

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

@step(replay_key=lambda connection, record_id: record_id)
async def load_record(connection: object, record_id: str) -> dict:
    return await connection.load(record_id)

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

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

راجع python/samples/03-workflows/functional/steps_and_checkpointing.py للحصول على مثال كامل.

RunContext

RunContext هو سياق التنفيذ الذي تم إدخاله في وظائف سير العمل والخطوة. تحتاج إليها فقط عند استخدام HITL أو حالة المفتاح/القيمة أو الأحداث المخصصة.

استيراده من agent_framework:

from agent_framework import RunContext, workflow

ctx.request_info() — الإنسان في الحلقة

ctx.request_info() تعليق سير العمل لانتظار الإدخال الخارجي:

@workflow
async def review_pipeline(topic: str, ctx: RunContext) -> str:
    draft = await write_draft(topic)
    feedback = await ctx.request_info(
        {"draft": draft, "instructions": "Please review this draft"},
        response_type=str,
        request_id="review_request",
    )
    return await revise_draft(draft, feedback)

البارامترات:

المعلمه Type Description
request_data Any البيانات الأساسية التي تصف الإدخال المطلوب (لوحة، نموذج Pydantic، سلسلة، ...).
response_type type Python المتوقع نوع الاستجابة.
request_id str | None معرف ثابت لهذا الطلب. إذا تم حذفه، يتم إنشاء معرف محدد auto::<index> من أمر الاتصال.

إعادة تشغيل الدلالات: عند التنفيذ الأول، request_info() يرفع إشارة داخلية توقف سير العمل مؤقتا. الإشارة غير مرئية أبدا للتعليمات البرمجية الخاصة بك. يتلقى WorkflowRunResult المتصل مع get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS.

لاستئناف العمل، اتصل .run(responses={request_id: value}) بنفس سير العمل المضمن. يعيد سير العمل التنفيذ من الأعلى، ويعيد request_info() القيمة المقدمة على الفور.

@step-مزينة الوظائف التي تم تشغيلها قبل التعليق إرجاع نتائجها المخزنة مؤقتا على استئناف بدلا من إعادة التنفيذ.

قبل أن يبلغ سير العمل عن الإيقاف المؤقت، ينتظر Agent Framework المكالمات المتزامنة @step النشطة لتسوية وتسجيل النتائج المكتملة لإعادة تشغيلها. في السيرة الذاتية، ترجع هذه الفروع نتائجها المخزنة مؤقتا بدلا من تكرار استدعاءات النموذج أو استدعاءات الأدوات أو عمليات الكتابة أو الآثار الجانبية الأخرى. يتم إلغاء مهام سير العمل المنفصلة المتبقية واستنزافها حتى لا يستمر العمل بعد إرجاع التشغيل المتوقف مؤقتا. ينطبق هذا السلوك على أنماط التزامن الأصلية مثل asyncio.gather و asyncio.TaskGroup.

بالنسبة إلى عدم الاستجابةNone ، request_info() يفرض القيمة response_type المقدمة ويتحقق من صحتها قبل إرجاعها. يمكن أن ينتج عن التعيين نموذج معلن، وتصبح Content السلسلة عندما response_type=Content. إذا لم يتمكن الإكراه من إنتاج النوع المطلوب، فإن السيرة الذاتية تثير ValueError. يحافظ الصريح None على السلوك الحالي: يسجل سير العمل تحذيرا ويرجع None.

معالجة الاستجابة:

workflow_instance = review_pipeline.build()

# Phase 1 — run until the workflow pauses
result1 = await workflow_instance.run("AI Safety")
assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS

requests = result1.get_request_info_events()
print(requests[0].request_id)  # "review_request"

# Phase 2 — resume with the human's answer
result2 = await workflow_instance.run(
    responses={"review_request": "Add more details about alignment research"}
)
print(result2.text)

راجع python/samples/03-workflows/functional/hitl_review.py للحصول على مثال كامل.

ctx.request_info() كما يتم دعم داخل @step الوظائف.

ctx.add_event() — أحداث مخصصة

يستخدم ctx.add_event() لإصدار أحداث خاصة بالتطبيق إلى جانب أحداث دورة حياة إطار العمل. للحصول على التفاصيل والأمثلة الكاملة، راجع إصدار الأحداث المخصصة.

ctx.get_state() / ctx.set_state() — حالة المفتاح/القيمة

استخدم ctx.get_state() و ctx.set_state() لتخزين القيم التي تستمر عبر مقاطعات HITL والمضمنة في نقاط التحقق. للحصول على التفاصيل الكاملة، راجع حالة سير العمل.

يجب أن تكون قيم الحالة قابلة للتسلسل JSON عند تكوين تخزين نقطة التحقق.

ctx.is_streaming()

إرجاع True وقت بدء التشغيل الحالي باستخدام stream=True. وظائف خطوة مفيدة داخل تريد ضبط سلوكها استنادا إلى وضع الدفق.

get_run_context()

استرداد نشط RunContext من أي مكان داخل سير عمل قيد التشغيل — مفيد في وظائف المساعد التي لا تعلن عن معلمة ctx :

from agent_framework import get_run_context

async def helper():
    ctx = get_run_context()
    if ctx is not None:
        ctx.set_state("helper_ran", True)

إرجاع None عند استدعاؤه خارج سير عمل قيد التشغيل.

التوازي مع asyncio.gather

استخدم التزامن Python القياسي ل fan-out/fan-in - لا توجد حاجة إلى بدائيات إطار العمل:

import asyncio
from agent_framework import workflow

@workflow
async def research_pipeline(topic: str) -> str:
    web, papers, news = await asyncio.gather(
        research_web(topic),
        research_papers(topic),
        research_news(topic),
    )
    return await synthesize([web, papers, news])

asyncio.gather يعمل أيضا عندما يتم تزيين الوظائف ب @step. يتم تخزين كل فرع مؤقتا بواسطة هوية إعادة التشغيل الخاصة به. أضف replay_key عندما تميز الوسيطات المبهمة أو الحالة الملتقطة المكالمات المتزامنة إلى نفس الخطوة.

راجع python/samples/03-workflows/functional/parallel_pipeline.py للحصول على مثال كامل.

استدعاء الوكلاء داخل مهام سير العمل

تعمل استدعاءات العامل كمكالمات دالة عادية داخل @workflow:

from agent_framework import Agent, workflow

writer = Agent(name="WriterAgent", instructions="Write a short poem.", client=client)
reviewer = Agent(name="ReviewerAgent", instructions="Review the poem.", client=client)

@workflow
async def poem_workflow(topic: str) -> str:
    poem = (await writer.run(f"Write a poem about: {topic}")).text
    review = (await reviewer.run(f"Review this poem: {poem}")).text
    return f"Poem:\n{poem}\n\nReview: {review}"

أضف @step إلى دوال استدعاء العامل عندما تريد تخزين نتائجها مؤقتا عبر سير HITL الذاتية أو عمليات استعادة نقطة التحقق:

from agent_framework import step

@step
async def write_poem(topic: str) -> str:
    return (await writer.run(f"Write a poem about: {topic}")).text

راجع python/samples/03-workflows/functional/agent_integration.py للحصول على مثال كامل.

.as_agent() — استخدام سير عمل كعامل

FunctionalWorkflowأنشئ ، ثم قم بتضمينه ككائن متوافق مع العامل مع .as_agent():

from agent_framework import workflow

@workflow
async def poem_workflow(topic: str) -> str:
    ...

# Build the workflow and wrap it as an agent
agent = poem_workflow.build().as_agent(name="PoemAgent")

# Use with the standard agent interface
response = await agent.run("Write a poem about the ocean")
print(response.text)

# Or use in a larger workflow or orchestration

.as_agent() FunctionalWorkflowAgent إرجاع الذي يعرض نفس run() الواجهة كعناصر عامل أخرى، ما يجعل مهام سير العمل الوظيفية قابلة للتكوين مع أي نظام يقبل العوامل.

المعلمه Type Description
name str | None اسم العرض للوكيل. الإعدادات الافتراضية لاسم سير العمل.
description str | None تجاوز الوصف الاختياري. الإعدادات الافتراضية لوصف سير العمل.
context_providers Sequence[Any] | None موفرو السياق الاختياريون لإقرانهم بالعامل.

Samples

توجد أمثلة قابلة للتشغيل في نماذج المجلدات التالية:

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

منشئ سير العمل وتنفيذه

المواضيع ذات الصلة:

لا تتوفر واجهة برمجة تطبيقات سير العمل الوظيفي ل C# في الوقت الحالي.

لا تتوفر واجهة برمجة تطبيقات سير العمل الوظيفي ل Go في الوقت الحالي. استخدم واجهات برمجة تطبيقات سير عمل الرسم البياني في منشئ سير العمل والتنفيذ.

  • تاريخ آخر تحديث في