AI Agent開發入門：Semantic Kernel構建智能郵件助手

點擊 “AladdinEdu，同學們用得起的【H卡】算力平臺”，H卡級別算力，80G大顯存，按量計費，靈活彈性，頂級配置，學生更享專屬優惠。

引言：AI Agent——下一代人機交互范式

在人工智能技術飛速發展的今天，基于大語言模型(LLM)的AI Agent正在成為人機交互的新范式。與傳統的單一功能AI應用不同，AI Agent能夠理解復雜指令、規劃執行路徑、使用工具解決問題，真正扮演"智能助手"的角色。微軟推出的Semantic Kernel作為一款輕量級SDK，為開發者提供了構建此類AI Agent的強大框架。

本文將指導你使用Semantic Kernel構建一個智能郵件助手，重點講解任務規劃鏈設計和工具函數集成兩大核心概念。通過這篇教程，你不僅將掌握一個實用項目的開發過程，更將理解AI Agent開發的核心范式，為構建更復雜的智能應用打下堅實基礎。

第一部分：Semantic Kernel核心概念解析

1.1 什么是Semantic Kernel？

Semantic Kernel(SK)是一個輕量級的SDK，它將大型語言模型(如GPT-4)、傳統編程語言(如C#、Python)和外部數據源與服務連接起來。通過SK，開發者可以創建能夠理解語義、執行復雜任務的AI應用。

SK的核心優勢包括：

插件架構：將代碼作為插件提供給LLM調用
規劃器：將復雜目標分解為可執行步驟
記憶管理：短期與長期記憶結合
模板化：使用模板動態生成提示詞

1.2 AI Agent開發的核心組件

構建一個完整的AI Agent需要以下核心組件：

任務理解：解析用戶意圖和需求
規劃能力：將復雜任務分解為步驟序列
工具使用：調用API、函數等外部資源
結果合成：將執行結果整合為連貫響應

1.3 環境準備與安裝

開始之前，確保你的開發環境已就緒：

# 創建并進入項目目錄
mkdir semantic-kernel-mail-assistant
cd semantic-kernel-mail-assistant# 創建虛擬環境（可選但推薦）
python -m venv .venv
source .venv/bin/activate  # Linux/MacOS
# 或
.venv\Scripts\activate      # Windows# 安裝Semantic Kernel
pip install semantic-kernel

此外，你需要準備一個LLM API密鑰（如OpenAI或Azure OpenAI服務）：

import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion# 初始化內核
kernel = sk.Kernel()# 配置LLM服務
api_key = "your_openai_api_key"
model_id = "gpt-3.5-turbo"  # 或 "gpt-4"kernel.add_chat_service("chat_completion", OpenAIChatCompletion(model_id, api_key)
)

第二部分：智能郵件助手設計與規劃

2.1 需求分析與功能設計

我們的智能郵件助手將具備以下核心功能：

郵件內容生成：根據主題和關鍵點撰寫郵件
郵件分類與優先級判斷：識別郵件重要程度
郵件回復建議：針對收到郵件生成回復建議
日程關聯：從郵件內容提取日程信息

2.2 任務規劃鏈設計原理

任務規劃是AI Agent的核心能力，它使Agent能夠將復雜用戶請求分解為一系列可執行步驟。在Semantic Kernel中，規劃器主要有兩種類型：

順序規劃器：將目標分解為線性步驟序列
行動規劃器：根據上下文動態選擇下一步行動

我們的郵件助手將采用順序規劃器處理大多數任務，以下是規劃鏈設計示例：

用戶請求："幫我給團隊發郵件通知下周項目評審會，時間周三下午3點，需要準備材料"
↓
任務分解：
1. 提取關鍵信息：時間、地點、要求
2. 生成郵件主題
3. 撰寫郵件正文
4. 添加提醒標記（如需）
5. 請求用戶確認發送

2.3 定義技能(Skills)與函數(Functions)

在Semantic Kernel中，功能被組織為技能(Skills)，每個技能包含多個函數(Functions)。我們的郵件助手需要以下技能：

EmailSkill：核心郵件處理功能
CalendarSkill：日程相關功能
PrioritySkill：優先級判斷功能

第三部分：實現智能郵件助手

3.1 創建工具函數

首先實現基礎工具函數，這些函數將被封裝為Semantic Kernel的本地函數：

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import datetime
from semantic_kernel.skill_definition import sk_functionclass EmailSkill:"""郵件處理技能"""def __init__(self, smtp_server, smtp_port, email_address, email_password):self.smtp_server = smtp_serverself.smtp_port = smtp_portself.email_address = email_addressself.email_password = email_password@sk_function(description="發送電子郵件",name="send_email")async def send_email_async(self, recipient: str, subject: str, body: str) -> str:"""發送郵件工具函數"""try:msg = MIMEMultipart()msg['From'] = self.email_addressmsg['To'] = recipientmsg['Subject'] = subjectmsg.attach(MIMEText(body, 'plain'))with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:server.starttls()server.login(self.email_address, self.email_password)server.send_message(msg)return f"郵件已成功發送至 {recipient}"except Exception as e:return f"發送郵件時出錯: {str(e)}"@sk_function(description="生成郵件正文",name="generate_email_content")async def generate_email_content_async(self, context: sk.SKContext) -> str:"""生成郵件內容的語義函數"""# 從上下文中獲取變量topic = context.variables.get("topic")key_points = context.variables.get("key_points")tone = context.variables.get("tone", "professional")# 使用LLM生成郵件內容prompt = f"""根據以下信息撰寫一封{tone}風格的郵件：主題：{topic}關鍵點：{key_points}請生成完整郵件正文，包括適當的問候語和結束語。"""# 調用LLM生成內容response = await kernel.create_semantic_function(prompt)(context=context)return response.result

3.2 集成語義函數

語義函數使用自然語言描述功能，由LLM執行：

# 創建郵件分類語義函數
classify_email_prompt = """
分析以下郵件內容，判斷其類別和優先級。
可選類別：['工作', '個人', '營銷', '通知', '垃圾郵件']
優先級：['高', '中', '低']郵件內容：{{$email_content}}請以JSON格式返回結果，包含category和priority字段。
"""classify_email_function = kernel.create_semantic_function(classify_email_prompt,function_name="classify_email",skill_name="EmailSkill",description="對郵件進行分類和優先級判斷"
)

3.3 構建任務規劃鏈

實現完整的郵件處理任務規劃鏈：

from semantic_kernel.planning import SequentialPlanner
from semantic_kernel.planning.sequential_planner import SequentialPlannerConfigasync def process_email_request(user_request):"""處理郵件請求的完整規劃鏈"""# 創建規劃器實例planner = SequentialPlanner(kernel)# 定義目標goal = f"""根據用戶請求處理郵件相關任務：{user_request}可能需要的步驟：1. 理解用戶意圖和需求2. 提取關鍵信息（收件人、主題、內容要點等）3. 生成適當的郵件內容4. 判斷是否需要設置優先級或提醒5. 發送郵件或提供預覽"""# 創建計劃plan = await planner.create_plan_async(goal)# 執行計劃result = await plan.invoke_async()return result.result

3.4 添加記憶與上下文管理

為了使AI Agent能夠處理多輪對話和上下文相關任務，需要實現記憶管理：

from semantic_kernel import ContextVariables, MemoryStorageBase
from semantic_kernel.memory.volatile_memory_store import VolatileMemoryStore# 初始化記憶存儲
kernel.register_memory_store(VolatileMemoryStore())async def save_conversation_context(user_id, conversation_context):"""保存對話上下文"""await kernel.memory.save_information_async(collection="conversations",text=conversation_context,id=user_id,additional_metadata={"timestamp": datetime.datetime.now().isoformat()})async def recall_conversation_context(user_id):"""回憶對話上下文"""memories = await kernel.memory.search_async(collection="conversations",query="最近對話",filter=lambda metadata: metadata["id"] == user_id,limit=1)if memories:return memories[0].textreturn None

第四部分：高級功能與優化

4.1 實現郵件處理工作流

創建一個完整的郵件處理工作流，集成多個技能：

async def handle_incoming_email(email_content, sender):"""處理收到的工作流"""# 步驟1：分類郵件classification_vars = ContextVariables()classification_vars["email_content"] = email_contentclassification = await classify_email_function.invoke_async(variables=classification_vars)# 解析分類結果import jsontry:classification_result = json.loads(classification.result)category = classification_result.get("category", "未知")priority = classification_result.get("priority", "中")except:category, priority = "未知", "中"# 步驟2：根據分類采取不同行動if category == "工作" and priority == "高":# 提取可能的時間信息time_vars = ContextVariables()time_vars["email_content"] = email_contenttime_extraction = await extract_time_function.invoke_async(variables=time_vars)# 添加到日歷if "時間" in time_extraction.result:calendar_result = await add_to_calendar_async(title="重要工作郵件待處理",description=f"來自{sender}的郵件：{email_content[:100]}...",time_info=time_extraction.result)# 生成回復建議reply_vars = ContextVariables()reply_vars["email_content"] = email_contentreply_suggestion = await generate_reply_suggestion_async(variables=reply_vars)return {"category": category,"priority": priority,"action": "需要及時處理","reply_suggestion": reply_suggestion.result,"calendar_event": calendar_result if calendar_result else None}# 其他分類處理邏輯...return {"category": category, "priority": priority, "action": "已歸檔"}

4.2 錯誤處理與韌性設計

確保AI Agent能夠優雅處理異常情況：

from semantic_kernel import KernelException
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponentialclass ResilientEmailSkill(EmailSkill):"""具有韌性的郵件技能"""@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))async def send_email_with_retry(self, recipient: str, subject: str, body: str) -> str:"""帶重試機制的郵件發送"""try:return await self.send_email_async(recipient, subject, body)except Exception as e:print(f"郵件發送失敗: {str(e)}，進行重試...")raiseasync def safe_send_email(self, recipient: str, subject: str, body: str, max_retries=3) -> dict:"""安全的郵件發送，包含詳細結果"""for attempt in range(max_retries):try:result = await self.send_email_with_retry(recipient, subject, body)return {"success": True, "message": result, "attempts": attempt + 1}except Exception as e:if attempt == max_retries - 1:return {"success": False, "error": str(e), "attempts": attempt + 1}await asyncio.sleep(2 ** attempt)  # 指數退避return {"success": False, "error": "未知錯誤", "attempts": max_retries}

4.3 性能優化與緩存策略

為提高響應速度，實現緩存機制：

from datetime import timedelta
from functools import lru_cache
import hashlibclass OptimizedEmailSkill(EmailSkill):"""性能優化的郵件技能"""def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.response_cache = {}def _generate_cache_key(self, function_name, *args):"""生成緩存鍵"""key_str = function_name + "|" + "|".join(str(arg) for arg in args)return hashlib.md5(key_str.encode()).hexdigest()async def cached_generate_content(self, topic, key_points, tone="professional", expiry_minutes=10):"""帶緩存的郵件內容生成"""cache_key = self._generate_cache_key("generate_content", topic, key_points, tone)# 檢查緩存if cache_key in self.response_cache:cached_time, response = self.response_cache[cache_key]if datetime.datetime.now() - cached_time < timedelta(minutes=expiry_minutes):return response# 生成新內容context_vars = ContextVariables()context_vars["topic"] = topiccontext_vars["key_points"] = key_pointscontext_vars["tone"] = toneresponse = await self.generate_email_content_async(context_vars)# 更新緩存self.response_cache[cache_key] = (datetime.datetime.now(), response)return response

第五部分：測試與部署

5.1 編寫單元測試

確保核心功能的可靠性：

import pytest
from unittest.mock import AsyncMock, patch@pytest.mark.asyncio
async def test_email_classification():"""測試郵件分類功能"""kernel = sk.Kernel()# 配置測試環境...test_email = "尊敬的同事，請于下周一下午2點參加項目評審會議，地點在301會議室。"classification_vars = ContextVariables()classification_vars["email_content"] = test_emailresult = await classify_email_function.invoke_async(variables=classification_vars)# 驗證結果assert "category" in result.resultassert "priority" in result.resultassert "工作" in result.result or "通知" in result.result@pytest.mark.asyncio
async def test_email_send_failure_handling():"""測試郵件發送失敗處理"""with patch('smtplib.SMTP') as mock_smtp:mock_instance = mock_smtp.return_value.__enter__.return_valuemock_instance.send_message.side_effect = Exception("SMTP錯誤")skill = ResilientEmailSkill("smtp.example.com", 587, "test@example.com", "password")result = await skill.safe_send_email("recipient@example.com", "Test", "Test content")assert not result["success"]assert result["attempts"] == 3

5.2 創建演示應用

構建一個簡單的命令行界面展示功能：

async def main_demo():"""主演示函數"""print("智能郵件助手演示")print("=" * 50)while True:print("\n請選擇功能:")print("1. 撰寫新郵件")print("2. 處理收到郵件")print("3. 退出")choice = input("請輸入選項 (1-3): ").strip()if choice == "1":recipient = input("收件人: ")topic = input("主題: ")key_points = input("關鍵要點: ")tone = input("語氣風格 (professional/casual): ") or "professional"print("\n生成郵件內容中...")content = await email_skill.cached_generate_content(topic, key_points, tone)print(f"\n生成的郵件內容:\n{content}")send_confirmation = input("\n是否發送? (y/n): ")if send_confirmation.lower() == 'y':result = await email_skill.safe_send_email(recipient, topic, content)if result["success"]:print("? 郵件發送成功")else:print(f"? 發送失敗: {result['error']}")elif choice == "2":email_content = input("粘貼收到的郵件內容: ")sender = input("發件人: ")print("\n分析郵件中...")result = await handle_incoming_email(email_content, sender)print(f"\n分析結果:")print(f"類別: {result['category']}")print(f"優先級: {result['priority']}")print(f建議操作: {result['action']}")if 'reply_suggestion' in result:print(f"\n回復建議:\n{result['reply_suggestion']}")elif choice == "3":print("感謝使用智能郵件助手!")breakelse:print("無效選項，請重新選擇")if __name__ == "__main__":import asyncioasyncio.run(main_demo())

5.3 部署考慮與最佳實踐

在生產環境部署時需注意：

安全性：妥善管理API密鑰和郵件憑據
性能監控：記錄LLM調用耗時和費用
速率限制：遵守LLM API的調用限制
用戶體驗：提供適當的加載狀態和錯誤提示

# 生產環境配置示例
class ProductionConfig:"""生產環境配置"""def __init__(self):self.llm_max_retries = 3self.llm_timeout = 30self.email_rate_limit = 10  # 每分鐘最多發送10封郵件self.cache_ttl = 3600  # 緩存1小時def setup_kernel(self):"""生產環境內核配置"""kernel = sk.Kernel()# 添加帶超時和重試的LLM服務kernel.add_chat_service("chat_completion",OpenAIChatCompletion(model_id="gpt-4",api_key=os.getenv("OPENAI_API_KEY"),timeout=self.llm_timeout,max_retries=self.llm_max_retries))return kernel