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.
Microsoft Agent Framework unterstützt das Erstellen von benutzerdefinierten Agents, indem sie von der AIAgent Klasse erben und die erforderlichen Methoden implementieren.
In diesem Artikel wird erläutert, wie Sie einen einfachen benutzerdefinierten Agent erstellen, der die Benutzereingaben in Großbuchstaben zurückgibt. In den meisten Fällen umfasst das Erstellen Ihres eigenen Agents komplexere Logik und Integration in einen KI-Dienst.
Erste Schritte
Fügen Sie dem Projekt die erforderlichen NuGet-Pakete hinzu.
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
Erstellen eines benutzerdefinierten Agents
Die Agentsitzung
Zum Erstellen eines benutzerdefinierten Agents benötigen Sie auch eine Sitzung, die zum Nachverfolgen des Status einer einzelnen Unterhaltung verwendet wird, einschließlich des Nachrichtenverlaufs und aller anderen Status, den der Agent verwalten muss.
Um den Einstieg zu erleichtern, können Sie von verschiedenen Basisklassen erben, die allgemeine Sitzungsspeichermechanismen implementieren.
-
InMemoryAgentSession– speichert den Chatverlauf im Arbeitsspeicher und kann in JSON serialisiert werden. -
ServiceIdAgentSession– speichert keinen Chatverlauf, ermöglicht ihnen jedoch die Zuordnung einer ID zur Sitzung, unter der der Chatverlauf extern gespeichert werden kann.
In diesem Beispiel verwenden Sie die InMemoryAgentSession Basisklasse für die benutzerdefinierte Sitzung.
internal sealed class CustomAgentSession : InMemoryAgentSession
{
internal CustomAgentSession() : base() { }
internal CustomAgentSession(JsonElement serializedSessionState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedSessionState, jsonSerializerOptions) { }
}
Die Agent-Klasse
Erstellen Sie als Nächstes die Agentklasse selbst, indem Sie von der AIAgent Klasse erben.
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
Erstellen von Sitzungen
Sitzungen werden immer über zwei Factorymethoden für die Agentklasse erstellt. Dadurch kann der Agent steuern, wie Sitzungen erstellt und deserialisiert werden. Agents können daher alle zusätzlichen Zustände oder Verhaltensweisen anfügen, die beim Erstellen der Sitzung erforderlich sind.
Es sind zwei Methoden erforderlich, die implementiert werden müssen:
public override Task<AgentSession> CreateSessionAsync(CancellationToken cancellationToken = default)
=> Task.FromResult<AgentSession>(new CustomAgentSession());
public override Task<AgentSession> DeserializeSessionAsync(JsonElement serializedSession, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
=> Task.FromResult<AgentSession>(new CustomAgentSession(serializedSession, jsonSerializerOptions));
Kern-Agent-Logik
Die Kernlogik des Agents besteht darin, eingabemeldungen zu übernehmen, den Text in Großbuchstaben zu konvertieren und als Antwortnachrichten zurückzugeben.
Fügen Sie die folgende Methode hinzu, um diese Logik zu enthalten.
Die Eingabemeldungen werden geklont, da verschiedene Aspekte der Eingabemeldungen geändert werden müssen, um gültige Antwortnachrichten zu sein. Zum Beispiel muss die Rolle in Assistant geändert werden.
private static IEnumerable<ChatMessage> CloneAndToUpperCase(IEnumerable<ChatMessage> messages, string agentName) => messages.Select(x =>
{
var messageClone = x.Clone();
messageClone.Role = ChatRole.Assistant;
messageClone.MessageId = Guid.NewGuid().ToString();
messageClone.AuthorName = agentName;
messageClone.Contents = x.Contents.Select(c => c is TextContent tc ? new TextContent(tc.Text.ToUpperInvariant())
{
AdditionalProperties = tc.AdditionalProperties,
Annotations = tc.Annotations,
RawRepresentation = tc.RawRepresentation
} : c).ToList();
return messageClone;
});
Agent-Ausführungsmethoden
Schließlich müssen Sie die beiden Kernmethoden implementieren, die verwendet werden, um den Agenten auszuführen: eine für Nicht-Streaming und eine für Streaming.
Für beide Methoden müssen Sie sicherstellen, dass eine Sitzung bereitgestellt wird, und falls nicht, eine neue Sitzung erstellen.
Nachrichten können abgerufen und an die ChatHistoryProvider Sitzung übergeben werden.
Wenn Sie dies nicht tun, kann der Benutzer keine Multi-Turn-Unterhaltung mit einem Agenten führen, und jede Ausführung wird als eine neue Interaktion betrachtet.
public override async Task<AgentResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
{
session ??= await this.CreateSessionAsync(cancellationToken);
// Get existing messages from the store
var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
// Notify the session of the input and output messages.
var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
{
ResponseMessages = responseMessages
};
await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);
return new AgentResponse
{
AgentId = this.Id,
ResponseId = Guid.NewGuid().ToString(),
Messages = responseMessages
};
}
public override async IAsyncEnumerable<AgentResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
session ??= await this.CreateSessionAsync(cancellationToken);
// Get existing messages from the store
var invokingContext = new ChatHistoryProvider.InvokingContext(messages);
var storeMessages = await typedSession.ChatHistoryProvider.InvokingAsync(invokingContext, cancellationToken);
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
// Notify the session of the input and output messages.
var invokedContext = new ChatHistoryProvider.InvokedContext(messages, storeMessages)
{
ResponseMessages = responseMessages
};
await typedSession.ChatHistoryProvider.InvokedAsync(invokedContext, cancellationToken);
foreach (var message in responseMessages)
{
yield return new AgentResponseUpdate
{
AgentId = this.Id,
AuthorName = this.DisplayName,
Role = ChatRole.Assistant,
Contents = message.Contents,
ResponseId = Guid.NewGuid().ToString(),
MessageId = Guid.NewGuid().ToString()
};
}
}
Tipp
Vollständige Runnable-Beispiele finden Sie in den .NET-Beispielen .
Den Agent verwenden
Wenn die AIAgent-Methoden alle ordnungsgemäß implementiert sind, wäre der Agent ein Standardagent und würde Standardbetriebsfunktionen eines Agents unterstützen.
Weitere Informationen zum Ausführen und Interagieren mit Agenten finden Sie in den Agenten-Einführungstutorials.
Microsoft Agent Framework unterstützt das Erstellen von benutzerdefinierten Agents, indem sie von der BaseAgent Klasse erben und die erforderlichen Methoden implementieren.
Dieses Dokument zeigt, wie Sie einen einfachen benutzerdefinierten Agent erstellen, der die Benutzereingabe mit einem Präfix wiedergibt. In den meisten Fällen umfasst das Erstellen Ihres eigenen Agents komplexere Logik und Integration in einen KI-Dienst.
Erste Schritte
Fügen Sie ihrem Projekt die erforderlichen Python-Pakete hinzu.
pip install agent-framework-core --pre
Erstellen eines benutzerdefinierten Agents
Das Agent-Protokoll
Das Framework stellt das SupportsAgentRun Protokoll bereit, das die Schnittstelle definiert, die alle Agents implementieren müssen. Benutzerdefinierte Agents können dieses Protokoll entweder direkt implementieren oder die BaseAgent Klasse zur Vereinfachung erweitern.
from typing import Any, Literal, overload
from collections.abc import Awaitable, Sequence
from agent_framework import (
AgentResponse,
AgentResponseUpdate,
AgentSession,
Message,
ResponseStream,
SupportsAgentRun,
)
class MyCustomAgent(SupportsAgentRun):
"""A custom agent that implements the SupportsAgentRun directly."""
@property
def id(self) -> str:
"""Returns the ID of the agent."""
...
@overload
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: Literal[False] = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse]: ...
@overload
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: Literal[True],
session: AgentSession | None = None,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
"""Execute the agent and return either an awaitable response or a ResponseStream."""
...
Tipp
Fügen Sie @overload Signaturen zu run() hinzu, sodass IDEs und statische Typenprüfer den Rückgabetyp basierend auf stream (Awaitable[AgentResponse] für stream=False und ResponseStream[AgentResponseUpdate, AgentResponse] für stream=True) ableiten.
Verwenden von BaseAgent
Der empfohlene Ansatz besteht darin, die BaseAgent Klasse zu erweitern, die allgemeine Funktionen bereitstellt und die Implementierung vereinfacht:
import asyncio
from collections.abc import AsyncIterable, Awaitable, Sequence
from typing import Any, Literal, overload
from agent_framework import (
AgentResponse,
AgentResponseUpdate,
AgentSession,
BaseAgent,
Content,
Message,
ResponseStream,
normalize_messages,
)
class EchoAgent(BaseAgent):
"""A simple custom agent that echoes user messages with a prefix."""
echo_prefix: str = "Echo: "
def __init__(
self,
*,
name: str | None = None,
description: str | None = None,
echo_prefix: str = "Echo: ",
**kwargs: Any,
) -> None:
super().__init__(
name=name,
description=description,
echo_prefix=echo_prefix,
**kwargs,
)
@overload
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: Literal[False] = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse]: ...
@overload
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: Literal[True],
session: AgentSession | None = None,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...
def run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
"""Execute the agent.
Args:
messages: The message(s) to process.
stream: If True, return a ResponseStream of updates.
session: The conversation session (optional).
Returns:
When stream=False: An awaitable AgentResponse.
When stream=True: A ResponseStream with AgentResponseUpdate items and final response support.
"""
if stream:
return ResponseStream(
self._run_stream(messages=messages, session=session, **kwargs),
finalizer=AgentResponse.from_updates,
)
return self._run(messages=messages, session=session, **kwargs)
async def _run(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
session: AgentSession | None = None,
**kwargs: Any,
) -> AgentResponse:
normalized_messages = normalize_messages(messages)
if not normalized_messages:
response_message = Message(
role="assistant",
contents=[Content.from_text("Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")],
)
else:
last_message = normalized_messages[-1]
echo_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"
response_message = Message(role="assistant", contents=[Content.from_text(echo_text)])
if session is not None:
stored = session.state.setdefault("memory", {}).setdefault("messages", [])
stored.extend(normalized_messages)
stored.append(response_message)
return AgentResponse(messages=[response_message])
async def _run_stream(
self,
messages: str | Message | Sequence[str | Message] | None = None,
*,
session: AgentSession | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentResponseUpdate]:
normalized_messages = normalize_messages(messages)
if not normalized_messages:
response_text = "Hello! I'm a custom echo agent. Send me a message and I'll echo it back."
else:
last_message = normalized_messages[-1]
response_text = f"{self.echo_prefix}{last_message.text}" if last_message.text else f"{self.echo_prefix}[Non-text message received]"
words = response_text.split()
for i, word in enumerate(words):
chunk_text = f" {word}" if i > 0 else word
yield AgentResponseUpdate(
contents=[Content.from_text(chunk_text)],
role="assistant",
)
await asyncio.sleep(0.1)
if session is not None:
complete_response = Message(role="assistant", contents=[Content.from_text(response_text)])
stored = session.state.setdefault("memory", {}).setdefault("messages", [])
stored.extend(normalized_messages)
stored.append(complete_response)
Den Agent verwenden
Wenn agent-Methoden alle ordnungsgemäß implementiert sind, unterstützt der Agent Standardvorgänge, einschließlich Streaming über ResponseStream:
stream = echo_agent.run("Stream this response", stream=True, session=echo_agent.create_session())
async for update in stream:
print(update.text or "", end="", flush=True)
final_response = await stream.get_final_response()
Weitere Informationen zum Ausführen und Interagieren mit Agenten finden Sie in den Agenten-Einführungstutorials.