Notitie
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen u aan te melden of de directory te wijzigen.
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen de mappen te wijzigen.
Microsoft Agent Framework ondersteunt het bouwen van aangepaste agents door te erven van de AIAgent klasse en door de vereiste methoden te implementeren.
In dit artikel wordt beschreven hoe u een eenvoudige aangepaste agent bouwt die gebruikersinvoer in hoofdletters weergeeft. In de meeste gevallen omvat het bouwen van uw eigen agent complexere logica en integratie met een AI-service.
Aan de slag komen
Voeg de vereiste NuGet-pakketten toe aan uw project.
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
Een aangepaste agent maken
De agentsessie
Als u een aangepaste agent wilt maken, hebt u ook een sessie nodig, die wordt gebruikt om de status van één gesprek bij te houden, inclusief de berichtgeschiedenis en elke andere status die de agent moet onderhouden.
Om het eenvoudig te maken om aan de slag te gaan, kunt u overnemen van verschillende basisklassen die algemene mechanismen voor sessieopslag implementeren.
-
InMemoryAgentSession- slaat de chatgeschiedenis op in het geheugen en kan worden geserialiseerd naar JSON. -
ServiceIdAgentSession- slaat geen chatgeschiedenis op, maar hiermee kunt u een id koppelen aan de sessie, waaronder de chatgeschiedenis extern kan worden opgeslagen.
In dit voorbeeld gebruikt u de InMemoryAgentSession basisklasse voor de aangepaste sessie.
internal sealed class CustomAgentSession : InMemoryAgentSession
{
internal CustomAgentSession() : base() { }
internal CustomAgentSession(JsonElement serializedSessionState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedSessionState, jsonSerializerOptions) { }
}
De agentklasse
Maak vervolgens de agentklasse zelf door deze over te nemen van de AIAgent klasse.
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
Sessies samenstellen
Sessies worden altijd gemaakt via twee factorymethoden in de agentklasse. Hierdoor kan de agent bepalen hoe sessies worden gemaakt en gedeserialiseerd. Agenten kunnen daarom eventuele aanvullende status of gedragingen aan de sessie koppelen wanneer deze bij het aanmaken nodig zijn.
Er moeten twee methoden worden geïmplementeerd:
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));
Kernagentlogica
De kernlogica van de agent is het nemen van invoerberichten, het converteren van de tekst naar hoofdletters en het retourneren als antwoordberichten.
Voeg de volgende methode toe om deze logica te bevatten.
De invoerberichten worden gekloond, omdat verschillende aspecten van de invoerberichten moeten worden gewijzigd om geldige antwoordberichten te zijn. De rol moet bijvoorbeeld worden gewijzigd in Assistant.
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;
});
Methoden voor agentuitvoering
Ten slotte moet u de twee kernmethoden implementeren die worden gebruikt om de agent uit te voeren: één voor niet-streaming en één voor streaming.
Voor beide methoden moet u ervoor zorgen dat er een sessie wordt opgegeven, en als dat niet het geval is, maakt u een nieuwe sessie aan.
Berichten kunnen worden opgehaald en doorgegeven aan de ChatHistoryProvider sessie.
Als u dit niet doet, kan de gebruiker geen meerfasen gesprek met de agent voeren en zal elke sessie een nieuwe interactie zijn.
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()
};
}
}
Aanbeveling
Zie de .NET-voorbeelden voor volledige runnable voorbeelden.
De agent gebruiken
Als de AIAgent methoden allemaal correct zijn geïmplementeerd, zou de agent een standaardagent AIAgent zijn en standaardagentbewerkingen ondersteunen.
Voor meer informatie over hoe je agents uitvoert en ermee werkt, raadpleeg de Aan de slag met agent-tutorials.
Microsoft Agent Framework ondersteunt het bouwen van aangepaste agents door te erven van de BaseAgent klasse en door de vereiste methoden te implementeren.
Dit document laat zien hoe u een eenvoudige aangepaste agent bouwt die gebruikersinvoer weergeeft met een voorvoegsel. In de meeste gevallen omvat het bouwen van uw eigen agent complexere logica en integratie met een AI-service.
Aan de slag komen
Voeg de vereiste Python-pakketten toe aan uw project.
pip install agent-framework-core --pre
Een aangepaste agent maken
Het agentprotocol
Het framework biedt het SupportsAgentRun protocol waarmee de interface wordt gedefinieerd die alle agents moeten implementeren. Aangepaste agents kunnen dit protocol rechtstreeks implementeren of de BaseAgent klasse uitbreiden voor het gemak.
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."""
...
Aanbeveling
Voeg @overload handtekeningen toe aan run() zodat IDE's en statische typecontroleurs het retourtype afleiden op basis van stream (Awaitable[AgentResponse] voor stream=False en ResponseStream[AgentResponseUpdate, AgentResponse] voor stream=True).
BaseAgent gebruiken
De aanbevolen aanpak is het uitbreiden van de BaseAgent klasse, die algemene functionaliteit biedt en de implementatie vereenvoudigt:
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)
De agent gebruiken
Als agentmethoden allemaal correct zijn geïmplementeerd, ondersteunt de agent standaardbewerkingen, inclusief streaming via 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()
Voor meer informatie over hoe je agents uitvoert en ermee werkt, raadpleeg de Aan de slag met agent-tutorials.