Jegyzet
Az oldalhoz való hozzáférés engedélyezést igényel. Próbálhatod be jelentkezni vagy könyvtárat váltani.
Az oldalhoz való hozzáférés engedélyezést igényel. Megpróbálhatod a könyvtár váltását.
A Microsoft Agent Framework az osztályból történő örökléssel és a szükséges metódusok megvalósításával támogatja az AIAgent egyéni ügynökök készítését.
Ez a cikk bemutatja, hogyan lehet egyszerű egyéni ügynököt létrehozni, amely a felhasználói bemenetet nagybetűkkel visszaadja. A legtöbb esetben a saját ügynök létrehozása összetettebb logikát és AI-szolgáltatással való integrációt igényel.
Első lépések
Adja hozzá a szükséges NuGet-csomagokat a projekthez.
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
Egyéni ügynök létrehozása
Az ügynökszál
Egyéni ügynök létrehozásához szükség van egy szálra is, amely egy beszélgetés állapotának nyomon követésére szolgál, beleértve az üzenetelőzményeket és az ügynök által karbantartandó egyéb állapotokat.
Az első lépések megkönnyítése érdekében örökölhet különböző alaposztályokat, amelyek közös száltárolási mechanizmusokat implementálnak.
-
InMemoryAgentThread- tárolja a csevegési előzményeket a memóriában, és szerializálható A JSON-ra. -
ServiceIdAgentThread- nem tárol csevegési előzményeket, de lehetővé teszi, hogy azonosítót társítson a szálhoz, amely alatt a csevegési előzmények külsőleg tárolhatók.
Ebben a példában az InMemoryAgentThread egyéni szál alaposztályaként fogjuk használni.
internal sealed class CustomAgentThread : InMemoryAgentThread
{
internal CustomAgentThread() : base() { }
internal CustomAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedThreadState, jsonSerializerOptions) { }
}
Az Ügynök osztály
Ezután hozza létre magát az ügynökosztályt az AIAgent osztálytól való örökléssel.
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
Szálak létrehozása
A szálak mindig két gyári módszerrel jönnek létre az ügynökosztályon. Ez lehetővé teszi az ügynök számára a szálak létrehozásának és deszerializálásának szabályozását. Az ügynökök ezért bármilyen további állapotot vagy viselkedést csatolhatnak a szálhoz annak létrehozásakor.
Két módszert kell implementálni:
public override AgentThread GetNewThread() => new CustomAgentThread();
public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null)
=> new CustomAgentThread(serializedThread, jsonSerializerOptions);
Alapvető ügynöklogika
Az ügynök alapvető logikája az, hogy bármilyen bemeneti üzenetet fogad, nagybetűssé konvertálja a szöveget, és válaszüzenetekként adja vissza őket.
A következő metódus hozzáadásával foglalja magában ezt a logikát.
A bemeneti üzenetek klónozva vannak, mivel a bemeneti üzenetek különböző aspektusait módosítani kell, hogy érvényes válaszüzenetek legyenek. A szerepkört a következőre kell módosítani: 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;
});
Ügynökvégrehajtási módszerek
Végül implementálnia kell az ügynök futtatásához használt két alapvető módszert: egyet a nem streameléshez, egyet pedig a streameléshez.
Mindkét módszer esetében gondoskodnia kell arról, hogy egy szál legyen megadva, és ha nem, hozzon létre egy új szálat.
A szál az új üzenetekkel a NotifyThreadOfNewMessagesAsync hívással frissíthető.
Ha ezt nem teszi meg, a felhasználó nem tud többfordulós beszélgetést folytatni az ügynökkel, és minden futtatás új interakció lesz.
public override async Task<AgentResponse> 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 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 ??= 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 AgentResponseUpdate
{
AgentId = this.Id,
AuthorName = this.DisplayName,
Role = ChatRole.Assistant,
Contents = message.Contents,
ResponseId = Guid.NewGuid().ToString(),
MessageId = Guid.NewGuid().ToString()
};
}
}
Az ügynök használata
Ha a AIAgent metódusok mindegyike megfelelően van implementálva, az ügynök egy szabvány AIAgent lesz, és támogatja a szabványos ügynöki műveleteket.
Az ügynökök futtatásával és használatával kapcsolatos további információkért tekintse meg az ügynök első lépéseket ismertető oktatóanyagait.
A Microsoft Agent Framework az BaseAgent osztályból örökölve és a szükséges metódusok implementálásával támogatja az egyéni ügynökök készítését.
Ez a dokumentum bemutatja, hogyan hozhat létre egy egyszerű egyéni ügynököt, amely egy előtaggal adja vissza a felhasználói bemenetet. A legtöbb esetben a saját ügynök létrehozása összetettebb logikát és AI-szolgáltatással való integrációt igényel.
Első lépések
Adja hozzá a szükséges Python-csomagokat a projekthez.
pip install agent-framework-core --pre
Egyéni ügynök létrehozása
Az ügynökprotokoll
A keretrendszer biztosítja azt a AgentProtocol protokollt, amely meghatározza azt a felületet, amelyet minden ügynöknek végre kell hajtania. Az egyéni ügynökök vagy közvetlenül implementálhatják ezt a protokollt, vagy a BaseAgent osztályt kiterjeszthetik a kényelmesebb használat érdekében.
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."""
...
A BaseAgent használata
Az ajánlott módszer az BaseAgent osztály kiterjesztése, amely általános funkciókat biztosít, és leegyszerűsíti a megvalósítást:
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)
Az ügynök használata
Ha az ügynökmetóciók mindegyike megfelelően van implementálva, az ügynök minden szabványos ügynökműveletet támogat.
Az ügynökök futtatásával és használatával kapcsolatos további információkért tekintse meg az ügynök első lépéseket ismertető oktatóanyagait.