Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
In diesem Lernprogramm wird veranschaulicht, wie KI-Agents mithilfe von Agent Framework in Workflows integriert werden. Sie lernen, Workflows zu erstellen, die die Leistungsfähigkeit spezialisierter KI-Agents für die Erstellung, Überprüfung und andere gemeinsame Aufgaben nutzen.
Was Sie erstellen werden
Sie erstellen einen Workflow, der:
- Verwendet den Azure Foundry Agent Service zum Erstellen intelligenter Agents
- Implementiert einen französischen Übersetzungsagenten, der Eingaben in Französisch übersetzt.
- Implementiert einen spanischen Übersetzungsagenten, der Französisch in Spanisch übersetzt
- Implementiert einen englischsprachigen Übersetzungs-Agent, der Spanisch zurück ins Englische übersetzt
- Verbindet Agents in einer sequenziellen Workflowpipeline
- Echtzeitaktualisierungen streamen, während Agenten Anfragen verarbeiten
- Veranschaulicht die ordnungsgemäße Ressourcenbereinigung für Azure Foundry-Agents.
Behandelte Konzepte
Voraussetzungen
- .NET 8.0 SDK oder höher
- Ein Azure Foundry-Projektendpunkt und -Modell konfiguriert
- Installierte und authentifizierte Azure CLI (für die Azure-Anmeldeinformationsauthentifizierung)
- Eine neue Konsolenanwendung
Schritt 1: Installieren von NuGet-Paketen
Installieren Sie zunächst die erforderlichen Pakete für Ihr .NET-Projekt:
dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Workflows --prerelease
Schritt 2: Einrichten des Azure Foundry-Clients
Konfigurieren Sie den Azure Foundry-Client mit Umgebungsvariablen und Authentifizierung:
using Azure.AI.Projects;
using Azure.AI.Projects.Agents;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
public static class Program
{
private static async Task Main()
{
// Set up the Azure AI Project client
var endpoint = Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("FOUNDRY_MODEL") ?? "gpt-4o-mini";
var aiProjectClient = new AIProjectClient(new Uri(endpoint), new AzureCliCredential());
Schritt 3: Erstellen der Agent Factory-Methode
Implementieren Sie eine Hilfsmethode zum Erstellen von Azure Foundry-Agents mit bestimmten Anweisungen:
/// <summary>
/// Creates a translation agent for the specified target language.
/// </summary>
/// <param name="targetLanguage">The target language for translation</param>
/// <param name="aiProjectClient">The AIProjectClient to create the agent</param>
/// <param name="model">The model to use for the agent</param>
/// <returns>A FoundryAgent configured for the specified language</returns>
private static async Task<FoundryAgent> GetTranslationAgentAsync(
string targetLanguage,
AIProjectClient aiProjectClient,
string model)
{
string agentName = $"{targetLanguage}Translator";
var version = await aiProjectClient.AgentAdministrationClient.CreateAgentVersionAsync(
agentName,
new ProjectsAgentVersionCreationOptions(
new DeclarativeAgentDefinition(model)
{
Instructions = $"You are a translation assistant that translates the provided text to {targetLanguage}."
}));
return aiProjectClient.AsAIAgent(version);
}
}
Schritt 4: Erstellen von spezialisierten Azure Foundry Agents
Erstellen Sie drei Übersetzungs-Agents mithilfe der Hilfsmethode:
// Create agents
AIAgent frenchAgent = await GetTranslationAgentAsync("French", aiProjectClient, deploymentName);
AIAgent spanishAgent = await GetTranslationAgentAsync("Spanish", aiProjectClient, deploymentName);
AIAgent englishAgent = await GetTranslationAgentAsync("English", aiProjectClient, deploymentName);
Schritt 5: Erstellen des Workflows
Verbinden Sie die Agents in einem sequenziellen Workflow mithilfe von WorkflowBuilder:
// Build the workflow by adding executors and connecting them
var workflow = new WorkflowBuilder(frenchAgent)
.AddEdge(frenchAgent, spanishAgent)
.AddEdge(spanishAgent, englishAgent)
.Build();
Schritt 6: Ausführen mit Streaming
Führen Sie den Workflow mit Streaming aus, um Echtzeitupdates von allen Agents zu beobachten:
// Execute the workflow
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));
// Must send the turn token to trigger the agents.
// The agents are wrapped as executors. When they receive messages,
// they will cache the messages and only start processing when they receive a TurnToken.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
if (evt is AgentResponseUpdateEvent executorComplete)
{
Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
}
}
Schritt 7: Ressourcenbereinigung
Bereinigen Sie die Azure Foundry-Agents nach der Verwendung ordnungsgemäß:
// Cleanup the agents created for the sample.
await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(frenchAgent.Name);
await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(spanishAgent.Name);
await aiProjectClient.AgentAdministrationClient.DeleteAgentAsync(englishAgent.Name);
}
Funktionsweise
-
Einrichten des Azure Foundry-Clients: Verwendet
AIProjectClientmit Azure CLI-Anmeldeinformationen für die Authentifizierung - Agent Creation: Erstellt versionsgebundene Agents in Azure Foundry mit spezifischen Anweisungen für die Übersetzung
- Sequenzielle Verarbeitung: Der französische Agent übersetzt zuerst Eingaben, dann spanischer Agent und dann englischer Agent
-
Token-Umschaltmuster: Agents speichern Nachrichten zwischen und verarbeiten sie nur, wenn sie ein
TurnTokenerhalten. -
Streamingupdates:
AgentResponseUpdateEventStellt Echtzeittokenupdates bereit, wenn Agents Antworten generieren - Ressourcenverwaltung: Ordnungsgemäße Bereinigung von Azure Foundry-Agents mithilfe der Verwaltungs-API
Wichtige Konzepte
- Azure Foundry Agent Service: Cloudbasierte KI-Agenten mit erweiterten Denkfähigkeiten
- AIProjectClient: Client zum Erstellen und Verwalten von Agents in Azure Foundry
-
WorkflowEvent: Ausgabeereignisse (
type="output") enthalten Agentausgabedaten (AgentResponseUpdatefür Streaming,AgentResponsefür Nicht-Streaming) - TurnToken: Signal, das die Agentverarbeitung nach dem Zwischenspeichern von Nachrichten auslöst
- Sequenzieller Workflow: Agents, die in einer Pipeline verbunden sind, wobei die Ausgabe von einem Agenten zum nächsten fließt
Vollständige Implementierung
Die vollständige Arbeitsimplementierung dieses Azure Foundry Agents-Workflows finden Sie im FoundryAgent-Program.cs Beispiel im Agent Framework-Repository.
Was Sie erstellen werden
Sie erstellen einen Workflow, der:
- Verwendet
FoundryChatClient, um intelligente Agenten zu erstellen - Implementiert einen Writer-Agent, der Inhalte basierend auf Aufforderungen erstellt.
- Implementiert einen Prüfer-Agent, der Feedback zu den Inhalten bereitstellt
- Verbindet Agents in einer sequenziellen Workflowpipeline
- Echtzeitaktualisierungen streamen, während Agenten Anfragen verarbeiten
Behandelte Konzepte
Voraussetzungen
- Python 3.11 oder höher
- Agent Framework installiert:
pip install agent-framework - Azure OpenAI-Antworten, die mit ordnungsgemäßen Umgebungsvariablen konfiguriert sind
- Azure CLI-Authentifizierung:
az login
Schritt 1: Importieren erforderlicher Abhängigkeiten
Importieren Sie zunächst die erforderlichen Komponenten für Workflows und Azure OpenAI Responses Agents:
import asyncio
import os
from agent_framework import AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
Schritt 2: Erstellen des Azure OpenAI-Antwortclients
Erstellen Sie einen freigegebenen Client, den Sie zum Erstellen mehrerer Agents verwenden können:
async def main() -> None:
client = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
)
Schritt 3: Erstellen von spezialisierten Agents
Erstellen Sie zwei spezialisierte Agents für die Inhaltserstellung und -überprüfung:
# Create a Writer agent that generates content
writer_agent = client.as_agent(
name="Writer",
instructions=(
"You are an excellent content writer. You create new content and edit contents based on the feedback."
),
)
# Create a Reviewer agent that provides feedback
reviewer_agent = client.as_agent(
name="Reviewer",
instructions=(
"You are an excellent content reviewer. "
"Provide actionable feedback to the writer about the provided content. "
"Provide the feedback in the most concise manner possible."
),
)
Schritt 4: Erstellen des Workflows
Verbinden Sie die Agents in einem sequenziellen Workflow mithilfe des Generators:
# Build the workflow with agents as executors
workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()
Schritt 5: Ausführen mit Streaming
Führen Sie den Workflow mit Streaming aus, um Echtzeitupdates von beiden Agents zu beobachten:
last_author: str | None = None
events = workflow.run("Create a slogan for a new electric SUV that is affordable and fun to drive.", stream=True)
async for event in events:
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
update = event.data
author = update.author_name
if author != last_author:
if last_author is not None:
print()
print(f"{author}: {update.text}", end="", flush=True)
last_author = author
else:
print(update.text, end="", flush=True)
Schritt 6: Ausführen der Hauptfunktion
Umschließen Sie alles in der Hauptfunktion mit der richtigen asynchronen Ausführung:
if __name__ == "__main__":
asyncio.run(main())
Funktionsweise
-
Clientsetup: Verwendet eine
FoundryChatClientmit Azure CLI-Anmeldeinformationen für die Authentifizierung. - Agent Creation: Erstellt Writer- und Reviewer-Agents aus derselben Clientkonfiguration.
- Sequenzielle Verarbeitung: Writer-Agent generiert zuerst Inhalt und übergibt ihn dann an den Prüfer-Agent.
-
Streamingupdates: Ausgabeereignisse (
type="output") mitAgentResponseUpdate-Daten stellen Echtzeit-Tokenaktualisierungen bereit, wenn Agents Antworten generieren.
Wichtige Konzepte
- FoundryChatClient: Freigegebener Client zum Erstellen von Workflow-Agents mit konsistenter Konfiguration.
-
WorkflowEvent: Ausgabeereignisse (
type="output") enthalten Agentenausgabedaten (AgentResponseUpdatefür Streaming,AgentResponsefür Nicht-Streaming). - Sequenzieller Workflow: Agenten, die in einer Pipeline verbunden sind, in der die Ausgabe von einem Agenten zum nächsten fließt.
Vollständige Implementierung
Die vollständige Arbeitsimplementierung finden Sie unter azure_ai_agents_streaming.py im Agent Framework-Repository.
Was Sie erstellen werden
Sie erstellen einen Workflow, der:
- Verwendet Microsoft Foundry Model-Deployment-Agents als Workflow-Runner
- Implementiert einen französischen Übersetzungsagenten
- Implementiert einen spanischen Übersetzungsagenten
- Implementiert einen englischsprachigen Übersetzungs-Agent
- Verbindet Agents in einer sequenziellen Workflowpipeline
- Echtzeitaktualisierungen streamen, während Agenten Anfragen verarbeiten
Behandelte Konzepte
Voraussetzungen
- Go 1.26 oder höher
- Microsoft Foundry-Projektendpunkt und Modellbereitstellung konfiguriert
- Authentifizierung über die Azure CLI oder eine andere Quelle für Azure-Anmeldeinformationen
Initialisieren Sie in einem neuen Verzeichnis ein Go-Modul, und installieren Sie Agent Framework:
go mod init example.com/agent-workflow
go get github.com/microsoft/agent-framework-go
Legen Sie den Projektendpunkt und den Namen Des bereitgestellten Modells fest:
$env:FOUNDRY_PROJECT_ENDPOINT = "https://<resource>.services.ai.azure.com/api/projects/<project>"
$env:FOUNDRY_MODEL = "<deployment-name>"
Schritt 1: Einrichten der Foundry-Konfiguration
In den folgenden Abschnitten wird die runExample Funktion in der vollständigen Implementierung aufgebaut. Beginnen Sie mit den Importen und der Konfiguration:
import (
"context"
"errors"
"fmt"
"log"
"os"
"strings"
"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
"github.com/microsoft/agent-framework-go/agent"
"github.com/microsoft/agent-framework-go/message"
"github.com/microsoft/agent-framework-go/provider/foundryprovider"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/agentworkflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
if endpoint == "" {
return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
}
model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
if model == "" {
return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
}
token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
return err
}
Warnung
azidentity.NewDefaultAzureCredential ist praktisch für die Entwicklung, erfordert aber sorgfältige Überlegungen in der Produktion. Berücksichtigen Sie in Produktionsumgebungen die Verwendung einer bestimmten Anmeldeinformation, wie azidentity.NewManagedIdentityCredential, um Latenzprobleme, unbeabsichtigtes Ausprobieren von Anmeldeinformationen und potenzielle Sicherheitsrisiken durch Fallbackmechanismen zu vermeiden.
Schritt 2: Erstellen der Agent Factory-Funktion
Erstellen Sie Agents mit bestimmten Übersetzungsanweisungen:
newTranslationAgent := func(language string) *agent.Agent {
return foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
Instructions: fmt.Sprintf(
"Translate the user's text to %s. Return only the translation.",
language,
),
Config: agent.Config{Name: language + "Agent"},
})
}
Schritt 3: Erstellen Sie spezielle Foundry-Agenten
frenchAgent := newTranslationAgent("French")
spanishAgent := newTranslationAgent("Spanish")
englishAgent := newTranslationAgent("English")
Schritt 4: Erstellen des Workflows
Agents können als Workflowausführer verwendet werden, wodurch KI-gestützte Workflowschritte aktiviert werden.
Binden Sie jeden Agent als Workflow-Executor, und verbinden Sie dann die Executors mit Edges:
cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
french := agentworkflow.New(frenchAgent, cfg)
spanish := agentworkflow.New(spanishAgent, cfg)
english := agentworkflow.New(englishAgent, cfg)
wf, err := workflow.NewBuilder(french).
AddEdge(french, spanish).
AddEdge(spanish, english).
WithIntermediateOutputFrom(french, spanish).
WithOutputFrom(english).
Build()
if err != nil {
return err
}
ForwardIncomingMessages: new(false) übergibt die Antwort jedes Agents nachgelagert, ohne auch die Eingabe dieses Agents weiterzuleiten. Jeder gehostete Executor verwaltet eine eigene Agentsitzung.
Registrieren Sie Französisch und Spanisch als Zwischenausgabequellen, damit Anrufer alle drei Agents beobachten können. Englisch bleibt die Quelle für die Terminalausgabe. Die Ausgabezuweisung steuert, was der Aufrufer sieht, nicht welcher Nachrichten-Flow entlang der Ränder fließt.
Schritt 5: Ausführen mit Streaming
Führen Sie den Workflow aus, und aktivieren Sie Updateereignisse mit einem workflow.TurnToken. Behandeln Sie Workflowfehlerereignisse sowie Fehler aus dem Ereignisiterator:
run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
if err != nil {
return err
}
defer func() {
err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()
emitEvents := true
if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
return err
} else if !accepted {
return fmt.Errorf("workflow doesn't accept turn tokens")
}
lastExecutorID := ""
for evt, streamErr := range run.WatchStream(ctx) {
if streamErr != nil {
return streamErr
}
switch event := evt.(type) {
case workflow.OutputEvent:
update, ok := event.Output.(*agent.ResponseUpdate)
if !ok || update.String() == "" {
continue
}
if event.ExecutorID != lastExecutorID {
if lastExecutorID != "" {
fmt.Println()
}
fmt.Printf("%s: ", event.ExecutorID)
lastExecutorID = event.ExecutorID
}
fmt.Print(update.String())
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
if lastExecutorID != "" {
fmt.Println()
}
Die Ausgabe enthält Übersetzungen in Französisch, Spanisch und Englisch, gruppiert nach Executor. Das vollständige Beispiel kehrt von runExample zurück, bevor ein schwerwiegender Fehler protokolliert wird, sodass die verzögerte Bereinigung des Workflows weiterhin ausgeführt wird.
Im Gegensatz zum C#-Beispiel erstellt dieses Go-Beispiel keine gespeicherten Agent-Versionen, die in Foundry gelöscht werden müssen.
Funktionsweise
- Clientsetup: Verwendet einen Foundry-Projektendpunkt, Modellbereitstellung und Azure Anmeldeinformationen.
- Agent Creation: Erstellt spezialisierte Agents mit sprachspezifischen Anweisungen.
-
Agent-Hosting: Verwendet
agentworkflow.New, um jeden Agent als Workflowausführer zu binden. - Sequenzielle Verarbeitung: Der französische Executor wird zuerst ausgeführt, dann der spanische und anschließend der englische.
-
Token-Umschaltmuster: Gehostete Agents puffern Nachrichten und werden ausgeführt, wenn sie ein
workflow.TurnTokenerhalten. - Streamingupdates: Registrierungen für Zwischen- und Endausgaben stellen Updates von allen drei Agents bereit.
-
Fehlerbehandlung: Der Host überprüft
ErrorEventundExecutorFailedEventund schließt den Workflow vor dem Melden eines Fehlers.
Wichtige Konzepte
-
Foundry-Agent: Ein
agent.Agent-Agent, der von einer Foundry-Modellbereitstellung unterstützt wird. - agentworkflow. Neu: Passt einen Agent für die Verwendung als Workflowausführer an.
- Workflow. TurnToken: Signal, das gehostete Agents zum Verarbeiten von gepufferten Nachrichten auslöst.
- Workflow OutputEvent: Enthält Agentantwortaktualisierungen und endgültige Workflowausgaben.
- Sequenzieller Workflow: Agenten, die in einer Pipeline verbunden sind, in der die Ausgabe von einem Agenten zum nächsten fließt.
Vollständige Implementierung
Speichern Sie dieses Beispiel als main.go in Ihrem Go-Modul:
package main
import (
"context"
"errors"
"fmt"
"log"
"os"
"strings"
"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
"github.com/microsoft/agent-framework-go/agent"
"github.com/microsoft/agent-framework-go/message"
"github.com/microsoft/agent-framework-go/provider/foundryprovider"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/agentworkflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func main() {
if err := runExample(context.Background()); err != nil {
log.Fatal(err)
}
}
func runExample(ctx context.Context) (err error) {
endpoint := strings.TrimSpace(os.Getenv("FOUNDRY_PROJECT_ENDPOINT"))
if endpoint == "" {
return fmt.Errorf("set FOUNDRY_PROJECT_ENDPOINT")
}
model := strings.TrimSpace(os.Getenv("FOUNDRY_MODEL"))
if model == "" {
return fmt.Errorf("set FOUNDRY_MODEL to your model deployment name")
}
token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
return err
}
newTranslationAgent := func(language string) *agent.Agent {
return foundryprovider.NewAgent(
endpoint,
token,
foundryprovider.ModelDeployment(model),
foundryprovider.AgentConfig{
Instructions: fmt.Sprintf(
"Translate the user's text to %s. Return only the translation.",
language,
),
Config: agent.Config{Name: language + "Agent"},
},
)
}
cfg := agentworkflow.Config{ForwardIncomingMessages: new(false)}
french := agentworkflow.New(newTranslationAgent("French"), cfg)
spanish := agentworkflow.New(newTranslationAgent("Spanish"), cfg)
english := agentworkflow.New(newTranslationAgent("English"), cfg)
wf, err := workflow.NewBuilder(french).
AddEdge(french, spanish).
AddEdge(spanish, english).
WithIntermediateOutputFrom(french, spanish).
WithOutputFrom(english).
Build()
if err != nil {
return err
}
run, err := inproc.Default.RunStreaming(ctx, wf, message.NewText("Hello World"))
if err != nil {
return err
}
defer func() {
err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()
emitEvents := true
if accepted, err := run.TrySendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
return err
} else if !accepted {
return fmt.Errorf("workflow doesn't accept turn tokens")
}
lastExecutorID := ""
for evt, streamErr := range run.WatchStream(ctx) {
if streamErr != nil {
return streamErr
}
switch event := evt.(type) {
case workflow.OutputEvent:
update, ok := event.Output.(*agent.ResponseUpdate)
if !ok || update.String() == "" {
continue
}
if event.ExecutorID != lastExecutorID {
if lastExecutorID != "" {
fmt.Println()
}
fmt.Printf("%s: ", event.ExecutorID)
lastExecutorID = event.ExecutorID
}
fmt.Print(update.String())
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
if lastExecutorID != "" {
fmt.Println()
}
return nil
}
Führen Sie das Beispiel aus:
go run .
Warnung
azidentity.NewDefaultAzureCredential ist praktisch für die Entwicklung, erfordert aber sorgfältige Überlegungen in der Produktion. Berücksichtigen Sie in Produktionsumgebungen die Verwendung einer bestimmten Anmeldeinformation, wie azidentity.NewManagedIdentityCredential, um Latenzprobleme, unbeabsichtigtes Ausprobieren von Anmeldeinformationen und potenzielle Sicherheitsrisiken durch Fallbackmechanismen zu vermeiden.
Tip
Sehen Sie sich das Beispiel für Agents in Workflows für eine andere runnable Übersetzungspipeline an.