Aracılığıyla paylaş

AutoGen'in Microsoft Agent Framework'e Geçiş Kılavuzu

AutoGen'den Microsoft Agent Framework Python SDK'sına geçiş için kapsamlı bir kılavuz.

İçindekiler Tablosu

Arka plan

AutoGen , büyük dil modellerini (LLM) kullanarak yapay zeka aracıları ve çok aracılı sistemler oluşturmaya yönelik bir çerçevedir. Microsoft Research'te bir araştırma projesi olarak başladı ve GroupChat ve olay odaklı aracı çalışma zamanı gibi çok aracılı düzenlemede çeşitli kavramlara öncülük etti. Proje, açık kaynak topluluğunun verimli bir işbirliği oldu ve birçok önemli özellik dış katkıda bulunanlardan geldi.

Microsoft Agent Framework , LLM'leri kullanarak yapay zeka aracıları ve iş akışları oluşturmaya yönelik yeni bir çok dilli SDK'dır. AutoGen'de öncü olan fikirlerin önemli bir evrimini temsil eder ve gerçek dünya kullanımından alınan dersleri birleştirir. Microsoft'un temel AutoGen ve Semantic Kernel ekipleri tarafından geliştirilmiştir ve ileriye dönük yapay zeka uygulamaları oluşturmak için yeni bir temel olacak şekilde tasarlanmıştır.

Bu kılavuzda pratik bir geçiş yolu açıklanmaktadır: Nelerin aynı kaldığını ve bir bakışta nelerin değiştiğini kapsayan bir başlangıçtır. Ardından model istemci kurulumunu, tek aracılı özellikleri ve son olarak somut kodla yan yana çok aracılı düzenlemeyi kapsar. Bu arada, Agent Framework deposunda çalıştırılabilir örneklerin bağlantıları her adımı doğrulamanıza yardımcı olur.

Temel Benzerlikler ve Farklar

Aynı Kalanlar

Temeller tanıdık. Yine de model istemcisi etrafında aracılar oluşturur, yönergeler sağlar ve araçlar eklersiniz. her iki kitaplık da işlev stili araçları, belirteç akışını, çok modüllü içeriği ve zaman uyumsuz G/Ç'yi destekler.

# 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 = ChatAgent(name="assistant", chat_client=client, tools=[my_tool])
result = await agent.run("Help me with this task")

Temel Farklılıklar

  1. Düzenleme stili: AutoGen, olay temelli bir çekirdeği üst düzey Teamile eşleştirir. Agent Framework, verileri kenarlar boyunca yönlendiren ve girişler hazır olduğunda yürütücüleri etkinleştiren, türlenmiş, graf tabanlı Workflow bir temel alır.

  2. Araçlar: AutoGen işlevleri ile FunctionToolsarmalar. Agent Framework kullanır @ai_function, şemaları otomatik olarak çıkarsar ve kod yorumlayıcı ve web araması gibi barındırılan araçlar ekler.

  3. Aracı davranışı: AssistantAgent artırmadığınız max_tool_iterationssürece tek dönüşlüdür. ChatAgent varsayılan olarak çok dönüşlüdür ve son yanıtı döndürene kadar araçları çağırmaya devam eder.

  4. Çalışma zamanı: AutoGen, tümleşik ve deneysel dağıtılmış çalışma zamanları sunar. Agent Framework bugün tek işlemli bileşime odaklanmaktadır; dağıtılmış yürütme planlanmıştır.

Model İstemcisi Oluşturma ve Yapılandırma

Her iki çerçeve de büyük yapay zeka sağlayıcıları için benzer ama aynı API'lere sahip olmayan model istemcileri sağlar.

Özellik AutoGen Ajan Çerçevesi
OpenAI İstemcisi OpenAIChatCompletionClient OpenAIChatClient
OpenAI Yanıt İstemcisi ❌ Yok OpenAIResponsesClient
Azure OpenAI AzureOpenAIChatCompletionClient AzureOpenAIChatClient
Azure OpenAI Yanıtları ❌ Yok AzureOpenAIResponsesClient
Azure AI AzureAIChatCompletionClient AzureAIAgentClient
Anthropic AnthropicChatCompletionClient 🚧 Planlanmış
Ollama OllamaChatCompletionClient 🚧 Planlanmış
Caching ChatCompletionCache Sarıcı 🚧 Planlanmış

AutoGen Model İstemcileri

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 OpenAIChatClient
from agent_framework.azure import AzureOpenAIChatClient

# OpenAI (reads API key from environment)
client = OpenAIChatClient(model_id="gpt-5")

# Azure OpenAI (uses environment or default credentials; see samples for auth options)
client = AzureOpenAIChatClient(model_id="gpt-5")

Ayrıntılı örnekler için bkz:

Yanıtlar API Desteği (Agent Framework Özel)

Agent Framework'ün ve AzureOpenAIResponsesClient AutoGen'de OpenAIResponsesClient kullanılamayan model ve yapılandırılmış yanıtların akıl yürütmesi için özel destek sağlar:

from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.openai import OpenAIResponsesClient

# Azure OpenAI with Responses API
azure_responses_client = AzureOpenAIResponsesClient(model_id="gpt-5")

# OpenAI with Responses API
openai_responses_client = OpenAIResponsesClient(model_id="gpt-5")

Yanıtlar API'si örnekleri için bkz:

özellik eşlemeyi Single-Agent

Bu bölüm, AutoGen ve Agent Framework arasında tek aracılı özellikleri eşler. Bir istemci mevcutken bir aracı oluşturun, araçlar ekleyin ve akış dışı yürütme ile akış yürütme arasında seçim yapın.

Temel Aracı Oluşturma ve Yürütme

Bir model istemcisi yapılandırıldıktan sonra, sonraki adım aracıları oluşturmaktır. Her iki çerçeve de benzer aracı soyutlamaları sağlar, ancak farklı varsayılan davranışlar ve yapılandırma seçenekleri sunar.

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?")

Agent Framework ChatAgent

from agent_framework import ChatAgent, ai_function
from agent_framework.openai import OpenAIChatClient

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

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

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

async def example():
    # Direct creation
    agent = ChatAgent(
        name="assistant",
        chat_client=client,
        instructions="You are a helpful assistant.",
        tools=[get_weather]  # Multi-turn by default
    )

    # Factory method (more convenient)
    agent = client.create_agent(
        name="assistant",
        instructions="You are a helpful assistant.",
        tools=[get_weather]
    )

    # Execution with runtime tool configuration
    result = await agent.run(
        "What's the weather?",
        tools=[get_time],  # Can add tools at runtime
        tool_choice="auto"
    )

Önemli Farklar:

  • Varsayılan davranış: ChatAgent Araç çağrıları aracılığıyla otomatik olarak yinelenirken AssistantAgent , açık max_tool_iterations ayar gerektirir
  • Çalışma zamanı yapılandırması: ChatAgent.run() çağırma başına özelleştirme için ve tools parametrelerini kabul eder tool_choice
  • Fabrika yöntemleri: Agent Framework doğrudan sohbet istemcilerinden uygun fabrika yöntemleri sağlar
  • Durum yönetimi: ChatAgent durum bilgisi yoktur ve konuşma geçmişini durumunun bir parçası olarak tutanın aksine AssistantAgent çağrılar arasında konuşma geçmişini korumaz

AgentThread ile Konuşma Durumunu Yönetme

ile ChatAgentgörüşmelere devam etmek için konuşma geçmişini yönetmek için kullanın AgentThread :

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

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

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

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

Varsayılan olarak durum bilgisi olmayan: hızlı tanıtım

# Without a thread (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 thread (shared context across calls)
thread = agent.get_new_thread()
print((await agent.run("What's 2+2?", thread=thread)).text)  # "4"
print((await agent.run("What about that number times 10?", thread=thread)).text)  # "40"

İş parçacığı yönetimi örnekleri için bkz:

OpenAI Yardımcısı Aracı Eşdeğerliği

Her iki çerçeve de OpenAI Yardımcısı API tümleştirmesi sağlar:

# AutoGen OpenAIAssistantAgent
from autogen_ext.agents.openai import OpenAIAssistantAgent

# Agent Framework has OpenAI Assistants support via OpenAIAssistantsClient
from agent_framework.openai import OpenAIAssistantsClient

OpenAI Yardımcısı örnekleri için bkz:

Akış Desteği

her iki çerçeve de, URI'leri yanıt vermeye devam etmek için istemcilerden ve aracılardan gerçek zamanlı olarak belirteç akışı sağlar.

AutoGen Akışı

# 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")

Agent Framework Akışı

# Assume we have client, agent, and tools from previous examples
async def streaming_example():
    # Chat client streaming
    async for chunk in client.get_streaming_response("Hello", tools=tools):
        if chunk.text:
            print(chunk.text, end="")

    # Agent streaming
    async for chunk in agent.run_stream("Hello"):
        if chunk.text:
            print(chunk.text, end="", flush=True)

İpucu: Agent Framework'te hem istemciler hem de aracılar aynı güncelleştirme şeklini verir; her iki durumda da okuyabilirsiniz chunk.text .

İleti Türleri ve Oluşturma

Etkili aracı iletişimi için iletilerin nasıl çalıştığını anlamak çok önemlidir. Her iki çerçeve de ayrı ileti sınıfları kullanan AutoGen ve birleşik bir ileti sistemi kullanan Aracı Çerçevesi ile ileti oluşturma ve işlemeye yönelik farklı yaklaşımlar sağlar.

AutoGen İleti Türleri

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()

Agent Framework İleti Türleri

from agent_framework import ChatMessage, TextContent, DataContent, UriContent, Role
import base64

# Text message
text_msg = ChatMessage(role=Role.USER, text="Hello")

# Supply real image bytes, or use a data: URI/URL via UriContent
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 = ChatMessage(
    role=Role.USER,
    contents=[
        TextContent(text="Describe this image"),
        DataContent(uri=image_uri, media_type="image/jpeg")
    ]
)

Önemli Farklar:

  • AutoGen bir alanla TextMessage ayrı ileti sınıfları (MultiModalMessage, source) kullanır
  • Agent Framework, yazılan içerik nesneleri ve bir alan ile birleştirilmiş ChatMessage bir role kullanır
  • Agent Framework iletileri dize kaynakları yerine sabit listesi (USER, ASSISTANT, SYSTEM, TOOL) kullanır Role

Araç Oluşturma ve Tümleştirme

Araçlar, aracı özelliklerini metin oluşturmanın ötesine genişletir. Çerçeveler araç oluşturmaya farklı yaklaşımlar uygular ve Agent Framework daha otomatik şema oluşturma sağlar.

AutoGen İşlev Aracı

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])

Aracı Çerçevesi @ai_function

from agent_framework import ai_function
from typing import Annotated
from pydantic import Field

@ai_function
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 = ChatAgent(name="assistant", chat_client=client, tools=[get_weather])

Ayrıntılı örnekler için bkz:

Barındırılan Araçlar (Agent Framework Özel)

Agent Framework, AutoGen'de kullanılamayan barındırılan araçlar sağlar:

from agent_framework import ChatAgent, HostedCodeInterpreterTool, HostedWebSearchTool
from agent_framework.azure import AzureOpenAIChatClient

# Azure OpenAI client with a model that supports hosted tools
client = AzureOpenAIChatClient(model_id="gpt-5")

# Code execution tool
code_tool = HostedCodeInterpreterTool()

# Web search tool
search_tool = HostedWebSearchTool()

agent = ChatAgent(
    name="researcher",
    chat_client=client,
    tools=[code_tool, search_tool]
)

Ayrıntılı örnekler için bkz:

Gereksinimler ve uyarılar:

  • Barındırılan araçlar yalnızca bunları destekleyen modellerde/hesaplarda kullanılabilir. Bu araçları etkinleştirmeden önce sağlayıcınız için yetkilendirmeleri ve model desteğini doğrulayın.
  • Yapılandırma sağlayıcıya göre farklılık gösterir; kurulum ve izinler için her örnekteki önkoşulları izleyin.
  • Her model barındırılan her aracı (örneğin, web araması ve kod yorumlayıcı) desteklemez. Ortamınızda uyumlu bir model seçin.

Uyarı

AutoGen yerel kod yürütme araçlarını destekler, ancak bu özellik gelecekteki Agent Framework sürümleri için planlanmaktadır.

Önemli Fark: Aracı Çerçevesi aracı düzeyinde araç yinelemesini otomatik olarak işler. AutoGen'in max_tool_iterations parametresinden farklı olarak, Agent Framework aracıları sonsuz döngüleri önlemek için yerleşik güvenlik mekanizmalarıyla varsayılan olarak tamamlanıncaya kadar araç yürütmeye devam eder.

MCP Sunucusu Desteği

Gelişmiş araç tümleştirmesi için her iki çerçeve de Model Bağlam Protokolü'ne (MCP) destek sağlayarak aracıların dış hizmetler ve veri kaynaklarıyla etkileşim kurmasını sağlar. Agent Framework daha kapsamlı yerleşik destek sağlar.

AutoGen MCP Desteği

AutoGen, uzantılar aracılığıyla temel MCP desteğine sahiptir (belirli uygulama ayrıntıları sürüme göre değişir).

Agent Framework MCP Desteği

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

# Create client for the example
client = OpenAIChatClient(model_id="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 = ChatAgent(name="assistant", chat_client=client, tools=[mcp_tool])

MCP örnekleri için bkz:

Araç Olarak Aracı Deseni

Güçlü desenlerden biri aracıları araç olarak kullanmak ve hiyerarşik aracı mimarilerini etkinleştirmektir. Her iki çerçeve de farklı uygulamalarla bu düzeni destekler.

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]
)

Agent Framework as_tool()

from agent_framework import ChatAgent

# Assume we have client from previous examples
# Create specialized agent
writer = ChatAgent(
    name="writer",
    chat_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 = ChatAgent(
    name="coordinator",
    chat_client=client,
    tools=[writer_tool]
)

Açık geçiş notu: AutoGen'de, aynı aracı örneğini çağırırken eşzamanlılık sorunlarını önlemek için aracıları araç olarak sarmalarken koordinatörün model istemcisini ayarlayın parallel_tool_calls=False . Aracılar varsayılan olarak durum bilgisi olmadığından Agent as_tool() Framework'te paralel araç çağrılarının devre dışı bırakılması gerekmez.

Ara Yazılım (Agent Framework Özelliği)

Agent Framework, AutoGen'in eksik olduğu ara yazılım özelliklerini sunar. Ara yazılım, günlüğe kaydetme, güvenlik ve performans izleme gibi güçlü çapraz kesme endişeleri sağlar.

from agent_framework import ChatAgent, AgentRunContext, FunctionInvocationContext
from typing import Callable, Awaitable

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

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

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

Avantajlar:

  • Güvenlik: Giriş doğrulama ve içerik filtreleme
  • Gözlemlenebilirlik: Günlüğe kaydetme, ölçümler ve izleme
  • Performans: Önbelleğe alma ve hız sınırlama
  • Hata işleme: Düzgün performans düşüşü ve yeniden deneme mantığı

Ayrıntılı ara yazılım örnekleri için bkz:

Özel Ajanlar

Bazen model destekli bir aracıyı hiç istemezsiniz; özel mantığa sahip belirlenimci veya API destekli bir aracı istersiniz. Her iki çerçeve de özel aracılar oluşturulmasını destekler, ancak desenler farklılık gösterir.

AutoGen: Alt Sınıf 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))

Notes:

  • Sohbet iletisiyle bir on_messages(...) uygulama Response ve döndürme.
  • İsteğe bağlı olarak çalıştırmalar arasındaki iç durumu temizlemek için uygulayın on_reset(...) .

Agent Framework: BaseAgent'ı Genişletme (iş parçacığı kullanan)

from collections.abc import AsyncIterable
from typing import Any
from agent_framework import (
    AgentRunResponse,
    AgentRunResponseUpdate,
    AgentThread,
    BaseAgent,
    ChatMessage,
    Role,
    TextContent,
)

class StaticAgent(BaseAgent):
    async def run(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AgentRunResponse:
        # Build a static reply
        reply = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text="Hello from AF custom agent")])

        # Persist conversation to the provided AgentThread (if any)
        if thread is not None:
            normalized = self._normalize_messages(messages)
            await self._notify_thread_of_new_messages(thread, normalized, reply)

        return AgentRunResponse(messages=[reply])

    async def run_stream(
        self,
        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
        *,
        thread: AgentThread | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[AgentRunResponseUpdate]:
        # Stream the same static response in a single chunk for simplicity
        yield AgentRunResponseUpdate(contents=[TextContent(text="Hello from AF custom agent")], role=Role.ASSISTANT)

        # Notify thread of input and the complete response once streaming ends
        if thread is not None:
            reply = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text="Hello from AF custom agent")])
            normalized = self._normalize_messages(messages)
            await self._notify_thread_of_new_messages(thread, normalized, reply)

Notes:

  • AgentThread konuşma durumunu harici olarak korur; kullanın agent.get_new_thread() ve öğesine run/run_streamgeçirin.
  • İş parçacığının değişimin her iki tarafına da sahip olması için çağırın self._notify_thread_of_new_messages(thread, input_messages, response_messages) .
  • Tam örneğe bakın: Özel Aracı

Şimdi, çerçevelerin en çok farklılık gösterdiği alan olan çok aracılı düzenlemeye göz atalım.

Çok Aracılı Özellik Eşleme

Programlama Modeline Genel Bakış

Çok aracılı programlama modelleri, iki çerçeve arasındaki en önemli farkı temsil eder.

AutoGen'in İkili Model Yaklaşımı

AutoGen iki programlama modeli sağlar:

  1. autogen-core: ve ileti abonelikleriyle RoutedAgent düşük düzeyli, olay odaklı programlama
  2. Team soyutlama: Üst düzey, çalıştırma odaklı model 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")

Zorluklar:

  • Düşük düzeyli model çoğu kullanıcı için çok karmaşık
  • Üst düzey model karmaşık davranışlar için sınırlayıcı hale gelebilir
  • İki model arasında köprü oluşturma, uygulama karmaşıklığını artırır

Agent Framework'ün Birleşik İş Akışı Modeli

Agent Framework, her iki yaklaşımın da en iyilerini birleştiren tek Workflow bir soyutlama sağlar:

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()
           .add_edge(agent1_executor, agent2_executor)
           .set_start_executor(agent1_executor)
           .build())

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

Ayrıntılı iş akışı örnekleri için bkz:

Avantajlar:

  • Birleşik model: Tüm karmaşıklık düzeyleri için tek soyutlama
  • Tür güvenliği: Kesin olarak yazılan girişler ve çıkışlar
  • Graf görselleştirmesi: Veri akışı gösterimini temizleme
  • Esnek bileşim: Aracıları, işlevleri ve alt iş akışlarını karıştırma

İş Akışı ve GraphFlow karşılaştırması

Agent Framework'ün Workflow soyutlaması AutoGen'in deneysel GraphFlow özelliğinden ilham alır, ancak tasarım felsefesinde önemli bir evrimi temsil eder:

  • GraphFlow: Uçların geçiş olduğu ve iletilerin tüm aracılara yayınlandığı denetim akışı; geçişler, yayınlanan ileti içeriğine göre koşullandırılır
  • İş akışı: İletilerin belirli kenarlar üzerinden yönlendirildiği ve yürütücülerin kenarlar tarafından etkinleştirildiği ve eşzamanlı yürütme desteğinin sağlandığı veri akışı.

Görsele Genel Bakış

Aşağıdaki diyagramda AutoGen'in denetim akışı GraphFlow (sol) ile Agent Framework'ün veri akışı İş Akışı (sağ) karşılaştırması yapılır. GraphFlow, aracıları koşullu geçişlere ve yayınlara sahip düğümler olarak modeller. Yazılan kenarlara bağlı iş akışı modelleri yürütücüleri (aracılar, işlevler veya alt iş akışları); ayrıca istek/yanıt duraklatmalarını ve denetim noktası oluşturmayı da destekler.

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

Uygulamada:

  • GraphFlow, aracıları düğüm olarak kullanır ve iletileri yayınlar; kenarlar koşullu geçişleri temsil eder.
  • İş akışı, yazılan iletileri kenarlar boyunca yönlendirir. Düğümler (yürütücüler) aracılar, saf işlevler veya alt iş akışları olabilir.
  • İstek/yanıt, bir iş akışının dış giriş için duraklatılmış olmasına olanak tanır; denetim noktası oluşturma ilerlemeyi sürdürür ve sürdürmeyi etkinleştirir.

Kod Karşılaştırması

1) Sıralı + Koşullu
# 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()
    .add_edge(writer_exec, reviewer_exec)
    .add_edge(reviewer_exec, editor_exec)
    .set_start_executor(writer_exec)
    .build()
)
2) Fan-out + Join (ALL vs ANY)
# 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()
    .add_edge(start, branch_b).add_edge(start, branch_c)
    .add_edge(branch_b, join_any).add_edge(branch_c, join_any)
    .set_start_executor(start)
    .build()
)

wf_all = (
    WorkflowBuilder()
    .add_edge(start, branch_b).add_edge(start, branch_c)
    .add_edge(branch_b, join_all).add_edge(branch_c, join_all)
    .set_start_executor(start)
    .build()
)
3) Hedeflenen Yönlendirme (yayın yok)
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()
    .add_edge(ingest, write)
    .add_edge(ingest, caption)
    .set_start_executor(ingest)
    .build()
)

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

Dikkate gerekenler:

  • GraphFlow iletileri yayınlar ve koşullu geçişleri kullanır. Birleştirme davranışı hedef tarafı activation ve kenar activation_group/activation_conditionbaşına aracılığıyla yapılandırılır (örneğin, her iki kenarı ile olarak join_dgruplandırma).activation_condition="any"
  • İş akışı verileri açıkça yönlendirir; aşağı akış yürütücülerini seçmek için kullanın target_id . Birleştirme davranışı, alıcı yürütücüde (örneğin, ilk girişte verim ve tümünü bekleme) veya düzenleme oluşturucuları/toplayıcıları aracılığıyla bulunur.
  • İş Akışı'ndaki yürütücüler serbest biçimlidir: bir ChatAgent, işlevi veya bir alt iş akışını sarmalar ve bunları aynı grafın içinde karıştırır.

Temel Farklılıklar

Aşağıdaki tabloda AutoGen'in GraphFlow ile Agent Framework İş Akışı arasındaki temel farklar özetlenmiştir:

Görünüş AutoGen GraphFlow Agent Framework İş Akışı
Akış Türü Denetim akışı (kenarlar geçişlerdir) Veri akışı (uçlar iletileri yönlendirir)
Düğüm Türleri Yalnızca aracılar Aracılar, işlevler, alt iş akışları
Etkinleştirme İleti yayını Edge tabanlı etkinleştirme
Tür Güvenliği Sınırlı Tüm gün boyunca güçlü yazma
Bestelenebilirlik Sınırlı Yüksek oranda birleştirilebilir

İç İçe Yerleştirme Desenleri

AutoGen Ekibi İç İçe Yerleştirme

# 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 iç içe yerleştirme özellikleri:

  • İç içe ekip dış ekipten tüm iletileri alır
  • İç içe ekip iletileri tüm dış ekip katılımcılarına yayınlanır
  • Tüm düzeylerde paylaşılan ileti bağlamı

Aracı Çerçevesi İş Akışı İç İçe Yerleştirme

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()
               .add_edge(specialist1_executor, specialist2_executor)
               .set_start_executor(specialist1_executor)
               .build())

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

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

Aracı Çerçevesi iç içe yerleştirme özellikleri:

  • Yalıtılmış giriş/çıkış WorkflowExecutor
  • İleti yayını yok - belirli bağlantılar üzerinden veri akışları
  • Her iş akışı düzeyi için bağımsız durum yönetimi

Grup Sohbeti Desenleri

Grup sohbeti desenleri, birden çok aracının karmaşık görevler üzerinde işbirliği yapmasını sağlar. Yaygın desenlerin çerçeveler arasında nasıl çevrildiği aşağıda anlatılır.

RoundRobinGroupChat Deseni

AutoGen Uygulaması:

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")

Aracı Çerçevesi Uygulaması:

from agent_framework import SequentialBuilder, WorkflowOutputEvent

# 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_stream("Discuss this topic"):
        if isinstance(event, WorkflowOutputEvent):
            conversation_history = event.data  # list[ChatMessage]

Ayrıntılı düzenleme örnekleri için bkz:

Eşzamanlı yürütme desenleri için Agent Framework şunları da sağlar:

from agent_framework import ConcurrentBuilder, WorkflowOutputEvent

# 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_stream("Process this in parallel"):
        if isinstance(event, WorkflowOutputEvent):
            results = event.data  # Combined results from all agents

Eşzamanlı yürütme örnekleri için bkz:

MagenticOneGroupChat Deseni

AutoGen Uygulaması:

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")

Aracı Çerçevesi Uygulaması:

from typing import cast
from agent_framework import (
    MAGENTIC_EVENT_TYPE_AGENT_DELTA,
    MAGENTIC_EVENT_TYPE_ORCHESTRATOR,
    AgentRunUpdateEvent,
    ChatAgent,
    ChatMessage,
    MagenticBuilder,
    WorkflowOutputEvent,
)
from agent_framework.openai import OpenAIChatClient

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

workflow = (
    MagenticBuilder()
    .participants(researcher=researcher, coder=coder)
    .with_standard_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_stream("Complex research task"):
        if isinstance(event, AgentRunUpdateEvent):
            props = event.data.additional_properties if event.data else None
            event_type = props.get("magentic_event_type") if props else None

            if event_type == MAGENTIC_EVENT_TYPE_ORCHESTRATOR:
                text = event.data.text if event.data else ""
                print(f"[ORCHESTRATOR]: {text}")
            elif event_type == MAGENTIC_EVENT_TYPE_AGENT_DELTA:
                agent_id = props.get("agent_id", event.executor_id) if props else event.executor_id
                if event.data and event.data.text:
                    print(f"[{agent_id}]: {event.data.text}", end="")

        elif isinstance(event, WorkflowOutputEvent):
            output_messages = cast(list[ChatMessage], event.data)
            if output_messages:
                output = output_messages[-1].text

Agent Framework Özelleştirme Seçenekleri:

Magentic iş akışı kapsamlı özelleştirme seçenekleri sağlar:

  • Yönetici yapılandırması: Özel yönergeler ve model ayarlarıyla ChatAgent kullanma
  • Yuvarlak sınırlar: max_round_count, max_stall_count, max_reset_count
  • Olay akışı: Meta verilerle magentic_event_type kullanma AgentRunUpdateEvent
  • Aracı özelleştirmesi: Aracı başına özel yönergeler ve araçlar
  • Döngüdeki insan: Gözden geçirmeyi planlama, araç onayı ve müdahaleyi durdurma
# Advanced customization example with human-in-the-loop
from typing import cast
from agent_framework import (
    MAGENTIC_EVENT_TYPE_AGENT_DELTA,
    MAGENTIC_EVENT_TYPE_ORCHESTRATOR,
    AgentRunUpdateEvent,
    ChatAgent,
    MagenticBuilder,
    MagenticHumanInterventionDecision,
    MagenticHumanInterventionKind,
    MagenticHumanInterventionReply,
    MagenticHumanInterventionRequest,
    RequestInfoEvent,
    WorkflowOutputEvent,
)
from agent_framework.openai import OpenAIChatClient

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

workflow = (
    MagenticBuilder()
    .participants(
        researcher=researcher_agent,
        coder=coder_agent,
        analyst=analyst_agent,
    )
    .with_standard_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_plan_review()           # Enable human plan review
    .with_human_input_on_stall()  # Enable human intervention on stalls
    .build()
)

# Handle human intervention requests during execution
async for event in workflow.run_stream("Complex task"):
    if isinstance(event, RequestInfoEvent) 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.send_responses_streaming({event.request_id: reply}):
                pass  # Handle continuation

Ayrıntılı Eflatun örnekleri için bkz:

Gelecek Desenleri

Agent Framework yol haritası şu anda geliştirme aşamasında olan birkaç AutoGen desenini içerir:

  • Swarm deseni: İletim tabanlı aracı koordinasyonu
  • SelectorGroupChat: LLM temelli konuşmacı seçimi

İstek Yanıtı ile Döngüde İnsan

Agent Framework'ün Workflow önemli yeni özelliklerinden biri, iş akışlarının yürütmeyi duraklatmasına ve devam etmeden önce dış girişi beklemesine olanak tanıyan istek ve yanıt kavramıdır. Bu özellik AutoGen'in Team soyutlamasında yoktur ve karmaşık döngüdeki insan desenlerini etkinleştirir.

AutoGen Sınırlamaları

AutoGen'in Team soyutlaması başlatıldıktan sonra sürekli çalışır ve insan girişi için yürütmeyi duraklatmak için yerleşik mekanizmalar sağlamaz. Döngüdeki insan işlevleri için çerçeve dışında özel uygulamalar gerekir.

Agent Framework Request-Response API'si

Agent Framework, tüm yürütücülerin dekoratör kullanarak ctx.request_info()@response_handler istek gönderebildiği ve yanıtları işleyebildiği yerleşik istek yanıtı özellikleri sağlar.

from agent_framework import (
    RequestInfoEvent, 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()
           .add_edge(agent_executor, reviewer)
           .set_start_executor(agent_executor)
           .build())

Döngüdeki İnsan İş Akışlarını Çalıştırma

Agent Framework, duraklatma-sürdürme döngüsünü işlemek için akış API'leri sağlar:

from agent_framework import RequestInfoEvent, WorkflowOutputEvent

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

    while not completed:
        # First iteration uses run_stream, subsequent use send_responses_streaming
        stream = (
            workflow.send_responses_streaming(pending_responses)
            if pending_responses
            else workflow.run_stream("initial input")
        )

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

        # Collect human requests and outputs
        for event in events:
            if isinstance(event, RequestInfoEvent):
                # 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 isinstance(event, WorkflowOutputEvent):
                print(f"Final result: {event.data}")
                completed = True

Döngüdeki insan iş akışı örnekleri için bkz:

Denetim Noktası Oluşturma ve İş Akışlarını Sürdürme

Agent Framework'ün Workflow AutoGen'in Team soyutlaması üzerindeki bir diğer önemli avantajı da denetim noktası oluşturma ve yürütmeyi sürdürmeye yönelik yerleşik destektir. Bu, iş akışlarının herhangi bir denetim noktasından duraklatılmasını, kalıcı olmasını ve sürdürülebilmesini sağlayarak hataya dayanıklılık sağlar ve uzun süre çalışan veya zaman uyumsuz iş akışlarını etkinleştirir.

AutoGen Sınırlamaları

AutoGen'in Team soyutlaması yerleşik denetim noktası oluşturma özellikleri sağlamaz. Tüm kalıcılık veya kurtarma mekanizmaları, genellikle karmaşık durum yönetimi ve serileştirme mantığı gerektiren dış olarak uygulanmalıdır.

Aracı Çerçevesi Denetim Noktası Oluşturma

Agent Framework aracılığıyla FileCheckpointStorage kapsamlı denetim noktası oluşturma ve with_checkpointing() üzerinde WorkflowBuilderyöntemi sağlar. Denetim noktaları yakalama:

  • Yürütücü durumu: Kullanan her yürütücü için yerel durum ctx.set_executor_state()
  • Paylaşılan durum: Kullanarak çapraz yürütücü durumu ctx.set_shared_state()
  • İleti kuyrukları: Yürütücüler arasında bekleyen iletiler
  • İş akışı konumu: Geçerli yürütme ilerleme durumu ve sonraki adımlar
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
        await ctx.set_shared_state("original_input", data)
        await ctx.set_shared_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()
           .add_edge(processing_executor, finalize_executor)
           .set_start_executor(processing_executor)
           .with_checkpointing(checkpoint_storage=checkpoint_storage)  # Enable checkpointing
           .build())

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

Kontrol Noktalarından Devam Etme

Agent Framework belirli denetim noktalarını listelemek, incelemek ve sürdürmek için API'ler sağlar:

from typing_extensions import Never

from agent_framework import (
    Executor,
    FileCheckpointStorage,
    WorkflowContext,
    WorkflowBuilder,
    get_checkpoint_summary,
    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()
           .add_edge(upper_executor, reverse_executor)
           .set_start_executor(upper_executor)
           .with_checkpointing(checkpoint_storage=checkpoint_storage)
           .build())

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

async def checkpoint_resume_example():
    # List available checkpoints
    checkpoints = await checkpoint_storage.list_checkpoints()

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

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

        # Create new workflow instance and resume
        new_workflow = create_workflow(checkpoint_storage)
        async for event in new_workflow.run_stream(
            checkpoint_id=chosen_checkpoint_id,
            checkpoint_storage=checkpoint_storage
        ):
            print(f"Resumed event: {event}")

Gelişmiş Denetim Noktası Oluşturma Özellikleri

Döngüde İnsan Tümleştirmesi ile Denetim Noktası:

Denetim noktası oluşturma, döngüdeki insan iş akışlarıyla sorunsuz çalışır ve iş akışlarının insan girişi için duraklatılıp daha sonra sürdürülebilmesini sağlar. Bekleyen istekleri içeren bir denetim noktasından devam ederken, bu istekler olaylar olarak yeniden gönderilir:

# 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_stream(
        checkpoint_id=checkpoint_id,
        checkpoint_storage=checkpoint_storage
    ):
        if isinstance(event, RequestInfoEvent):
            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.send_responses_streaming(responses):
        print(f"Event: {event}")

Önemli Avantajlar

AutoGen ile karşılaştırıldığında, Agent Framework'ün denetim noktası şunları sağlar:

  • Otomatik kalıcılık: El ile durum yönetimi gerekmez
  • Ayrıntılı kurtarma: Herhangi bir üst sınırdan sürdürme
  • Durum yalıtımı: Ayrı yürütücü-yerel ve paylaşılan durum
  • Döngüde insan tümleştirmesi: İnsan girişiyle sorunsuz duraklatma-sürdürme
  • Hataya dayanıklılık: Hatalardan veya kesintilerden sağlam kurtarma

Pratik Örnekler

Kapsamlı denetim noktası örnekleri için bkz:

Observability

Hem AutoGen hem de Agent Framework gözlemlenebilirlik özellikleri sağlar, ancak farklı yaklaşımlar ve özellikler sunar.

AutoGen Gözlemlenebilirliği

AutoGen, izleme özelliği ile OpenTelemetry için yerel desteğe sahiptir:

  • Çalışma zamanı izleme: SingleThreadedAgentRuntime ve GrpcWorkerAgentRuntime
  • Araç yürütme: BaseToolexecute_tool GenAI semantik kurallarını izleyen span'larla
  • Aracı işlemleri: BaseChatAgent ve create_agent aralıkları ile 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)

Agent Framework Gözlemlenebilirliği

Agent Framework, birden çok yaklaşım aracılığıyla kapsamlı gözlemlenebilirlik sağlar:

  • Sıfır kod kurulumu: Ortam değişkenleri aracılığıyla otomatik izleme
  • El ile yapılandırma: Özel parametrelerle programlı kurulum
  • Zengin telemetri: Aracılar, iş akışları ve araç yürütme izleme
  • Konsol çıkışı: Yerleşik konsol günlüğü ve görselleştirmesi
from agent_framework import ChatAgent
from agent_framework.observability import setup_observability
from agent_framework.openai import OpenAIChatClient

# Zero-code setup via environment variables
# Set ENABLE_OTEL=true
# Set OTLP_ENDPOINT=http://localhost:4317

# Or manual setup
setup_observability(
    otlp_endpoint="http://localhost:4317"
)

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

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

Önemli Farklar:

  • Kurulum karmaşıklığı: Agent Framework daha basit sıfır kod kurulum seçenekleri sunar
  • Kapsam: Agent Framework, iş akışı düzeyinde gözlemlenebilirlik de dahil olmak üzere daha geniş kapsamlı bir kapsam sağlar
  • Görselleştirme: Agent Framework yerleşik konsol çıkışı ve geliştirme kullanıcı arabirimini içerir
  • Yapılandırma: Agent Framework daha esnek yapılandırma seçenekleri sunar

Ayrıntılı gözlemlenebilirlik örnekleri için bkz:

Sonuç

Bu geçiş kılavuzu, temel aracı oluşturma işleminden karmaşık çok aracılı iş akışlarına kadar her şeyi kapsayan AutoGen ile Microsoft Agent Framework arasında kapsamlı bir eşleme sağlar. Geçiş için önemli ayrıntılar:

  • Tek aracılı geçiş , Agent Framework'teki benzer API'ler ve gelişmiş özelliklerle basittir
  • Çok aracılı desenler , olay temelli mimarilerden veri akışı tabanlı mimarilere yaklaşımınızı yeniden değerlendirmenizi gerektirir, ancak GraphFlow'a zaten aşinaysanız geçiş daha kolay olur
  • Agent Framework ara yazılım, barındırılan araçlar ve yazılan iş akışları gibi ek özellikler sunar

Ek örnekler ve ayrıntılı uygulama kılavuzu için Agent Framework örnekleri dizinine bakın.

Ek Örnek Kategoriler

Agent Framework, diğer birçok önemli alanda örnekler sağlar:

Sonraki Adımlar

Hızlı Başlangıç Kılavuzu

  • Last updated on