Migrationshandbuch für v0.2 bis v0.4#

Dies ist ein Migrationshandbuch für Benutzer der v0.2.* Versionen von autogen-agentchat zur v0.4 Version, die einen neuen Satz von APIs und Funktionen einführt. Die v0.4 Version enthält Breaking Changes. Bitte lesen Sie dieses Handbuch sorgfältig durch. Wir pflegen die v0.2 Version weiterhin im 0.2 Branch; wir empfehlen jedoch dringend, auf die v0.4 Version zu aktualisieren.

Hinweis

Wir haben keinen Administratorzugriff mehr auf das pyautogen PyPI-Paket, und die Releases dieses Pakets stammen seit Version 0.2.34 nicht mehr von Microsoft. Um die v0.2 Version von AutoGen weiter zu verwenden, installieren Sie sie mit autogen-agentchat~=0.2. Bitte lesen Sie unsere Erklärung bezüglich Forks.

Was ist v0.4?#

Seit der Veröffentlichung von AutoGen im Jahr 2023 haben wir intensiv auf die Meinungen unserer Community und Benutzer von kleinen Start-ups bis hin zu großen Unternehmen gehört und viel Feedback gesammelt. Basierend auf diesem Feedback haben wir AutoGen v0.4 entwickelt, eine komplette Neuentwicklung, die eine asynchrone, ereignisgesteuerte Architektur verwendet, um Probleme wie Beobachtbarkeit, Flexibilität, interaktive Kontrolle und Skalierbarkeit anzugehen.

Die v0.4 API ist geschichtet: die Core API ist die Basisschicht, die ein skalierbares, ereignisgesteuertes Actor-Framework zur Erstellung von Agenten-Workflows bietet; die AgentChat API baut auf Core auf und bietet ein aufgabenorientiertes High-Level-Framework zum Erstellen interaktiver Agenten-Anwendungen. Sie ist ein Ersatz für AutoGen v0.2.

Der Großteil dieses Handbuchs konzentriert sich auf die AgentChat API von v0.4; Sie können jedoch auch Ihr eigenes High-Level-Framework nur mit der Core API erstellen.

Neu bei AutoGen?#

Springen Sie direkt zum AgentChat Tutorial, um mit v0.4 loszulegen.

Was ist in diesem Handbuch?#

Wir bieten eine detaillierte Anleitung, wie Sie Ihren bestehenden Code von v0.2 auf v0.4 migrieren.

Siehe die einzelnen Funktionen unten für detaillierte Informationen zur Migration.

Die folgenden Funktionen, die derzeit in v0.2 vorhanden sind, werden in zukünftigen Releases von v0.4.* Versionen bereitgestellt

  • Modell-Client Kosten #4835

  • Lehrbarer Agent

  • RAG-Agent

Wir werden dieses Handbuch aktualisieren, sobald die fehlenden Funktionen verfügbar sind.

Modell-Client#

In v0.2 konfigurieren Sie den Modell-Client wie folgt und erstellen das OpenAIWrapper Objekt.

from autogen.oai import OpenAIWrapper

config_list = [
    {"model": "gpt-4o", "api_key": "sk-xxx"},
    {"model": "gpt-4o-mini", "api_key": "sk-xxx"},
]

model_client = OpenAIWrapper(config_list=config_list)

Hinweis: In AutoGen 0.2 hat der OpenAI-Client Konfigurationen in der Liste ausprobiert, bis eine funktionierte. 0.4 erwartet stattdessen, dass eine spezifische Modellkonfiguration ausgewählt wird.

In v0.4 bieten wir zwei Möglichkeiten, einen Modell-Client zu erstellen.

Verwenden der Komponentenkonfiguration#

AutoGen 0.4 verfügt über ein generisches Komponentenkonfigurationssystem. Modell-Clients sind ein großartiger Anwendungsfall dafür. Nachfolgend sehen Sie, wie Sie einen OpenAI Chat Completion Client erstellen.


from autogen_core.models import ChatCompletionClient

config = {
    "provider": "OpenAIChatCompletionClient",
    "config": {
        "model": "gpt-4o",
        "api_key": "sk-xxx" # os.environ["...']
    }
}

model_client = ChatCompletionClient.load_component(config)

Direkter Zugriff auf die Modell-Client-Klasse#

Open AI

from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key="sk-xxx")

Azure OpenAI

from autogen_ext.models.openai import AzureOpenAIChatCompletionClient

model_client = AzureOpenAIChatCompletionClient(
    azure_deployment="gpt-4o",
    azure_endpoint="https://<your-endpoint>.openai.azure.com/",
    model="gpt-4o",
    api_version="2024-09-01-preview",
    api_key="sk-xxx",
)

Lesen Sie mehr über OpenAIChatCompletionClient.

Modell-Client für OpenAI-kompatible APIs#

Sie können den OpenAIChatCompletionClient verwenden, um eine Verbindung zu einer OpenAI-kompatiblen API herzustellen, aber Sie müssen die base_url und model_info angeben.

from autogen_ext.models.openai import OpenAIChatCompletionClient

custom_model_client = OpenAIChatCompletionClient(
    model="custom-model-name",
    base_url="https://custom-model.com/reset/of/the/path",
    api_key="placeholder",
    model_info={
        "vision": True,
        "function_calling": True,
        "json_output": True,
        "family": "unknown",
        "structured_output": True,
    },
)

Hinweis: Wir testen nicht alle OpenAI-kompatiblen APIs, und viele von ihnen funktionieren anders als die OpenAI-API, auch wenn sie angeben, sie zu unterstützen. Bitte testen Sie sie vor der Verwendung.

Lesen Sie mehr über Modell-Clients im AgentChat Tutorial und detailliertere Informationen in der Core API-Dokumentation.

Unterstützung für andere gehostete Modelle wird in Zukunft hinzugefügt.

Modell-Client-Cache#

In v0.2 können Sie den Cache-Seed über den Parameter cache_seed in der LLM-Konfiguration festlegen. Der Cache ist standardmäßig aktiviert.

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
    "cache_seed": 42,
}

In v0.4 ist der Cache standardmäßig nicht aktiviert. Um ihn zu verwenden, müssen Sie einen ChatCompletionCache Wrapper um den Modell-Client verwenden.

Sie können einen DiskCacheStore oder RedisStore verwenden, um den Cache zu speichern.

pip install -U "autogen-ext[openai, diskcache, redis]"

Hier ist ein Beispiel für die Verwendung von diskcache für lokales Caching

import asyncio
import tempfile

from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.models.cache import ChatCompletionCache, CHAT_CACHE_VALUE_TYPE
from autogen_ext.cache_store.diskcache import DiskCacheStore
from diskcache import Cache


async def main():
    with tempfile.TemporaryDirectory() as tmpdirname:
        # Initialize the original client
        openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")

        # Then initialize the CacheStore, in this case with diskcache.Cache.
        # You can also use redis like:
        # from autogen_ext.cache_store.redis import RedisStore
        # import redis
        # redis_instance = redis.Redis()
        # cache_store = RedisCacheStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
        cache_store = DiskCacheStore[CHAT_CACHE_VALUE_TYPE](Cache(tmpdirname))
        cache_client = ChatCompletionCache(openai_model_client, cache_store)

        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print response from OpenAI
        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print cached response
        await openai_model_client.close()


asyncio.run(main())

Assistenten-Agent#

In v0.2 erstellen Sie einen Assistenten-Agenten wie folgt

from autogen.agentchat import AssistantAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant.",
    llm_config=llm_config,
)

In v0.4 ist es ähnlich, aber Sie müssen model_client anstelle von llm_config angeben.

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key="sk-xxx", seed=42, temperature=0)

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant.",
    model_client=model_client,
)

Die Verwendung ist jedoch etwas anders. In v0.4 rufen Sie anstelle von assistant.send die Methoden assistant.on_messages oder assistant.on_messages_stream auf, um eingehende Nachrichten zu verarbeiten. Darüber hinaus sind die Methoden on_messages und on_messages_stream asynchron, und letztere gibt einen asynchronen Generator zurück, um die inneren Gedanken des Agenten zu streamen.

Hier sehen Sie, wie Sie den Assistenten-Agenten in v0.4 direkt aufrufen können, fortgesetzt aus dem obigen Beispiel

import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    response = await assistant.on_messages([TextMessage(content="Hello!", source="user")], cancellation_token)
    print(response)

    await model_client.close()

asyncio.run(main())

Der CancellationToken kann verwendet werden, um die Anfrage asynchron abzubrechen, wenn Sie cancellation_token.cancel() aufrufen, was dazu führt, dass der await auf dem Aufruf von on_messages eine CancelledError auslöst.

Lesen Sie mehr im Agent Tutorial und über AssistantAgent.

Multi-Modaler Agent#

Der AssistantAgent in v0.4 unterstützt Multi-Modale Eingaben, wenn der Modell-Client dies unterstützt. Die vision Fähigkeit des Modell-Clients wird verwendet, um zu bestimmen, ob der Agent Multi-Modale Eingaben unterstützt.

import asyncio
from pathlib import Path
from autogen_agentchat.messages import MultiModalMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken, Image
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    message = MultiModalMessage(
        content=["Here is an image:", Image.from_file(Path("test.png"))],
        source="user",
    )
    response = await assistant.on_messages([message], cancellation_token)
    print(response)

    await model_client.close()

asyncio.run(main())

Benutzer-Proxy#

In v0.2 erstellen Sie einen Benutzer-Proxy wie folgt

from autogen.agentchat import UserProxyAgent

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config=False,
    llm_config=False,
)

Dieser Benutzer-Proxy würde Eingaben vom Benutzer über die Konsole entgegennehmen und beenden, wenn die eingehende Nachricht mit "TERMINATE" endet.

In v0.4 ist ein Benutzer-Proxy einfach ein Agent, der nur Benutzereingaben entgegennimmt, es sind keine weiteren speziellen Konfigurationen erforderlich. Sie können einen Benutzer-Proxy wie folgt erstellen

from autogen_agentchat.agents import UserProxyAgent

user_proxy = UserProxyAgent("user_proxy")

Siehe UserProxyAgent für weitere Details und wie die Eingabefunktion mit Timeout angepasst werden kann.

RAG-Agent#

In v0.2 gab es das Konzept von lehrbaren Agenten sowie RAG-Agenten, die eine Datenbankkonfiguration übernehmen konnten.

teachable_agent = ConversableAgent(
    name="teachable_agent",
    llm_config=llm_config
)

# Instantiate a Teachability object. Its parameters are all optional.
teachability = Teachability(
    reset_db=False,
    path_to_db_dir="./tmp/interactive/teachability_db"
)

teachability.add_to_agent(teachable_agent)

In v0.4 können Sie einen RAG-Agenten mithilfe der Klasse Memory implementieren. Insbesondere können Sie eine Speicherklasse definieren und diese als Parameter an den Assistenten-Agenten übergeben. Sehen Sie sich das Memory-Tutorial für weitere Details an.

Diese klare Trennung der Verantwortlichkeiten ermöglicht es Ihnen, einen Speicher zu implementieren, der jede gewünschte Datenbank oder jedes Speichersystem verwendet (Sie müssen von der Klasse Memory erben) und ihn mit einem Assistenten-Agenten zu verwenden. Das folgende Beispiel zeigt, wie ein ChromaDB Vektorspeicher mit dem Assistenten-Agenten verwendet wird. Darüber hinaus sollte Ihre Anwendungslogik bestimmen, wie und wann Inhalte zum Speicher hinzugefügt werden. Sie können beispielsweise memory.add für jede Antwort des Assistenten-Agenten aufrufen oder einen separaten LLM-Aufruf verwenden, um zu bestimmen, ob der Inhalt zum Speicher hinzugefügt werden soll.


# ...
# example of a ChromaDBVectorMemory class
chroma_user_memory = ChromaDBVectorMemory(
    config=PersistentChromaDBVectorMemoryConfig(
        collection_name="preferences",
        persistence_path=os.path.join(str(Path.home()), ".chromadb_autogen"),
        k=2,  # Return top  k results
        score_threshold=0.4,  # Minimum similarity score
    )
)

# you can add logic such as a document indexer that adds content to the memory store

assistant_agent = AssistantAgent(
    name="assistant_agent",
    model_client=OpenAIChatCompletionClient(
        model="gpt-4o",
    ),
    tools=[get_weather],
    memory=[chroma_user_memory],
)

Konversationsfähiger Agent und Registrierung von Antworten#

In v0.2 können Sie einen konversationsfähigen Agenten erstellen und eine Antwortfunktion wie folgt registrieren

from typing import Any, Dict, List, Optional, Tuple, Union
from autogen.agentchat import ConversableAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

conversable_agent = ConversableAgent(
    name="conversable_agent",
    system_message="You are a helpful assistant.",
    llm_config=llm_config,
    code_execution_config={"work_dir": "coding"},
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
)

def reply_func(
    recipient: ConversableAgent,
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[Any] = None,
) -> Tuple[bool, Union[str, Dict, None]]:
    # Custom reply logic here
    return True, "Custom reply"

# Register the reply function
conversable_agent.register_reply([ConversableAgent], reply_func, position=0)

# NOTE: An async reply function will only be invoked with async send.

Anstatt zu erraten, was die reply_func tut, alle ihre Parameter und was die position sein sollte, können wir in v0.4 einfach einen benutzerdefinierten Agenten erstellen und die Methoden on_messages, on_reset und produced_message_types implementieren.

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

class CustomAgent(BaseChatAgent):
    async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
        return Response(chat_message=TextMessage(content="Custom reply", source=self.name))

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

    @property
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
        return (TextMessage,)

Sie können dann den benutzerdefinierten Agenten auf die gleiche Weise wie den AssistantAgent verwenden. Sehen Sie sich das Tutorial zu benutzerdefinierten Agenten für weitere Details an.

Agentenstatus speichern und laden#

In v0.2 gibt es keine integrierte Möglichkeit, den Status eines Agenten zu speichern und zu laden: Sie müssen ihn selbst implementieren, indem Sie das Attribut chat_messages von ConversableAgent exportieren und es über den Parameter chat_messages wieder importieren.

In v0.4 können Sie die Methoden save_state und load_state auf Agenten aufrufen, um ihren Status zu speichern und zu laden.

import asyncio
import json
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    response = await assistant.on_messages([TextMessage(content="Hello!", source="user")], cancellation_token)
    print(response)

    # Save the state.
    state = await assistant.save_state()

    # (Optional) Write state to disk.
    with open("assistant_state.json", "w") as f:
        json.dump(state, f)

    # (Optional) Load it back from disk.
    with open("assistant_state.json", "r") as f:
        state = json.load(f)
        print(state) # Inspect the state, which contains the chat history.

    # Carry on the chat.
    response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)
    print(response)

    # Load the state, resulting the agent to revert to the previous state before the last message.
    await assistant.load_state(state)

    # Carry on the same chat again.
    response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)
    # Close the connection to the model client.
    await model_client.close()

asyncio.run(main())

Sie können auch save_state und load_state für beliebige Teams aufrufen, wie z.B. RoundRobinGroupChat, um den Status des gesamten Teams zu speichern und zu laden.

Zwei-Agenten-Chat#

In v0.2 können Sie einen Zwei-Agenten-Chat für die Codeausführung wie folgt erstellen

from autogen.coding import LocalCommandLineCodeExecutor
from autogen.agentchat import AssistantAgent, UserProxyAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant. Write all code in python. Reply only 'TERMINATE' if the task is done.",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"code_executor": LocalCommandLineCodeExecutor(work_dir="coding")},
    llm_config=False,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

chat_result = user_proxy.initiate_chat(assistant, message="Write a python script to print 'Hello, world!'")
# Intermediate messages are printed to the console directly.
print(chat_result)

Um das gleiche Verhalten in v0.4 zu erzielen, können Sie den AssistantAgent und den CodeExecutorAgent zusammen in einem RoundRobinGroupChat verwenden.

import asyncio
from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant. Write all code in python. Reply only 'TERMINATE' if the task is done.",
        model_client=model_client,
    )

    code_executor = CodeExecutorAgent(
        name="code_executor",
        code_executor=LocalCommandLineCodeExecutor(work_dir="coding"),
    )

    # The termination condition is a combination of text termination and max message termination, either of which will cause the chat to terminate.
    termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(10)

    # The group chat will alternate between the assistant and the code executor.
    group_chat = RoundRobinGroupChat([assistant, code_executor], termination_condition=termination)

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a python script to print 'Hello, world!'")
    # `Console` is a simple UI to display the stream.
    await Console(stream)
    
    # Close the connection to the model client.
    await model_client.close()

asyncio.run(main())

Tool-Nutzung#

In v0.2 mussten Sie für einen Chatbot mit Tool-Nutzung zwei Agenten haben, einen zum Aufrufen des Tools und einen zur Ausführung des Tools. Sie mussten einen Zwei-Agenten-Chat für jede Benutzeranfrage initiieren.

from autogen.agentchat import AssistantAgent, UserProxyAgent, register_function

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

tool_caller = AssistantAgent(
    name="tool_caller",
    system_message="You are a helpful assistant. You can call tools to help user.",
    llm_config=llm_config,
    max_consecutive_auto_reply=1, # Set to 1 so that we return to the application after each assistant reply as we are building a chatbot.
)

tool_executor = UserProxyAgent(
    name="tool_executor",
    human_input_mode="NEVER",
    code_execution_config=False,
    llm_config=False,
)

def get_weather(city: str) -> str:
    return f"The weather in {city} is 72 degree and sunny."

# Register the tool function to the tool caller and executor.
register_function(get_weather, caller=tool_caller, executor=tool_executor)

while True:
    user_input = input("User: ")
    if user_input == "exit":
        break
    chat_result = tool_executor.initiate_chat(
        tool_caller,
        message=user_input,
        summary_method="reflection_with_llm", # To let the model reflect on the tool use, set to "last_msg" to return the tool call result directly.
    )
    print("Assistant:", chat_result.summary)

In v0.4 benötigen Sie im Grunde nur einen Agenten – den AssistantAgent – um sowohl den Tool-Aufruf als auch die Tool-Ausführung zu handhaben.

import asyncio
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage

def get_weather(city: str) -> str: # Async tool is possible too.
    return f"The weather in {city} is 72 degree and sunny."

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)
    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant. You can call tools to help user.",
        model_client=model_client,
        tools=[get_weather],
        reflect_on_tool_use=True, # Set to True to have the model reflect on the tool use, set to False to return the tool call result directly.
    )
    while True:
        user_input = input("User: ")
        if user_input == "exit":
            break
        response = await assistant.on_messages([TextMessage(content=user_input, source="user")], CancellationToken())
        print("Assistant:", response.chat_message.to_text())
    await model_client.close()

asyncio.run(main())

Wenn Sie Tool-fähige Agenten in einem Gruppenchat wie RoundRobinGroupChat verwenden, machen Sie einfach dasselbe wie oben, um Tools zu den Agenten hinzuzufügen, und erstellen Sie einen Gruppenchat mit den Agenten.

Chat-Ergebnis#

In v0.2 erhalten Sie ein ChatResult Objekt von der Methode initiate_chat. Zum Beispiel

chat_result = tool_executor.initiate_chat(
    tool_caller,
    message=user_input,
    summary_method="reflection_with_llm",
)
print(chat_result.summary) # Get LLM-reflected summary of the chat.
print(chat_result.chat_history) # Get the chat history.
print(chat_result.cost) # Get the cost of the chat.
print(chat_result.human_input) # Get the human input solicited by the chat.

Siehe ChatResult Docs für weitere Details.

In v0.4 erhalten Sie ein TaskResult Objekt von einer Methode run oder run_stream. Das TaskResult Objekt enthält die messages, welche die Nachrichten-Historie des Chats ist, einschließlich privater Nachrichten (Tool-Aufrufe etc.) und öffentlicher Nachrichten beider Agenten.

Es gibt einige bemerkenswerte Unterschiede zwischen TaskResult und ChatResult

  • Die Liste messages in TaskResult verwendet ein anderes Nachrichtenformat als die Liste ChatResult.chat_history.

  • Es gibt kein Feld summary. Es liegt an der Anwendung zu entscheiden, wie der Chat mithilfe der Liste messages zusammengefasst wird.

  • human_input wird nicht im TaskResult Objekt bereitgestellt, da die Benutzereingabe aus der Liste messages durch Filterung nach dem Feld source extrahiert werden kann.

  • cost wird nicht im TaskResult Objekt bereitgestellt, jedoch können Sie die Kosten basierend auf der Token-Nutzung berechnen. Es wäre eine großartige Community-Erweiterung, die Kostenberechnung hinzuzufügen. Siehe Community-Erweiterungen.

Konvertierung zwischen v0.2 und v0.4 Nachrichten#

Sie können die folgenden Konvertierungsfunktionen verwenden, um zwischen einer v0.4-Nachricht in autogen_agentchat.base.TaskResult.messages und einer v0.2-Nachricht in ChatResult.chat_history zu konvertieren.

from typing import Any, Dict, List, Literal

from autogen_agentchat.messages import (
    BaseAgentEvent,
    BaseChatMessage,
    HandoffMessage,
    MultiModalMessage,
    StopMessage,
    TextMessage,
    ToolCallExecutionEvent,
    ToolCallRequestEvent,
    ToolCallSummaryMessage,
)
from autogen_core import FunctionCall, Image
from autogen_core.models import FunctionExecutionResult


def convert_to_v02_message(
    message: BaseAgentEvent | BaseChatMessage,
    role: Literal["assistant", "user", "tool"],
    image_detail: Literal["auto", "high", "low"] = "auto",
) -> Dict[str, Any]:
    """Convert a v0.4 AgentChat message to a v0.2 message.

    Args:
        message (BaseAgentEvent | BaseChatMessage): The message to convert.
        role (Literal["assistant", "user", "tool"]): The role of the message.
        image_detail (Literal["auto", "high", "low"], optional): The detail level of image content in multi-modal message. Defaults to "auto".

    Returns:
        Dict[str, Any]: The converted AutoGen v0.2 message.
    """
    v02_message: Dict[str, Any] = {}
    if isinstance(message, TextMessage | StopMessage | HandoffMessage | ToolCallSummaryMessage):
        v02_message = {"content": message.content, "role": role, "name": message.source}
    elif isinstance(message, MultiModalMessage):
        v02_message = {"content": [], "role": role, "name": message.source}
        for modal in message.content:
            if isinstance(modal, str):
                v02_message["content"].append({"type": "text", "text": modal})
            elif isinstance(modal, Image):
                v02_message["content"].append(modal.to_openai_format(detail=image_detail))
            else:
                raise ValueError(f"Invalid multimodal message content: {modal}")
    elif isinstance(message, ToolCallRequestEvent):
        v02_message = {"tool_calls": [], "role": "assistant", "content": None, "name": message.source}
        for tool_call in message.content:
            v02_message["tool_calls"].append(
                {
                    "id": tool_call.id,
                    "type": "function",
                    "function": {"name": tool_call.name, "args": tool_call.arguments},
                }
            )
    elif isinstance(message, ToolCallExecutionEvent):
        tool_responses: List[Dict[str, str]] = []
        for tool_result in message.content:
            tool_responses.append(
                {
                    "tool_call_id": tool_result.call_id,
                    "role": "tool",
                    "content": tool_result.content,
                }
            )
        content = "\n\n".join([response["content"] for response in tool_responses])
        v02_message = {"tool_responses": tool_responses, "role": "tool", "content": content}
    else:
        raise ValueError(f"Invalid message type: {type(message)}")
    return v02_message


def convert_to_v04_message(message: Dict[str, Any]) -> BaseAgentEvent | BaseChatMessage:
    """Convert a v0.2 message to a v0.4 AgentChat message."""
    if "tool_calls" in message:
        tool_calls: List[FunctionCall] = []
        for tool_call in message["tool_calls"]:
            tool_calls.append(
                FunctionCall(
                    id=tool_call["id"],
                    name=tool_call["function"]["name"],
                    arguments=tool_call["function"]["args"],
                )
            )
        return ToolCallRequestEvent(source=message["name"], content=tool_calls)
    elif "tool_responses" in message:
        tool_results: List[FunctionExecutionResult] = []
        for tool_response in message["tool_responses"]:
            tool_results.append(
                FunctionExecutionResult(
                    call_id=tool_response["tool_call_id"],
                    content=tool_response["content"],
                    is_error=False,
                    name=tool_response["name"],
                )
            )
        return ToolCallExecutionEvent(source="tools", content=tool_results)
    elif isinstance(message["content"], list):
        content: List[str | Image] = []
        for modal in message["content"]:  # type: ignore
            if modal["type"] == "text":  # type: ignore
                content.append(modal["text"])  # type: ignore
            else:
                content.append(Image.from_uri(modal["image_url"]["url"]))  # type: ignore
        return MultiModalMessage(content=content, source=message["name"])
    elif isinstance(message["content"], str):
        return TextMessage(content=message["content"], source=message["name"])
    else:
        raise ValueError(f"Unable to convert message: {message}")

Gruppenchat#

In v0.2 mussten Sie eine GroupChat Klasse erstellen und sie an einen GroupChatManager übergeben und einen Teilnehmer haben, der ein Benutzer-Proxy ist, um den Chat zu initiieren. Für ein einfaches Szenario eines Schreibers und Kritikers können Sie Folgendes tun

from autogen.agentchat import AssistantAgent, GroupChat, GroupChatManager

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

writer = AssistantAgent(
    name="writer",
    description="A writer.",
    system_message="You are a writer.",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("APPROVE"),
)

critic = AssistantAgent(
    name="critic",
    description="A critic.",
    system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
    llm_config=llm_config,
)

# Create a group chat with the writer and critic.
groupchat = GroupChat(agents=[writer, critic], messages=[], max_round=12)

# Create a group chat manager to manage the group chat, use round-robin selection method.
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config, speaker_selection_method="round_robin")

# Initiate the chat with the editor, intermediate messages are printed to the console directly.
result = editor.initiate_chat(
    manager,
    message="Write a short story about a robot that discovers it has feelings.",
)
print(result.summary)

In v0.4 können Sie RoundRobinGroupChat verwenden, um das gleiche Verhalten zu erzielen.

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    writer = AssistantAgent(
        name="writer",
        description="A writer.",
        system_message="You are a writer.",
        model_client=model_client,
    )

    critic = AssistantAgent(
        name="critic",
        description="A critic.",
        system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
        model_client=model_client,
    )

    # The termination condition is a text termination, which will cause the chat to terminate when the text "APPROVE" is received.
    termination = TextMentionTermination("APPROVE")

    # The group chat will alternate between the writer and the critic.
    group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination, max_turns=12)

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a short story about a robot that discovers it has feelings.")
    # `Console` is a simple UI to display the stream.
    await Console(stream)
    # Close the connection to the model client.
    await model_client.close()

asyncio.run(main())

Für die LLM-basierte Sprecherwahl können Sie stattdessen SelectorGroupChat verwenden. Siehe Tutorial zu Selector Group Chat und SelectorGroupChat für weitere Details.

Hinweis: In v0.4 müssen Sie Funktionen nicht mehr auf einem Benutzer-Proxy registrieren, um Tools in einem Gruppenchat zu verwenden. Sie können die Tool-Funktionen einfach dem AssistantAgent übergeben, wie im Abschnitt Tool-Nutzung gezeigt. Der Agent ruft die Tools automatisch auf, wenn sie benötigt werden. Wenn Ihr Tool keine korrekt formatierten Antworten liefert, können Sie den Parameter reflect_on_tool_use verwenden, damit das Modell die Tool-Nutzung reflektiert.

Gruppenchat mit Wiederaufnahme#

In v0.2 ist ein Gruppenchat mit Wiederaufnahme etwas kompliziert. Sie müssen die Gruppenchat-Nachrichten explizit speichern und beim Wiederaufnehmen des Chats wieder laden. Siehe Gruppenchat mit Wiederaufnahme in v0.2 für weitere Details.

In v0.4 können Sie einfach erneut run oder run_stream mit demselben Gruppenchat-Objekt aufrufen, um den Chat fortzusetzen. Zum Exportieren und Laden des Status können Sie die Methoden save_state und load_state verwenden.

import asyncio
import json
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

def create_team(model_client : OpenAIChatCompletionClient) -> RoundRobinGroupChat:
    writer = AssistantAgent(
        name="writer",
        description="A writer.",
        system_message="You are a writer.",
        model_client=model_client,
    )

    critic = AssistantAgent(
        name="critic",
        description="A critic.",
        system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
        model_client=model_client,
    )

    # The termination condition is a text termination, which will cause the chat to terminate when the text "APPROVE" is received.
    termination = TextMentionTermination("APPROVE")

    # The group chat will alternate between the writer and the critic.
    group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination)

    return group_chat


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)
    # Create team.
    group_chat = create_team(model_client)

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a short story about a robot that discovers it has feelings.")
    # `Console` is a simple UI to display the stream.
    await Console(stream)

    # Save the state of the group chat and all participants.
    state = await group_chat.save_state()
    with open("group_chat_state.json", "w") as f:
        json.dump(state, f)

    # Create a new team with the same participants configuration.
    group_chat = create_team(model_client)

    # Load the state of the group chat and all participants.
    with open("group_chat_state.json", "r") as f:
        state = json.load(f)
    await group_chat.load_state(state)

    # Resume the chat.
    stream = group_chat.run_stream(task="Translate the story into Chinese.")
    await Console(stream)

    # Close the connection to the model client.
    await model_client.close()

asyncio.run(main())

Gruppenchat-Status speichern und laden#

In v0.2 müssen Sie die Gruppenchat-Nachrichten explizit speichern und beim Wiederaufnehmen des Chats wieder laden.

In v0.4 können Sie einfach die Methoden save_state und load_state auf dem Gruppenchat-Objekt aufrufen. Siehe Gruppenchat mit Wiederaufnahme für ein Beispiel.

Gruppenchat mit Tool-Nutzung#

Im Gruppenchat von v0.2 müssen Sie bei Tool-Beteiligung die Tool-Funktionen auf einem Benutzer-Proxy registrieren und den Benutzer-Proxy in den Gruppenchat aufnehmen. Die von anderen Agenten durchgeführten Tool-Aufrufe werden an den Benutzer-Proxy zur Ausführung weitergeleitet.

Wir haben zahlreiche Probleme mit diesem Ansatz beobachtet, wie z. B. dass die Weiterleitung von Tool-Aufrufen nicht wie erwartet funktioniert und die Tool-Aufrufsanfrage und -ergebnis von Modellen ohne Unterstützung für Funktionsaufrufe nicht akzeptiert werden können.

In v0.4 ist es nicht mehr notwendig, die Tool-Funktionen auf einem Benutzer-Proxy zu registrieren, da die Tools direkt im AssistantAgent ausgeführt werden, der die Antwort vom Tool im Gruppenchat veröffentlicht. Der Gruppenchat-Manager muss also nicht mehr an der Weiterleitung von Tool-Aufrufen beteiligt sein.

Siehe Tutorial zu Selector Group Chat für ein Beispiel zur Verwendung von Tools in einem Gruppenchat.

Gruppenchat mit benutzerdefiniertem Selektor (Stateflow)#

Im Gruppenchat von v0.2 kann die speaker_selection_method, wenn sie auf eine benutzerdefinierte Funktion gesetzt ist, die Standard-Auswahlmethode überschreiben. Dies ist nützlich für die Implementierung einer zustandsbasierten Auswahlmethode. Weitere Details finden Sie unter Benutzerdefinierte Sprecherwahl in v0.2.

In v0.4 können Sie SelectorGroupChat mit selector_func verwenden, um das gleiche Verhalten zu erzielen. Die selector_func ist eine Funktion, die den aktuellen Nachrichtenverlauf des Gruppenchats nimmt und den Namen des nächsten Sprechers zurückgibt. Wenn None zurückgegeben wird, wird die LLM-basierte Auswahlmethode verwendet.

Hier ist ein Beispiel für die Verwendung der zustandsbasierten Auswahlmethode zur Implementierung eines Web-Such-/Analyse-Szenarios.

import asyncio
from typing import Sequence
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Note: This example uses mock tools instead of real APIs for demonstration purposes
def search_web_tool(query: str) -> str:
    if "2006-2007" in query:
        return """Here are the total points scored by Miami Heat players in the 2006-2007 season:
        Udonis Haslem: 844 points
        Dwayne Wade: 1397 points
        James Posey: 550 points
        ...
        """
    elif "2007-2008" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2007-2008 is 214."
    elif "2008-2009" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2008-2009 is 398."
    return "No data found."


def percentage_change_tool(start: float, end: float) -> float:
    return ((end - start) / start) * 100

def create_team(model_client : OpenAIChatCompletionClient) -> SelectorGroupChat:
    planning_agent = AssistantAgent(
        "PlanningAgent",
        description="An agent for planning tasks, this agent should be the first to engage when given a new task.",
        model_client=model_client,
        system_message="""
        You are a planning agent.
        Your job is to break down complex tasks into smaller, manageable subtasks.
        Your team members are:
            Web search agent: Searches for information
            Data analyst: Performs calculations

        You only plan and delegate tasks - you do not execute them yourself.

        When assigning tasks, use this format:
        1. <agent> : <task>

        After all tasks are complete, summarize the findings and end with "TERMINATE".
        """,
    )

    web_search_agent = AssistantAgent(
        "WebSearchAgent",
        description="A web search agent.",
        tools=[search_web_tool],
        model_client=model_client,
        system_message="""
        You are a web search agent.
        Your only tool is search_tool - use it to find information.
        You make only one search call at a time.
        Once you have the results, you never do calculations based on them.
        """,
    )

    data_analyst_agent = AssistantAgent(
        "DataAnalystAgent",
        description="A data analyst agent. Useful for performing calculations.",
        model_client=model_client,
        tools=[percentage_change_tool],
        system_message="""
        You are a data analyst.
        Given the tasks you have been assigned, you should analyze the data and provide results using the tools provided.
        """,
    )

    # The termination condition is a combination of text mention termination and max message termination.
    text_mention_termination = TextMentionTermination("TERMINATE")
    max_messages_termination = MaxMessageTermination(max_messages=25)
    termination = text_mention_termination | max_messages_termination

    # The selector function is a function that takes the current message thread of the group chat
    # and returns the next speaker's name. If None is returned, the LLM-based selection method will be used.
    def selector_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
        if messages[-1].source != planning_agent.name:
            return planning_agent.name # Always return to the planning agent after the other agents have spoken.
        return None

    team = SelectorGroupChat(
        [planning_agent, web_search_agent, data_analyst_agent],
        model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"), # Use a smaller model for the selector.
        termination_condition=termination,
        selector_func=selector_func,
    )
    return team

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    team = create_team(model_client)
    task = "Who was the Miami Heat player with the highest points in the 2006-2007 season, and what was the percentage change in his total rebounds between the 2007-2008 and 2008-2009 seasons?"
    await Console(team.run_stream(task=task))

asyncio.run(main())

Verschachtelter Chat#

Verschachtelter Chat ermöglicht es Ihnen, ein ganzes Team oder einen anderen Agenten in einen Agenten zu verschachteln. Dies ist nützlich für die Erstellung einer hierarchischen Struktur von Agenten oder "Informationssilos", da die verschachtelten Agenten nicht direkt mit anderen Agenten außerhalb derselben Gruppe kommunizieren können.

In v0.2 wird der verschachtelte Chat durch die Verwendung der Methode register_nested_chats in der Klasse ConversableAgent unterstützt. Sie müssen die verschachtelte Sequenz von Agenten mithilfe von Dictionaries angeben. Siehe Verschachtelter Chat in v0.2 für weitere Details.

In v0.4 ist der verschachtelte Chat ein Implementierungsdetail eines benutzerdefinierten Agenten. Sie können einen benutzerdefinierten Agenten erstellen, der ein Team oder einen anderen Agenten als Parameter übernimmt und die Methode on_messages implementiert, um das verschachtelte Team oder den Agenten auszulösen. Es liegt an der Anwendung zu entscheiden, wie Nachrichten vom und zum verschachtelten Team oder Agenten übergeben oder transformiert werden.

Das folgende Beispiel zeigt einen einfachen verschachtelten Chat, der Zahlen zählt.

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

class CountingAgent(BaseChatAgent):
    """An agent that returns a new number by adding 1 to the last number in the input messages."""
    async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
        if len(messages) == 0:
            last_number = 0 # Start from 0 if no messages are given.
        else:
            assert isinstance(messages[-1], TextMessage)
            last_number = int(messages[-1].content) # Otherwise, start from the last number.
        return Response(chat_message=TextMessage(content=str(last_number + 1), source=self.name))

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

    @property
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
        return (TextMessage,)

class NestedCountingAgent(BaseChatAgent):
    """An agent that increments the last number in the input messages
    multiple times using a nested counting team."""
    def __init__(self, name: str, counting_team: RoundRobinGroupChat) -> None:
        super().__init__(name, description="An agent that counts numbers.")
        self._counting_team = counting_team

    async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
        # Run the inner team with the given messages and returns the last message produced by the team.
        result = await self._counting_team.run(task=messages, cancellation_token=cancellation_token)
        # To stream the inner messages, implement `on_messages_stream` and use that to implement `on_messages`.
        assert isinstance(result.messages[-1], TextMessage)
        return Response(chat_message=result.messages[-1], inner_messages=result.messages[len(messages):-1])

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        # Reset the inner team.
        await self._counting_team.reset()

    @property
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
        return (TextMessage,)

async def main() -> None:
    # Create a team of two counting agents as the inner team.
    counting_agent_1 = CountingAgent("counting_agent_1", description="An agent that counts numbers.")
    counting_agent_2 = CountingAgent("counting_agent_2", description="An agent that counts numbers.")
    counting_team = RoundRobinGroupChat([counting_agent_1, counting_agent_2], max_turns=5)
    # Create a nested counting agent that takes the inner team as a parameter.
    nested_counting_agent = NestedCountingAgent("nested_counting_agent", counting_team)
    # Run the nested counting agent with a message starting from 1.
    response = await nested_counting_agent.on_messages([TextMessage(content="1", source="user")], CancellationToken())
    assert response.inner_messages is not None
    for message in response.inner_messages:
        print(message)
    print(response.chat_message)

asyncio.run(main())

Sie sollten die folgende Ausgabe sehen

source='counting_agent_1' models_usage=None content='2' type='TextMessage'
source='counting_agent_2' models_usage=None content='3' type='TextMessage'
source='counting_agent_1' models_usage=None content='4' type='TextMessage'
source='counting_agent_2' models_usage=None content='5' type='TextMessage'
source='counting_agent_1' models_usage=None content='6' type='TextMessage'

Sie können sich SocietyOfMindAgent für eine komplexere Implementierung ansehen.

Sequenzieller Chat#

In v0.2 wird der sequenzielle Chat durch die Verwendung der Funktion initiate_chats unterstützt. Sie nimmt eine Liste von Dictionary-Konfigurationen für jeden Schritt der Sequenz entgegen. Siehe Sequenzieller Chat in v0.2 für weitere Details.

Basierend auf dem Feedback aus der Community ist die Funktion initiate_chats zu einschränkend und nicht flexibel genug, um die vielfältigen Szenarien zu unterstützen, die Benutzer implementieren möchten. Wir stellen oft fest, dass Benutzer Schwierigkeiten haben, die Funktion initiate_chats zum Laufen zu bringen, wenn sie die Schritte einfach mit grundlegendem Python-Code zusammenfügen können. Daher bieten wir in v0.4 keine integrierte Funktion für sequenzielle Chats in der AgentChat API an.

Stattdessen können Sie einen ereignisgesteuerten sequenziellen Workflow mit der Core API erstellen und die anderen Komponenten der AgentChat API verwenden, um jeden Schritt des Workflows zu implementieren. Siehe ein Beispiel für einen sequenziellen Workflow im Core API Tutorial.

Wir erkennen an, dass das Konzept des Workflows im Zentrum vieler Anwendungen steht, und wir werden in Zukunft mehr integrierte Unterstützung für Workflows bereitstellen.

GPTAssistantAgent#

In v0.2 ist GPTAssistantAgent eine spezielle Agentenklasse, die auf der OpenAI Assistant API basiert.

In v0.4 ist das Äquivalent die Klasse OpenAIAssistantAgent. Sie unterstützt denselben Funktionsumfang wie der GPTAssistantAgent in v0.2 mit zusätzlichen Funktionen wie anpassbaren Threads und Datei-Uploads. Siehe OpenAIAssistantAgent für weitere Details.

Langkontext-Verarbeitung#

In v0.2 kann ein langer Kontext, der das Kontextfenster des Modells überschreitet, durch die Verwendung der transforms-Funktion, die nach der Erstellung zu einem ConversableAgent hinzugefügt wird, verarbeitet werden.

Das Feedback unserer Community hat uns zu der Überzeugung geführt, dass diese Funktion essentiell ist und eine integrierte Komponente von AssistantAgent sein sollte und für jeden benutzerdefinierten Agenten verwendet werden kann.

In v0.4 führen wir die Basisklasse ChatCompletionContext ein, die die Nachrichten-Historie verwaltet und eine virtuelle Ansicht der Historie bereitstellt. Anwendungen können integrierte Implementierungen wie BufferedChatCompletionContext verwenden, um die an das Modell gesendete Nachrichten-Historie zu begrenzen, oder eigene Implementierungen bereitstellen, die unterschiedliche virtuelle Ansichten erstellen.

Um BufferedChatCompletionContext in einem AssistantAgent in einem Chatbot-Szenario zu verwenden.

import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_core.model_context import BufferedChatCompletionContext
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
        model_context=BufferedChatCompletionContext(buffer_size=10), # Model can only view the last 10 messages.
    )
    while True:
        user_input = input("User: ")
        if user_input == "exit":
            break
        response = await assistant.on_messages([TextMessage(content=user_input, source="user")], CancellationToken())
        print("Assistant:", response.chat_message.to_text())
    
    await model_client.close()

asyncio.run(main())

In diesem Beispiel kann der Chatbot nur die letzten 10 Nachrichten im Verlauf lesen.

Observability und Kontrolle#

In AgentChat v0.4 können Sie die Agenten beobachten, indem Sie die Methode on_messages_stream verwenden, die einen asynchronen Generator zurückgibt, um die inneren Gedanken und Aktionen des Agenten zu streamen. Für Teams können Sie die Methode run_stream verwenden, um die interne Konversation zwischen den Agenten im Team zu streamen. Ihre Anwendung kann diese Streams verwenden, um die Agenten und Teams in Echtzeit zu beobachten.

Sowohl die Methoden on_messages_stream als auch run_stream nehmen ein CancellationToken als Parameter entgegen, das verwendet werden kann, um den Ausgabe-Stream asynchron abzubrechen und den Agenten oder das Team zu stoppen. Für Teams können Sie auch Abbruchbedingungen verwenden, um das Team zu stoppen, wenn eine bestimmte Bedingung erfüllt ist. Weitere Details finden Sie im Tutorial zur Abbruchbedingung.

Im Gegensatz zu v0.2, das mit einem speziellen Protokollierungsmodul geliefert wurde, verwendet die v0.4 API einfach das logging-Modul von Python, um Ereignisse wie Modellclientaufrufe zu protokollieren. Weitere Details finden Sie unter Logging in der Core API-Dokumentation.

Code-Ausführer#

Die Code-Ausführer in v0.2 und v0.4 sind nahezu identisch, außer dass die v0.4-Ausführer die asynchrone API unterstützen. Sie können auch CancellationToken verwenden, um eine Codeausführung abzubrechen, wenn diese zu lange dauert. Weitere Details finden Sie im Tutorial zu Befehlszeilen-Code-Ausführern in der Core API-Dokumentation.

Wir haben auch ACADynamicSessionsCodeExecutor hinzugefügt, das Azure Container Apps (ACA) dynamische Sitzungen für die Codeausführung verwenden kann. Weitere Details finden Sie in der ACA Dynamic Sessions Code Executor-Dokumentation.