Nuta
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować się zalogować lub zmienić katalog.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
Program Microsoft Agent Framework obsługuje tworzenie agentów niestandardowych przez dziedziczenie po AIAgent klasie i implementowanie wymaganych metod.
W tym dokumencie pokazano, jak utworzyć prostego agenta niestandardowego, który powtarza dane wejściowe użytkownika zapisane wielkimi literami. W większości przypadków tworzenie własnego agenta wymaga bardziej złożonej logiki i integracji z usługą sztucznej inteligencji.
Wprowadzenie
Dodaj wymagane pakiety NuGet do projektu.
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
Tworzenie agenta niestandardowego
Wątek agenta
Aby utworzyć agenta niestandardowego, potrzebny jest również wątek, który służy do śledzenia stanu jednej konwersacji, w tym historii komunikatów i dowolnego innego stanu, który musi obsługiwać agent.
Aby ułatwić rozpoczęcie pracy, można dziedziczyć z różnych klas bazowych, które implementują powszechne mechanizmy przechowywania wątków.
-
InMemoryAgentThread— przechowuje historię czatów w pamięci i może być serializowany do formatu JSON. -
ServiceIdAgentThread— nie przechowuje żadnej historii czatów, ale umożliwia skojarzenie identyfikatora z wątkiem, w którym historia czatu może być przechowywana zewnętrznie.
W tym przykładzie użyjemy InMemoryAgentThread jako klasy bazowej dla naszego wątku niestandardowego.
internal sealed class CustomAgentThread : InMemoryAgentThread
{
internal CustomAgentThread() : base() { }
internal CustomAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedThreadState, jsonSerializerOptions) { }
}
Klasa agenta
Następnie chcemy utworzyć samą klasę agenta, dziedzicząc z AIAgent klasy.
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
Konstruowanie wątków
Wątki są zawsze tworzone za pomocą dwóch metod fabrycznych klasy agenta. Dzięki temu agent może kontrolować sposób tworzenia i deserializacji wątków. W związku z tym agenci mogą dołączać wszelkie dodatkowe stany lub zachowania potrzebne do wątku przy jego tworzeniu.
Do zaimplementowania są dwie metody:
public override AgentThread GetNewThread() => new CustomAgentThread();
public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null)
=> new CustomAgentThread(serializedThread, jsonSerializerOptions);
Podstawowa logika agenta
Podstawową logiką agenta jest przyjęcie dowolnych wiadomości wejściowych, przekonwertowanie ich tekstu na wielkie litery i zwrócenie ich jako odpowiedzi.
Chcemy dodać następującą metodę, aby zawierała tę logikę.
Klonujemy komunikaty wejściowe, ponieważ różne aspekty komunikatów wejściowych muszą zostać zmodyfikowane w celu wprowadzania prawidłowych komunikatów odpowiedzi. Przykładowo, trzeba zmienić rolę 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 działania agenta
Na koniec musimy zaimplementować dwie podstawowe metody, które są używane do uruchamiania agenta. Jeden dla nieprzesyłania strumieniowego i jeden dla przesyłania strumieniowego.
W przypadku obu metod musimy upewnić się, że podano wątek, a jeśli nie, utworzymy nowy wątek.
Następnie można zaktualizować wątek przy użyciu nowych komunikatów, wywołując polecenie NotifyThreadOfNewMessagesAsync.
Jeśli tego nie zrobimy, użytkownik nie będzie mógł przeprowadzić wielokrotnej rozmowy z agentem, a każda sesja będzie nową interakcją.
public override async Task<AgentRunResponse> RunAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
{
thread ??= this.GetNewThread();
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), cancellationToken);
return new AgentRunResponse
{
AgentId = this.Id,
ResponseId = Guid.NewGuid().ToString(),
Messages = responseMessages
};
}
public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IEnumerable<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
thread ??= this.GetNewThread();
List<ChatMessage> responseMessages = CloneAndToUpperCase(messages, this.DisplayName).ToList();
await NotifyThreadOfNewMessagesAsync(thread, messages.Concat(responseMessages), cancellationToken);
foreach (var message in responseMessages)
{
yield return new AgentRunResponseUpdate
{
AgentId = this.Id,
AuthorName = this.DisplayName,
Role = ChatRole.Assistant,
Contents = message.Contents,
ResponseId = Guid.NewGuid().ToString(),
MessageId = Guid.NewGuid().ToString()
};
}
}
Korzystanie z agenta
AIAgent Jeśli wszystkie metody są prawidłowo zaimplementowane, agent będzie standardowym AIAgent i obsługuje standardowe operacje agenta.
Aby uzyskać więcej informacji na temat uruchamiania agentów i interakcji z nimi, zobacz samouczki wprowadzające Agenta.
Program Microsoft Agent Framework obsługuje tworzenie agentów niestandardowych przez dziedziczenie po BaseAgent klasie i implementowanie wymaganych metod.
W tym dokumencie pokazano, jak utworzyć prostego agenta niestandardowego, który odtwarza dane wejściowe użytkownika, dodając prefiks. W większości przypadków tworzenie własnego agenta wymaga bardziej złożonej logiki i integracji z usługą sztucznej inteligencji.
Wprowadzenie
Dodaj wymagane pakiety języka Python do projektu.
pip install agent-framework-core --pre
Tworzenie agenta niestandardowego
Protokół agenta
Platforma udostępnia AgentProtocol protokół definiujący interfejs, który każdy agent musi implementować. Agenci niestandardowi mogą zaimplementować ten protokół bezpośrednio lub rozszerzyć klasę BaseAgent dla wygody.
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, 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,
) -> AgentRunResponse:
"""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[AgentRunResponseUpdate]:
"""Execute the agent and yield streaming response updates."""
...
Korzystanie z agenta BaseAgent
Zalecaną metodą jest rozszerzenie BaseAgent klasy, która zapewnia typowe funkcje i upraszcza implementację:
from agent_framework import (
BaseAgent,
AgentRunResponse,
AgentRunResponseUpdate,
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,
) -> AgentRunResponse:
"""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 AgentRunResponse 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 AgentRunResponse(messages=[response_message])
async def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentRunResponseUpdate]:
"""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:
AgentRunResponseUpdate 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 AgentRunResponseUpdate(
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)
Korzystanie z agenta
Jeśli wszystkie metody agenta są prawidłowo implementowane, agent będzie obsługiwał wszystkie standardowe operacje agenta.
Aby uzyskać więcej informacji na temat uruchamiania agentów i interakcji z nimi, zobacz samouczki wprowadzające Agenta.