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