Microsoft Agent Framework 支援透過從類別 AIAgent 繼承並實作必要的方法來建立自訂代理。
本文說明如何建立一個簡單的自訂代理程式，並以大寫字母形式回應使用者的輸入。 在大多數情況下，建立自己的代理程式將涉及更複雜的邏輯以及與 AI 服務的整合。
使用者入門
將必要的 NuGet 套件新增至您的專案。
dotnet add package Microsoft.Agents.AI.Abstractions --prerelease
建立自訂代理人
代理線程
若要建立自訂代理程式，您還需要一個執行緒，用於追蹤單一交談的狀態，包括訊息歷史記錄，以及代理程式需要維護的任何其他狀態。
為了方便入門，您可以繼承自實作通用執行緒儲存機制的各種基類。
-
InMemoryAgentThread- 將聊天記錄儲存在記憶體中，並可序列化為 JSON。 -
ServiceIdAgentThread- 不儲存聊天紀錄，但允許你將 ID 與該串關聯，並可將聊天記錄儲存在外部。
在這個例子中，你會使用 InMemoryAgentThread 作為自訂執行緒的基底類別。
internal sealed class CustomAgentThread : InMemoryAgentThread
{
internal CustomAgentThread() : base() { }
internal CustomAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
: base(serializedThreadState, jsonSerializerOptions) { }
}
Agent 類別
接著，透過從 AIAgent 類別繼承，建立代理程式類別本身。
internal sealed class UpperCaseParrotAgent : AIAgent
{
}
建構執行緒
執行緒一律會透過代理程式類別上的兩個 Factory 方法建立。 這允許代理程式控制執行緒的建立和還原序列化方式。 因此，代理程式可以在建構時將所需的任何其他狀態或行為附加到執行緒。
需要實作兩種方法：
public override AgentThread GetNewThread() => new CustomAgentThread();
public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null)
=> new CustomAgentThread(serializedThread, jsonSerializerOptions);
核心代理程式邏輯
代理的核心邏輯是接收任何輸入訊息，將其文字轉換為大寫，並以回應訊息的形式回傳。
新增以下方法來包含此邏輯。
輸入訊息會被複製，因為輸入訊息的各個面向必須修改才能成為有效的回應訊息。 例如，角色必須改為 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;
});
代理程式執行方法
最後，您需要實作兩個核心方法來執行代理程式：一種是非串流模式，另一種是串流模式。
兩種方法都需要確保有執行緒，如果沒有，就建立新的執行緒。
可以透過呼叫 NotifyThreadOfNewMessagesAsync 來更新執行緒的新消息。
如果不這麼做，使用者將無法與代理進行多回合對話，每次執行都會是全新的互動。
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()
};
}
}
使用代理程式
如果方法 AIAgent 都正確實作，代理程式將是標準 AIAgent 並支援標準代理程式作業。
想了解更多如何執行及與代理互動的資訊，請參閱代理 入門教學。
Microsoft Agent Framework 支援透過從類別 BaseAgent 繼承並實作所需的方法來建立自訂代理程式。
本文檔介紹如何構建一個簡單的自定義代理，以使用字首回顯使用者輸入。 在大多數情況下，建立自己的代理程式將涉及更複雜的邏輯以及與 AI 服務的整合。
使用者入門
將必要的 Python 套件新增至您的專案。
pip install agent-framework-core --pre
建立自訂代理人
代理程式通訊協定
該框架提供了 AgentProtocol 定義所有代理必須實現的接口的協議。 自訂代理程式可以直接實作此通訊協定，也可以擴充 BaseAgent 類別以方便使用。
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."""
...
使用 BaseAgent
建議的方法是擴充 BaseAgent 類別，它提供了通用功能並簡化了實作：
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)
使用代理程式
如果代理程式方法都正確地實作，則代理程式將支援所有標準代理程式作業。
想了解更多如何執行及與代理互動的資訊，請參閱代理 入門教學。