Co-creation-projects/bichchibui5-hub-EmailSmartAssistant/EmailSmartAssistant_HelloAgents.ipynb
"""
基于HelloAgents框架构建的智能邮件处理系统，能够自动分类邮件、生成回复草稿、提取关键信息并设置智能提醒。 项目采用ReAct智能体范式，结合多个专业工具，实现邮件处理的全流程自动化。
核心功能：
# 安装依赖
!pip install -q hello-agents[all]
!pip install -q pandas numpy matplotlib seaborn
!pip install -q jieba textblob langdetect
!pip install -q python-dotenv rich
# 导入必要的库
from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.tools import BaseTool
import os
import json
import re
from datetime import datetime, timedelta
from typing import Dict, List, Any
import pandas as pd
import numpy as np
from dotenv import load_dotenv
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
import warnings
warnings.filterwarnings('ignore')
console = Console()
print("✅ 库导入成功！")
# 加载环境变量
load_dotenv()
# 设置API密钥（如果需要）
# os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# os.environ["ANTHROPIC_API_KEY"] = "your-api-key-here"
print("✅ 环境配置完成！")
class EmailClassifierTool(BaseTool):
"""邮件分类工具"""
name = "email_classifier"
description = "对邮件进行智能分类，包括类型、优先级和发件人类型判断"
def __init__(self):
super().__init__()
self.classification_rules = {
'work_keywords': ['会议', '项目', '工作', '任务', '汇报', 'meeting', 'project', 'work', 'task', 'urgent'],
'customer_keywords': ['客户', '咨询', '购买', '服务', 'customer', 'inquiry', 'purchase', 'service'],
'personal_keywords': ['个人', '家庭', '朋友', 'personal', 'family', 'friend', '聚餐'],
'spam_keywords': ['广告', '推广', '营销', '优惠', 'advertisement', 'promotion', 'marketing', '折扣']
}
def run(self, email_data: str) -> str:
"""执行邮件分类"""
try:
# 解析邮件数据
email_info = json.loads(email_data)
subject = email_info.get('subject', '').lower()
body = email_info.get('body', '').lower()
sender = email_info.get('sender', '').lower()
text_content = f"{subject} {body}"
# 分类逻辑
classification = self._classify_email(text_content, sender)
return json.dumps(classification, ensure_ascii=False, indent=2)
except Exception as e:
return f"分类失败: {str(e)}"
def _classify_email(self, text_content: str, sender: str) -> Dict[str, str]:
"""内部分类逻辑"""
# 检查垃圾邮件
spam_score = sum(1 for keyword in self.classification_rules['spam_keywords']
if keyword in text_content)
if spam_score >= 2:
return {'type': 'spam', 'priority': 'low', 'sender_type': 'external'}
# 计算各类型得分
work_score = sum(1 for keyword in self.classification_rules['work_keywords']
if keyword in text_content)
customer_score = sum(1 for keyword in self.classification_rules['customer_keywords']
if keyword in text_content)
personal_score = sum(1 for keyword in self.classification_rules['personal_keywords']
if keyword in text_content)
# 确定类型
scores = {'work': work_score, 'customer': customer_score, 'personal': personal_score}
email_type = max(scores, key=scores.get) if max(scores.values()) > 0 else 'other'
# 确定优先级
priority = 'high' if any(word in text_content for word in ['紧急', 'urgent', 'asap', '重要']) else 'medium'
if email_type == 'spam':
priority = 'low'
# 确定发件人类型
if 'company.com' in sender:
sender_type = 'colleague'
elif 'noreply' in sender or 'no-reply' in sender:
sender_type = 'system'
elif email_type == 'customer':
sender_type = 'customer'
else:
sender_type = 'external'
return {
'type': email_type,
'priority': priority,
'sender_type': sender_type
}
print("✅ 邮件分类工具定义完成")
class InformationExtractorTool(BaseTool):
"""信息提取工具"""
name = "information_extractor"
description = "从邮件中提取关键信息，包括日期、时间、联系方式、待办事项等"
def __init__(self):
super().__init__()
self.date_patterns = [
r'\d{4}-\d{1,2}-\d{1,2}',
r'\d{1,2}月\d{1,2}日',
r'\d{1,2}/\d{1,2}'
]
self.time_patterns = [
r'\d{1,2}:\d{2}',
r'\d{1,2}点',
r'\d{1,2} PM',
r'\d{1,2} AM'
]
def run(self, email_data: str) -> str:
"""执行信息提取"""
try:
email_info = json.loads(email_data)
body = email_info.get('body', '')
extracted_info = self._extract_information(body)
return json.dumps(extracted_info, ensure_ascii=False, indent=2)
except Exception as e:
return f"信息提取失败: {str(e)}"
def _extract_information(self, body: str) -> Dict[str, List[str]]:
"""内部信息提取逻辑"""
# 提取日期
dates = []
for pattern in self.date_patterns:
dates.extend(re.findall(pattern, body))
# 提取时间
times = []
for pattern in self.time_patterns:
times.extend(re.findall(pattern, body))
# 提取联系方式
phones = re.findall(r'1[3-9]\d{9}', body)
emails = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', body)
# 提取待办事项
todo_keywords = ['需要', '请', '准备', 'need', 'please', 'prepare', '确认']
sentences = body.replace('。', '.').split('.')
todos = []
for sentence in sentences:
if any(keyword in sentence for keyword in todo_keywords):
clean_sentence = sentence.strip()
if len(clean_sentence) > 5:
todos.append(clean_sentence)
return {
'dates': list(set(dates)),
'times': list(set(times)),
'phones': phones,
'emails': emails,
'todos': todos[:3] # 最多3个
}
print("✅ 信息提取工具定义完成")
class ReplyGeneratorTool(BaseTool):
"""回复生成工具"""
name = "reply_generator"
description = "根据邮件内容和分类结果生成合适的回复草稿"
def __init__(self):
super().__init__()
self.reply_templates = {
'work': {
'zh': '感谢您的邮件。关于{subject}，我已收到您的信息。我将在24小时内回复您详细的反馈。如有紧急事项，请随时联系我。\n\n此致\n敬礼',
'en': 'Thank you for your email regarding {subject}. I have received your information and will provide detailed feedback within 24 hours. Please feel free to contact me if there are any urgent matters.\n\nBest regards'
},
'customer': {
'zh': '尊敬的客户，\n\n感谢您对我们产品/服务的关注。关于您咨询的{subject}，我们将安排专业人员在24小时内为您提供详细解答。\n\n如有其他问题，欢迎随时联系我们。\n\n此致\n敬礼',
'en': 'Dear Valued Customer,\n\nThank you for your interest in our products/services. Regarding your inquiry about {subject}, we will arrange for a professional to provide you with detailed answers within 24 hours.\n\nPlease feel free to contact us if you have any other questions.\n\nBest regards'
},
'general': {
'zh': '您好，\n\n已收到您的邮件，我将仔细阅读并在24小时内回复。\n\n谢谢！',
'en': 'Hello,\n\nI have received your email and will read it carefully and reply within 24 hours.\n\nThank you!'
}
}
def run(self, input_data: str) -> str:
"""执行回复生成"""
try:
data = json.loads(input_data)
email_info = data.get('email', {})
classification = data.get('classification', {})
if classification.get('type') == 'spam':
return json.dumps({'message': '垃圾邮件不生成回复'}, ensure_ascii=False)
reply = self._generate_reply(email_info, classification)
return json.dumps(reply, ensure_ascii=False, indent=2)
except Exception as e:
return f"回复生成失败: {str(e)}"
def _generate_reply(self, email_info: Dict, classification: Dict) -> Dict[str, str]:
"""内部回复生成逻辑"""
# 检测语言
body = email_info.get('body', '')
is_chinese = any('\u4e00' <= char <= '\u9fff' for char in body)
lang = 'zh' if is_chinese else 'en'
# 选择模板
email_type = classification.get('type', 'general')
template_type = email_type if email_type in ['work', 'customer'] else 'general'
template = self.reply_templates[template_type][lang]
# 生成回复
subject = email_info.get('subject', '')
reply_content = template.format(subject=subject)
return {
'to': email_info.get('sender', ''),
'subject': f"Re: {subject}",
'content': reply_content,
'language': lang,
'template_type': template_type
}
print("✅ 回复生成工具定义完成")
class ReminderCreatorTool(BaseTool):
"""提醒创建工具"""
name = "reminder_creator"
description = "根据提取的信息创建智能提醒"
def run(self, input_data: str) -> str:
"""执行提醒创建"""
try:
data = json.loads(input_data)
email_info = data.get('email', {})
extracted_info = data.get('extracted_info', {})
classification = data.get('classification', {})
reminders = self._create_reminders(email_info, extracted_info, classification)
return json.dumps(reminders, ensure_ascii=False, indent=2)
except Exception as e:
return f"提醒创建失败: {str(e)}"
def _create_reminders(self, email_info: Dict, extracted_info: Dict, classification: Dict) -> List[Dict]:
"""内部提醒创建逻辑"""
reminders = []
# 只为高优先级和中优先级邮件创建提醒
if classification.get('priority') not in ['high', 'medium']:
return reminders
# 为日期创建提醒
for date_str in extracted_info.get('dates', []):
try:
# 简单的日期解析
if '-' in date_str and len(date_str) == 10:
target_date = datetime.strptime(date_str, '%Y-%m-%d')
reminder_date = target_date - timedelta(days=1)
if reminder_date > datetime.now():
reminders.append({
'type': 'date_reminder',
'email_subject': email_info.get('subject', ''),
'reminder_date': reminder_date.isoformat(),
'target_date': target_date.isoformat(),
'message': f"提醒：{email_info.get('subject', '')} - 明天到期（{date_str}）"
})
except:
continue
# 为待办事项创建提醒
for todo in extracted_info.get('todos', []):
reminder_date = datetime.now() + timedelta(hours=2)
reminders.append({
'type': 'todo_reminder',
'email_subject': email_info.get('subject', ''),
'reminder_date': reminder_date.isoformat(),
'message': f"待办事项提醒：{todo[:50]}..."
})
return reminders
print("✅ 提醒创建工具定义完成")
# 创建LLM（使用本地模型或API）
try:
llm = HelloAgentsLLM(
model_name="gpt-3.5-turbo", # 可以替换为其他模型
temperature=0.1
)
print("✅ LLM创建成功")
except Exception as e:
print(f"⚠️ LLM创建失败，使用模拟模式: {e}")
llm = None
# 创建智能体
system_prompt = """
你是一个专业的邮件处理助手，具备以下能力：
1. 邮件分类：能够准确识别邮件类型（工作、客户、个人、垃圾邮件）和优先级
2. 信息提取：从邮件中提取关键信息如日期、时间、联系方式、待办事项
3. 回复生成：根据邮件内容和分类生成合适的回复草稿
4. 提醒创建：基于提取的信息创建智能提醒
处理邮件时，请按以下步骤进行：
1. 首先使用email_classifier工具对邮件进行分类
2. 然后使用information_extractor工具提取关键信息
3. 根据分类结果使用reply_generator工具生成回复草稿
4. 最后使用reminder_creator工具创建相应的提醒
请确保处理过程专业、准确，并提供清晰的结果说明。
"""
if llm:
agent = SimpleAgent(
name="智能邮件助手",
llm=llm,
system_prompt=system_prompt
)
else:
# 创建一个模拟智能体用于演示
class MockAgent:
def __init__(self, name):
self.name = name
self.tools = {}
def add_tool(self, tool):
self.tools[tool.name] = tool
def run(self, query):
return self._mock_process(query)
def _mock_process(self, query):
# 模拟智能体处理流程
results = []
# 模拟邮件数据
if "演示" in query or "demo" in query.lower():
demo_email = {
"subject": "紧急：项目进度汇报会议安排",
"sender": "[email protected]",
"body": "各位同事，请准备明天下午2点的项目进度汇报会议。需要准备本周工作总结和下周计划。截止时间：2024-01-16 14:00。请确认参会。"
}
email_json = json.dumps(demo_email, ensure_ascii=False)
else:
# 尝试解析用户输入的邮件数据
email_json = query
# 1. 邮件分类
if 'email_classifier' in self.tools:
classification_result = self.tools['email_classifier'].run(email_json)
results.append(f"📋 邮件分类结果：\n{classification_result}")
# 2. 信息提取
if 'information_extractor' in self.tools:
extraction_result = self.tools['information_extractor'].run(email_json)
results.append(f"\n🔍 信息提取结果：\n{extraction_result}")
# 3. 回复生成
if 'reply_generator' in self.tools:
try:
email_data = json.loads(email_json)
classification_data = json.loads(classification_result) if 'email_classifier' in self.tools else {}
reply_input = json.dumps({
'email': email_data,
'classification': classification_data
}, ensure_ascii=False)
reply_result = self.tools['reply_generator'].run(reply_input)
results.append(f"\n✍️ 回复草稿：\n{reply_result}")
except:
results.append("\n✍️ 回复生成跳过")
# 4. 提醒创建
if 'reminder_creator' in self.tools:
try:
email_data = json.loads(email_json)
classification_data = json.loads(classification_result) if 'email_classifier' in self.tools else {}
extraction_data = json.loads(extraction_result) if 'information_extractor' in self.tools else {}
reminder_input = json.dumps({
'email': email_data,
'classification': classification_data,
'extracted_info': extraction_data
}, ensure_ascii=False)
reminder_result = self.tools['reminder_creator'].run(reminder_input)
results.append(f"\n⏰ 提醒创建结果：\n{reminder_result}")
except:
results.append("\n⏰ 提醒创建跳过")
return "\n".join(results)
agent = MockAgent("智能邮件助手")
print("✅ 智能体创建成功")
# 添加工具
agent.add_tool(EmailClassifierTool())
agent.add_tool(InformationExtractorTool())
agent.add_tool(ReplyGeneratorTool())
agent.add_tool(ReminderCreatorTool())
print("✅ 工具添加完成")
print(f"智能体 '{agent.name}' 已配置以下工具：")
if hasattr(agent, 'tools'):
for tool_name in agent.tools.keys():
print(f" - {tool_name}")
else:
print(" - email_classifier")
print(" - information_extractor")
print(" - reply_generator")
print(" - reminder_creator")
# 示例1:基础功能演示
print("=== 示例1:基础功能演示 ===")
console.print(Panel.fit("🚀 开始演示智能邮件助手的基础功能", style="blue"))
# 使用演示邮件数据
demo_query = "演示邮件处理功能"
try:
result = agent.run(demo_query)
console.print(Panel(result, title="📧 处理结果", style="green"))
except Exception as e:
console.print(f"❌ 处理失败: {str(e)}", style="red")
# 示例2:复杂场景演示
print("\n=== 示例2:复杂场景演示 ===")
console.print(Panel.fit("🎯 演示处理客户咨询邮件", style="blue"))
# 客户咨询邮件示例
customer_email = {
"subject": "产品功能咨询和演示预约",
"sender": "[email protected]",
"body": "您好，我对贵公司的智能邮件助手产品很感兴趣。希望了解更多功能详情，并预约一次产品演示。我的联系方式是13800138000，邮箱是[email protected]。希望能在本周五之前安排演示，谢谢！"
}
customer_query = json.dumps(customer_email, ensure_ascii=False)
try:
result = agent.run(customer_query)
console.print(Panel(result, title="📧 客户邮件处理结果", style="green"))
except Exception as e:
console.print(f"❌ 处理失败: {str(e)}", style="red")
# 示例3:英文邮件处理
print("\n=== 示例3:英文邮件处理 ===")
console.print(Panel.fit("🌍 演示处理英文邮件", style="blue"))
# 英文邮件示例
english_email = {
"subject": "Urgent: Quarterly Report Meeting",
"sender": "[email protected]",
"body": "Hi team, we need to schedule an urgent meeting tomorrow at 3 PM to discuss the quarterly results. Please prepare your reports and confirm attendance by 5 PM today. This is very important for our Q4 planning."
}
english_query = json.dumps(english_email, ensure_ascii=False)
try:
result = agent.run(english_query)
console.print(Panel(result, title="📧 英文邮件处理结果", style="green"))
except Exception as e:
console.print(f"❌ 处理失败: {str(e)}", style="red")
# 示例4:批量邮件处理演示
print("\n=== 示例4:批量邮件处理演示 ===")
console.print(Panel.fit("📦 演示批量处理多封邮件", style="blue"))
# 多封邮件示例
batch_emails = [
{
"subject": "系统维护通知",
"sender": "[email protected]",
"body": "系统将于2024-01-20 02:00-04:00进行维护升级，期间服务可能中断。请提前做好准备工作。"
},
{
"subject": "限时优惠！立即购买享受8折优惠",
"sender": "[email protected]",
"body": "亲爱的用户，我们的产品正在进行限时促销活动！现在购买可享受8折优惠，机会难得，不要错过！"
},
{
"subject": "周末聚餐安排",
"sender": "[email protected]",
"body": "这个周末我们一起聚餐吧，时间定在周六晚上7点，地点在市中心的那家川菜馆。请确认是否能参加。"
}
]
# 处理统计
batch_results = []
processing_stats = {'total': 0, 'work': 0, 'customer': 0, 'personal': 0, 'spam': 0, 'other': 0}
for i, email in enumerate(batch_emails, 1):
console.print(f"\n📧 处理邮件 {i}/{len(batch_emails)}: {email['subject'][:30]}...", style="cyan")
try:
email_query = json.dumps(email, ensure_ascii=False)
result = agent.run(email_query)
# 简单统计（从结果中提取分类信息）
processing_stats['total'] += 1
if 'work' in result:
processing_stats['work'] += 1
elif 'customer' in result:
processing_stats['customer'] += 1
elif 'personal' in result:
processing_stats['personal'] += 1
elif 'spam' in result:
processing_stats['spam'] += 1
else:
processing_stats['other'] += 1
batch_results.append(result)
console.print("✅ 处理完成", style="green")
except Exception as e:
console.print(f"❌ 处理失败: {str(e)}", style="red")
batch_results.append(f"处理失败: {str(e)}")
# 显示批量处理统计
stats_table = Table(title="📊 批量处理统计")
stats_table.add_column("类型", style="cyan")
stats_table.add_column("数量", style="magenta")
for category, count in processing_stats.items():
stats_table.add_row(category, str(count))
console.print(stats_table)
# 性能评估
import time
console.print(Panel.fit("📈 开始性能评估", style="blue"))
# 测试数据
test_emails = [
{"subject": "会议通知", "sender": "[email protected]", "body": "明天下午2点开会"},
{"subject": "客户咨询", "sender": "[email protected]", "body": "想了解产品功能"},
{"subject": "广告推广", "sender": "[email protected]", "body": "限时优惠，立即购买"},
{"subject": "朋友聚会", "sender": "[email protected]", "body": "周末一起吃饭"},
{"subject": "系统通知", "sender": "[email protected]", "body": "系统维护通知"}
]
# 预期分类结果
expected_types = ['work', 'customer', 'spam', 'personal', 'other']
# 性能测试
start_time = time.time()
correct_classifications = 0
total_processed = 0
for i, (email, expected_type) in enumerate(zip(test_emails, expected_types)):
try:
email_query = json.dumps(email, ensure_ascii=False)
result = agent.run(email_query)
# 简单的准确率评估（检查结果中是否包含预期类型）
if expected_type in result.lower():
correct_classifications += 1
total_processed += 1
except Exception as e:
console.print(f"测试邮件 {i+1} 处理失败: {str(e)}", style="red")
end_time = time.time()
processing_time = end_time - start_time
# 计算性能指标
accuracy = (correct_classifications / total_processed * 100) if total_processed > 0 else 0
avg_time_per_email = processing_time / total_processed if total_processed > 0 else 0
# 显示性能结果
performance_table = Table(title="📊 性能评估结果")
performance_table.add_column("指标", style="cyan")
performance_table.add_column("数值", style="magenta")
performance_table.add_row("总处理邮件数", str(total_processed))
performance_table.add_row("分类准确率", f"{accuracy:.1f}%")
performance_table.add_row("总处理时间", f"{processing_time:.2f}秒")
performance_table.add_row("平均处理时间", f"{avg_time_per_email:.2f}秒/封")
performance_table.add_row("处理速度", f"{1/avg_time_per_email:.1f}封/秒" if avg_time_per_email > 0 else "N/A")
console.print(performance_table)
"""
问题：需要同时处理中英文邮件，并生成对应语言的回复 解决方案：
问题：邮件中的日期、时间格式多样，难以准确提取 解决方案：
问题：多个工具之间需要传递数据，确保处理流程的连贯性 解决方案：
本项目成功展示了如何使用HelloAgents框架构建一个完整的智能邮件处理系统。通过模块化的工具设计和智能体协调，实现了邮件处理的全流程自动化，为用户节省了大量的邮件处理时间，提升了工作效率。
项目不仅具有实用价值，还为其他类似的文本处理和自动化任务提供了可参考的技术架构和实现方案。 """
# 项目完成提示
console.print(Panel.fit(
"🎉 智能邮件助手项目演示完成！\n\n"
"✨ 主要成果：\n"
"• 实现了完整的邮件智能处理流程\n"
"• 支持中英文邮件的自动分类和回复生成\n"
"• 基于HelloAgents框架的模块化架构\n"
"• 提供了丰富的演示和性能评估\n\n"
"🚀 下一步：\n"
"• 集成真实的LLM模型提升智能化水平\n"
"• 连接真实邮箱进行实际应用测试\n"
"• 根据使用反馈持续优化功能",
title="项目总结",
style="bold green"
))
print("\n" + "="*50)
print("感谢使用智能邮件助手！")
print("项目地址：https://github.com/EmailSmartAssistant")
print("="*50)