منفذ العامل

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

نظرة عامة

يقوم "منفذ العامل" بسد الفجوة بين تجريد العامل ونموذج تنفيذ سير العمل. تكنولوجيا المعلومات:

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

كيفية عملها

في C#، يقوم محرك سير العمل داخليا بإنشاء AIAgentHostExecutor لكل AIAgent تمت إضافته إلى سير العمل. يقوم هذا المنفذ المتخصص بتوسيع ChatProtocolExecutor نمط الرمز المميز بدوره واستخدامه :

التخزين المؤقت للرسائل — عند وصول الرسائل من منفذين آخرين، يجمعها منفذ العامل. إذا ForwardIncomingMessages تم تمكين (الافتراضي)، يتم أيضا إعادة توجيه الرسائل الواردة إلى منفذي انتقال البيانات من الخادم.
تشغيل الرمز المميز — يعالج العامل رسائله المخزنة مؤقتا فقط بعد تلقي TurnToken.
استدعاء العامل — يستدعي RunAsync المنفذ (غير متدفق) أو RunStreamingAsync (دفق) على العامل الأساسي.
الناتج - إذا تم تمكين أحداث الدفق، يتم إرجاع كل تزايدي AgentResponseUpdate كإخراج سير عمل. إذا EmitAgentResponseEvents تم تمكينه، يتم أيضا إرجاع التجميع AgentResponse كإخراج سير عمل.
المراسلة النهائية — يتم إرسال رسائل استجابة العامل إلى منفذي انتقال البيانات من الخادم المتصلين.
تمرير الرمز المميز - بعد إكمال دوره، يرسل المنفذ انتقال بيانات جديد TurnToken بحيث يمكن للعامل التالي في السلسلة بدء المعالجة.

Tip

قد تتطلب بعض السيناريوهات منفذ وكيل أكثر تخصصا؛ على سبيل المثال، تستخدم تنسيقات التسليم مخصصا HandoffAgentExecutor مع منطق توجيه مخصص.

الإنشاء الضمني مقابل الإنشاء الصريح

عند تمرير AIAgent إلى WorkflowBuilder، يلتف إطار العمل تلقائيا في AIAgentBinding، والذي ينشئ الأساسي AIAgentHostExecutor. لا تحتاج إلى إنشاء مثيل لمنفذ العامل مباشرة.

AIAgent writerAgent = /* create your agent */;
AIAgent reviewerAgent = /* create your agent */;

// Agents are automatically wrapped — no manual executor creation required
var workflow = new WorkflowBuilder(writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

يمكنك أيضا استخدام أساليب المساعد على AgentWorkflowBuilder للأنماط الشائعة:

// Build a sequential pipeline of agents
var workflow = AgentWorkflowBuilder.BuildSequential(writerAgent, reviewerAgent);

تكوين مخصص

لتخصيص كيفية تصرف منفذ العامل، استخدم BindAsExecutor مع AIAgentHostOptions:

var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
    EmitAgentResponseEvents = true,
    ReassignOtherAgentsAsUsers = true,
    ForwardIncomingMessages = true,
};

ExecutorBinding writerBinding = writerAgent.BindAsExecutor(options);
var workflow = new WorkflowBuilder(writerBinding)
    .AddEdge(writerBinding, reviewerAgent)
    .Build();

أنواع المدخلات

يقبل منفذ العامل في C# أنواع إدخال متعددة: stringو ChatMessageو IEnumerable<ChatMessage>. يتم تحويل مدخلات السلسلة تلقائيا إلى ChatMessage مثيلات مع User الدور. يتم تجميع جميع الرسائل الواردة حتى TurnToken يتم تلقي، وعند هذه النقطة يقوم المنفذ بمعالجة الدفعة. عند ReassignOtherAgentsAsUsers تمكين (الافتراضي)، تتم إعادة تعيين الرسائل من عوامل أخرى إلى User الدور بحيث يعاملها النموذج الأساسي كإدخالات للمستخدم، بينما تحتفظ Assistant الرسائل من العامل الحالي بالدور.

الإخراج والتسلسل

بعد أن يكمل العامل دوره، يقوم المنفذ:

يرسل رسائل استجابة العامل إلى جميع منفذي انتقال البيانات من الخادم المتصلين.
إعادة توجيه جديد TurnToken حتى يتمكن العامل التالي في السلسلة من بدء المعالجة.

وهذا يجعل وكلاء التسلسل مباشرين — ما عليك سوى توصيلهم بالحواف:

var workflow = new WorkflowBuilder(frenchTranslator)
    .AddEdge(frenchTranslator, spanishTranslator)
    .AddEdge(spanishTranslator, englishTranslator)
    .Build();

سلوك الدفق

يتم التحكم في سلوك الدفق بواسطة EmitAgentUpdateEvents الخيار على AIAgentHostOptions، أو ديناميكيا عبر TurnToken:

عند التمكين — يستدعي RunStreamingAsync المنفذ العامل وينتج كل AgentResponseUpdate منها كحدث إخراج سير عمل. يوفر هذا تحديثات الرمز المميز في الوقت الحقيقي.
عند التعطيل — يستدعي RunAsync المنفذ وينتج استجابة كاملة واحدة.

// Enable streaming events at the configuration level
var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
};

// Or enable streaming dynamically via TurnToken
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

جلسات العمل المشتركة

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

خيارات التكوين

AIAgentHostOptions يتحكم في سلوك منفذ العامل:

خيار	Default	Description
`EmitAgentUpdateEvents`	`null`	إصدار أحداث تحديث البث أثناء التنفيذ. `TurnToken` له الأسبقية إذا تم تعيينه. إذا كان كلاهما ، `null`يتم تعطيل الدفق.
`EmitAgentResponseEvents`	`false`	إرسال استجابة العامل المجمعة كحدث إخراج سير عمل.
`InterceptUserInputRequests`	`false`	اعترضها `UserInputRequestContent` وتوجيهها كرسالة سير عمل للتعامل معها.
`InterceptUnterminatedFunctionCalls`	`false`	اعترض `FunctionCallContent` دون نتيجة مقابلة وقم بتوجيهها كرسالة سير عمل.
`ReassignOtherAgentsAsUsers`	`true`	أعد تعيين الرسائل من عوامل أخرى إلى `User` الدور بحيث يعاملها النموذج على أنها مدخلات المستخدم.
`ForwardIncomingMessages`	`true`	إعادة توجيه الرسائل الواردة إلى منفذي انتقال البيانات من الخادم قبل الرسائل التي أنشأها العامل.

التحقق

يدعم منفذ العامل نقاط التحقق لسير العمل طويل الأمد. عند أخذ نقطة تحقق، يقوم المنفذ بالتسلسل:

حالة جلسة العامل (عبر SerializeSessionAsync).
تكوين انبعاث حدث الدور الحالي (موجود فقط في حين أن الطلبات معلقة ولم ينتج عن المنفذ بعد الواردة TurnToken).
أي طلبات إدخال مستخدم معلقة وطلبات استدعاء دالة.

عند الاستعادة، يقوم المنفذ بإلغاء تسلسل جلسة العمل وحالة الطلب المعلقة، ما يسمح باستئناف سير العمل من حيث توقف.

كيفية عملها

تقوم AgentExecutor الفئة بتضمين عامل يقوم بتنفيذ SupportsAgentRun البروتوكول. عندما يتلقى المنفذ رسالة:

تسوية الرسائل — تتم تسوية الإدخال في قائمة Message الكائنات وإضافته إلى ذاكرة التخزين المؤقت الداخلية للمنفذ. يقبل المنفذ أنواع إدخال متعددة — strو AgentExecutorRequestMessagelist[str | Message]AgentExecutorResponse — يتم توجيه كل منها إلى معالج مخصص يقوم بتطبيع الإدخال قبل التخزين المؤقت.
استدعاء العامل — يستدعي agent.run() المنفذ مع الرسائل المخزنة مؤقتا، ويحدد تلقائيا وضع الدفق أو غير المتدفق استنادا إلى وضع تنفيذ سير العمل.
انبعاث الإخراج - في وضع الدفق، يتم إرجاع كل AgentResponseUpdate منها كحدث إخراج سير عمل. في وضع عدم الدفق، يتم إنتاج واحد AgentResponse .
إرسال انتقال البيانات من الخادم — بعد اكتمال العامل، يرسل AgentExecutorResponse المنفذ إلى جميع منفذي انتقال البيانات من الخادم المتصلين. تتضمن هذه الاستجابة محفوظات المحادثات الكاملة، مما يتيح تسلسلا سلسا.
إعادة تعيين ذاكرة التخزين المؤقت - يتم مسح ذاكرة التخزين المؤقت للرسائل الداخلية للمنفذ بعد استدعاء العامل، مما يضمن أن كل استدعاء عامل يعالج الرسائل الجديدة المستلمة فقط منذ الاستدعاء الأخير.

Tip

قد تتطلب بعض السيناريوهات منفذ وكيل أكثر تخصصا؛ على سبيل المثال، تستخدم تنسيقات التسليم منفذا مخصصا مع منطق توجيه مخصص.

الإنشاء الضمني مقابل الإنشاء الصريح

يقوم WorkflowBuilder تلقائيا بتضمين العوامل في AgentExecutor الحالات عند تمرير عامل مباشرة. بالنسبة لمعظم مهام سير العمل، يكون الإنشاء الضمني كافيا:

from agent_framework import WorkflowBuilder

writer_agent = client.as_agent(name="Writer", instructions="...")
reviewer_agent = client.as_agent(name="Reviewer", instructions="...")

# Agents are automatically wrapped — no manual AgentExecutor creation required
workflow = (
    WorkflowBuilder(start_executor=writer_agent)
    .add_edge(writer_agent, reviewer_agent)
    .build()
)

إنشاء صريح

AgentExecutor إنشاء بشكل صريح عندما تحتاج إلى:

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

from agent_framework import AgentExecutor

writer_executor = AgentExecutor(writer_agent, id="my-writer")
reviewer_executor = AgentExecutor(reviewer_agent, id="my-reviewer")

workflow = (
    WorkflowBuilder(start_executor=writer_executor)
    .add_edge(writer_executor, reviewer_executor)
    .build()
)

معلمات الدالة الإنشائية:

المعلمه	Type	Description
`agent`	`SupportsAgentRun`	العامل المراد التفافه.
`session`	`AgentSession \\| None`	جلسة العمل لاستخدامها في تشغيل العامل. إذا `None`، يتم إنشاء جلسة عمل جديدة من العامل.
`id`	`str \\| None`	معرف المنفذ الفريد. الإعدادات الافتراضية لاسم العامل إذا كانت متوفرة.
`context_mode`	`"full" \\| "last_agent" \\| "custom" \\| None`	يتحكم في كيفية معالجة سياق المحادثة عند تلقي من `AgentExecutorResponse` عامل المصدر. الإعدادات الافتراضية ل `"full"`، والتي توفر المحادثة الكاملة للعامل المصدر (الإدخال + الاستجابة). راجع أوضاع السياق.
`context_filter`	`Callable[[list[Message]], list[Message]] \\| None`	دالة تصفية مخصصة لتحديد الرسائل التي يجب تضمينها. مطلوب إذا كان `context_mode` هو `"custom"`.

Tip

معرف المنفذ هو أيضا المفتاح المستخدم عند استهداف workflow.run(function_invocation_kwargs=...) أو client_kwargs= في وكلاء فرديين. إذا حذفت id، يستخدم سير العمل اسم العامل الملتف.

أنواع المدخلات

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

نوع الإدخال	معالج	عامل المشغلات	Description
`AgentExecutorRequest`	`run`	شرطي	نوع الإدخال المتعارف عليه. يحتوي على قائمة بالرسائل وعلامة `should_respond` تتحكم في ما إذا كان العامل يعمل أم لا.
`str`	`from_str`	دائما	يقبل مطالبة سلسلة أولية.
`Message`	`from_message`	دائما	يقبل كائنا واحدا `Message` .
`list[str \\| Message]`	`from_messages`	دائما	يقبل قائمة بالسلاسل أو `Message` الكائنات كت سياق محادثة.
`AgentExecutorResponse`	`from_response`	دائما	يقبل استجابة منفذ وكيل سابق، ما يتيح التسلسل المباشر.

استخدام AgentExecutorRequest

AgentExecutorRequest هو نوع الإدخال المتعارف عليه ويوفر أكبر قدر من التحكم:

from agent_framework import AgentExecutorRequest, Message

# Create a request with messages
request = AgentExecutorRequest(
    messages=[Message(role="user", contents=["Hello, world!"])],
    should_respond=True,
)

# Run the workflow
result = await workflow.run(request)

should_respond تتحكم العلامة في ما إذا كان العامل يعالج الرسائل على الفور أو ببساطة يقوم بالتخزين المؤقت لها في وقت لاحق:

True (افتراضي) — يعمل العامل وينتج استجابة.
False — تتم إضافة الرسائل إلى ذاكرة التخزين المؤقت ولكن العامل لا يعمل. هذا مفيد للتحميل المسبق لسياق المحادثة قبل تشغيل استجابة.

الإخراج والتسلسل

بعد اكتمال العامل، يرسل المنفذ انتقال البيانات AgentExecutorResponse من الخادم. تحتوي فئة البيانات هذه على:

الحقل	Type	Description
`executor_id`	`str`	معرف المنفذ الذي أنتج الاستجابة.
`agent_response`	`AgentResponse`	استجابة العامل الأساسي (لم يتم تغييرها من العميل).
`full_conversation`	`list[Message]`	سياق المحادثة الكامل (المدخلات السابقة + مخرجات العامل) للتسلسل.

عند ربط منفذي العامل، يتلقى AgentExecutorResponse منفذ انتقال البيانات من الخادم عبر from_response المعالج. بشكل افتراضي، يستخدم full_conversation الحقل للحفاظ على محفوظات المحادثات الكاملة، مما يمنع عوامل انتقال البيانات من الخادم من فقدان السياق السابق. يمكنك تغيير هذا السلوك باستخدام أوضاع السياق:

spam_detector = AgentExecutor(create_spam_detector_agent())
email_assistant = AgentExecutor(create_email_assistant_agent())

# The email_assistant receives the spam_detector's full conversation context
workflow = (
    WorkflowBuilder(start_executor=spam_detector)
    .add_edge(spam_detector, email_assistant)
    .build()
)

سلوك الدفق

يتكيف AgentExecutor تلقائيا مع وضع تنفيذ سير العمل:

stream=True — يستدعي agent.run(stream=True) وينتج كل AgentResponseUpdate منها كحدث إخراج سير عمل. بعد اكتمال البث، يتم تجميع التحديثات في نسخة كاملة AgentResponse لإرسال انتقال البيانات من الخادم.
stream=False (افتراضي) — يستدعي agent.run(stream=False) وينتج حدثا واحدا AgentResponse كإخراج سير عمل.

# Streaming mode — receive incremental updates
events = workflow.run("Write a story about a cat.", stream=True)
async for event in events:
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        print(event.data.text, end="", flush=True)

# Non-streaming mode — receive complete response
result = await workflow.run("Write a story about a cat.")

# Retrieve terminal AgentResponse objects from the result
outputs = result.get_outputs()
for output in outputs:
    if isinstance(output, AgentResponse):
        print(output.text)

# Retrieve intermediate outputs (progress / observational emissions)
intermediate_outputs = result.get_intermediate_outputs()
for item in intermediate_outputs:
    print(f"Intermediate: {item}")

أوضاع السياق

عندما يتم ربط العوامل معا، تتحكم المعلمة context_mode على AgentExecutor سياق المحادثة الذي يستهلكه العامل عندما يتلقى AgentExecutorResponse من عامل المصدر عبر from_response المعالج.

الأوضاع المتوفرة

وضع	السلوك
`"full"` (افتراضي)	يستهلك العامل المحادثة الكاملة للعامل المصدر — كل من رسائل الإدخال المقدمة إلى العامل المصدر ورسائل الاستجابة الخاصة به.
`"last_agent"`	يستهلك العامل رسائل استجابة العامل المصدر فقط، باستثناء الإدخال الذي تم توفيره للعامل المصدر.
`"custom"`	تحدد الدالة التي يوفرها `context_filter` المستخدم الرسائل التي يستهلكها العامل. يتطلب المعلمة `context_filter` .

استخدام `last_agent` الوضع

استخدم "last_agent" عندما يجب أن يركز كل عامل فقط على تحويل إخراج العامل السابق دون التأثر بتحولات المحادثة السابقة. وهذا مفيد لمسارات الترجمة والتحسين التدريجي والتحويلات المتسلسلة المماثلة:

from agent_framework import AgentExecutor, WorkflowBuilder

# Each agent consumes only the previous agent's response messages
french_executor = AgentExecutor(french_agent, context_mode="last_agent")
spanish_executor = AgentExecutor(spanish_agent, context_mode="last_agent")

workflow = (
    WorkflowBuilder(start_executor=writer_agent)
    .add_edge(writer_agent, french_executor)
    .add_edge(french_executor, spanish_executor)
    .build()
)

مع context_mode="last_agent"، يستهلك المترجم الفرنسي رسائل استجابة الكاتب فقط (باستثناء مطالبة المستخدم الأصلية التي كانت مدخلات للكاتب)، ويستهلك المترجم الأسباني رسائل استجابة المترجم الفرنسي فقط.

استخدام `custom` الوضع

للتحكم الدقيق في السياق الذي يستهلكه العامل، استخدم context_mode="custom" مع دالة context_filter . يتلقى عامل التصفية المحادثة الكاملة ك list[Message] وإرجاع المجموعة الفرعية التي تمت تصفيتها:

from agent_framework import AgentExecutor, Message

def keep_user_and_last_agent(messages: list[Message]) -> list[Message]:
    """Keep only user messages and the last agent's response."""
    user_msgs = [m for m in messages if m.role == "user"]
    agent_msgs = [m for m in messages if m.role == "assistant"]
    return user_msgs + agent_msgs[-1:] if agent_msgs else user_msgs

executor = AgentExecutor(
    my_agent,
    context_mode="custom",
    context_filter=keep_user_and_last_agent,
)

أوضاع السياق في SequentialBuilder

SequentialBuilder يوفر التنسيق معلمة ملائمة chain_only_agent_responses تقوم بتكوين جميع المشاركين في الوكيل لاستخدام context_mode="last_agent"، بحيث يستهلك كل عامل رسائل استجابة العامل السابق فقط:

from agent_framework.orchestrations import SequentialBuilder

workflow = SequentialBuilder(
    participants=[writer, translator, reviewer],
    chain_only_agent_responses=True,
).build()

للحصول على مثال كامل، راجع sequential_chain_only_agent_responses.py في مستودع إطار عمل العامل.

جلسات العمل المشتركة

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

from agent_framework import AgentExecutor

# Create a shared session from one agent
shared_session = writer_agent.create_session()

# Both executors share the same session
writer_executor = AgentExecutor(writer_agent, session=shared_session)
reviewer_executor = AgentExecutor(reviewer_agent, session=shared_session)

Note

لا يدعم جميع الوكلاء الجلسات المشتركة. عادة، يمكن فقط للوكلاء من نفس نوع الموفر مشاركة جلسة عمل.

التحقق

AgentExecutor يدعم نقاط التحقق لحفظ الحالة واستعادتها في مهام سير العمل طويلة الأمد. عند أخذ نقطة تحقق، يقوم المنفذ بالتسلسل:

ذاكرة التخزين المؤقت للرسائل الداخلية.
محفوظات المحادثات الكاملة.
حالة جلسة العامل.
أي طلبات واستجابات إدخال مستخدم معلقة.

عند الاستعادة، يقوم المنفذ بإلغاء تسلسل هذه الحالة، مما يسمح باستئناف سير العمل من حيث توقف.

تحذير

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

كيفية عملها

انتقل إلى وكلاء المضيفين كمنفذي سير العمل مع workflow/agentworkflow. يستخدم المنفذ المستضاف نمط رمز الدور التالي :

التخزين المؤقت للرسائل — عند وصول الرسائل من منفذين آخرين، يجمعها العامل المستضاف. إذا تم تمكين إعادة توجيه الرسائل (الافتراضي)، تتم أيضا إعادة توجيه الرسائل الواردة إلى منفذي انتقال البيانات من الخادم.
تشغيل الرمز المميز — يعالج العامل المستضاف رسائله المخزنة مؤقتا فقط بعد تلقي workflow.TurnToken.
استدعاء العامل — يستدعي المنفذ العامل الأساسي من خلال Run ويختار سلوك الدفق من agentworkflow.Config أو TurnToken.
الناتج - إذا تم تمكين أحداث التحديث، يتم إرجاع كل *agent.ResponseUpdate منها كإخراج سير عمل. إذا تم تمكين أحداث الاستجابة، يتم إرجاع التجميع *agent.Response كإخراج سير عمل.
المراسلة النهائية — يتم إرسال رسائل استجابة العامل إلى منفذي انتقال البيانات من الخادم المتصلين.
تشغيل تمرير الرمز المميز — بعد اكتمال الدور، يرسل المنفذ انتقالا جديدا workflow.TurnToken حتى يتمكن العامل المستضاف التالي من بدء المعالجة.

تكوين مخصص

تخصيص كيفية تصرف منفذ العامل المستضاف عن طريق إنشاء الربط بقيمة agentworkflow.Newagentworkflow.Config و:

hostedAgent := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents: true,
    DisableForwardIncomingMessages: true,
})

wf, err := workflow.NewBuilder(hostedAgent).
    WithOutputFrom(hostedAgent).
    Build()
if err != nil {
    return err
}

Tip

راجع العوامل في نموذج مهام سير العمل للحصول على مثال كامل قابل للتشغيل.

أنواع المدخلات

يقبل منفذ العامل المستضاف الإدخالات string*message.Message[]*message.Messageوiter.Seq[*message.Message]. يتم تحويل مدخلات السلسلة إلى message.Message مثيلات مع User الدور. يتم تخزين إدخالات الرسائل مؤقتا حتى يتلقى workflow.TurnTokenالمنفذ ، ما يؤدي إلى تشغيل العامل المستضاف على الدفعة المتراكمة.

run, err := inproc.Default.RunStreaming(ctx, wf, nil)
if err != nil {
    return err
}
defer run.Close(ctx)

if err := run.SendMessage(ctx, "Summarize this deployment plan."); err != nil {
    return err
}
if err := run.SendMessage(ctx, message.NewText("Include risk notes.")); err != nil {
    return err
}
if err := run.SendMessage(ctx, []*message.Message{message.NewText("Keep it concise.")}); err != nil {
    return err
}

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

الإخراج والتسلسل

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

french := agentworkflow.New(frenchAgent, agentworkflow.Config{})
spanish := agentworkflow.New(spanishAgent, agentworkflow.Config{})
english := agentworkflow.New(englishAgent, agentworkflow.Config{})

wf, err := workflow.NewBuilder(french).
    AddEdge(french, spanish).
    AddEdge(spanish, english).
    Build()
if err != nil {
    return err
}

سلوك الدفق

اضبط EmitUpdateEvents على agentworkflow.Config، أو أرسل workflow.TurnToken مع EmitEvents مجموعة، لإرسال تحديثات استجابة العامل من خلال أحداث إخراج سير العمل.

hostedAgent := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents: true,
})

wf, err := workflow.NewBuilder(hostedAgent).
    WithOutputFrom(hostedAgent).
    Build()
if err != nil {
    return err
}

run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Write a status update."))
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        if update, ok := output.Output.(*agent.ResponseUpdate); ok {
            fmt.Print(update.String())
        }
    }
}

خيارات التكوين

agentworkflow.Config يتحكم في سلوك منفذ العامل المستضاف:

خيار	Default	Description
`EmitUpdateEvents`	`false`	إرسال قيم دفق `*agent.ResponseUpdate` أثناء التنفيذ. `workflow.TurnToken.EmitEvents` له الأسبقية عند التعيين.
`EmitResponseEvents`	`false`	إصدار مجمع `*agent.Response` كحدث إخراج سير عمل.
`InterceptUserInputRequests`	`false`	اعترضها `ToolApprovalRequestContent` وتوجيهها كرسالة سير عمل للتعامل معها.
`InterceptUnterminatedFunctionCalls`	`false`	اعترض القيم التي لم يتم `FunctionCallContent` حلها وقم بتوجيهها كرسائل سير عمل.
`DisableReassignOtherAgentsAsUsers`	`false`	الاحتفاظ بأدوار المساعد الواردة من عوامل أخرى بدلا من إعادة تعيينها إلى دور المستخدم.
`DisableForwardIncomingMessages`	`false`	توقف عن إعادة توجيه الرسائل الواردة إلى منفذي انتقال البيانات من الخادم قبل الرسائل التي أنشأها العامل المستضاف.

hostedAgent := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents:                  true,
    EmitResponseEvents:                true,
    InterceptUserInputRequests:        true,
    InterceptUnterminatedFunctionCalls: true,
    DisableReassignOtherAgentsAsUsers: false,
    DisableForwardIncomingMessages:    false,
})

التحقق

يشارك الوكلاء المستضافون في نقاط فحص سير العمل. agentworkflow.New تسجيل نقاط التحقق واستعادة خطافات على المنفذ. عند أخذ نقطة تحقق، يخزن المضيف:

حالة JSON للعامل المستضاف agent.Session .
إعداد انبعاث الحدث في الدور الحالي.
الموافقة على الأداة المعلقة وحالة طلب استدعاء الدالة.

عند الاستعادة، يعيد المضيف إنشاء جلسة العامل ويستعيد معالجات الطلب المعلقة قبل متابعة سير العمل. تمكين نقاط التحقق من خلال بيئة تنفيذ سير العمل، على سبيل المثال مع inproc.Default.WithCheckpointing(...)؛ لا يلزم خيار agentworkflow.Config .

checkpointManager := checkpoint.NewInMemoryManager()
environment := inproc.Default.WithCheckpointing(checkpointManager)

var checkpoints []workflow.CheckpointInfo
run, err := environment.RunStreaming(ctx, wf, message.NewText("Start the review."))
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

for evt, err := range run.WatchUntilHalt(ctx) {
    if err != nil {
        return err
    }
    if completed, ok := evt.(workflow.SuperStepCompletedEvent); ok && completed.CompletionInfo != nil {
        if completed.CompletionInfo.CheckpointInfo != nil {
            checkpoints = append(checkpoints, *completed.CompletionInfo.CheckpointInfo)
        }
    }
}

if len(checkpoints) == 0 {
    return fmt.Errorf("no checkpoints were created")
}

resumedRun, err := environment.ResumeStreaming(ctx, wf, checkpoints[len(checkpoints)-1])
if err != nil {
    return err
}
defer resumedRun.Close(ctx)

Note

لا يزال من الممكن أن يكون للجلسات المدعومة من الموفر حدود متانة خاصة بموفر الخدمة. تلتقط agent.Session نقاط التحقق الحالة المتوفرة لمضيف Go، وليس حالة الخدمة الخارجية التي لا يقوم الموفر بتسلسلها في جلسة العمل.

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

الوكلاء في مهام سير العمل

الملاحظات

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

Last updated on 2026-09-29

منفذ العامل

نظرة عامة

كيفية عملها

الإنشاء الضمني مقابل الإنشاء الصريح

تكوين مخصص

أنواع المدخلات

الإخراج والتسلسل

سلوك الدفق

جلسات العمل المشتركة

خيارات التكوين

التحقق

كيفية عملها

الإنشاء الضمني مقابل الإنشاء الصريح

إنشاء صريح

أنواع المدخلات

استخدام AgentExecutorRequest

الإخراج والتسلسل

سلوك الدفق

أوضاع السياق

الأوضاع المتوفرة

استخدام last_agent الوضع

استخدام custom الوضع

أوضاع السياق في SequentialBuilder

جلسات العمل المشتركة

التحقق

كيفية عملها

تكوين مخصص

أنواع المدخلات

الإخراج والتسلسل

سلوك الدفق

خيارات التكوين

التحقق

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

الملاحظات

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

استخدام `last_agent` الوضع

استخدام `custom` الوضع