Udostępnij przez

Przewodnik migracji autogenu do programu Microsoft Agent Framework

Kompleksowy przewodnik po migracji z narzędzia AutoGen do zestawu SDK języka Python programu Microsoft Agent Framework.

Spis treści

Kontekst

AutoGen to struktura do tworzenia agentów sztucznej inteligencji i systemów z wieloma agentami przy użyciu dużych modeli językowych (LLMs). Zaczęło się od projektu badawczego w firmie Microsoft Research i zapoczątkowało kilka pojęć w orkiestracji obejmującej wiele agentów, takich jak GroupChat i środowisko uruchomieniowe agenta opartego na zdarzeniach. Projekt był owocną współpracą społeczności open source i wiele ważnych funkcji pochodzi od zewnętrznych współautorów.

Microsoft Agent Framework to nowy zestaw SDK dla wielu języków do tworzenia agentów i przepływów pracy sztucznej inteligencji przy użyciu usługi LLMs. Reprezentuje znaczącą ewolucję pomysłów pionierskich w rozwiązaniu AutoGen i obejmuje wnioski wyciągnięte z rzeczywistego użycia. Jest ona opracowywana przez podstawowe zespoły autogenów i semantycznych jądra w firmie Microsoft i została zaprojektowana jako nowa podstawa do tworzenia aplikacji sztucznej inteligencji w przyszłości.

W tym przewodniku opisano praktyczną ścieżkę migracji: zaczyna się od oglądu tego, co pozostaje takie same i jakie zmiany w skrócie. Następnie obejmuje konfigurację klienta modelu, funkcje pojedynczego agenta i aranżację wielu agentów przy użyciu konkretnego kodu równoległego. Po drodze linki do przykładów możliwych do uruchomienia w repozytorium Platformy agentów ułatwiają zweryfikowanie każdego kroku.

Kluczowe podobieństwa i różnice

Co pozostaje takie same

Podstawy są znane. Nadal tworzysz agentów wokół klienta modelu, udostępniasz instrukcje i dołączasz narzędzia. Obie biblioteki obsługują narzędzia stylu funkcji, przesyłanie strumieniowe tokenów, zawartość wielomodalną i asynchroniczne operacje we/wy.

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

Kluczowe różnice

  1. Styl aranżacji: Funkcja AutoGen łączy rdzenie sterowane zdarzeniami z wysokim poziomem Team. Struktura agentów koncentruje się na typowych, opartych Workflow na grafach, które kieruje dane wzdłuż krawędzi i aktywuje funkcje wykonawcze, gdy dane wejściowe są gotowe.

  2. Narzędzia: Funkcja AutoGen opakowuje funkcje za pomocą polecenia FunctionTool. Struktura agenta automatycznie używa @ai_functionschematów wnioskowania i dodaje hostowane narzędzia, takie jak interpreter kodu i wyszukiwanie w Internecie.

  3. Zachowanie agenta: AssistantAgent jest jednokrotne, chyba że zwiększysz wartość max_tool_iterations. ChatAgent funkcja jest domyślnie włączona w trybie multi-turn i do momentu, gdy będzie mogła zwrócić ostateczną odpowiedź.

  4. Środowisko uruchomieniowe: Rozwiązanie AutoGen oferuje osadzone i eksperymentalne środowiska uruchomieniowe rozproszone. Struktura agenta koncentruje się obecnie na kompozycji jednoprocesowej; planowane jest wykonanie rozproszone.

Tworzenie i konfiguracja klienta modelu

Obie struktury udostępniają klientów modelu dla głównych dostawców sztucznej inteligencji, z podobnymi, ale nie identycznymi interfejsami API.

Funkcja Autogen Struktura agenta
Klient OpenAI OpenAIChatCompletionClient OpenAIChatClient
Klient odpowiedzi interfejsu OpenAI ❌ Niedostępne OpenAIResponsesClient
Azure OpenAI AzureOpenAIChatCompletionClient AzureOpenAIChatClient
Odpowiedzi usługi Azure OpenAI ❌ Niedostępne AzureOpenAIResponsesClient
Azure AI AzureAIChatCompletionClient AzureAIAgentClient
Antropiczny AnthropicChatCompletionClient 🚧 Rozmyślny
Ollama OllamaChatCompletionClient 🚧 Rozmyślny
Caching ChatCompletionCache Otoki 🚧 Rozmyślny

Klienci modelu autogenu

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

Aby uzyskać szczegółowe przykłady, zobacz:

Obsługa interfejsu API odpowiedzi (wyłącznie struktura agentów)

Struktura agenta AzureOpenAIResponsesClient i OpenAIResponsesClient zapewniają wyspecjalizowaną obsługę modeli rozumowania i odpowiedzi ze strukturą, które nie są dostępne w rozwiązaniu AutoGen:

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

Przykłady interfejsu API odpowiedzi można znaleźć w następujących tematach:

mapowanie funkcji Single-Agent

Ta sekcja mapuje funkcje pojedynczego agenta między narzędziami AutoGen i Agent Framework. Po utworzeniu klienta utwórz agenta, dołącz narzędzia i wybierz między wykonywaniem przesyłania strumieniowego i przesyłania strumieniowego.

Tworzenie i wykonywanie podstawowego agenta

Po skonfigurowaniu klienta modelu następnym krokiem jest utworzenie agentów. Obie struktury zapewniają podobne abstrakcje agentów, ale z różnymi domyślnymi zachowaniami i opcjami konfiguracji.

Asystent autogenówAgent

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

Kluczowe różnice:

  • Domyślne zachowanie: ChatAgent automatycznie iteruje za pośrednictwem wywołań narzędzi, a jednocześnie AssistantAgent wymaga jawnego max_tool_iterations ustawienia
  • Konfiguracja środowiska uruchomieniowego: ChatAgent.run() akceptuje tools i tool_choice parametry dostosowywania poszczególnych wywołań
  • Metody fabryki: Struktura agenta zapewnia wygodne metody fabryki bezpośrednio od klientów czatów
  • Zarządzanie stanem: ChatAgent jest bezstanowe i nie utrzymuje historii konwersacji między wywołaniami, w przeciwieństwie do AssistantAgent tej, która utrzymuje historię konwersacji w ramach swojego stanu

Zarządzanie stanem konwersacji za pomocą agentaWczycie

Aby kontynuować konwersacje z usługą , użyj polecenia ChatAgent do zarządzania historią AgentThreadkonwersacji:

# 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

Bezstanowy domyślnie: szybki pokaz

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

Przykłady zarządzania wątkami można znaleźć w następujących tematach:

Równoważność agenta Asystenta OpenAI

Obie struktury zapewniają integrację interfejsu API asystenta openAI:

# AutoGen OpenAIAssistantAgent
from autogen_ext.agents.openai import OpenAIAssistantAgent

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

Aby zapoznać się z przykładami asystenta OpenAI, zobacz:

Obsługa przesyłania strumieniowego

Obie struktury przesyłają strumieniowo tokeny w czasie rzeczywistym — od klientów i agentów — aby zapewnić dynamiczne działanie interfejsów użytkownika.

Automatyczne generowanie przesyłania strumieniowego

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

Przesyłanie strumieniowe platformy agentów

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

Porada: w programie Agent Framework zarówno klienci, jak i agenci dają ten sam kształt aktualizacji; Możesz przeczytać chunk.text w obu przypadkach.

Typy komunikatów i tworzenie

Zrozumienie sposobu działania komunikatów ma kluczowe znaczenie dla efektywnej komunikacji agenta. Obie struktury zapewniają różne podejścia do tworzenia i obsługi komunikatów, a rozwiązanie AutoGen korzysta z oddzielnych klas komunikatów i struktury agentów przy użyciu ujednoliconego systemu komunikatów.

Typy komunikatów autogenu

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

Typy komunikatów platformy agentów

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

Kluczowe różnice:

  • Funkcja AutoGen używa oddzielnych klas komunikatów (TextMessage, MultiModalMessage) z polem source
  • Struktura agentów używa ujednoliconego ChatMessage obiektu zawartości typizowanej role i pola
  • Komunikaty platformy agentów używają Role wyliczenia (USER, ASSISTANT, SYSTEM, TOOL) zamiast źródeł ciągów

Tworzenie i integracja narzędzi

Narzędzia rozszerzają możliwości agenta poza generowanie tekstu. Struktury przyjmują różne podejścia do tworzenia narzędzi, a platforma Agent Framework zapewnia bardziej zautomatyzowane generowanie schematu.

AutoGen, funkcjaTool

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

Struktura agenta @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])

Aby uzyskać szczegółowe przykłady, zobacz:

Hostowane narzędzia (wyłącznie struktura agentów)

Struktura agenta udostępnia hostowane narzędzia, które nie są dostępne w narzędziu AutoGen:

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

Aby uzyskać szczegółowe przykłady, zobacz:

Wymagania i zastrzeżenia:

  • Hostowane narzędzia są dostępne tylko na modelach/kontach, które je obsługują. Przed włączeniem tych narzędzi sprawdź uprawnienia i obsługę modelu dla dostawcy.
  • Konfiguracja różni się od dostawcy; postępuj zgodnie z wymaganiami wstępnymi w każdym przykładzie, aby skonfigurować i uprawnienia.
  • Nie każdy model obsługuje każde hostowane narzędzie (na przykład wyszukiwanie w Internecie a interpreter kodu). Wybierz zgodny model w swoim środowisku.

Uwaga / Notatka

Narzędzie AutoGen obsługuje lokalne narzędzia do wykonywania kodu, ale ta funkcja jest planowana dla przyszłych wersji programu Agent Framework.

Kluczowa różnica: Platforma agenta automatycznie obsługuje iterację narzędzi na poziomie agenta. W przeciwieństwie do parametru max_tool_iterations autogenu agenci programu Agent Framework kontynuują wykonywanie narzędzi domyślnie do momentu ukończenia, z wbudowanymi mechanizmami bezpieczeństwa, aby zapobiec nieskończonym pętlom.

Obsługa serwera MCP

W przypadku zaawansowanej integracji narzędzi obie struktury obsługują protokół MCP (Model Context Protocol), umożliwiając agentom interakcję z usługami zewnętrznymi i źródłami danych. Platforma Agent Framework zapewnia bardziej kompleksową wbudowaną obsługę.

Obsługa autogenu MCP

Funkcja AutoGen ma podstawową obsługę mcp za pośrednictwem rozszerzeń (szczegóły implementacji różnią się w zależności od wersji).

Obsługa programu Agent Framework MCP

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

Aby zapoznać się z przykładami mcp, zobacz:

Wzorzec agenta jako narzędzia

Jednym z zaawansowanych wzorców jest używanie samych agentów jako narzędzi, co umożliwia hierarchiczne architektury agentów. Obie struktury obsługują ten wzorzec z różnymi implementacjami.

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

Uwaga dotycząca jawnej migracji: w narzędziu AutoGen ustaw parallel_tool_calls=False klienta modelu koordynatora podczas opakowywania agentów jako narzędzi, aby uniknąć problemów ze współbieżnością podczas wywoływania tego samego wystąpienia agenta. W programie Agent Framework as_tool() nie wymaga domyślnie wyłączania równoległych wywołań narzędzi, ponieważ agenci są domyślnie bezstanowi.

Oprogramowanie pośredniczące (funkcja platformy agentów)

Struktura agenta wprowadza funkcje oprogramowania pośredniczącego, których brakuje w narzędziu AutoGen. Oprogramowanie pośredniczące umożliwia zaawansowane zagadnienia obejmujące krzyżowe, takie jak rejestrowanie, zabezpieczenia i monitorowanie wydajności.

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

Korzyści:

  • Zabezpieczenia: Sprawdzanie poprawności danych wejściowych i filtrowanie zawartości
  • Możliwość obserwowania: rejestrowanie, metryki i śledzenie
  • Wydajność: buforowanie i ograniczanie szybkości
  • Obsługa błędów: łagodne obniżenie i logika ponawiania prób

Aby uzyskać szczegółowe przykłady oprogramowania pośredniczącego, zobacz:

Agenci dostosowani

Czasami nie potrzebujesz w ogóle agenta opartego na modelu — chcesz mieć agenta deterministycznego lub opartego na interfejsie API z logiką niestandardową. Obie struktury obsługują tworzenie agentów niestandardowych, ale wzorce różnią się.

Autogen: Podklasa 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:

  • Zaimplementuj on_messages(...) i zwróć element Response z wiadomością czatu.
  • Opcjonalnie zaimplementuj on_reset(...) , aby wyczyścić stan wewnętrzny między przebiegami.

Struktura agenta: rozszerzanie agenta BaseAgent (obsługa wątków)

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:

  • AgentThreadutrzymuje stan konwersacji zewnętrznie; użyj agent.get_new_thread() polecenia i przekaż go dorun/run_stream .
  • Wywołaj self._notify_thread_of_new_messages(thread, input_messages, response_messages) tak, aby wątek miał obie strony wymiany.
  • Zobacz pełny przykład: Agent niestandardowy

Następnie przyjrzyjmy się orkiestracji wielu agentów — obszarowi, w którym struktury różnią się najbardziej.

Mapowanie funkcji wielu agentów

Omówienie modelu programowania

Modele programowania obejmujące wiele agentów reprezentują najważniejszą różnicę między dwiema strukturami.

Podejście podwójnego modelu autogenu

Funkcja AutoGen udostępnia dwa modele programowania:

  1. autogen-core: Programowanie oparte na zdarzeniach niskiego poziomu z subskrypcjami RoutedAgent komunikatów i
  2. Team abstrakcja: model skoncentrowany na wysokim poziomie oparty na uruchomieniu oparty na 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")

Wyzwania:

  • Model niskiego poziomu jest zbyt złożony dla większości użytkowników
  • Model wysokiego poziomu może stać się ograniczeniem złożonych zachowań
  • Mostkowanie między dwoma modelami zwiększa złożoność implementacji

Ujednolicony model przepływu pracy platformy Agent Framework

Struktura agenta zapewnia pojedynczą Workflow abstrakcję, która łączy najlepsze z obu podejść:

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

Aby uzyskać szczegółowe przykłady przepływów pracy, zobacz:

Korzyści:

  • Ujednolicony model: pojedyncza abstrakcja dla wszystkich poziomów złożoności
  • Bezpieczeństwo typów: silnie typizowane dane wejściowe i wyjściowe
  • Wizualizacja grafu: reprezentacja przepływu danych wyczyść
  • Elastyczna kompozycja: Mieszanie agentów, funkcji i podrzędnych przepływów pracy

Przepływ pracy a program GraphFlow

Abstrakcja struktury agenta Workflow jest inspirowana funkcją eksperymentalną GraphFlow autogenu, ale stanowi znaczącą ewolucję filozofii projektowania:

  • GraphFlow: oparty na przepływie sterowania, w którym krawędzie są przejściami, a komunikaty są emitowane do wszystkich agentów; przejścia są warunkowe dla zawartości wiadomości rozgłaszanej
  • Przepływ pracy: przepływ danych, na podstawie którego komunikaty są kierowane przez określone krawędzie i funkcje wykonawcze są aktywowane przez krawędzie z obsługą współbieżnego wykonywania.

Przegląd wizualizacji

Na poniższym diagramie kontrastuje przepływ sterowania programu AutoGen GraphFlow (po lewej) z przepływem danych platformy Agent Framework (po prawej). Program GraphFlow modeluje agentów jako węzły z przejściami warunkowymi i emisjami. Funkcje wykonawcze modeli przepływu pracy (agenci, funkcje lub podrzędne przepływy pracy) połączone z typowymi krawędziami; obsługuje również wstrzymywanie żądań/odpowiedzi i tworzenie punktów kontrolnych.

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

W praktyce:

  • Narzędzie GraphFlow używa agentów jako węzłów i emituje komunikaty; krawędzie reprezentują przejścia warunkowe.
  • Trasy przepływu pracy typizowane komunikaty wzdłuż krawędzi. Węzły (funkcje wykonawcze) mogą być agentami, czystymi funkcjami lub podrzędnymi przepływami pracy.
  • Żądanie/odpowiedź umożliwia wstrzymanie przepływu pracy dla danych wejściowych zewnętrznych; Tworzenie punktów kontrolnych utrzymuje postęp i włącza wznawianie.

Porównanie kodu

1) Sekwencyjne i warunkowe
# 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 (WSZYSTKIE 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) Routing docelowy (bez emisji)
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")

Co należy zauważyć:

  • Program GraphFlow emituje komunikaty i używa przejść warunkowych. Zachowanie sprzężenia jest konfigurowane za pośrednictwem strony activation docelowej i krawędzi activation_group/activation_condition(na przykład grupowanie obu krawędzi w element join_d za pomocą elementu ).activation_condition="any"
  • Przepływ pracy kieruje dane jawnie; użyj target_id polecenia , aby wybrać podrzędne funkcje wykonawcze. Zachowanie sprzężenia działa w funkcji wykonawczej odbierającej (na przykład wydajność przy pierwszym wejściu a oczekiwaniu na wszystkie) lub za pośrednictwem konstruktorów orkiestracji/agregatorów.
  • Funkcje wykonawcze w przepływie pracy są wolne: zawijanie ChatAgentfunkcji lub podrzędnego przepływu pracy i mieszanie ich w obrębie tego samego grafu.

Kluczowe różnice

W poniższej tabeli podsumowano podstawowe różnice między przepływem pracy programu GraphFlow a platformą Agent Framework:

Aspekt AutoGen GraphFlow Przepływ pracy platformy agentów
Typ przepływu Przepływ sterowania (krawędzie są przejściami) Przepływ danych (krawędzie kierują komunikaty)
Typy węzłów Tylko agenci Agenci, funkcje, podrzędne przepływy pracy
Aktywacja Emisja komunikatu Aktywacja oparta na brzegu sieci
Bezpieczeństwo typów Limited Silne wpisywanie przez cały czas
Możliwość komponowania Limited Wysoce komponowalne

Wzorce zagnieżdżenia

Zagnieżdżanie zespołu autogenów

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

Cechy zagnieżdżania autogenów:

  • Zagnieżdżony zespół odbiera wszystkie komunikaty od zespołu zewnętrznego
  • Zagnieżdżone komunikaty zespołu są emitowane do wszystkich uczestników zespołu zewnętrznego
  • Kontekst udostępnionej wiadomości na wszystkich poziomach

Zagnieżdżanie przepływu pracy platformy agentów

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

Cechy zagnieżdżania struktury agentów:

  • Izolowane dane wejściowe/wyjściowe za pośrednictwem WorkflowExecutor
  • Brak emisji komunikatów — dane przepływa przez określone połączenia
  • Niezależne zarządzanie stanem dla każdego poziomu przepływu pracy

Wzorce czatu grupowego

Wzorce czatów grupowych umożliwiają wielu agentom współpracę nad złożonymi zadaniami. Poniżej przedstawiono sposób tłumaczenia typowych wzorców między strukturami.

Wzorzec RoundRobinGroupChat

Implementacja autogenu:

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

Implementacja platformy agentów:

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]

Aby uzyskać szczegółowe przykłady aranżacji, zobacz:

W przypadku wzorców współbieżnych wykonywania platforma Agent Framework udostępnia również następujące elementy:

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

Aby zapoznać się z przykładami współbieżnych wykonywania, zobacz:

Wzorzec MagenticOneGroupChat

Implementacja autogenu:

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

Implementacja platformy agentów:

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

Opcje dostosowywania struktury agenta:

Przepływ pracy Magentic oferuje rozbudowane opcje dostosowywania:

  • Konfiguracja menedżera: używanie agenta ChatAgent z niestandardowymi instrukcjami i ustawieniami modelu
  • Limity zaokrągleń: max_round_count, , max_stall_countmax_reset_count
  • Przesyłanie strumieniowe zdarzeń: używanie z AgentRunUpdateEvent metadanymi magentic_event_type
  • Specjalizacja agenta: niestandardowe instrukcje i narzędzia na agenta
  • Człowiek w pętli: Przegląd planu, zatwierdzanie narzędzi i interwencja w martwym punkcie
# 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

Aby uzyskać szczegółowe przykłady magentyczne, zobacz:

Przyszłe wzorce

Harmonogram działania platformy agenta zawiera kilka wzorców autogenów, które są obecnie opracowywane:

  • Wzorzec Swarm: Koordynacja agenta oparta na handoff
  • SelectorGroupChat: wybór głośnika opartego na funkcji LLM

Human-in-the-Loop z odpowiedzią na żądanie

Kluczową nową funkcją w programie Agent Framework Workflow jest koncepcja żądania i odpowiedzi, która umożliwia przepływom pracy wstrzymanie wykonywania i oczekiwanie na zewnętrzne dane wejściowe przed kontynuowaniem. Ta funkcja nie jest obecna w abstrakcji autogenu Team i umożliwia zaawansowane wzorce pętli człowieka w pętli.

Ograniczenia autogenu

Abstrakcja autogenu Team jest uruchamiana w sposób ciągły po uruchomieniu i nie udostępnia wbudowanych mechanizmów wstrzymywania wykonywania danych wejściowych przez człowieka. Każda funkcja pętli typu człowiek-in-the-loop wymaga niestandardowych implementacji poza strukturą.

Interfejs API Request-Response platformy Agent

Struktura agenta udostępnia wbudowane funkcje odpowiedzi na żądania, w których dowolny wykonawca może wysyłać żądania przy użyciu ctx.request_info() funkcji dekoratora @response_handler i obsługiwać odpowiedzi.

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

Uruchamianie przepływów pracy typu human-in-the-loop

Struktura agenta udostępnia interfejsy API przesyłania strumieniowego do obsługi cyklu wstrzymywania wznawiania:

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

Aby zapoznać się z przykładami przepływu pracy typu human-in-the-loop, zobacz:

Punkty kontrolne i wznawianie przepływów pracy

Kolejną kluczową zaletą abstrakcji programu Agent Framework Workflow nad abstrakcją autogenu Team jest wbudowana obsługa tworzenia punktów kontrolnych i wznawiania wykonywania. Dzięki temu przepływy pracy mogą być wstrzymywane, utrwalane i wznawiane później z dowolnego punktu kontrolnego, zapewniając odporność na uszkodzenia i włączając długotrwałe lub asynchroniczne przepływy pracy.

Ograniczenia autogenu

Abstrakcja autogenu Team nie zapewnia wbudowanych funkcji tworzenia punktów kontrolnych. Wszelkie mechanizmy trwałości lub odzyskiwania muszą być implementowane zewnętrznie, często wymagają złożonego zarządzania stanem i logiki serializacji.

Sprawdzanie punktów kontrolnych struktury agentów

Struktura agenta zapewnia kompleksowe punkty kontrolne za pomocą metody FileCheckpointStorage i with_checkpointing() metody w systemie WorkflowBuilder. Przechwytywanie punktów kontrolnych:

  • Stan funkcji wykonawczej: stan lokalny dla każdego wykonawcy przy użyciu polecenia ctx.set_executor_state()
  • Stan współużytkowany: stan wykonywania krzyżowego przy użyciu polecenia ctx.set_shared_state()
  • Kolejki komunikatów: Oczekujące komunikaty między funkcjami wykonawczych
  • Pozycja przepływu pracy: Bieżący postęp wykonywania i następne kroki
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}")

Wznawianie z punktów kontrolnych

Platforma Agent Framework udostępnia interfejsy API do wyświetlania listy, inspekcji i wznawiania pracy z określonych punktów kontrolnych:

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

Zaawansowane funkcje tworzenia punktów kontrolnych

Punkt kontrolny z integracją human-in-the-loop:

Tworzenie punktów kontrolnych bezproblemowo współdziała z przepływami pracy typu human-in-the-loop, co umożliwia wstrzymanie przepływów pracy dla danych wejściowych przez ludzi i wznowienie ich później. Wznawianie z punktu kontrolnego zawierającego oczekujące żądania spowoduje ponowne emitowanie tych żądań jako zdarzenia:

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

Najważniejsze korzyści

W porównaniu z rozwiązaniem AutoGen funkcja tworzenia punktów kontrolnych platformy Agent Framework zapewnia następujące elementy:

  • Automatyczna trwałość: nie jest wymagane ręczne zarządzanie stanem
  • Szczegółowe odzyskiwanie: wznawianie z dowolnej granicy superkroku
  • Izolacja stanu: oddzielny stan funkcji wykonawczej lokalnej i udostępnionej
  • Integracja z człowiekiem w pętli: bezproblemowe wznawianie wstrzymywania z danymi wejściowymi człowieka
  • Odporność na uszkodzenia: niezawodne odzyskiwanie po awariach lub przerwach

Przykłady praktyczne

Aby uzyskać kompleksowe przykłady tworzenia punktów kontrolnych, zobacz:

Observability

Zarówno rozwiązanie AutoGen, jak i struktura agentów zapewniają możliwości obserwacji, ale z różnymi podejściami i funkcjami.

Obserwowanie autogenu

Funkcja AutoGen ma natywną obsługę technologii OpenTelemetry z instrumentacją dla:

  • Śledzenie środowiska uruchomieniowego: SingleThreadedAgentRuntime i GrpcWorkerAgentRuntime
  • Wykonywanie narzędzi: BaseTool z zakresami execute_tool następujących konwencji semantycznych GenAI
  • Operacje agenta: BaseChatAgent z zakresami create_agent i 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)

Możliwość obserwowania struktury agentów

Struktura agentów zapewnia kompleksową obserwację za pomocą wielu podejść:

  • Konfiguracja kodu zerowego: automatyczna instrumentacja za pomocą zmiennych środowiskowych
  • Konfiguracja ręczna: Konfiguracja programowa z parametrami niestandardowymi
  • Bogata telemetria: Agenci, przepływy pracy i śledzenie wykonywania narzędzi
  • Dane wyjściowe konsoli: wbudowane rejestrowanie i wizualizacja konsoli
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

Kluczowe różnice:

  • Złożoność instalacji: Struktura agentów oferuje prostsze opcje instalacji kodu zerowego
  • Zakres: Struktura agentów zapewnia szerszy zakres, w tym możliwość obserwowania na poziomie przepływu pracy
  • Wizualizacja: Struktura agenta zawiera wbudowane dane wyjściowe konsoli i interfejs użytkownika programowania
  • Konfiguracja: Struktura agenta oferuje bardziej elastyczne opcje konfiguracji

Aby uzyskać szczegółowe przykłady obserwacji, zobacz:

Podsumowanie

Ten przewodnik migracji zawiera kompleksowe mapowanie między rozwiązaniem AutoGen i programem Microsoft Agent Framework, obejmujące wszystko, od tworzenia podstawowego agenta po złożone przepływy pracy obejmujące wiele agentów. Najważniejsze wnioski dotyczące migracji:

  • Migracja pojedynczego agenta jest prosta, z podobnymi interfejsami API i rozszerzonymi możliwościami w programie Agent Framework
  • Wzorce wielu agentów wymagają ponownego przemyślenia podejścia od architektur opartych na zdarzeniach po architektury oparte na przepływie danych, ale jeśli znasz już platformę GraphFlow, przejście będzie łatwiejsze
  • Struktura agentów oferuje dodatkowe funkcje, takie jak oprogramowanie pośredniczące, hostowane narzędzia i typizowane przepływy pracy

Aby uzyskać dodatkowe przykłady i szczegółowe wskazówki dotyczące implementacji, zapoznaj się z katalogem przykładów platformy Agent Framework .

Dodatkowe kategorie przykładów

Struktura agentów udostępnia przykłady w kilku innych ważnych obszarach:

Dalsze kroki

Przewodnik Szybki start

  • Last updated on