Jimeng AI Studio实现智能体(Agent)开发：自动化任务实战

Jimeng AI Studio实现智能体(Agent)开发：自动化任务实战

你是不是也遇到过这样的场景？每天上班，打开电脑，就要面对一堆重复、琐碎但又不得不做的任务：从各个网站抓取数据、整理成表格、生成报告、再发送邮件……这些工作技术含量不高，却极其耗费时间和精力，让人感觉像个“人肉机器人”。

好消息是，现在有了更聪明的办法。借助Jimeng AI Studio，我们可以开发自己的智能体（Agent），让AI来帮我们处理这些自动化任务。这不再是简单的脚本，而是一个能理解目标、规划步骤、执行任务，甚至在遇到问题时能自己想办法的“数字员工”。

今天，我就来分享一个实战案例：如何用Jimeng AI Studio，从零开始构建一个能自动完成“市场信息日报”的智能体。整个过程不需要复杂的底层框架搭建，我们聚焦于如何利用平台能力，快速实现一个真正能跑起来的自动化系统。

1. 为什么选择Jimeng AI Studio来开发智能体？

在开始动手之前，我们先聊聊为什么是Jimeng AI Studio。市面上做AI开发的平台不少，但专门为智能体开发优化过的环境，能省去我们很多麻烦。

首先，它把环境配置的难题解决了。你不需要自己去折腾Python版本、依赖库冲突，或者为GPU内存发愁。平台提供了预置好的运行环境，开箱即用，这对于快速验证想法至关重要。

其次，它内置了对大模型API的友好支持。开发智能体，核心是让大模型充当“大脑”。Jimeng AI Studio让你能方便地调用各种主流模型，无论是进行复杂的任务规划，还是解析网页内容，都能找到合适的“工具”。

最重要的是，它提供了清晰的开发框架。智能体不是单次问答，它需要记忆、需要规划、需要调用外部工具。平台提供的Agent和Skill等概念，正好对应了这些核心组件，让我们能用更结构化的方式去思考和实现。

简单来说，选择Jimeng AI Studio，意味着我们可以把精力集中在“让智能体做什么”和“怎么做”的逻辑设计上，而不是浪费在“怎么能让它跑起来”的环境问题上。

2. 实战目标：设计一个市场信息日报自动化智能体

我们的目标是创建一个名为MarketDigestAgent的智能体。它的核心任务是：每天自动从指定的几个科技资讯网站抓取头条新闻，提取关键信息（如标题、摘要、链接），整理成一份格式清晰的日报，并通过邮件发送给指定人员。

这个任务看似简单，但拆解开来，智能体需要具备多种能力：

1. 任务理解与规划：理解“生成市场日报”这个抽象指令，并将其分解为具体的步骤。
2. 信息获取：能够访问互联网，获取实时信息。
3. 信息处理：从网页中准确提取所需的结构化数据。
4. 内容整合：将零散的信息组织成一份连贯、易读的报告。
5. 结果交付：将最终报告通过邮件发送出去。

接下来，我们就围绕这些能力，在Jimeng AI Studio中一步步构建我们的智能体。

3. 核心模块一：定义智能体的“技能”(Skill)

智能体的能力来源于其掌握的“技能”。在Jimeng AI Studio中，我们可以把每个独立的功能封装成一个Skill。对于我们的日报智能体，我设计了三个核心技能。

3.1 技能一：网络信息抓取技能

这个技能负责从给定的网址获取网页内容。我们使用requests库和BeautifulSoup来简化操作。

# skill_web_fetcher.py import requests from bs4 import BeautifulSoup from typing import Dict, Any class WebFetcherSkill: """技能：从指定URL抓取并解析网页正文内容""" def __init__(self): self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } def execute(self, url: str) -> Dict[str, Any]: """执行技能：抓取网页并提取核心文本。 Args: url: 目标网页地址 Returns: 包含状态、标题和正文内容的字典 """ try: print(f"[WebFetcher] 正在抓取: {url}") response = requests.get(url, headers=self.headers, timeout=10) response.raise_for_status() response.encoding = response.apparent_encoding soup = BeautifulSoup(response.text, 'html.parser') # 尝试获取标题 title = soup.title.string if soup.title else "无标题" # 简单提取正文：通常位于<p>标签或<article>标签内 main_content = soup.find('article') if not main_content: main_content = soup.find('main') if not main_content: # 如果找不到特定标签，则获取所有段落文本 paragraphs = soup.find_all('p') content_text = ' '.join([p.get_text(strip=True) for p in paragraphs[:10]]) # 取前10段 else: content_text = main_content.get_text(strip=True, separator=' ') # 限制内容长度，避免过长 content_preview = content_text[:500] + "..." if len(content_text) > 500 else content_text return { "status": "success", "title": title, "content_preview": content_preview, "source_url": url } except requests.RequestException as e: print(f"[WebFetcher] 抓取失败: {e}") return { "status": "error", "error": str(e), "source_url": url } except Exception as e: print(f"[WebFetcher] 解析失败: {e}") return { "status": "error", "error": f"解析错误: {str(e)}", "source_url": url } # 技能使用示例 if __name__ == "__main__": fetcher = WebFetcherSkill() result = fetcher.execute("https://example-tech-news.com") print(result)

3.2 技能二：关键信息提取技能

抓取到网页内容后，我们需要从中提取出我们关心的信息，比如新闻标题、核心观点。这里我们调用大模型来完成这项理解性工作。

# skill_info_extractor.py import json from jimeng_ai import ChatClient # 假设Jimeng AI Studio提供此类客户端 class InfoExtractorSkill: """技能：使用大模型从文本中提取结构化关键信息""" def __init__(self, model_name="gpt-4"): self.client = ChatClient(model=model_name) self.system_prompt = """你是一个专业的信息分析助手。你的任务是从提供的文本中提取出对科技市场、产品动态、行业趋势有价值的信息。 请严格按照JSON格式输出，包含以下字段： - `article_title`: 文章的主标题。 - `key_points`: 一个数组，包含3-5个最核心的观点或事实摘要。 - `potential_impact`: 简要分析这条信息可能对市场或相关公司产生的影响（1-2句话）。 - `tags`: 给文章打上几个关键词标签，如 ['人工智能', '融资', '新产品']。 如果文本中不包含明确新闻，请将`key_points`设为空数组。""" def execute(self, raw_text: str, source_title: str) -> Dict[str, Any]: """执行技能：提取关键信息。 Args: raw_text: 原始文本内容 source_title: 文章来源标题，用于上下文 Returns: 结构化提取的信息字典 """ print(f"[InfoExtractor] 正在分析来自 '{source_title}' 的内容...") user_prompt = f""" 请分析以下文本内容，提取关键信息： 来源标题：{source_title} 文本内容：{raw_text[:3000]} # 限制长度避免token超限 请输出JSON。 """ try: response = self.client.chat_completion( system=self.system_prompt, messages=[{"role": "user", "content": user_prompt}], response_format="json" ) extracted_info = json.loads(response['content']) extracted_info["extraction_status"] = "success" return extracted_info except json.JSONDecodeError: print("[InfoExtractor] 模型返回非JSON格式，尝试解析...") # 简易回退：手动提取标题和首段 return { "article_title": source_title, "key_points": [raw_text[:200]], "potential_impact": "需人工复核", "tags": ["未分类"], "extraction_status": "fallback" } except Exception as e: print(f"[InfoExtractor] 信息提取失败: {e}") return { "article_title": source_title, "key_points": [], "potential_impact": "信息提取失败", "tags": ["提取错误"], "extraction_status": "error" } # 技能使用示例 if __name__ == "__main__": extractor = InfoExtractorSkill(model_name="qwen-max") # 示例使用Qwen模型 sample_text = "某科技公司今日宣布推出新一代AI芯片，能效比提升50%，预计将用于数据中心..." result = extractor.execute(sample_text, "科技公司发布新款AI芯片") print(json.dumps(result, indent=2, ensure_ascii=False))

3.3 技能三：报告生成与邮件发送技能

信息都齐备后，需要整合成报告并发送。这个技能分为两部分：用大模型润色报告格式，然后用SMTP协议发送邮件。

# skill_report_sender.py import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime import json from jimeng_ai import ChatClient class ReportSenderSkill: """技能：生成格式化的日报并发送邮件""" def __init__(self, smtp_config: Dict, sender_email: str, model_name="gpt-4"): self.smtp_config = smtp_config # 包含host, port, username, password self.sender_email = sender_email self.client = ChatClient(model=model_name) def _generate_report_content(self, extracted_data_list: List[Dict]) -> str: """调用大模型，将提取的信息生成易读的日报内容""" print("[ReportSender] 正在生成日报内容...") data_summary = json.dumps(extracted_data_list, ensure_ascii=False, indent=2) system_prompt = """你是一位专业的市场分析报告编辑。请将提供的JSON格式的原始信息，整理成一份简洁、专业、易读的每日市场信息摘要。 报告结构建议： 1. 开头：今日摘要（几句话概括整体动向）。 2. 主体：分条列举重要新闻，每条包含标题、核心要点和简要影响分析。 3. 结尾：趋势观察或重点关注建议。 语言风格：客观、清晰、精炼。""" user_prompt = f"""今天是{datetime.now().strftime('%Y年%m月%d日')}。以下是从多个来源提取的市场信息原始数据： {data_summary} 请根据以上信息，生成今日市场信息日报的正文内容。""" try: response = self.client.chat_completion( system=system_prompt, messages=[{"role": "user", "content": user_prompt}] ) return response['content'] except Exception as e: print(f"[ReportSender] 报告生成失败，使用备用模板: {e}") # 备用模板 return self._fallback_report_template(extracted_data_list) def _fallback_report_template(self, data_list): """备用报告模板""" report_lines = [f"# 市场信息日报 {datetime.now().strftime('%Y-%m-%d')}", "\n"] for i, item in enumerate(data_list, 1): report_lines.append(f"## {i}. {item.get('article_title', '未知标题')}") points = item.get('key_points', []) for point in points[:3]: # 最多取3点 report_lines.append(f"- {point}") report_lines.append("") # 空行 return "\n".join(report_lines) def execute(self, report_content: str, recipient_emails: List[str], subject_prefix="市场信息日报"): """执行技能：发送邮件。 Args: report_content: 报告正文 recipient_emails: 收件人邮箱列表 subject_prefix: 邮件主题前缀 """ subject = f"{subject_prefix} - {datetime.now().strftime('%Y%m%d')}" # 创建邮件 msg = MIMEMultipart() msg['From'] = self.sender_email msg['To'] = ", ".join(recipient_emails) msg['Subject'] = subject # 添加正文 msg.attach(MIMEText(report_content, 'plain', 'utf-8')) try: print(f"[ReportSender] 正在发送邮件给 {len(recipient_emails)} 位收件人...") with smtplib.SMTP(self.smtp_config['host'], self.smtp_config['port']) as server: server.starttls() # 安全连接 server.login(self.smtp_config['username'], self.smtp_config['password']) server.send_message(msg) print("[ReportSender] 邮件发送成功！") return {"status": "success", "sent_to": recipient_emails} except Exception as e: print(f"[ReportSender] 邮件发送失败: {e}") return {"status": "error", "error": str(e)} # 技能配置与使用示例 if __name__ == "__main__": # 注意：在实际环境中，密码等敏感信息应通过环境变量或配置管理加载 smtp_config = { 'host': 'smtp.example.com', 'port': 587, 'username': 'your_email@example.com', 'password': 'your_password' # 建议使用应用专用密码 } sender = ReportSenderSkill( smtp_config=smtp_config, sender_email='noreply@yourcompany.com', model_name="qwen-max" ) # 假设有一些提取好的数据 sample_data = [{"article_title": "测试新闻", "key_points": ["这是一个测试点"]}] content = sender._generate_report_content(sample_data) # 执行发送 (实际运行时取消注释) # result = sender.execute(content, ['colleague@example.com']) # print(result)

4. 核心模块二：组装智能体大脑与工作流

有了这些技能，就像有了手和脚。现在需要为智能体装上“大脑”，让它知道在什么情况下使用什么技能，并管理整个执行流程。我们在Jimeng AI Studio中创建一个主控智能体。

# market_digest_agent.py import time from typing import List, Dict, Any from skill_web_fetcher import WebFetcherSkill from skill_info_extractor import InfoExtractorSkill from skill_report_sender import ReportSenderSkill class MarketDigestAgent: """市场信息日报智能体：协调多个技能完成自动化任务""" def __init__(self, target_urls: List[str], recipient_emails: List[str], smtp_config: Dict): """初始化智能体，装配所有技能。 Args: target_urls: 需要监控的资讯网站URL列表 recipient_emails: 日报接收邮箱列表 smtp_config: 邮件发送服务配置 """ print("="*50) print("初始化 MarketDigestAgent...") print("="*50) self.target_urls = target_urls self.recipient_emails = recipient_emails # 装配技能 self.web_fetcher = WebFetcherSkill() self.info_extractor = InfoExtractorSkill(model_name="qwen-max") # 使用Qwen模型进行信息提取 self.report_sender = ReportSenderSkill(smtp_config, sender_email=smtp_config['username'], model_name="qwen-max") # 使用Qwen模型润色报告 # 记忆与状态 self.execution_log = [] self.daily_data = [] def plan_tasks(self) -> List[Dict]: """任务规划：将总目标分解为具体步骤。""" print("[Agent] 正在规划今日任务...") tasks = [ {"step": 1, "action": "fetch_news", "description": "从目标网站抓取最新内容"}, {"step": 2, "action": "extract_info", "description": "从抓取的内容中提取关键信息"}, {"step": 3, "action": "generate_report", "description": "整合信息生成日报"}, {"step": 4, "action": "send_email", "description": "发送日报邮件"} ] return tasks def execute(self) -> Dict[str, Any]: """执行智能体的主要工作流。""" print(f"\n[Agent] 开始执行市场信息日报生成任务，共{len(self.target_urls)}个数据源。") start_time = time.time() tasks = self.plan_tasks() overall_status = "success" error_messages = [] # 步骤1: 抓取新闻 print(f"\n>>> 执行步骤1: {tasks[0]['description']}") raw_articles = [] for url in self.target_urls: print(f" 处理: {url}") result = self.web_fetcher.execute(url) self.execution_log.append({"action": "web_fetch", "url": url, "result": result["status"]}) if result["status"] == "success": raw_articles.append(result) else: error_messages.append(f"抓取失败 {url}: {result.get('error')}") overall_status = "partial_failure" if not raw_articles: print("[Agent] 错误：所有数据源抓取均失败，任务终止。") return {"status": "failure", "errors": error_messages} # 步骤2: 提取信息 print(f"\n>>> 执行步骤2: {tasks[1]['description']}") for article in raw_articles: extracted = self.info_extractor.execute( raw_text=article["content_preview"], source_title=article["title"] ) extracted["source_url"] = article["source_url"] self.daily_data.append(extracted) self.execution_log.append({ "action": "info_extract", "source": article["title"], "status": extracted.get("extraction_status", "unknown") }) print(f" 已分析: {article['title'][:50]}...") # 步骤3: 生成报告 print(f"\n>>> 执行步骤3: {tasks[2]['description']}") report_content = self.report_sender._generate_report_content(self.daily_data) # 步骤4: 发送邮件 print(f"\n>>> 执行步骤4: {tasks[3]['description']}") send_result = self.report_sender.execute( report_content=report_content, recipient_emails=self.recipient_emails ) self.execution_log.append({"action": "send_email", "result": send_result["status"]}) if send_result["status"] == "error": error_messages.append(f"邮件发送失败: {send_result.get('error')}") overall_status = "failure" # 任务总结 end_time = time.time() duration = round(end_time - start_time, 2) summary = { "status": overall_status, "total_sources": len(self.target_urls), "successful_fetches": len([a for a in raw_articles if a["status"]=="success"]), "articles_processed": len(self.daily_data), "recipients": len(self.recipient_emails), "time_elapsed_seconds": duration, "errors": error_messages if error_messages else None, "report_preview": report_content[:200] + "..." # 报告预览 } print(f"\n[Agent] 任务执行完毕！总耗时: {duration}秒") print(f"状态: {overall_status.upper()}") print(f"处理文章: {summary['articles_processed']}篇") return summary # 主程序：配置并运行智能体 if __name__ == "__main__": # 1. 配置信息（实际使用时，建议从配置文件或环境变量读取） TARGET_URLS = [ "https://example-tech-news-1.com/latest", "https://example-tech-news-2.com/headlines", # ... 添加更多你关注的资讯源 ] RECIPIENT_EMAILS = ["team@yourcompany.com"] # 警告：敏感信息务必妥善管理！ SMTP_CONFIG = { 'host': 'smtp.your-email-provider.com', 'port': 587, 'username': 'your-robot-email@domain.com', 'password': 'YOUR_APP_SPECIFIC_PASSWORD' # 切勿在代码中硬编码，使用环境变量 } # 2. 创建智能体实例 agent = MarketDigestAgent( target_urls=TARGET_URLS, recipient_emails=RECIPIENT_EMAILS, smtp_config=SMTP_CONFIG ) # 3. 执行任务 final_result = agent.execute() # 4. 输出结果摘要 print("\n" + "="*50) print("任务执行摘要:") print("="*50) for key, value in final_result.items(): if key != 'report_preview': print(f" {key}: {value}") print(f"\n报告开头预览:\n{final_result.get('report_preview')}")

5. 核心模块三：让智能体更聪明——异常处理与优化

一个健壮的智能体不能一遇到问题就崩溃。我们需要为它增加一些“常识”和“应变能力”。这里主要讲两个关键点：错误处理和任务优化。

错误处理策略：在上面的代码中，你已经看到每个Skill的execute方法都返回包含status字段的字典。智能体主控流程会根据这个状态决定是继续、跳过还是终止。例如，某个网站暂时无法访问，智能体会记录错误，但继续处理其他网站，而不是整个任务失败。

任务优化建议：

1. 设置重试机制：对于网络请求这类可能临时失败的操作，可以添加简单的重试逻辑。

   # 在WebFetcherSkill.execute中可加入重试 def execute_with_retry(self, url: str, max_retries=2): for attempt in range(max_retries): result = self.execute(url) if result["status"] == "success": return result print(f"第{attempt+1}次尝试失败，等待后重试...") time.sleep(2) return result # 返回最后一次失败结果

2. 结果缓存：如果某些信息源更新不频繁，可以缓存上一次的结果，避免重复提取相同内容，节省时间和API调用次数。
3. 动态任务规划：更高级的智能体可以根据历史执行效果（如某个网站经常失败）动态调整任务列表，甚至询问大模型“下一步该怎么办”。

在Jimeng AI Studio中部署与调度：代码写好了，怎么让它每天自动运行？Jimeng AI Studio通常支持两种方式：

• 手动触发：直接在平台的Notebook或脚本环境中运行你的market_digest_agent.py文件。
• 定时任务：利用平台提供的任务调度功能（如果支持），设置每天上午9点自动执行这个智能体。你也可以在代码外层包装一个简单的循环和sleep，但使用系统调度器更可靠。

6. 总结与展望

通过这个实战项目，我们完整地走了一遍在Jimeng AI Studio上开发一个功能型智能体的流程。从定义三个核心Skill（抓取、提取、发送），到组装成一个能规划执行的Agent，再到思考如何让它更健壮。

整个过程给我的感受是，Jimeng AI Studio确实降低了智能体开发的门槛。我们不需要从零开始构建复杂的Agent框架，而是可以像搭积木一样，把关注点放在业务逻辑和技能实现上。平台处理了环境、模型调用等底层细节，让我们能更专注于“让AI做什么”。

这个MarketDigestAgent只是一个起点。你可以很容易地扩展它：

• 增加数据源：除了新闻网站，可以接入行业报告、社交媒体趋势、竞品动态。
• 丰富技能：加入数据分析技能，自动生成图表；加入推送技能，将日报发送到企业微信或钉钉。
• 提升智能：让智能体不仅能汇总信息，还能基于历史数据做简单的趋势预测，或者在报告开头生成一份“执行官摘要”。

智能体开发的魅力就在于，你可以从一个明确的小任务开始，逐步赋予它更多的能力和自主性。Jimeng AI Studio提供了一个不错的起点，让你能快速验证想法，把自动化从简单的“如果-那么”规则，升级为能理解、能规划、能应对变化的智能系统。不妨就从你手头最重复的那项工作开始，试试看能打造出一个怎样的“数字助手”吧。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。