AutoGen to Microsoft دليل ترحيل إطار عمل العامل

دليل شامل للترحيل من AutoGen إلى Microsoft Agent Framework Python SDK.

جدول المحتويات

خلفية

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

Microsoft Agent Framework هو SDK جديد متعدد اللغات لبناء وكلاء الذكاء الاصطناعي ومهام سير العمل باستخدام LLMs. وهو يمثل تطورا كبيرا للأفكار الرائدة في AutoGen ويتضمن الدروس المستفادة من الاستخدام في العالم الحقيقي. تم تطويره من قبل فرق AutoGen الأساسية وفرق نواة دلالية في Microsoft، وتم تصميمه ليكون أساسا جديدا لبناء تطبيقات الذكاء الاصطناعي من الآن فصاعدا.

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

أوجه التشابه والاختلافات الرئيسية

ما الذي يبقى كما هو

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

# Both frameworks follow similar patterns
# AutoGen
agent = AssistantAgent(name="assistant", model_client=client, tools=[my_tool])
result = await agent.run(task="Help me with this task")

# Agent Framework
agent = Agent(name="assistant", client=client, tools=[my_tool])
result = await agent.run("Help me with this task")

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

  1. نمط التزامن: يقوم AutoGen بإقران نواة مستندة إلى الحدث بمستوى Teamعال . يركز إطار عمل العامل على رسم بياني Workflow مكتوب يوجه البيانات على طول الحواف وينشط المنفذين عندما تكون المدخلات جاهزة.

  2. الأدوات: يقوم AutoGen بتضمين الوظائف مع FunctionTool. يستخدم @toolإطار عمل العامل ، ويستنتج المخططات تلقائيا، ويضيف أدوات مستضافة مثل مترجم التعليمات البرمجية والبحث على الويب.

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

  4. وقت التشغيل: يوفر AutoGen أوقات تشغيل موزعة مضمنة وتجريبية. يركز إطار عمل العامل على تكوين عملية واحدة اليوم؛ تم التخطيط للتنفيذ الموزع.

إنشاء عميل النموذج وتكوينه

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

الميزة AutoGen إطار عمل الوكيل
عميل OpenAI OpenAIChatCompletionClient OpenAIChatCompletionClient
عميل استجابات OpenAI ❌ غير متوفر OpenAIChatClient
Azure OpenAI AzureOpenAIChatCompletionClient OpenAIChatCompletionClient
Azure استجابات OpenAI ❌ غير متوفر OpenAIChatClient
ذكاء Azure الاصطناعي AzureAIChatCompletionClient FoundryChatClient / FoundryAgent
الأنثروبي AnthropicChatCompletionClient 🚧 مخطط له
أولاما OllamaChatCompletionClient 🚧 مخطط له
التخزين المؤقت ChatCompletionCache برنامج تضمين 🚧 مخطط له

عملاء نموذج AutoGen

from autogen_ext.models.openai import OpenAIChatCompletionClient, AzureOpenAIChatCompletionClient

# OpenAI
client = OpenAIChatCompletionClient(
    model="gpt-5",
    api_key="your-key"
)

# Azure OpenAI
client = AzureOpenAIChatCompletionClient(
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    azure_deployment="gpt-5",
    api_version="2024-12-01",
    api_key="your-key"
)

Agent Framework ChatClients

from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import AzureCliCredential

# OpenAI (reads API key from environment)
client = OpenAIChatCompletionClient(model="gpt-5")

# Azure OpenAI (pass explicit Azure routing inputs)
client = OpenAIChatCompletionClient(
    model="gpt-5",
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01",
    credential=AzureCliCredential(),
)

للحصول على أمثلة مفصلة، راجع:

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

يوفر إطار عمل OpenAIChatClient العامل دعم واجهة برمجة تطبيقات الاستجابات لكل من توجيه OpenAI المباشر Azure OpenAI، بما في ذلك نماذج المنطق والاستجابات المنظمة غير المتوفرة في AutoGen:

from agent_framework.openai import OpenAIChatClient
from azure.identity import AzureCliCredential

# Azure OpenAI with Responses API
azure_responses_client = OpenAIChatClient(
    model="gpt-5",
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01",
    credential=AzureCliCredential(),
)

# OpenAI with Responses API
openai_responses_client = OpenAIChatClient(model="gpt-5")

للحصول على أمثلة واجهة برمجة تطبيقات الاستجابات، راجع:

تعيين ميزات Single-Agent

يعين هذا القسم ميزات عامل واحد بين AutoGen و Agent Framework. مع وجود عميل في مكانه، أنشئ وكيلا، وأرفق الأدوات، واختر بين التنفيذ غير المتدفق والتدفق.

إنشاء العامل الأساسي وتنفيذه

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

AutoGen AssistantAgent

from autogen_agentchat.agents import AssistantAgent

agent = AssistantAgent(
    name="assistant",
    model_client=client,
    system_message="You are a helpful assistant.",
    tools=[my_tool],
    max_tool_iterations=1  # Single-turn by default
)

# Execution
result = await agent.run(task="What's the weather?")

عامل إطار عمل العامل

from agent_framework import Agent, tool
from agent_framework.openai import OpenAIChatClient

# Create simple tools for the example
@tool
def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Weather in {location}: sunny"

@tool
def get_time() -> str:
    """Get current time."""
    return "Current time: 2:30 PM"

# Create client
client = OpenAIChatClient(model="gpt-5")

async def example():
    # Direct creation with default options
    agent = Agent(
        name="assistant",
        client=client,
        instructions="You are a helpful assistant.",
        tools=[get_weather],  # Multi-turn by default
        default_options={
            "temperature": 0.7,
            "max_tokens": 1000,
        }
    )

    # Factory method (more convenient)
    agent = client.as_agent(
        name="assistant",
        instructions="You are a helpful assistant.",
        tools=[get_weather],
        default_options={"temperature": 0.7}
    )

    # Execution with runtime tool and options configuration
    result = await agent.run(
        "What's the weather?",
        tools=[get_time],  # Can add tools at runtime (keyword arg)
        options={"tool_choice": "auto"}  # Other options go in options dict
    )

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

  • السلوك الافتراضي: Agent يتكرر تلقائيا من خلال استدعاءات الأدوات، بينما AssistantAgent يتطلب إعدادا صريحا max_tool_iterations
  • تكوين وقت التشغيل: Agent.run() يقبل كوسيطة tools كلمة أساسية وخيارات أخرى عبر معلمة options الإملاء لتخصيص كل استدعاء
  • نظام الخيارات: يستخدم إطار عمل العامل الخيارات المستندة إلى TypedDict (على سبيل المثال، OpenAIChatOptions) لسلامة النوع والإكمال التلقائي ل IDE. يتم تمرير الخيارات عبر default_options في البناء وفي options وقت التشغيل
  • أساليب المصنع: يوفر إطار عمل العامل أساليب مصنع ملائمة مباشرة من عملاء الدردشة
  • إدارة الحالة: Agent عديمة الحالة ولا تحتفظ بمحفوظات المحادثات بين استدعاءات، على عكس AssistantAgent ما يحافظ على محفوظات المحادثات كجزء من حالتها

إدارة حالة المحادثة باستخدام AgentSession

لمتابعة المحادثات باستخدام Agent، استخدم AgentSession لإدارة محفوظات المحادثات:

# Assume we have an agent from previous examples
async def conversation_example():
    # Create a new session that will be reused
    session = agent.create_session()

    # First interaction - session is empty
    result1 = await agent.run("What's 2+2?", session=session)
    print(result1.text)  # "4"

    # Continue conversation - session contains previous messages
    result2 = await agent.run("What about that number times 10?", session=session)
    print(result2.text)  # "40" (understands "that number" refers to 4)

    # AgentSession can use external storage, similar to ChatCompletionContext in AutoGen

عديم الحالة بشكل افتراضي: عرض توضيحي سريع

# Without a session (two independent invocations)
r1 = await agent.run("What's 2+2?")
print(r1.text)  # for example, "4"

r2 = await agent.run("What about that number times 10?")
print(r2.text)  # Likely ambiguous without prior context; cannot be "40"

# With a session (shared context across calls)
session = agent.create_session()
print((await agent.run("What's 2+2?", session=session)).text)  # "4"
print((await agent.run("What about that number times 10?", session=session)).text)  # "40"

للحصول على أمثلة جلسة المحادثة، راجع:

تكافؤ وكيل مساعد OpenAI

لا يزال AutoGen يعرض OpenAIAssistantAgent، ولكن إرشادات إطار عمل العامل الحالي Python لم تعد تستخدم سطحا خاصا Python المساعدين. قم بالترحيل إلى عميل الاستجابات لعمل OpenAI المباشر أو Azure OpenAI، أو استخدمه FoundryAgent عندما تحتاج إلى وكيل مدار بواسطة الخدمة:

from agent_framework.openai import OpenAIChatClient
from agent_framework.foundry import FoundryAgent

للحصول على أمثلة Python حالية قابلة للمقارنة، راجع:

دعم البث المباشر

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

AutoGen Streaming

# Model client streaming
async for chunk in client.create_stream(messages):
    if isinstance(chunk, str):
        print(chunk, end="")

# Agent streaming
async for event in agent.run_stream(task="Hello"):
    if isinstance(event, ModelClientStreamingChunkEvent):
        print(event.content, end="")
    elif isinstance(event, TaskResult):
        print("Final result received")

تدفق إطار عمل العامل

# Assume we have client, agent, and tools from previous examples
async def streaming_example():
    # Chat client streaming - tools go in options dict
    async for chunk in client.get_response(
        "Hello",
        options={"tools": tools},
        stream=True,
    ):
        if chunk.text:
            print(chunk.text, end="")

    # Agent streaming - tools can be keyword arg on agents
    async for chunk in agent.run("Hello", tools=tools, stream=True):
        if chunk.text:
            print(chunk.text, end="", flush=True)

تلميح: في إطار عمل العامل، ينتج عن كل من العملاء والوكلاء نفس شكل التحديث؛ يمكنك القراءة chunk.text في كلتا الحالتين. لاحظ أنه بالنسبة لعملاء الدردشة، tools ينتقل إلى options الإملاء، بينما بالنسبة للوكلاء، tools يبقى وسيطة كلمة أساسية مباشرة.

أنواع الرسائل وإنشاءها

يعد فهم كيفية عمل الرسائل أمرا بالغ الأهمية للاتصال الفعال للوكيل. يوفر كلا الإطارين أساليب مختلفة لإنشاء الرسائل ومعالجتها، مع AutoGen باستخدام فئات رسائل منفصلة وإطار عمل العامل باستخدام نظام رسائل موحد.

أنواع الرسائل "إنشاء تلقائي"

from autogen_agentchat.messages import TextMessage, MultiModalMessage
from autogen_core.models import UserMessage

# Text message
text_msg = TextMessage(content="Hello", source="user")

# Multi-modal message
multi_modal_msg = MultiModalMessage(
    content=["Describe this image", image_data],
    source="user"
)

# Convert to model format for use with model clients
user_message = text_msg.to_model_message()

أنواع رسائل إطار عمل العامل

from agent_framework import Message, Content
import base64

# Text message
text_msg = Message(role="user", contents=["Hello"])

# Supply real image bytes, or use a data: URI/URL via Content.from_uri()
image_bytes = b"<your_image_bytes>"
image_b64 = base64.b64encode(image_bytes).decode()
image_uri = f"data:image/jpeg;base64,{image_b64}"

# Multi-modal message with mixed content
multi_modal_msg = Message(
    role="user",
    contents=[
        Content.from_text(text="Describe this image"),
        Content.from_uri(uri=image_uri, media_type="image/jpeg")
    ]
)

الفروقات الرئيسية:

  • يستخدم AutoGen فئات رسائل منفصلة (TextMessage، MultiModalMessage) مع source حقل
  • يستخدم إطار عمل العامل موحدا Message مع كائنات المحتوى المطبعية وحقل role
  • تستخدم Role رسائل إطار عمل العامل قائمة تعداد (المستخدم، المساعد، النظام، الأداة) بدلا من مصادر السلسلة

إنشاء الأداة وتكاملها

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

AutoGen FunctionTool

from autogen_core.tools import FunctionTool

async def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Weather in {location}: sunny"

# Manual tool creation
tool = FunctionTool(
    func=get_weather,
    description="Get weather information"
)

# Use with agent
agent = AssistantAgent(name="assistant", model_client=client, tools=[tool])

إطار عمل العامل @tool

from agent_framework import tool
from typing import Annotated
from pydantic import Field

@tool
def get_weather(
    location: Annotated[str, Field(description="The location to get weather for")]
) -> str:
    """Get weather for a location."""
    return f"Weather in {location}: sunny"

# Direct use with agent (automatic conversion)
agent = Agent(name="assistant", client=client, tools=[get_weather])

للحصول على أمثلة مفصلة، راجع:

الأدوات المستضافة (إطار عمل العامل حصري)

يوفر إطار عمل العامل أدوات مستضافة غير متوفرة في AutoGen:

from agent_framework.openai import OpenAIChatClient

# Responses client with a model that supports hosted tools
client = OpenAIChatClient(model="gpt-5")

# Hosted tools are created from the client
code_tool = client.get_code_interpreter_tool()
search_tool = client.get_web_search_tool()

agent = client.as_agent(
    name="researcher",
    instructions="Use the available hosted tools to research answers.",
    tools=[code_tool, search_tool]
)

للحصول على أمثلة مفصلة، راجع:

المتطلبات والمحاذير:

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

Note

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

الفرق الرئيسي: يعالج إطار عمل العامل تكرار الأدوات تلقائيا على مستوى العامل. على عكس معلمة max_tool_iterations AutoGen، يستمر وكلاء Agent Framework في تنفيذ الأدوات حتى الاكتمال بشكل افتراضي، مع آليات أمان مضمنة لمنع التكرارات الحلقية اللانهائية.

دعم خادم MCP

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

دعم AutoGen MCP

يحتوي AutoGen على دعم MCP أساسي من خلال الملحقات (تختلف تفاصيل التنفيذ المحددة حسب الإصدار).

دعم MCP لإطار العامل

from agent_framework import Agent, MCPStdioTool, MCPStreamableHTTPTool, MCPWebsocketTool
from agent_framework.openai import OpenAIChatClient

# Create client for the example
client = OpenAIChatClient(model="gpt-5")

# Stdio MCP server
mcp_tool = MCPStdioTool(
    name="filesystem",
    command="uvx mcp-server-filesystem",
    args=["/allowed/directory"]
)

# HTTP streaming MCP
http_mcp = MCPStreamableHTTPTool(
    name="http_mcp",
    url="http://localhost:8000/sse"
)

# WebSocket MCP
ws_mcp = MCPWebsocketTool(
    name="websocket_mcp",
    url="ws://localhost:8000/ws"
)

agent = Agent(name="assistant", client=client, tools=[mcp_tool])

للحصول على أمثلة MCP، راجع:

نمط العامل كأداة

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

AutoGen AgentTool

from autogen_agentchat.tools import AgentTool

# Create specialized agent
writer = AssistantAgent(
    name="writer",
    model_client=client,
    system_message="You are a creative writer."
)

# Wrap as tool
writer_tool = AgentTool(agent=writer)

# Use in coordinator (requires disabling parallel tool calls)
coordinator_client = OpenAIChatCompletionClient(
    model="gpt-5",
    parallel_tool_calls=False
)
coordinator = AssistantAgent(
    name="coordinator",
    model_client=coordinator_client,
    tools=[writer_tool]
)

as_tool إطار عمل العامل()

from agent_framework import Agent

# Assume we have client from previous examples
# Create specialized agent
writer = Agent(
    name="writer",
    client=client,
    instructions="You are a creative writer."
)

# Convert to tool
writer_tool = writer.as_tool(
    name="creative_writer",
    description="Generate creative content",
    arg_name="request",
    arg_description="What to write"
)

# Use in coordinator
coordinator = Agent(
    name="coordinator",
    client=client,
    tools=[writer_tool]
)

ملاحظة الترحيل الصريح: في AutoGen، قم بتعيين parallel_tool_calls=False على عميل نموذج المنسق عند التفاف العوامل كأدوات لتجنب مشكلات التزامن عند استدعاء نفس مثيل العامل. في إطار عمل العامل، as_tool() لا يتطلب تعطيل استدعاءات الأدوات المتوازية لأن العوامل عديمة الحالة بشكل افتراضي.

البرنامج الوسيط (ميزة إطار عمل العامل)

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

from agent_framework import Agent, AgentContext, FunctionInvocationContext
from typing import Callable, Awaitable

# Assume we have client from previous examples
async def logging_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]]
) -> None:
    print(f"Agent {context.agent.name} starting")
    await call_next()
    print(f"Agent {context.agent.name} completed")

async def security_middleware(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]]
) -> None:
    if "password" in str(context.arguments):
        print("Blocking function call with sensitive data")
        return  # Don't call call_next()
    await call_next()

agent = Agent(
    name="secure_agent",
    client=client,
    middleware=[logging_middleware, security_middleware]
)

Benefits:

  • الأمان: التحقق من صحة الإدخال وتصفية المحتوى
  • إمكانية الملاحظة: التسجيل والمقاييس والتتبع
  • الأداء: التخزين المؤقت وتحديد المعدل
  • معالجة الأخطاء: تدهور رشيق ومنطق إعادة المحاولة

للحصول على أمثلة تفصيلية للبرامج الوسيطة، راجع:

عوامل مخصصة

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

AutoGen: الفئة الفرعية BaseChatAgent

from typing import Sequence
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import BaseChatMessage, TextMessage, StopMessage
from autogen_core import CancellationToken

class StaticAgent(BaseChatAgent):
    def __init__(self, name: str = "static", description: str = "Static responder") -> None:
        super().__init__(name, description)

    @property
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:  # Which message types this agent produces
        return (TextMessage,)

    async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
        # Always return a static response
        return Response(chat_message=TextMessage(content="Hello from AutoGen custom agent", source=self.name))

ملاحظات:

  • تنفيذ on_messages(...) وإرجاع Response مع رسالة دردشة.
  • تنفيذ on_reset(...) اختياريا لمسح الحالة الداخلية بين عمليات التشغيل.

إطار عمل العامل: توسيع BaseAgent (تشغيل مركزي)

from collections.abc import AsyncIterable, Awaitable, Sequence
from typing import Any, Literal, overload
from agent_framework import (
    AgentResponse,
    AgentResponseUpdate,
    AgentSession,
    BaseAgent,
    Message,
    Content,
    ResponseStream,
    normalize_messages,
)

class StaticAgent(BaseAgent):
    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: str | Message | Sequence[str | Message] | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        normalized_messages = normalize_messages(messages)
        response_text = "Hello from AF custom agent"

        async def _run_non_streaming() -> AgentResponse:
            reply = Message(role="assistant", contents=[Content.from_text(response_text)])

            if session is not None:
                stored = session.state.setdefault("memory", {}).setdefault("messages", [])
                stored.extend(normalized_messages)
                stored.append(reply)

            return AgentResponse(messages=[reply])

        async def _run_streaming() -> AsyncIterable[AgentResponseUpdate]:
            yield AgentResponseUpdate(contents=[Content.from_text(response_text)], role="assistant")

            if session is not None:
                reply = Message(role="assistant", contents=[Content.from_text(response_text)])
                stored = session.state.setdefault("memory", {}).setdefault("messages", [])
                stored.extend(normalized_messages)
                stored.append(reply)

        if stream:
            return ResponseStream(_run_streaming(), finalizer=AgentResponse.from_updates)
        return _run_non_streaming()

ملاحظات:

  • SupportsAgentRunلتلبية ، نفذ run(...) مع عقد الإرجاع الدفق وغير المتدفق.
  • BaseAgent يوفر create_session() / get_session()؛ احتفظ بالحالة المخصصة في session.state.
  • استمر في حالة المحادثة المخصصة في session.state (أو عبر موفري المحفوظات/السياق) حتى تبقى عبر المنعطفات.
  • راجع النموذج الكامل: عامل مخصص

بعد ذلك، لنلق نظرة على التنسيق متعدد العوامل - المنطقة التي تختلف فيها أطر العمل بشكل أكبر.

تعيين ميزة متعددة العوامل

نظرة عامة على نموذج البرمجة

تمثل نماذج البرمجة متعددة العوامل الفرق الأكثر أهمية بين الإطارين.

نهج النموذج المزدوج ل AutoGen

يوفر AutoGen نموذجين للبرمجة:

  1. autogen-core: برمجة منخفضة المستوى ومحركها الحدث مع RoutedAgent اشتراكات الرسائل
  2. Team التجريد: نموذج عالي المستوى يركز على التشغيل مبني على autogen-core
# Low-level autogen-core (complex)
class MyAgent(RoutedAgent):
    @message_handler
    async def handle_message(self, message: TextMessage, ctx: MessageContext) -> None:
        # Handle specific message types
        pass

# High-level Team (easier but limited)
team = RoundRobinGroupChat(
    participants=[agent1, agent2],
    termination_condition=StopAfterNMessages(5)
)
result = await team.run(task="Collaborate on this task")

التحديات:

  • النموذج منخفض المستوى معقد جدا لمعظم المستخدمين
  • يمكن أن يصبح النموذج عالي المستوى مقيدا للسلوكيات المعقدة
  • يضيف الجسر بين النموذجين تعقيد التنفيذ

نموذج سير العمل الموحد لإطار عمل العامل

يوفر إطار عمل العامل تجريدا واحدا Workflow يجمع بين أفضل النهجين:

from agent_framework import WorkflowBuilder, executor, WorkflowContext
from typing_extensions import Never

# Assume we have agent1 and agent2 from previous examples
@executor(id="agent1")
async def agent1_executor(input_msg: str, ctx: WorkflowContext[str]) -> None:
    response = await agent1.run(input_msg)
    await ctx.send_message(response.text)

@executor(id="agent2")
async def agent2_executor(input_msg: str, ctx: WorkflowContext[Never, str]) -> None:
    response = await agent2.run(input_msg)
    await ctx.yield_output(response.text)  # Final output

# Build typed data flow graph
workflow = (WorkflowBuilder(start_executor=agent1_executor)
           .add_edge(agent1_executor, agent2_executor)
           .build())

# Example usage (would be in async context)
# result = await workflow.run("Initial input")

للحصول على أمثلة مفصلة لسير العمل، راجع:

Benefits:

  • نموذج موحد: تجريد واحد لجميع مستويات التعقيد
  • أمان النوع: المدخلات والمخرجات التي تم كتابتها بقوة
  • تصور الرسم البياني: مسح تمثيل تدفق البيانات
  • التركيب المرن: خلط العوامل والوظائف وسير العمل الفرعي

سير العمل مقابل GraphFlow

يستوحى تجريد إطار العامل Workflow من الميزة التجريبية GraphFlow ل AutoGen، ولكنه يمثل تطورا كبيرا في فلسفة التصميم:

  • GraphFlow: تدفق التحكم استنادا إلى حيث تكون الحواف انتقالات ويتم بث الرسائل إلى جميع العوامل؛ تكون الانتقالات مشروطة بمحتوى الرسالة التي تم بثها
  • سير العمل: تدفق البيانات المستند إلى حيث يتم توجيه الرسائل من خلال حواف معينة ويتم تنشيط المنفذين بواسطة الحواف، مع دعم التنفيذ المتزامن.

نظرة عامة مرئية

يتباين الرسم التخطيطي أدناه مع تدفق التحكم في تدفق AutoGen GraphFlow (إلى اليسار) مع سير عمل تدفق البيانات ل Agent Framework (لليمين). نماذج GraphFlow عوامل كعقد مع انتقالات وبث شرطي. منفذو نماذج سير العمل (عوامل أو وظائف أو مهام سير عمل فرعية) متصلة بواسطة حواف مكتوبة؛ كما أنه يدعم الإيقاف المؤقت للطلب/الاستجابة ونقاط التفتيش.

flowchart LR

  subgraph AutoGenGraphFlow
    direction TB
    U[User / Task] --> A[Agent A]
    A -->|success| B[Agent B]
    A -->|retry| C[Agent C]
    A -. broadcast .- B
    A -. broadcast .- C
  end

  subgraph AgentFrameworkWorkflow
    direction TB
    I[Input] --> E1[Executor 1]
    E1 -->|"str"| E2[Executor 2]
    E1 -->|"image"| E3[Executor 3]
    E3 -->|"str"| E2
    E2 --> OUT[(Final Output)]
  end

  R[Request / Response Gate]
  E2 -. request .-> R
  R -. resume .-> E2

  CP[Checkpoint]
  E1 -. save .-> CP
  CP -. load .-> E1

في الممارسة العملية:

  • يستخدم GraphFlow العوامل كعقد ويبث الرسائل؛ تمثل الحواف انتقالات شرطية.
  • يوجه سير العمل الرسائل التي تكتب على طول الحواف. يمكن أن تكون العقد (المنفذون) عوامل أو وظائف خالصة أو مهام سير عمل فرعية.
  • يتيح الطلب/الاستجابة إيقاف سير العمل مؤقتا للإدخال الخارجي؛ تستمر نقاط التفتيش في التقدم وتمكن السيرة الذاتية.

مقارنة التعليمات البرمجية

1) تسلسلي + شرطي
# AutoGen GraphFlow (fluent builder) — writer → reviewer → editor (conditional)
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow

writer = AssistantAgent(name="writer", description="Writes a draft", model_client=client)
reviewer = AssistantAgent(name="reviewer", description="Reviews the draft", model_client=client)
editor = AssistantAgent(name="editor", description="Finalizes the draft", model_client=client)

graph = (
    DiGraphBuilder()
    .add_node(writer).add_node(reviewer).add_node(editor)
    .add_edge(writer, reviewer)  # always
    .add_edge(reviewer, editor, condition=lambda msg: "approve" in msg.to_model_text())
    .set_entry_point(writer)
).build()

team = GraphFlow(participants=[writer, reviewer, editor], graph=graph)
result = await team.run(task="Draft a short paragraph about solar power")

# Agent Framework Workflow — sequential executors with conditional logic
from agent_framework import WorkflowBuilder, executor, WorkflowContext
from typing_extensions import Never

@executor(id="writer")
async def writer_exec(task: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(f"Draft: {task}")

@executor(id="reviewer")
async def reviewer_exec(draft: str, ctx: WorkflowContext[str]) -> None:
    decision = "approve" if "solar" in draft.lower() else "revise"
    await ctx.send_message(f"{decision}:{draft}")

@executor(id="editor")
async def editor_exec(msg: str, ctx: WorkflowContext[Never, str]) -> None:
    if msg.startswith("approve:"):
        await ctx.yield_output(msg.split(":", 1)[1])
    else:
        await ctx.yield_output("Needs revision")

workflow_seq = (
    WorkflowBuilder(start_executor=writer_exec)
    .add_edge(writer_exec, reviewer_exec)
    .add_edge(reviewer_exec, editor_exec)
    .build()
)
2) Fan-out + Join (الكل مقابل أي)
# AutoGen GraphFlow — A → (B, C) → D with ALL/ANY join
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
A, B, C, D = agent_a, agent_b, agent_c, agent_d

# ALL (default): D runs after both B and C
g_all = (
    DiGraphBuilder()
    .add_node(A).add_node(B).add_node(C).add_node(D)
    .add_edge(A, B).add_edge(A, C)
    .add_edge(B, D).add_edge(C, D)
    .set_entry_point(A)
).build()

# ANY: D runs when either B or C completes
g_any = (
    DiGraphBuilder()
    .add_node(A).add_node(B).add_node(C).add_node(D)
    .add_edge(A, B).add_edge(A, C)
    .add_edge(B, D, activation_group="join_d", activation_condition="any")
    .add_edge(C, D, activation_group="join_d", activation_condition="any")
    .set_entry_point(A)
).build()

# Agent Framework Workflow — A → (B, C) → aggregator (ALL vs ANY)
from agent_framework import WorkflowBuilder, executor, WorkflowContext
from typing_extensions import Never

@executor(id="A")
async def start(task: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(f"B:{task}", target_id="B")
    await ctx.send_message(f"C:{task}", target_id="C")

@executor(id="B")
async def branch_b(text: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(f"B_done:{text}")

@executor(id="C")
async def branch_c(text: str, ctx: WorkflowContext[str]) -> None:
    await ctx.send_message(f"C_done:{text}")

@executor(id="join_any")
async def join_any(msg: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(f"First: {msg}")  # ANY join (first arrival)

@executor(id="join_all")
async def join_all(msg: str, ctx: WorkflowContext[str, str]) -> None:
    state = await ctx.get_executor_state() or {"items": []}
    state["items"].append(msg)
    await ctx.set_executor_state(state)
    if len(state["items"]) >= 2:
        await ctx.yield_output(" | ".join(state["items"]))  # ALL join

wf_any = (
    WorkflowBuilder(start_executor=start)
    .add_edge(start, branch_b).add_edge(start, branch_c)
    .add_edge(branch_b, join_any).add_edge(branch_c, join_any)
    .build()
)

wf_all = (
    WorkflowBuilder(start_executor=start)
    .add_edge(start, branch_b).add_edge(start, branch_c)
    .add_edge(branch_b, join_all).add_edge(branch_c, join_all)
    .build()
)
3) التوجيه المستهدف (بدون بث)
from agent_framework import WorkflowBuilder, executor, WorkflowContext
from typing_extensions import Never

@executor(id="ingest")
async def ingest(task: str, ctx: WorkflowContext[str]) -> None:
    # Route selectively using target_id
    if task.startswith("image:"):
        await ctx.send_message(task.removeprefix("image:"), target_id="vision")
    else:
        await ctx.send_message(task, target_id="writer")

@executor(id="writer")
async def write(text: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(f"Draft: {text}")

@executor(id="vision")
async def caption(image_ref: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(f"Caption: {image_ref}")

workflow = (
    WorkflowBuilder(start_executor=ingest)
    .add_edge(ingest, write)
    .add_edge(ingest, caption)
    .build()
)

# Example usage (async):
# await workflow.run("Summarize the benefits of solar power")
# await workflow.run("image:https://example.com/panel.jpg")

ما يجب ملاحظته:

  • يبث GraphFlow الرسائل ويستخدم الانتقالات الشرطية. يتم تكوين سلوك الانضمام عبر الجانب activation الهدف ولكل حافة activation_groupactivation_condition/(على سبيل المثال، تجميع الحافتين إلى join_d مع ).activation_condition="any"
  • سير العمل يوجه البيانات بشكل صريح؛ استخدم target_id لتحديد منفذي انتقال البيانات من الخادم. يعيش سلوك الانضمام في المنفذ المتلقي (على سبيل المثال، العائد على الإدخال الأول مقابل الانتظار للجميع)، أو عبر منشئي/مجمعات التزامن.
  • المنفذون في سير العمل هم نموذج حر: التفاف Agentأو دالة أو سير عمل فرعي ومزجها ضمن نفس الرسم البياني.

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

يلخص الجدول أدناه الاختلافات الأساسية بين GraphFlow الخاص ب AutoGen وسير عمل إطار عمل العامل:

الجانب AutoGen GraphFlow سير عمل إطار عمل العامل
نوع التدفق تدفق التحكم (الحواف هي انتقالات) تدفق البيانات (رسائل توجيه الحواف)
أنواع العقد العوامل فقط العوامل والوظائف ومهام سير العمل الفرعية
التنشيط بث الرسالة التنشيط المستند إلى الحافة
أمان النوع Limited كتابة قوية في جميع أنحاء
إمكانية الإنشاء Limited قابل للتكوين بدرجة كبيرة

أنماط التداخل

تداخل فريق AutoGen

# Inner team
inner_team = RoundRobinGroupChat(
    participants=[specialist1, specialist2],
    termination_condition=StopAfterNMessages(3)
)

# Outer team with nested team as participant
outer_team = RoundRobinGroupChat(
    participants=[coordinator, inner_team, reviewer],  # Team as participant
    termination_condition=StopAfterNMessages(10)
)

# Messages are broadcasted to all participants including nested team
result = await outer_team.run("Complex task requiring collaboration")

خصائص تداخل AutoGen:

  • يتلقى الفريق المتداخل جميع الرسائل من الفريق الخارجي
  • يتم بث رسائل الفريق المتداخلة لجميع المشاركين في الفريق الخارجي
  • سياق الرسالة المشتركة عبر جميع المستويات

تداخل سير عمل إطار العامل

from agent_framework import WorkflowExecutor, WorkflowBuilder

# Assume we have executors from previous examples
# specialist1_executor, specialist2_executor, coordinator_executor, reviewer_executor

# Create sub-workflow
sub_workflow = (WorkflowBuilder(start_executor=specialist1_executor)
               .add_edge(specialist1_executor, specialist2_executor)
               .build())

# Wrap as executor
sub_workflow_executor = WorkflowExecutor(
    workflow=sub_workflow,
    id="sub_process"
)

# Use in parent workflow
parent_workflow = (WorkflowBuilder(start_executor=coordinator_executor)
                  .add_edge(coordinator_executor, sub_workflow_executor)
                  .add_edge(sub_workflow_executor, reviewer_executor)
                  .build())

خصائص تداخل إطار العامل:

  • الإدخال/الإخراج المعزول من خلال WorkflowExecutor
  • لا يوجد بث للرسائل - تتدفق البيانات من خلال اتصالات محددة
  • إدارة الحالة المستقلة لكل مستوى سير عمل

أنماط الدردشة الجماعية

تمكن أنماط الدردشة الجماعية العديد من الوكلاء من التعاون في المهام المعقدة. فيما يلي كيفية ترجمة الأنماط الشائعة بين الأطر.

نمط RoundRobinGroupChat

تنفيذ AutoGen:

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import StopAfterNMessages

team = RoundRobinGroupChat(
    participants=[agent1, agent2, agent3],
    termination_condition=StopAfterNMessages(10)
)
result = await team.run("Discuss this topic")

تنفيذ إطار عمل العامل:

from agent_framework.orchestrations import SequentialBuilder

# Assume we have agent1, agent2, agent3 from previous examples
# Sequential workflow through participants
workflow = SequentialBuilder(participants=[agent1, agent2, agent3]).build()

# Example usage (would be in async context)
async def sequential_example():
    # Each agent appends to shared conversation
    async for event in workflow.run("Discuss this topic", stream=True):
        if event.type == "output":
            conversation_history = event.data  # list[Message]

للحصول على أمثلة تزامن مفصلة، راجع:

بالنسبة لأنماط التنفيذ المتزامنة، يوفر إطار عمل العامل أيضا:

from agent_framework.orchestrations import ConcurrentBuilder

# Assume we have agent1, agent2, agent3 from previous examples
# Concurrent workflow for parallel processing
workflow = (ConcurrentBuilder(participants=[agent1, agent2, agent3])
           .build())

# Example usage (would be in async context)
async def concurrent_example():
    # All agents process the input concurrently
    async for event in workflow.run("Process this in parallel", stream=True):
        if event.type == "output":
            results = event.data  # Combined results from all agents

للحصول على أمثلة التنفيذ المتزامن، راجع:

نمط MagenticOneGroupChat

تنفيذ AutoGen:

from autogen_agentchat.teams import MagenticOneGroupChat

team = MagenticOneGroupChat(
    participants=[researcher, coder, executor],
    model_client=coordinator_client,
    termination_condition=StopAfterNMessages(20)
)
result = await team.run("Complex research and analysis task")

تنفيذ إطار عمل العامل:

from typing import cast
from agent_framework import (
    AgentResponseUpdate,
    Agent,
    Message,
)
from agent_framework.orchestrations import (
    MAGENTIC_EVENT_TYPE_AGENT_DELTA,
    MAGENTIC_EVENT_TYPE_ORCHESTRATOR,
    MagenticBuilder,
)
from agent_framework.openai import OpenAIChatClient

# Create a manager agent for orchestration
manager_agent = Agent(
    name="MagenticManager",
    description="Orchestrator that coordinates the workflow",
    instructions="You coordinate a team to complete complex tasks efficiently.",
    client=OpenAIChatClient(),
)

workflow = MagenticBuilder(
    participants=[researcher, coder],
    manager_agent=manager_agent,
    max_round_count=20,
    max_stall_count=3,
    max_reset_count=2,
).build()

# Example usage (would be in async context)
async def magentic_example():
    output: str | None = None
    async for event in workflow.run("Complex research task", stream=True):
        if event.type == "output":
            output_messages = cast(list[Message], event.data)
            if output_messages:
                output = output_messages[-1].text

خيارات تخصيص إطار عمل العامل:

يوفر سير العمل Magentic خيارات تخصيص واسعة النطاق:

  • تكوين المدير: استخدام عامل مع إرشادات مخصصة وإعدادات النموذج
  • حدود التقريب: max_round_count، ، max_stall_countmax_reset_count
  • دفق الأحداث: استخدام أحداث الإخراج (event.type == "output") مع AgentResponseUpdate البيانات للبث
  • تخصص العامل: إرشادات وأدوات مخصصة لكل عامل
  • Human-in-the-loop: مراجعة الخطة والموافقة على الأدوات وتدخل المماطلة
# Advanced customization example with human-in-the-loop
from typing import cast
from agent_framework import (
    AgentResponseUpdate,
    Agent,
    WorkflowEvent,
)
from agent_framework.orchestrations import (
    MAGENTIC_EVENT_TYPE_AGENT_DELTA,
    MAGENTIC_EVENT_TYPE_ORCHESTRATOR,
    MagenticBuilder,
    MagenticHumanInterventionDecision,
    MagenticHumanInterventionKind,
    MagenticHumanInterventionReply,
    MagenticHumanInterventionRequest,
)
from agent_framework.openai import OpenAIChatClient

# Create manager agent with custom configuration
manager_agent = Agent(
    name="MagenticManager",
    description="Orchestrator for complex tasks",
    instructions="Custom orchestration instructions...",
    client=OpenAIChatClient(model="gpt-4o"),
)

workflow = (
    MagenticBuilder(
        participants=[researcher_agent, coder_agent, analyst_agent],
        enable_plan_review=True,
        manager_agent=manager_agent,
        max_round_count=15,      # Limit total rounds
        max_stall_count=2,       # Trigger stall handling
        max_reset_count=1,       # Allow one reset on failure
    )
    .with_human_input_on_stall()  # Enable human intervention on stalls
    .build()
)

# Handle human intervention requests during execution
async for event in workflow.run("Complex task", stream=True):
    if event.type == "request_info" and event.request_type is MagenticHumanInterventionRequest:
        req = cast(MagenticHumanInterventionRequest, event.data)
        if req.kind == MagenticHumanInterventionKind.PLAN_REVIEW:
            # Review and approve the plan
            reply = MagenticHumanInterventionReply(
                decision=MagenticHumanInterventionDecision.APPROVE
            )
            async for ev in workflow.run(responses={event.request_id: reply}, stream=True):
                pass  # Handle continuation

للحصول على أمثلة أرجوانية مفصلة، راجع:

أنماط المستقبل

يتضمن مخطط إطار عمل العامل عدة أنماط AutoGen قيد التطوير حاليا:

  • نمط Swarm: تنسيق الوكيل المستند إلى التسليم
  • SelectorGroupChat: تحديد مكبر الصوت المستند إلى LLM

Human-in-the-Loop مع استجابة الطلب

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

قيود AutoGen

يتم تشغيل تجريد AutoGen Team بشكل مستمر بمجرد البدء ولا يوفر آليات مضمنة لإيقاف التنفيذ مؤقتا للإدخال البشري. تتطلب أي وظيفة بشرية في التكرار الحلقي تطبيقات مخصصة خارج إطار العمل.

واجهة برمجة تطبيقات Request-Response إطار العامل

يوفر إطار عمل العامل قدرات استجابة الطلب المضمنة حيث يمكن لأي منفذ إرسال الطلبات باستخدام ctx.request_info() الاستجابات ومعالجتها @response_handler باستخدام مصمم الديكور.

from agent_framework import (
    WorkflowBuilder, WorkflowContext,
    Executor, handler, response_handler
)
from dataclasses import dataclass

# Assume we have agent_executor defined elsewhere

# Define typed request payload
@dataclass
class ApprovalRequest:
    """Request human approval for agent output."""
    content: str = ""
    agent_name: str = ""

# Workflow executor that requests human approval
class ReviewerExecutor(Executor):

    @handler
    async def review_content(
        self,
        agent_response: str,
        ctx: WorkflowContext
    ) -> None:
        # Request human input with structured data
        approval_request = ApprovalRequest(
            content=agent_response,
            agent_name="writer_agent"
        )
        await ctx.request_info(request_data=approval_request, response_type=str)

    @response_handler
    async def handle_approval_response(
        self,
        original_request: ApprovalRequest,
        decision: str,
        ctx: WorkflowContext
    ) -> None:
        decision_lower = decision.strip().lower()
        original_content = original_request.content

        if decision_lower == "approved":
            await ctx.yield_output(f"APPROVED: {original_content}")
        else:
            await ctx.yield_output(f"REVISION NEEDED: {decision}")

# Build workflow with human-in-the-loop
reviewer = ReviewerExecutor(id="reviewer")

workflow = (WorkflowBuilder(start_executor=agent_executor)
           .add_edge(agent_executor, reviewer)
           .build())

تشغيل مهام سير العمل البشرية في التكرار الحلقي

يوفر إطار عمل العامل واجهات برمجة التطبيقات المتدفقة لمعالجة دورة الإيقاف المؤقت:

# Assume we have workflow defined from previous examples
async def run_with_human_input():
    pending_responses = None
    completed = False

    while not completed:
        # First iteration starts the workflow; subsequent iterations pass responses back
        stream = (
            workflow.run(responses=pending_responses, stream=True)
            if pending_responses
            else workflow.run("initial input", stream=True)
        )

        events = [event async for event in stream]
        pending_responses = None

        # Collect human requests and outputs
        for event in events:
            if event.type == "request_info":
                # Display request to human and collect response
                request_data = event.data  # ApprovalRequest instance
                print(f"Review needed: {request_data.content}")

                human_response = input("Enter 'approved' or revision notes: ")
                pending_responses = {event.request_id: human_response}

            elif event.type == "output":
                print(f"Final result: {event.data}")
                completed = True

للحصول على أمثلة سير عمل بشري في التكرار الحلقي، راجع:

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

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

قيود AutoGen

لا يوفر تجريد AutoGen Team قدرات نقاط تفتيش مضمنة. يجب تنفيذ أي آليات استمرار أو استرداد خارجي، وغالبا ما تتطلب إدارة الحالة المعقدة ومنطق التسلسل.

نقاط التحقق في إطار عمل العامل

يوفر إطار عمل العامل نقاط تفتيش شاملة من خلال FileCheckpointStorage ومعلمة الدالة checkpoint_storage الإنشائية على WorkflowBuilder. التقاط نقاط التحقق:

  • حالة المنفذ: الحالة المحلية لكل منفذ يستخدم ctx.set_executor_state()
  • الحالة: حالة المنفذ المشترك باستخدام ctx.set_state()
  • قوائم انتظار الرسائل: الرسائل المعلقة بين المنفذين
  • موضع سير العمل: تقدم التنفيذ الحالي والخطوات التالية
from agent_framework import (
    FileCheckpointStorage, WorkflowBuilder, WorkflowContext,
    Executor, handler
)
from typing_extensions import Never

class ProcessingExecutor(Executor):
    @handler
    async def process(self, data: str, ctx: WorkflowContext[str]) -> None:
        # Process the data
        result = f"Processed: {data.upper()}"
        print(f"Processing: '{data}' -> '{result}'")

        # Persist executor-local state
        prev_state = await ctx.get_executor_state() or {}
        count = prev_state.get("count", 0) + 1
        await ctx.set_executor_state({
            "count": count,
            "last_input": data,
            "last_output": result
        })

        # Persist shared state for other executors
        ctx.set_state("original_input", data)
        ctx.set_state("processed_output", result)

        await ctx.send_message(result)

class FinalizeExecutor(Executor):
    @handler
    async def finalize(self, data: str, ctx: WorkflowContext[Never, str]) -> None:
        result = f"Final: {data}"
        await ctx.yield_output(result)

# Configure checkpoint storage
checkpoint_storage = FileCheckpointStorage(storage_path="./checkpoints")
processing_executor = ProcessingExecutor(id="processing")
finalize_executor = FinalizeExecutor(id="finalize")

# Build workflow with checkpointing enabled
workflow = (WorkflowBuilder(start_executor=processing_executor, checkpoint_storage=checkpoint_storage)
           .add_edge(processing_executor, finalize_executor)
           .build())

# Example usage (would be in async context)
async def checkpoint_example():
    # Run workflow - checkpoints are created automatically
    async for event in workflow.run("input data", stream=True):
        print(f"Event: {event}")

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

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

from typing_extensions import Never

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

class UpperCaseExecutor(Executor):
    @handler
    async def process(self, text: str, ctx: WorkflowContext[str]) -> None:
        result = text.upper()
        await ctx.send_message(result)

class ReverseExecutor(Executor):
    @handler
    async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
        result = text[::-1]
        await ctx.yield_output(result)

def create_workflow(checkpoint_storage: FileCheckpointStorage):
    """Create a workflow with two executors and checkpointing."""
    upper_executor = UpperCaseExecutor(id="upper")
    reverse_executor = ReverseExecutor(id="reverse")

    return (WorkflowBuilder(start_executor=upper_executor, checkpoint_storage=checkpoint_storage)
           .add_edge(upper_executor, reverse_executor)
           .build())

# Assume we have checkpoint_storage from previous examples
checkpoint_storage = FileCheckpointStorage(storage_path="./checkpoints")

async def checkpoint_resume_example():
    # Create workflow instance to get its configured name
    new_workflow = create_workflow(checkpoint_storage)

    # List available checkpoints
    checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=new_workflow.name)

    # Display checkpoint information
    for checkpoint in checkpoints:
        print(f"Checkpoint {checkpoint.checkpoint_id}: iteration={checkpoint.iteration_count}")

    # Resume from a specific checkpoint
    if checkpoints:
        chosen_checkpoint_id = checkpoints[0].checkpoint_id

        async for event in new_workflow.run(
            checkpoint_id=chosen_checkpoint_id,
            checkpoint_storage=checkpoint_storage,
            stream=True,
        ):
            print(f"Resumed event: {event}")

ميزات نقاط التحقق المتقدمة

نقطة تحقق مع تكامل Human-in-the-Loop:

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

# Assume we have workflow, checkpoint_id, and checkpoint_storage from previous examples
async def resume_with_pending_requests_example():
    # Resume from checkpoint - pending requests will be re-emitted
    request_info_events = []
    async for event in workflow.run(
        checkpoint_id=checkpoint_id,
        checkpoint_storage=checkpoint_storage,
        stream=True,
    ):
        if event.type == "request_info":
            request_info_events.append(event)

    # Handle re-emitted pending request
    responses = {}
    for event in request_info_events:
        response = handle_request(event.data)
        responses[event.request_id] = response

    # Send response back to workflow
    async for event in workflow.run(responses=responses, stream=True):
        print(f"Event: {event}")

الفوائد الرئيسة

مقارنة ب AutoGen، توفر نقاط التفتيش في إطار عمل العامل ما يلي:

  • الثبات التلقائي: لا يلزم إدارة الحالة اليدوية
  • الاسترداد متعدد المستويات: استئناف من أي حد فائقة
  • عزل الحالة: فصل المنفذ المحلي والحالة المشتركة
  • تكامل Human-in-the-loop: استئناف مؤقت سلس مع الإدخال البشري
  • التسامح مع الخطأ: استرداد قوي من حالات الفشل أو الانقطاع

أمثلة عملية

للحصول على أمثلة شاملة لنقاط التفتيش، راجع:

قابلية الرصد

يوفر كل من AutoGen و Agent Framework قدرات المراقبة، ولكن مع نهج وميزات مختلفة.

إمكانية مراقبة AutoGen

يحتوي AutoGen على دعم أصلي ل OpenTelemetry مع الأجهزة من أجل:

  • تتبع وقت التشغيل: SingleThreadedAgentRuntime و GrpcWorkerAgentRuntime
  • تنفيذ الأداة: BaseTool مع execute_tool امتدادات تلي اصطلاحات GenAI الدلالية
  • عمليات العامل: BaseChatAgent مع create_agent امتدادات و invoke_agent
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from autogen_core import SingleThreadedAgentRuntime

# Configure OpenTelemetry
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)

# Pass to runtime
runtime = SingleThreadedAgentRuntime(tracer_provider=tracer_provider)

إمكانية مراقبة إطار عمل العامل

يوفر إطار عمل العامل إمكانية مراقبة شاملة من خلال نهج متعددة:

  • إعداد التعليمات البرمجية الصفرية: الأجهزة التلقائية عبر متغيرات البيئة
  • التكوين اليدوي: الإعداد البرمجي باستخدام معلمات مخصصة
  • بيانات تتبع الاستخدام الغنية: العوامل ومهام سير العمل وتعقب تنفيذ الأدوات
  • إخراج وحدة التحكم: تسجيل وحدة التحكم المضمنة والتصور
from agent_framework import Agent
from agent_framework.observability import configure_otel_providers
from agent_framework.openai import OpenAIChatClient

# Zero-code setup via environment variables
# Set OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317

# Or manual setup
configure_otel_providers()

# Create client for the example
client = OpenAIChatClient(model="gpt-5")

async def observability_example():
    # Observability is automatically applied to all agents and workflows
    agent = Agent(name="assistant", client=client)
    result = await agent.run("Hello")  # Automatically traced

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

  • تعقيد الإعداد: يوفر إطار عمل العامل خيارات إعداد أبسط بدون تعليمات برمجية
  • النطاق: يوفر إطار عمل العامل تغطية أوسع بما في ذلك إمكانية المراقبة على مستوى سير العمل
  • التصور: يتضمن إطار عمل العامل إخراج وحدة التحكم المضمنة وواجهة مستخدم التطوير
  • التكوين: يوفر إطار عمل العامل خيارات تكوين أكثر مرونة

للحصول على أمثلة تفصيلية لقابلية الملاحظة، راجع:

خاتمة

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

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

للحصول على أمثلة إضافية وإرشادات تنفيذ مفصلة، راجع دليل نماذج إطار العامل .

فئات عينة إضافية

يوفر إطار عمل العامل عينات عبر العديد من المجالات الهامة الأخرى:

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

دليل التشغيل السريع

  • Last updated on