AgentApplication 是使用 Agents SDK 建構代理的核心組件。
AgentApplication 是所有進入活動的入口,包括使用者訊息、對話生命週期事件、自適應卡片互動、OAuth 回撥。
代理人本質上是一個 AgentApplication。 你配置處理程序來設定代理的功能描述。 SDK 負責路由、狀態管理以及執行所需的基礎設施。
AgentApplication 的運作方式
每個代理都有生命週期,從某個通路(Microsoft Teams、機器人服務或自訂客戶端)將活動送達你的代理端點開始。
AgentApplication 位於該生命週期的核心:
Channel → Hosting layer → AgentApplication → Your handlers
使用 Agents SDK 建置的代理程式處理層如下:
- 主機層接收 HTTP 請求並進行認證。
- 它透過
AgentApplication 管線處理所接收的活動。
- 你的聯絡人是根據匹配的路線被叫來的。
你的代理程式會在處理常式執行前載入回合狀態。 之後,代理人會儲存回合狀態。
核心概念
活動
Agents SDK 裡的所有流程都是呈現為活動。 活動是一種結構化的訊息,代表發生過的事情。 活動有一種類型,例如 message、event、invoke、conversationUpdate 等等。 它攜帶的有效載荷與該類型相符。
AgentApplication 接收活動並將其導向適當的處理者。
Routes
路由將 選擇器 與 處理器配對。 選擇器會判斷路線是否符合當前活動。 當路線匹配時,handler 會執行你的邏輯。
設定代理時請註冊路由。 它們可以匹配:
- 包含特定文字或符合正則表達式的訊息
- 任何特定類型的活動
- 對話生命週期事件(成員新增,成員移除)
- 調適型卡片動作
- 自訂條件
當有活動到達時,系統會依序評估路線,直到找到匹配的路線。 預設情況下,只有一條路線運行。
轉彎狀態
AgentApplication 管理_turn狀態—結構化儲存區域分成不同範圍:
| 範圍類型 |
描述 |
| 交談 |
在對話中分享給所有使用者,回合間持續存在 |
| User |
涵蓋所有對話且範圍限定於單一使用者 |
| 溫度 |
只限當前回合——從未持續存在 |
系統會在處理器執行前自動載入狀態,並自動儲存。
回合上下文
當處理器執行時,會接收一個 回合上下文。 Turn context 是當前活動、介面卡連線及發送回應的工具的快照。 回合上下文是您與當前互動的接口。
Middleware
AgentApplication 支援 中介軟體管線。 中介軟體是一連串元件,會在處理程序執行前後的每回合。 中介軟體可以檢查、轉換或提前終止活動流程。 常見用途包括日誌記錄、認證檢查及請求正規化。
建立代理人
子職業 AgentApplication 並在建構器中註冊你的操作者。 主機框架會自動注入 AgentApplicationOptions。
public class MyAgent : AgentApplication
{
public MyAgent(AgentApplicationOptions options) : base(options)
{
OnConversationUpdate(ConversationUpdateEvents.MembersAdded, WelcomeAsync);
OnActivity(ActivityTypes.Message, OnMessageAsync, rank: RouteRank.Last);
}
private async Task WelcomeAsync(ITurnContext context, ITurnState state, CancellationToken ct)
{
foreach (var member in context.Activity.MembersAdded)
{
if (member.Id != context.Activity.Recipient.Id)
{
await context.SendActivityAsync("Hello! How can I help you?", cancellationToken: ct);
}
}
}
private async Task OnMessageAsync(ITurnContext context, ITurnState state, CancellationToken ct)
{
await context.SendActivityAsync($"You said: {context.Activity.Text}", cancellationToken: ct);
}
}
在 Program.cs 註冊你的代理。
WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddHttpClient();
builder.Services.AddSingleton<IStorage, MemoryStorage>();
builder.Services.AddAgent<MyAgent>();
builder.Services.AddAgentAspNetAuthentication(builder.Configuration);
WebApplication app = builder.Build();
app.UseAuthentication();
app.UseAuthorization();
app.MapAgentApplicationEndpoints(requireAuth: !app.Environment.IsDevelopment());
app.Run();
使用 fluent API 實例化 AgentApplication 並在實例上註冊處理器。 要啟動伺服器,請從 Express 主機套件呼叫 startServer 。
import { AgentApplication, MemoryStorage, TurnContext, TurnState } from '@microsoft/agents-hosting'
import { startServer } from '@microsoft/agents-hosting-express'
const app = new AgentApplication<TurnState>({ storage: new MemoryStorage() })
app.onConversationUpdate('membersAdded', async (context: TurnContext) => {
await context.sendActivity('Hello! How can I help you?')
})
app.onActivity('message', async (context: TurnContext, state: TurnState) => {
await context.sendActivity(`You said: ${context.activity.text}`)
})
startServer(app)
手動設定 Express,而不使用 startServer:
import express from 'express'
import { CloudAdapter, authorizeJWT, AuthConfiguration } from '@microsoft/agents-hosting'
const authConfig: AuthConfiguration = loadAuthConfigFromEnv()
const adapter = new CloudAdapter(authConfig)
const expressApp = express()
expressApp.use(express.json())
expressApp.use(authorizeJWT(authConfig))
expressApp.post('/api/messages', async (req, res) => {
await adapter.process(req, res, async (context) => {
await app.run(context)
})
})
使用型別參數來實例化AgentApplication,並使用裝飾器來註冊處理器。
from microsoft_agents.hosting.core import AgentApplication, TurnState, TurnContext, MemoryStorage
from microsoft_agents.hosting.aiohttp import CloudAdapter, start_agent_process
from aiohttp.web import Request, Response, Application, run_app
AGENT_APP = AgentApplication[TurnState](
storage=MemoryStorage(),
adapter=CloudAdapter()
)
@AGENT_APP.conversation_update("membersAdded")
async def on_members_added(context: TurnContext, state: TurnState):
await context.send_activity("Hello! How can I help you?")
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, state: TurnState):
await context.send_activity(f"You said: {context.activity.text}")
if __name__ == "__main__":
async def messages(req: Request) -> Response:
return await start_agent_process(req, AGENT_APP, AGENT_APP.adapter)
app = Application()
app.router.add_post("/api/messages", messages)
app["agent_app"] = AGENT_APP
app["adapter"] = AGENT_APP.adapter
run_app(app, host="localhost", port=3978)
註冊活動處理程式
處理訊息
依精確文字匹配訊息(不區分大小寫):
OnMessage("help", async (context, state, ct) =>
{
await context.SendActivityAsync("Here's what I can do...", cancellationToken: ct);
});
使用正則表達式匹配訊息:
OnMessage(new Regex(@"^order\s+\d+$", RegexOptions.IgnoreCase), async (context, state, ct) =>
{
await context.SendActivityAsync("Looking up your order...", cancellationToken: ct);
});
app.onMessage('help', async (context: TurnContext, state: TurnState) => {
await context.sendActivity("Here's what I can do...")
})
使用正則表達式匹配訊息:
app.onMessage(/^order\s+\d+$/i, async (context: TurnContext, state: TurnState) => {
await context.sendActivity('Looking up your order...')
})
@AGENT_APP.message("/help")
async def on_help(context: TurnContext, state: TurnState):
await context.send_activity("Here's what I can do...")
使用正則表達式匹配訊息:
import re
@AGENT_APP.message(re.compile(r"^order\s+\d+$", re.IGNORECASE))
async def on_order(context: TurnContext, state: TurnState):
await context.send_activity("Looking up your order...")
處理對話更新
為對話生命週期事件(例如成員加入或離開)註冊處理程式。
OnConversationUpdate(ConversationUpdateEvents.MembersAdded, async (context, state, ct) =>
{
foreach (var member in context.Activity.MembersAdded)
{
if (member.Id != context.Activity.Recipient.Id)
{
await context.SendActivityAsync("Welcome!", cancellationToken: ct);
}
}
});
OnConversationUpdate(ConversationUpdateEvents.MembersRemoved, async (context, state, ct) =>
{
// Called when participants leave the conversation
});
app.onConversationUpdate('membersAdded', async (context: TurnContext) => {
for (const member of context.activity.membersAdded ?? []) {
if (member.id !== context.activity.recipient.id) {
await context.sendActivity('Welcome!')
}
}
})
app.onConversationUpdate('membersRemoved', async (context: TurnContext) => {
// Called when participants leave the conversation
})
@AGENT_APP.conversation_update("membersAdded")
async def on_members_added(context: TurnContext, state: TurnState):
for member in context.activity.members_added or []:
if member.id != context.activity.recipient.id:
await context.send_activity("Welcome!")
@AGENT_APP.conversation_update("membersRemoved")
async def on_members_removed(context: TurnContext, state: TurnState):
pass # Called when participants leave the conversation
處理任何活動類型
根據活動類型字符串進行匹配,實現對路由的完全控制。
OnActivity(ActivityTypes.Message, async (context, state, ct) =>
{
// Handles all message activities
});
OnActivity(ActivityTypes.Event, async (context, state, ct) =>
{
// Handles event activities
});
用 ActivityTypes 常數代替硬編碼字串。
app.onActivity('message', async (context: TurnContext, state: TurnState) => {
// Handles all message activities
})
app.onActivity('event', async (context: TurnContext, state: TurnState) => {
// Handles event activities
})
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, state: TurnState):
pass # Handles all message activities
@AGENT_APP.activity("event")
async def on_event(context: TurnContext, state: TurnState):
pass # Handles event activities
控制路径評估順序
系統會在你註冊路由時就依固定的評估順序進行排序,而不是在執行期間。 排序分成兩個層級:
路由類型:系統依類型分組路由,且無論排名如何,都會先評估較高優先權的類型,優先於較低優先權的類型:
| 優先順序 |
路線類型 |
| 1(最高) |
代理調用路由 |
| 2 |
呼叫路由(自適應卡片動作、OAuth 回撥及其他時間敏感呼叫) |
| 3 |
能動路徑 |
| 4(最低) |
其他路線 |
排名:在每個路線類型群組內,系統依排名值排序路線。 較低的數值會先被評估。
註冊處理器時使用 RouteRank 常數設定階級。
| 常數 |
Value |
意義 |
RouteRank.First |
0 |
在同組中的所有其他路線之前進行評估 |
RouteRank.Unspecified |
32767 |
當未指定排名時,預設為 |
RouteRank.Last |
65535 |
在該組所有路線後進行評估 |
預設情況下,評估會在符合的第一個路徑停止。 使用 RouteRank.Last 作為通用回退,處理那些未被更具體路由匹配的所有情況。
// Specific handlers use the default rank
OnMessage("status", HandleStatusAsync);
OnMessage("help", HandleHelpAsync);
// Catch-all — handles anything not matched above
OnActivity(ActivityTypes.Message, HandleUnknownMessageAsync, rank: RouteRank.Last);
// Specific handlers use the default rank
app.onMessage('status', handleStatusAsync)
app.onMessage('help', handleHelpAsync)
// Catch-all — handles anything not matched above
app.onActivity('message', handleUnknownMessageAsync, RouteRank.Last)
@AGENT_APP.message("/status")
async def on_status(context: TurnContext, state: TurnState):
await context.send_activity("Status: OK")
@AGENT_APP.message("/help")
async def on_help(context: TurnContext, state: TurnState):
await context.send_activity("Here's what I can do...")
# Catch-all — handles anything not matched above (registered last)
@AGENT_APP.activity("message")
async def on_unknown(context: TurnContext, state: TurnState):
await context.send_activity("I didn't understand that. Type /help for options.")
啟用生命週期鉤子
寄存器邏輯在每次執行時操作,無論是在路由匹配前或後。 這些鉤子對於記錄記錄、交叉切割問題及錯誤處理非常有用。
OnBeforeTurn(async (context, state, ct) =>
{
logger.LogInformation("Turn started: {Type}", context.Activity.Type);
return true; // Return false to abort the turn
});
OnAfterTurn(async (context, state, ct) =>
{
logger.LogInformation("Turn completed");
return true; // Return false to skip state saving
});
OnTurnError(async (context, state, exception, ct) =>
{
logger.LogError(exception, "Turn error");
await context.SendActivityAsync("Something went wrong. Please try again.", cancellationToken: ct);
});
當 OnBeforeTurn 返回 false時,該回合即中止,且不執行任何路線。 當OnAfterTurn回傳false時,轉向狀態不會被保存。
app.beforeTurn(async (context: TurnContext, state: TurnState) => {
console.log(`Turn started: ${context.activity.type}`)
return true // Return false to abort the turn
})
app.afterTurn(async (context: TurnContext, state: TurnState) => {
console.log('Turn completed')
return true // Return false to skip state saving
})
app.onError(async (context: TurnContext, error: Error) => {
console.error('Turn error:', error)
await context.sendActivity('Something went wrong. Please try again.')
})
@AGENT_APP.before_turn
async def on_before_turn(context: TurnContext, state: TurnState) -> bool:
print(f"Turn started: {context.activity.type}")
return True # Return False to abort the turn
@AGENT_APP.after_turn
async def on_after_turn(context: TurnContext, state: TurnState) -> bool:
print("Turn completed")
return True # Return False to skip state saving
@AGENT_APP.turn_error
async def on_turn_error(context: TurnContext, state: TurnState, error: Exception):
print(f"Turn error: {error}")
await context.send_activity("Something went wrong. Please try again.")
使用轉向狀態
代理會在你的處理器執行前自動載入回合狀態,然後在執行後儲存。 傳遞給處理器的回合狀態物件會讓你存取不同的範圍,讓你能讀寫跨回合持續存在或在當前回合短暫的資料:
-
對話範圍:指對話中所有回合共享的資料
-
使用者範圍:針對每位使用者資料
-
暫時範圍:只需在當前回合內存在的資料
OnActivity(ActivityTypes.Message, async (context, state, ct) =>
{
// Conversation scope — persisted per conversation
var count = state.Conversation.GetValue<int>("messageCount", () => 0);
state.Conversation.SetValue("messageCount", count + 1);
// User scope — persisted per user
var name = state.User.GetValue<string>("displayName");
// Temp scope — current turn only
state.Temp.SetValue("parsedInput", context.Activity.Text?.Trim());
await context.SendActivityAsync($"Message #{count + 1}: {context.Activity.Text}", cancellationToken: ct);
});
app.onActivity('message', async (context: TurnContext, state: TurnState) => {
// Conversation scope — persisted per conversation
let count: number = state.getValue('conversation.messageCount') ?? 0
state.setValue('conversation.messageCount', count + 1)
// User scope — persisted per user
const name: string = state.getValue('user.displayName')
await context.sendActivity(`Message #${count + 1}: ${context.activity.text}`)
})
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, state: TurnState):
# Conversation scope — persisted per conversation
count = state.get_value("conversation.message_count") or 0
state.set_value("conversation.message_count", count + 1)
# User scope — persisted per user
name = state.get_value("user.display_name")
await context.send_activity(f"Message #{count + 1}: {context.activity.text}")
備註
使用 MemoryStorage 進行本地開發與測試。 對於生產部署,尤其是多個實例執行的部署,建議使用持久儲存服務提供者,例如 Azure Cosmos DB 或 Azure Blob 儲存體。 請參見 在代理中使用存儲供應商。
後續步驟