autogen_ext.agents.azure#

class AzureAIAgent(name: str, description: str, project_client: AIProjectClient, deployment_name: str, instructions: str, tools: Iterable[Literal['file_search', 'code_interpreter', 'bing_grounding', 'azure_ai_search', 'azure_function'] | BingGroundingToolDefinition | CodeInterpreterToolDefinition | AzureAISearchToolDefinition | FileSearchToolDefinition | AzureFunctionToolDefinition | Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, agent_id: str | None = None, thread_id: str | None = None, metadata: Dict[str, str] | None = None, response_format: AgentsResponseFormat | None = None, temperature: float | None = None, tool_resources: ToolResources | None = None, top_p: float | None = None)[Quelle]#

Bases: BaseChatAgent

Azure AI Assistant Agent für AutoGen.

Installation

pip install "autogen-ext[azure]"  # For Azure AI Foundry Agent Service

Dieser Agent nutzt die Azure AI Assistant API, um KI-Assistenten mit Funktionen wie

Codeinterpretation und -ausführung
Grundlage mit Bing-Suche
Datei-Handling und -suche
Benutzerdefinierte Funktionsaufrufe
Mehrstufige Konversationen

zu erstellen. Der Agent integriert sich in das Messaging-System von AutoGen und bietet eine nahtlose Möglichkeit, Azure AI-Funktionen innerhalb des AutoGen-Frameworks zu nutzen. Er unterstützt Tools wie Code-Interpreter, Dateisuche und verschiedene Grounding-Mechanismen.

Der Agentenname muss ein gültiger Python-Bezeichner sein

Er muss mit einem Buchstaben (A-Z, a-z) oder einem Unterstrich (_) beginnen.
Er darf nur Buchstaben, Ziffern (0-9) oder Unterstriche enthalten.
Er darf kein Python-Schlüsselwort sein.
Er darf keine Leerzeichen oder Sonderzeichen enthalten.
Er darf nicht mit einer Ziffer beginnen.

Hier erfahren Sie, wie Sie einen neuen gesicherten Agenten mit benutzerverwalteter Identität erstellen: https://learn.microsoft.com/en-us/azure/ai-services/agents/how-to/virtual-networks

Examples

Verwenden Sie AzureAIAgent, um einen Agenten zu erstellen, der mit Bing grundiert ist

import asyncio
import os

from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential
from azure.ai.agents.models import BingGroundingTool
import dotenv


async def bing_example():
    async with DefaultAzureCredential() as credential:
        async with AIProjectClient(  # type: ignore
            credential=credential, endpoint=os.getenv("AZURE_PROJECT_ENDPOINT", "")
        ) as project_client:
            conn = await project_client.connections.get(name=os.getenv("BING_CONNECTION_NAME", ""))

            bing_tool = BingGroundingTool(conn.id)
            agent_with_bing_grounding = AzureAIAgent(
                name="bing_agent",
                description="An AI assistant with Bing grounding",
                project_client=project_client,
                deployment_name="gpt-4o",
                instructions="You are a helpful assistant.",
                tools=bing_tool.definitions,
                metadata={"source": "AzureAIAgent"},
            )

            # For the bing grounding tool to return the citations, the message must contain an instruction for the model to do return them.
            # For example: "Please provide citations for the answers"

            result = await agent_with_bing_grounding.on_messages(
                messages=[
                    TextMessage(
                        content="What is Microsoft\'s annual leave policy? Provide citations for your answers.",
                        source="user",
                    )
                ],
                cancellation_token=CancellationToken(),
                message_limit=5,
            )
            print(result)


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(bing_example())

Verwenden Sie AzureAIAgent, um einen Agenten mit Dateisuchfunktion zu erstellen

import asyncio
import os
import tempfile
import urllib.request

import dotenv
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential


async def file_search_example():
    # Download README.md from GitHub
    readme_url = "https://raw.githubusercontent.com/microsoft/autogen/refs/heads/main/README.md"
    temp_file = None

    try:
        # Create a temporary file to store the downloaded README
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".md")
        urllib.request.urlretrieve(readme_url, temp_file.name)
        print(f"Downloaded README.md to {temp_file.name}")

        async with DefaultAzureCredential() as credential:
            async with AIProjectClient(  # type: ignore
                credential=credential, endpoint=os.getenv("AZURE_PROJECT_ENDPOINT", "")
            ) as project_client:
                agent_with_file_search = AzureAIAgent(
                    name="file_search_agent",
                    description="An AI assistant with file search capabilities",
                    project_client=project_client,
                    deployment_name="gpt-4.1-mini",
                    instructions="You are a helpful assistant.",
                    tools=["file_search"],
                    metadata={"source": "AzureAIAgent"},
                )

                ct: CancellationToken = CancellationToken()
                # Use the downloaded README file for file search
                await agent_with_file_search.on_upload_for_file_search(
                    file_paths=[temp_file.name],
                    vector_store_name="file_upload_index",
                    vector_store_metadata={"source": "AzureAIAgent"},
                    cancellation_token=ct,
                    vector_store_polling_interval=60,
                )
                result = await agent_with_file_search.on_messages(
                    messages=[
                        TextMessage(
                            content="Hello, what is AutoGen and what capabilities does it have?", source="user"
                        )
                    ],
                    cancellation_token=ct,
                    message_limit=5,
                )
                print(result)
    finally:
        # Clean up the temporary file
        if temp_file and os.path.exists(temp_file.name):
            os.unlink(temp_file.name)
            print(f"Removed temporary file {temp_file.name}")


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(file_search_example())

Verwenden Sie AzureAIAgent, um einen Agenten mit Code-Interpreter-Funktion zu erstellen

import asyncio
import os

import dotenv
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential


async def code_interpreter_example():
    async with DefaultAzureCredential() as credential:
        async with AIProjectClient(  # type: ignore
            credential=credential, endpoint=os.getenv("AZURE_PROJECT_ENDPOINT", "")
        ) as project_client:
            agent_with_code_interpreter = AzureAIAgent(
                name="code_interpreter_agent",
                description="An AI assistant with code interpreter capabilities",
                project_client=project_client,
                deployment_name="gpt-4.1-mini",
                instructions="You are a helpful assistant.",
                tools=["code_interpreter"],
                metadata={"source": "AzureAIAgent"},
            )

            await agent_with_code_interpreter.on_upload_for_code_interpreter(
                file_paths="/workspaces/autogen/python/packages/autogen-core/docs/src/user-guide/core-user-guide/cookbook/data/nifty_500_quarterly_results.csv",
                cancellation_token=CancellationToken(),
                polling_interval=5,
            )

            result = await agent_with_code_interpreter.on_messages(
                messages=[
                    TextMessage(
                        content="Aggregate the number of stocks per industry and give me a markdown table as a result?",
                        source="user",
                    )
                ],
                cancellation_token=CancellationToken(),
            )

            print(result)


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(code_interpreter_example())

property produced_message_types: Sequence[type[Annotated[TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: Die Nachrichtentypen, die der Assistentenagent erzeugt.

property thread_id: str#

property description: str#: Die Beschreibung des Agenten. Diese wird vom Team verwendet, um Entscheidungen darüber zu treffen, welche Agenten verwendet werden sollen. Die Beschreibung sollte die Fähigkeiten des Agenten und die Art und Weise, wie mit ihm interagiert werden kann, beschreiben.

property agent_id: str#

property deployment_name: str#

property instructions: str#

property tools: List[ToolDefinition]#

Ruft die Liste der dem Agenten zur Verfügung stehenden Tools ab.

Gibt zurück:: List[ToolDefinition] – Die Liste der Tool-Definitionen.

async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken | None = None, message_limit: int = 1) → Response[Quelle]#

Verarbeitet eingehende Nachrichten und gibt eine Antwort vom Azure AI-Agenten zurück.

Diese Methode ist der primäre Einstiegspunkt für die Interaktion mit dem Agenten. Sie delegiert an on_messages_stream und gibt die endgültige Antwort zurück.

Parameter:

messages (Sequence[BaseChatMessage]) – Die zu verarbeitenden Nachrichten
cancellation_token (CancellationToken) – Token zur Abbruchbehandlung
message_limit (int, optional) – Maximale Anzahl von Nachrichten, die aus dem Thread abgerufen werden sollen

Gibt zurück:

Response – Die Antwort des Agenten, einschließlich der Chatnachricht und aller inneren Ereignisse

Löst aus:

AssertionError – Wenn der Stream kein Endergebnis zurückgibt

Verarbeitet eingehende Nachrichten und gibt Streaming-Antworten vom Azure AI-Agenten zurück.

Diese Methode behandelt den vollständigen Interaktionsfluss mit dem Azure AI-Agenten: 1. Verarbeitung von Eingabenachrichten 2. Erstellung und Überwachung eines Laufs 3. Handhabung von Tool-Aufrufen und deren Ergebnissen 4. Abrufen und Zurückgeben der endgültigen Antwort des Agenten.

Die Methode gibt während der Verarbeitung Ereignisse aus (z. B. Tool-Aufrufe) und gibt schließlich die vollständige Antwort mit der Nachricht des Agenten zurück.

Parameter:

messages (Sequence[BaseChatMessage]) – Die zu verarbeitenden Nachrichten
cancellation_token (CancellationToken) – Token zur Abbruchbehandlung
message_limit (int, optional) – Maximale Anzahl von Nachrichten, die aus dem Thread abgerufen werden sollen
polling_interval (float, optional) – Zeit zum Schlafen zwischen Abfragen des Laufstatus

Gibt:

AgentEvent | ChatMessage | Response – Ereignisse während der Verarbeitung und die endgültige Antwort

Löst aus:

ValueError – Wenn der Lauf fehlschlägt oder keine Nachricht vom Assistenten empfangen wird

async handle_text_message(content: str, cancellation_token: CancellationToken | None = None) → None[Quelle]#

Eine Textnachricht verarbeiten, indem sie zum Gesprächsverlauf hinzugefügt wird.

Parameter:

content (str) – Der Textinhalt der Nachricht
cancellation_token (CancellationToken) – Token zur Abbruchbehandlung

Gibt zurück:

Keine

async on_reset(cancellation_token: CancellationToken) → None[Quelle]#

Setzt die Konversation des Agenten zurück, indem ein neuer Thread erstellt wird.

Diese Methode ermöglicht das Zurücksetzen einer Konversation, ohne die Agentendefinition oder -fähigkeiten zu verlieren. Sie erstellt einen neuen Thread für neue Konversationen.

Hinweis: Derzeit unterstützt die Azure AI Agent API keine Nachrichtenlöschung, daher wird stattdessen ein neuer Thread erstellt.

Parameter:: cancellation_token (CancellationToken) – Token zur Abbruchbehandlung

async save_state() → Mapping[str, Any][Quelle]#

Speichert den aktuellen Zustand des Agenten für die zukünftige Wiederherstellung.

Diese Methode serialisiert den Zustand des Agenten, einschließlich IDs für den Agenten, Thread, Nachrichten und zugehörige Ressourcen wie Vektorspeicher und hochgeladene Dateien.

Gibt zurück:: Mapping[str, Any] – Ein Wörterbuch mit den serialisierten Zustandsdaten

async load_state(state: Mapping[str, Any]) → None[Quelle]#

Lädt einen zuvor gespeicherten Zustand in diesen Agenten.

Diese Methode deserialisiert und stellt einen zuvor gespeicherten Agentenzustand wieder her und konfiguriert den Agenten so, dass er eine frühere Konversation oder Sitzung fortsetzt.

Parameter:: state (Mapping[str, Any]) – Das zuvor gespeicherte Zustands-Dictionary

async on_upload_for_code_interpreter(file_paths: str | Iterable[str], cancellation_token: CancellationToken | None = None, polling_interval: float = 0.5) → None[Quelle]#

Lädt Dateien hoch, die mit dem Code-Interpreter-Tool verwendet werden sollen.

Diese Methode lädt Dateien für das Code-Interpreter-Tool des Agenten hoch und aktualisiert die Tool-Ressourcen des Threads, um diese Dateien einzuschließen.

Parameter:

file_paths (str | Iterable[str]) – Pfad(e) zur/zu den hochzuladenden Datei(en)
cancellation_token (Optional[CancellationToken]) – Token zur Abbruchbehandlung
polling_interval (float) – Zeit zum Schlafen zwischen Abfragen des Dateistatus

Löst aus:

ValueError – Wenn der Datei-Upload fehlschlägt oder der Agent nicht über die Code-Interpreter-Fähigkeit verfügt

async on_upload_for_file_search(file_paths: str | Iterable[str], cancellation_token: CancellationToken, vector_store_name: str | None = None, data_sources: List[VectorStoreDataSource] | None = None, expires_after: VectorStoreExpirationPolicy | None = None, chunking_strategy: VectorStoreChunkingStrategyRequest | None = None, vector_store_metadata: Dict[str, str] | None = None, vector_store_polling_interval: float = 1) → None[source]#

Lädt Dateien für das Datei-Suchwerkzeug hoch.

Diese Methode behandelt das Hochladen von Dateien für die Datei-Suchfunktion, erstellt bei Bedarf einen Vektor-Speicher und aktualisiert die Konfiguration des Agenten zur Verwendung des Vektor-Speichers.

Parameter:

file_paths (str | Iterable[str]) – Pfad(e) zur/zu den hochzuladenden Datei(en)
cancellation_token (CancellationToken) – Token zur Abbruchbehandlung
vector_store_name (Optional[str]) – Name, der dem Vektor-Speicher zugewiesen werden soll, falls ein neuer erstellt wird
data_sources (Optional[List[VectorStoreDataSource]]) – Zusätzliche Datenquellen für den Vektor-Speicher
expires_after (Optional[VectorStoreExpirationPolicy]) – Ablaufrichtlinie für den Inhalt des Vektor-Speichers
chunking_strategy (Optional[VectorStoreChunkingStrategyRequest]) – Strategie zum Aufteilen des Datei-Inhalts
vector_store_metadata (Optional[Dict[str, str]]) – Zusätzliche Metadaten für den Vektor-Speicher
vector_store_polling_interval (float) – Zeit, die zwischen dem Abfragen des Vektor-Speicher-Status geschlafen wird

Löst aus:

ValueError – Wenn die Datei-Suche für diesen Agenten nicht aktiviert ist oder der Datei-Upload fehlschlägt

async close() → None[source]#: Schließt den Azure AI Agenten und gibt alle Ressourcen frei.