Poznámka:
Přístup k této stránce vyžaduje autorizaci. Můžete se zkusit přihlásit nebo změnit adresáře.
Přístup k této stránce vyžaduje autorizaci. Můžete zkusit změnit adresáře.
Microsoft Agent Framework podporuje vytváření vlastních agentů tím, že dědí z AIAgent třídy a implementuje požadované metody.
Tento článek ukazuje, jak vytvořit jednoduchého vlastního agenta, který zopakuje vstup uživatele velkými písmeny. Ve většině případů bude sestavování vlastního agenta zahrnovat složitější logiku a integraci se službou AI.
Začínáme
Přidejte do projektu požadované balíčky NuGet.
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
Vytvoření vlastního agenta
Vlákno agenta
Pokud chcete vytvořit vlastního agenta, potřebujete také vlákno, které slouží ke sledování stavu jedné konverzace, včetně historie zpráv a jakéhokoli jiného stavu, který musí agent udržovat.
Abyste mohli snadno začít, můžete dědit z různých základních tříd, které implementují běžné mechanismy pro ukládání vláken.
-
InMemoryAgentThread– uloží historii chatu do paměti a dá se serializovat do formátu JSON. -
ServiceIdAgentThread- neukládá žádnou historii chatu, ale umožňuje přidružit ID k vláknu, pod kterým se dá historie chatu ukládat externě.
V tomto příkladu InMemoryAgentThread použijete jako základní třídu pro vlastní vlákno.
internal sealed class CustomAgentThread : InMemoryAgentThread
{
internal CustomAgentThread() : base() { }
internal CustomAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedThreadState, jsonSerializerOptions) { }
}
Třída Agent
Dále vytvořte samotnou třídu agenta tím, že zdědíte třídu AIAgent.
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
Vytváření vláken
Vlákna se vždy vytvářejí prostřednictvím dvou továrních metod ve třídě agenta. To umožňuje agentu řídit, jak se vlákna vytvářejí a deserializují. Agenti proto mohou připojit jakýkoli další stav nebo chování potřebné k vláknu při vytváření.
Je potřeba implementovat dvě metody:
public override Task<AgentThread> GetNewThreadAsync(CancellationToken cancellationToken = default)
=> Task.FromResult<AgentThread>(new CustomAgentThread());
public override Task<AgentThread> DeserializeThreadAsync(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
=> Task.FromResult<AgentThread>(new CustomAgentThread(serializedThread, jsonSerializerOptions));
Logika centrálního agenta
Základní logikou agenta je vzít všechny vstupní zprávy, převést jejich text na velká a velká písmena a vrátit je jako zprávy odpovědi.
Přidejte následující metodu, která bude obsahovat tuto logiku.
Vstupní zprávy se naklonují, protože různé aspekty vstupních zpráv musí být změněny tak, aby byly platné zprávy odpovědí. Například role musí být změněna na 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;
});
Metody spuštění agenta
Nakonec je potřeba implementovat dvě základní metody, které se používají ke spuštění agenta: jednu pro jiné než streamování a jednu pro streamování.
U obou metod musíte zajistit, aby bylo vlákno k dispozici, a pokud ne, vytvořte nové vlákno.
Vlákno lze pak aktualizovat novými zprávami voláním NotifyThreadOfNewMessagesAsync.
Pokud to neuděláte, uživatel nebude moct mít s agentem vícenásobnou konverzaci a každé spuštění bude mít novou interakci.
public override async Task<AgentResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
{
thread ??= await this.GetNewThreadAsync(cancellationToken);
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), cancellationToken);
return new AgentResponse
{
AgentId = this.Id,
ResponseId = Guid.NewGuid().ToString(),
Messages = responseMessages
};
}
public override async IAsyncEnumerable<AgentResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
thread ??= await this.GetNewThreadAsync(cancellationToken);
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), 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()
};
}
}
Použití agenta
AIAgent Pokud jsou všechny metody implementované správně, agent by byl standardní AIAgent a podporuje standardní operace agenta.
Další informace o tom, jak spouštět agenty a pracovat s nimi, najdete v úvodních kurzech agenta.
Microsoft Agent Framework podporuje vytváření vlastních agentů tím, že dědí z BaseAgent třídy a implementuje požadované metody.
Tento dokument ukazuje, jak vytvořit zjednodušeného vlastního agenta, který vrací vstup uživatele s danou předponou. Ve většině případů bude sestavování vlastního agenta zahrnovat složitější logiku a integraci se službou AI.
Začínáme
Přidejte do projektu požadované balíčky Pythonu.
pip install agent-framework-core --pre
Vytvoření vlastního agenta
Protokol Agenta
Architektura poskytuje AgentProtocol protokol, který definuje rozhraní, které musí implementovat všichni agenti. Vlastní agenti mohou buď implementovat tento protokol přímo, nebo pro snadnější práci rozšířit třídu BaseAgent.
from agent_framework import AgentProtocol, AgentResponse, AgentResponseUpdate, AgentThread, ChatMessage
from collections.abc import AsyncIterable
from typing import Any
class MyCustomAgent(AgentProtocol):
"""A custom agent that implements the AgentProtocol directly."""
@property
def id(self) -> str:
"""Returns the ID of the agent."""
...
async def run(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentResponse:
"""Execute the agent and return a complete response."""
...
def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentResponseUpdate]:
"""Execute the agent and yield streaming response updates."""
...
Použití baseagentu
Doporučeným přístupem je rozšířit BaseAgent třídu, která poskytuje běžné funkce a zjednodušuje implementaci:
from agent_framework import (
BaseAgent,
AgentResponse,
AgentResponseUpdate,
AgentThread,
ChatMessage,
Role,
TextContent,
)
from collections.abc import AsyncIterable
from typing import Any
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:
"""Initialize the EchoAgent.
Args:
name: The name of the agent.
description: The description of the agent.
echo_prefix: The prefix to add to echoed messages.
**kwargs: Additional keyword arguments passed to BaseAgent.
"""
super().__init__(
name=name,
description=description,
echo_prefix=echo_prefix,
**kwargs,
)
async def run(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentResponse:
"""Execute the agent and return a complete response.
Args:
messages: The message(s) to process.
thread: The conversation thread (optional).
**kwargs: Additional keyword arguments.
Returns:
An AgentResponse containing the agent's reply.
"""
# Normalize input messages to a list
normalized_messages = self._normalize_messages(messages)
if not normalized_messages:
response_message = ChatMessage(
role=Role.ASSISTANT,
contents=[TextContent(text="Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")],
)
else:
# For simplicity, echo the last user message
last_message = normalized_messages[-1]
if last_message.text:
echo_text = f"{self.echo_prefix}{last_message.text}"
else:
echo_text = f"{self.echo_prefix}[Non-text message received]"
response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=echo_text)])
# Notify the thread of new messages if provided
if thread is not None:
await self._notify_thread_of_new_messages(thread, normalized_messages, response_message)
return AgentResponse(messages=[response_message])
async def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentResponseUpdate]:
"""Execute the agent and yield streaming response updates.
Args:
messages: The message(s) to process.
thread: The conversation thread (optional).
**kwargs: Additional keyword arguments.
Yields:
AgentResponseUpdate objects containing chunks of the response.
"""
# Normalize input messages to a list
normalized_messages = self._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:
# For simplicity, echo the last user message
last_message = normalized_messages[-1]
if last_message.text:
response_text = f"{self.echo_prefix}{last_message.text}"
else:
response_text = f"{self.echo_prefix}[Non-text message received]"
# Simulate streaming by yielding the response word by word
words = response_text.split()
for i, word in enumerate(words):
# Add space before word except for the first one
chunk_text = f" {word}" if i > 0 else word
yield AgentResponseUpdate(
contents=[TextContent(text=chunk_text)],
role=Role.ASSISTANT,
)
# Small delay to simulate streaming
await asyncio.sleep(0.1)
# Notify the thread of the complete response if provided
if thread is not None:
complete_response = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)])
await self._notify_thread_of_new_messages(thread, normalized_messages, complete_response)
Použití agenta
Pokud jsou všechny metody agenta implementované správně, agent by podporoval všechny standardní operace agenta.
Další informace o tom, jak spouštět agenty a pracovat s nimi, najdete v úvodních kurzech agenta.