网站类型定位分析,网站关键词在哪里修改,企业微信app下载,app定制化开发前言 在上一篇文章中#xff0c;我们学习了如何利用MetaGPT框架构建单智能体和多智能体#xff0c;并通过一个技术文档撰写Agent和课后作业较为完整的理解一个Agent的需求分析和开发流程#xff1b;但是技术要和应用结合才能得到更广泛的推广#xff1b;在本文中#xff0… 前言 在上一篇文章中我们学习了如何利用MetaGPT框架构建单智能体和多智能体并通过一个技术文档撰写Agent和课后作业较为完整的理解一个Agent的需求分析和开发流程但是技术要和应用结合才能得到更广泛的推广在本文中作者将以一个Github订阅智能体为例带领读者感受Agent应用的潜力 一、介绍 不知各位读者有没有阅读GitHub Trending的习惯GitHub Trending 是 GitHub 上一个专门显示当前最受欢迎项目的页面。它根据项目在过去一个月内的星标数、分支数和贡献者数等因素进行排名展示出最热门的开源项目。无论对程序员、技术极客以及科技博主GitHub Trending都是一个绝好的信息源,其类似新闻的热榜阅读GitHub Trending有如下的好处 了解最新技术趋势最热门的编程语言、框架和工具帮助开发者保持对技术发展的敏锐。发现优秀项目很多优秀开源项目通常具有创新性、实用性和可拓展性可以为我们自己的项目提供灵感和参考。学习先进经验通过阅读热门项目的代码和文档可以学习到其中的设计思想、编码技巧和最佳实践提高自己的编程能力。参与开源社区通过参与热门项目的讨论和贡献代码可以与其他开发者建立联系扩展自己的社交圈子并为开源社区做出贡献。 因此及时高效地从GitHub中获取最新的技术发展信息和资源对于开发者很重要。而今天我们就以这个为我们的需求构建我们的OSS 订阅Agent。 通过本文你将学习到 基于Python构建GitHub Trending爬虫Action基于MetaGPT构建仓库内容生成Action构建Discord机器人构建邮箱订阅的论文总结Agent 二、OSS订阅Agent实现 1.需求分析 开始之前我们首先需要知道MetaGPT中订阅Agent机制SubscriptionRunner模块; SubscriptionRunner模块是metagpt中的一个模块它提供了SubscriptionRunner类用于提供一个Role的定时运行方式。基于SubscriptionRunner类我们可以定时触发运行一个Role然后将Role的执行输出通知给用户。下面是一个参考案例代码 import asyncio from metagpt.subscription import SubscriptionRunner from metagpt.roles import Searcher from metagpt.schema import Message# 定义触发器每24小时抛出一个Message async def trigger():while True:yield Message(the latest news about OpenAI)await asyncio.sleep(3600 * 24)# 回调函数实例话一个订阅器实例当触发器触发后打印返回的内容 async def callback(msg: Message):print(msg.content)# 主函数开始运行该计划任务 async def main():pb SubscriptionRunner()await pb.subscribe(Searcher(), trigger(), callback)await pb.run()asyncio.run(main()) 这个模块中我们的OSS订阅智能体主要包含三个要素 Role智能体本身Trigger触发器Callback回调函数 我们整理一下实现的思路 1.构建一个爬虫Action,其会从Github Trending将我们感兴趣的开源类目信息爬取下来;2.触发Trigger即什么情况下我们的Agent才会运行即按照哪些机制判断是否爬取如定时爬取或者页面更新时爬取本文中以每天早上9点定时爬取为例3.回调函数callback当Agent启动爬取到页面数据,后续执行的操作是什么本文中我们将爬取的信息根据Prompt模版整理为一个每日推送的文章然后将文章信息发送到我们的订阅平台(Discord、Telegram或QQ邮箱); 理清思路以后我们开始逐个实现 2.OSSAgent Role实现 按照我们上一篇文章的思路我们这里分别需要构建爬虫Crawl Action和模版输出Action: (1)Crawl Action实现 ①爬取方式确定 实现数据爬取有三种方式 基于API最高效最稳定但是经过作者的检索Github并没有提供Github Trending的API接口这个方案放弃基于自动化页面抓取稳定低效开发较复杂例如使用selenium等框架模拟人类行为进行数据抓取可以获得大多数页面数据但是需要配置环境可选:基于网页爬虫稳定情况看爬取目标但是实现较为简单爬取静态页面高效可以用这个 我们选择了使用基于网页爬取当然这也是需要一定的Python爬虫基础和Web基础了解页面元素构成和各种元素选择器作者常用的爬虫框架组合是常见的requests和BeautifulSoup前者用于数据爬取后者用于数据解析今天学习的时候发现官方文档使用的是aiohttp作者并不熟悉这个第三方库与GPT-4o学习了一下总结了其与requests的异同和特点 区别 同步和异步requests 是一个同步库它发送请求后会阻塞代码执行直到收到响应aiohttp 是一个异步库它发送请求后可以继续执行其他代码收到响应后再处理。性能由于 aiohttp 是异步库在处理大量请求时它的性能比 requests 更好因为它可以在等待响应的同时处理其他请求。功能虽然 requests 和 aiohttp 都可以发送 HTTP 请求但 aiohttp 还支持 websocket、中间件等功能。 联系 都是 Python 中用于发送 HTTP 请求的库。都支持常见的 HTTP 方法如 GET、POST、PUT、DELETE 等。都支持 SSL 加密和代理服务器。 特点 requests 的特点 简单易用requests 的 API 简单易用只需几行代码就可以发送 HTTP 请求。 支持各种数据格式requests 支持 JSON、表单数据、多部分文件上传等各种数据格式。 自动处理重定向和 cookierequests 可以自动处理重定向和 cookie无需手动处理。 aiohttp 的特点 异步处理aiohttp 基于 Python 的 async/await 语法支持异步处理多个请求。 高性能aiohttp 使用异步 I/O 模型在处理大量请求时表现出色。 支持 websocketaiohttp 支持 websocket 协议可以用于实时通信。 总而言之aiohttp库看起来最大的特点就是异步高效了对于MetaGPT其框架中处处都能看到异步的痕迹这对Agent数据爬取处理更友好我们今天就使用aiohttp库进行数据爬取 如果爬取目标为国外网页但代码在国内本地运行请修改metagpt项目文件夹下的config/key.yaml中配置代理服务器地址 GLOBAL_PROXY: http://127.0.0.1:8118 # 改成自己的代理服务器地址 ②确定爬取目标 打开[GitHub Trending 页面]选择我们要爬取的数据 作者这里要爬取的数据有 仓库名称仓库链接仓库简介仓库Star数仓库fork数仓库的readme文件(如果有) 选择筛选条件: 语言python、nodejs时间范围This week关注目标repositories 细心的读者会发现当我们选择不同的筛选条件时我们请求的页面URL也是不同的这说明页面筛选是Get请求因此我们如果我们要爬取对应的数据应该先筛选条件再确定请求的URL: ##### ③解析网页 我们进入到上一步选择的URL页面然后按下F12进入到开发者页面按照下图流程点击确定爬取的目标元素 由图可知我们要爬取的目标中每一个仓库项目信息都存储在class为Box-row的元素中,选择一个元素逐级点击查看每个元素内部可以看到我们需要的这些数据都存储在这些元素中我们只需要使用aiphttp库爬虫请求目标页面目标页面则返回整个页面的字符串数据然后我们使用BeautifulSoup库解析获得的数据为结构化元素树格式然后按照这些元素的定位符如id和class定位数据通过循环然后将这些数据爬取下来 整理一下我们需要的数据的定位符: 单仓库所有信息元素类Box-row仓库名称h2标签下的a标签下的文本,可写作h2 a仓库链接h2标签下的a标签下的href属性值与https://github/com/组合作者a标签下class为text-normal的元素文本简介p标签下文本Stara标签下class包含Link--muted的第一个元素文本写作a.Link--muted[0]Forka标签下class包含Link--muted的第二个元素文本写作a.Link--muted[1]本周Star数量span标签下class包含d-inline-block的元素文本写作span.d-inline-block 页面分析完成开始撰写代码爬取!!! ④仓库列表数据爬取 代码学习非一日之功但是在AI发达的当下读者可以直接用我的代码来询问AI我的思路 先爬取页面提取所有class属性值为Box-row的元素列表存储起来即获得所有仓库信息对获取的列表信息进行循环遍历对每个元素进行深入元素提取包括该仓库的name,url,description,language, 爬取网页并进行提取的代码如下: import aiohttp # 导入 aiohttp 库用于发送异步 HTTP 请求 import asyncio # 导入 asyncio 库用于处理异步任务 from bs4 import BeautifulSoup # 导入 BeautifulSoup 库用于解析 HTML 文档async def fetch_html(url):# 定义一个异步函数 fetch_html用于获取 URL 对应的 HTML 内容async with aiohttp.ClientSession() as session:# 使用 aiohttp 创建一个会话上下文管理器async with session.get(url) as response:# 使用会话对象发送 GET 请求并在响应对象上使用上下文管理器return await response.text()# 返回响应内容的文本格式async def parse_github_trending(html):# 定义一个异步函数 parse_github_trending用于解析 GitHub Trending 页面的 HTML 内容soup BeautifulSoup(html, html.parser)# 使用 BeautifulSoup 解析 HTML 内容生成一个 BeautifulSoup 对象repositories []# 创建一个空列表用于存储解析后的仓库信息for article in soup.select(article.Box-row):# 使用 CSS 选择器选择所有类名为 Box-row 的 article 标签repo_info {}# 创建一个空字典用于存储单个仓库的信息repo_info[name] article.select_one(h2 a).text.strip()# 获取仓库名称使用 CSS 选择器选择 h2 标签下的 a 标签并提取其文本内容repo_info[url] article.select_one(h2 a)[href].strip()# 获取仓库 URL使用 CSS 选择器选择 h2 标签下的 a 标签并提取其 href 属性值# Descriptiondescription_element article.select_one(p)# 使用 CSS 选择器选择 p 标签repo_info[description] description_element.text.strip() if description_element else None# 获取仓库描述如果存在 p 标签则提取其文本内容否则为 None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])# 使用 CSS 选择器选择 itemprop 属性为 programmingLanguage 的 span 标签repo_info[language] language_element.text.strip() if language_element else None# 获取仓库语言如果存在该标签则提取其文本内容否则为 None# Stars and Forksstars_element article.select(a.Link--muted)[0]# 使用 CSS 选择器选择类名为 Link--muted 的 a 标签并选择第一个元素forks_element article.select(a.Link--muted)[1]# 使用 CSS 选择器选择类名为 Link--muted 的 a 标签并选择第二个元素repo_info[stars] stars_element.text.strip()# 获取仓库星标数提取第一个元素的文本内容repo_info[forks] forks_element.text.strip()# 获取仓库分支数提取第二个元素的文本内容# weeks Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)# 使用 CSS 选择器选择类名为 d-inline-block 和 float-sm-right 的 span 标签repo_info[week_stars] today_stars_element.text.strip() if today_stars_element else None# 获取本周星标数如果存在该标签则提取其文本内容否则为 Nonerepositories.append(repo_info)# 将仓库信息添加到列表中return repositories# 返回解析后的仓库信息列表 async def main():url https://github.com/trending/python?sinceweeklyhtml await fetch_html(url)repos await parse_github_trending(html)# 格式化输出仓库信息format_str {:5} {:50} {:10} {:10} {:10} {}print(format_str.format(Rank, Name, Language, Stars, Forks, Description))for i, repo in enumerate(repos, start1):print(format_str.format(i, repo[name], repo[language], repo[stars], repo[forks],repo[description]))if __name__ __main__:asyncio.run(main()) 运行python文件输出如下 ⑤ readme文件读取 按照本次学习我们爬取到页面数据即可但是对于一个真正的Github用户我们想要知道的是该仓库的**技术栈**其**解决了哪些任务核心思想是什么以及如何部署和贡献**而这些信息基本都被保存在readme.md文件中,因此作者在这里补充一个获取readme文件的代码 相关配置 1.获取token:进入到Personal Access Tokens (Classic) (github.com)页面然后生成你的Token,需要进行相关配置 2.安装pyGithub仓库 pip install PyGithub 代码实现 import aiohttp # 导入 aiohttp 库用于发送异步 HTTP 请求 import asyncio # 导入 asyncio 库用于处理异步任务 from bs4 import BeautifulSoup # 导入 BeautifulSoup 库用于解析 HTML 文档 from github import Github import os from dotenv import load_dotenv load_dotenv() # 加载我们配置在.env文件中的环境变量async def fetch_html(url):# 定义一个异步函数 fetch_html用于获取 URL 对应的 HTML 内容async with aiohttp.ClientSession() as session:# 使用 aiohttp 创建一个会话上下文管理器async with session.get(url) as response:# 使用会话对象发送 GET 请求并在响应对象上使用上下文管理器return await response.text()# 返回响应内容的文本格式async def parse_github_trending(html):# 定义一个异步函数 parse_github_trending用于解析 GitHub Trending 页面的 HTML 内容soup BeautifulSoup(html, html.parser)# 使用 BeautifulSoup 解析 HTML 内容生成一个 BeautifulSoup 对象repositories []# 创建一个空列表用于存储解析后的仓库信息g Github(os.getenv(GITHUB_ACCESS_TOKEN))# g Github(your_access_token)# 使用您的 GitHub 访问令牌创建一个 Github 对象for article in soup.select(article.Box-row):# 使用 CSS 选择器选择所有类名为 Box-row 的 article 标签repo_info {}# 创建一个空字典用于存储单个仓库的信息repo_info[name] article.select_one(h2 a).text.strip().replace(\n, ).replace( , )# 获取仓库名称使用 CSS 选择器选择 h2 标签下的 a 标签并提取其文本内容print(repo_info[name])repo_info[url] https://github.comarticle.select_one(h2 a)[href].strip()# 获取仓库 URL使用 CSS 选择器选择 h2 标签下的 a 标签并提取其 href 属性值,并拼接成完整的 GitHub 仓库链接# 获取 README.md 文件try:# 尝试获取仓库的 README.md 文件repo g.get_repo(repo_info[name])contents repo.get_contents(README.md, refmain)repo_info[readme] contents.decoded_content.decode()print(repo_info[readme])except Exception as e:# 如果仓库没有 README.md 文件或发生其他错误则打印错误信息并跳过该仓库print(fError getting README.md file for {repo_info[name] }: {e})continue# Descriptiondescription_element article.select_one(p)# 使用 CSS 选择器选择 p 标签repo_info[description] description_element.text.strip() if description_element else None# 获取仓库描述如果存在 p 标签则提取其文本内容否则为 None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])# 使用 CSS 选择器选择 itemprop 属性为 programmingLanguage 的 span 标签repo_info[language] language_element.text.strip() if language_element else None# 获取仓库语言如果存在该标签则提取其文本内容否则为 None# Stars and Forksstars_element article.select(a.Link--muted)[0]# 使用 CSS 选择器选择类名为 Link--muted 的 a 标签并选择第一个元素forks_element article.select(a.Link--muted)[1]# 使用 CSS 选择器选择类名为 Link--muted 的 a 标签并选择第二个元素repo_info[stars] stars_element.text.strip()# 获取仓库星标数提取第一个元素的文本内容repo_info[forks] forks_element.text.strip()# 获取仓库分支数提取第二个元素的文本内容# weeks Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)# 使用 CSS 选择器选择类名为 d-inline-block 和 float-sm-right 的 span 标签repo_info[week_stars] today_stars_element.text.strip() if today_stars_element else None# 获取本周星标数如果存在该标签则提取其文本内容否则为 Nonerepositories.append(repo_info)# 将仓库信息添加到列表中return repositories# 返回解析后的仓库信息列表async def main():url https://github.com/trending/python?sinceweeklyhtml await fetch_html(url)repos await parse_github_trending(html)# 格式化输出仓库信息format_str {:5} {:50} {:10} {:10} {:10} {}print(format_str.format(Rank, Name, Language, Stars, Forks, Description))for i, repo in enumerate(repos, start1):print(format_str.format(i, repo[name], repo[language], repo[stars], repo[forks], repo[description]))if __name__ __main__:asyncio.run(main()) 运行结果如下 可以看到我们已经成功获得了目标页面readme的数据 现在我们再稍微对其进行完善修改将其修改为一个crawlAction类以便我们的Agent使用代码如下 import aiohttp from bs4 import BeautifulSoup import asyncio # 导入 asyncio 库用于处理异步任务 from github import Github from metagpt.actions.action import Action from metagpt.config import CONFIG# 定义一个爬虫动作类继承自Action类 class CrawlAction(Action):async def run(self, url: str https://github.com/trending/python?sinceweekly):# 定义一个异步方法run用于执行爬虫动作# URL默认为GitHub上按星标排名的Python项目周榜单async with aiohttp.ClientSession() as client:# 使用aiohttp创建一个会话对象clientasync with client.get(url) as response:# 使用client对象发送GET请求并使用代理配置response.raise_for_status()# 如果响应状态码不是200则raise_for_status()会抛出HTTPError异常html await response.text()# 获取响应的HTML文本soup BeautifulSoup(html, html.parser)# 使用BeautifulSoup解析HTML文本生成一个BeautifulSoup对象repositories []# 创建一个空列表用于存储爬取到的仓库信息g Github(os.getenv(GITHUB_ACCESS_TOKEN))# 使用您的 GitHub 访问令牌创建一个 Github 对象for article in soup.select(article.Box-row):# 使用 CSS 选择器选择所有类名为 Box-row 的 article 标签repo_info {}# 创建一个空字典用于存储单个仓库的信息repo_info[name] article.select_one(h2 a).text.strip().replace(\n, ).replace( , )# 获取仓库名称使用 CSS 选择器选择 h2 标签下的 a 标签并提取其文本内容print(repo_info[name])repo_info[url] https://github.comarticle.select_one(h2 a)[href].strip()# 获取仓库 URL使用 CSS 选择器选择 h2 标签下的 a 标签并提取其 href 属性值,并拼接成完整的 GitHub 仓库链接# 获取 README.md 文件try:# 尝试获取仓库的 README.md 文件repo g.get_repo(repo_info[name])contents repo.get_contents(README.md, refmain)repo_info[readme] contents.decoded_content.decode()print(repo_info[readme])except Exception as e:# 如果仓库没有 README.md 文件或发生其他错误则打印错误信息并跳过该仓库print(fError getting README.md file for {repo_info[name] }: {e})repo_info[readme] Nonecontinue# Descriptiondescription_element article.select_one(p)repo_info[description] description_element.text.strip() if description_element else None# 获取仓库描述信息如果不存在则设置为None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] language_element.text.strip() if language_element else None# 获取仓库使用的编程语言如果不存在则设置为None# Stars and Forksstars_element article.select(a.Link--muted)[0]forks_element article.select(a.Link--muted)[1]repo_info[stars] stars_element.text.strip()repo_info[forks] forks_element.text.strip()# 获取仓库的星标数和分支数# weeks Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[week_stars] today_stars_element.text.strip() if today_stars_element else None# 获取仓库在本周新增的星标数如果不存在则设置为Noneif repo_info.get(readme) ! None:repositories.append(repo_info)# 将仓库信息添加到列表中return json.dumps(repositories)# 使用json.dumps将字典转换为JSON字符串并存储到字符串中 返回爬取到的仓库信息列表 至此Crawl Action成功实现,相比于原项目我们这里返回的是每个仓库信息的列表 作者开发的时候发现metagpt似乎会将action链中每个action的返回转为字符串格式例如上述repositories是列表到导致下一步处理困难因此我这里提前打包为字符串再下一步中解析可以避免这个问题 (2)AnalysisOSSRepository Action实现 当我们获取到Github订阅页面的数据后我们可以让LLM将我们上文中爬取的页面数据和对应仓库的readme信息进行分析了; 这个Action实现起来比较简单之前的文章中我们都已经撰写过类似的Action因此我们这里说一说思路后直接给出完整代码我们的思路是 对单个仓库信息为元素遍历整个仓库列表每个仓库我们要将我们仓库的信息列为prompt模版字符串 设定LLM的处理逻辑和输出格式,如关注重点与输出markdown格式 从LLM的输出中提取出改仓库的名称链接作用技术栈实现思路部署方式。相关链接等这些信息 完整代码如下: from typing import Any from metagpt.actions.action import Action# Actions 的实现 class AnalysisOSSRepository(Action):def prompt_format(self,repo_info):question # 需求 您是一名 GitHub 仓库分析师旨在为用户提供有见地的、个性化的仓库分析。根据上下文填写以下缺失的信息生成吸引人并有信息量的标题确保用户发现与其兴趣相符的仓库。关于仓库的标题 仓库分析深入探索 xxx项目的特点和优势基于基本内容如网页链接了解其背后的作用技术栈实现思路部署方式等信息。 --- 格式示例 项目名称 项目地址 xxx 仓库介绍 xxx 是一个用于 xxx 的开源项目。它使用 xxx 技术栈实现采用 xxx 的实现思路。 特点和优势 特点1特点2 部署和使用 可以通过 部署方式如Docker,本地部署云服务器 的方式部署和使用该项目。详细信息可以参考以下链接 执行代码 pip install … --- 当前已有信息如下: 项目名称:{repository_name} 项目地址:{repository_URL} 项目Star:{repository_star} 项目Fork:{repository_fork} 项目语言{repository_language} 项目readme:{repository_readme} .format(repository_namerepo_info[name], repository_URLrepo_info[url], repository_starrepo_info[stars], repository_forkrepo_info[forks], repository_languagerepo_info[language], repository_readmerepo_info[readme])return questionasync def run(self, repo_info_list: Any):repo_summary_list []for repo_info in json.loads(repo_info_list):repository_info self.prompt_format(repo_info)summary await self._aask(repository_info)repo_summary_list.append(summary)return repo_summary_list 提示词工程在这里很重要 (3)OSSWatchAgent实现 我们已经实现了两个主要的Action,我现在我只需要设计好我们的OSS订阅Agent即可代码如下 import asyncio import os from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional, List from github import Github import aiohttp import discord from aiocron import crontab from bs4 import BeautifulSoup from pydantic import BaseModel, Field from pytz import BaseTzInfo import jsonfrom metagpt.actions.action import Action from metagpt.logs import logger from metagpt.roles import Role from metagpt.schema import Message# 修复 SubscriptionRunner 未完全定义的问题 from metagpt.environment import Environment as _ # noqa: F401 from dotenv import load_dotenv load_dotenv() # 加载我们配置在.env文件中的环境变量# 定义一个爬虫动作类继承自Action类 class CrawlAction(Action):async def run(self, url: str https://github.com/trending/python?sinceweekly):# 定义一个异步方法run用于执行爬虫动作# URL默认为GitHub上按星标排名的Python项目周榜单async with aiohttp.ClientSession() as client:# 使用aiohttp创建一个会话对象clientasync with client.get(url) as response:# 使用client对象发送GET请求并使用代理配置response.raise_for_status()# 如果响应状态码不是200则raise_for_status()会抛出HTTPError异常html await response.text()# 获取响应的HTML文本soup BeautifulSoup(html, html.parser)# 使用BeautifulSoup解析HTML文本生成一个BeautifulSoup对象repositories []# 创建一个空列表用于存储爬取到的仓库信息g Github(os.getenv(GITHUB_ACCESS_TOKEN))# 使用您的 GitHub 访问令牌创建一个 Github 对象for article in soup.select(article.Box-row):# 使用 CSS 选择器选择所有类名为 Box-row 的 article 标签repo_info {}# 创建一个空字典用于存储单个仓库的信息repo_info[name] article.select_one(h2 a).text.strip().replace(\n, ).replace( , )# 获取仓库名称使用 CSS 选择器选择 h2 标签下的 a 标签并提取其文本内容print(repo_info[name])repo_info[url] https://github.comarticle.select_one(h2 a)[href].strip()# 获取仓库 URL使用 CSS 选择器选择 h2 标签下的 a 标签并提取其 href 属性值,并拼接成完整的 GitHub 仓库链接# 获取 README.md 文件try:# 尝试获取仓库的 README.md 文件repo g.get_repo(repo_info[name])contents repo.get_contents(README.md, refmain)repo_info[readme] contents.decoded_content.decode()print(repo_info[readme])except Exception as e:# 如果仓库没有 README.md 文件或发生其他错误则打印错误信息并跳过该仓库print(fError getting README.md file for {repo_info[name] }: {e})repo_info[readme] Nonecontinue# Descriptiondescription_element article.select_one(p)repo_info[description] description_element.text.strip() if description_element else None# 获取仓库描述信息如果不存在则设置为None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] language_element.text.strip() if language_element else None# 获取仓库使用的编程语言如果不存在则设置为None# Stars and Forksstars_element article.select(a.Link--muted)[0]forks_element article.select(a.Link--muted)[1]repo_info[stars] stars_element.text.strip()repo_info[forks] forks_element.text.strip()# 获取仓库的星标数和分支数# weeks Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[week_stars] today_stars_element.text.strip() if today_stars_element else None# 获取仓库在本周新增的星标数如果不存在则设置为Noneif repo_info.get(readme) ! None:repositories.append(repo_info)# 将仓库信息添加到列表中return json.dumps(repositories)# 使用json.dumps将字典转换为JSON字符串并存储到字符串中 返回爬取到的仓库信息列表# Actions 的实现 class AnalysisOSSRepository(Action):def prompt_format(self,repo_info):question # 需求 您是一名 GitHub 仓库分析师旨在为用户提供有见地的、个性化的仓库分析。根据上下文填写以下缺失的信息生成吸引人并有信息量的标题确保用户发现与其兴趣相符的仓库。关于仓库的标题 仓库分析深入探索 xxx项目的特点和优势基于基本内容如网页链接了解其背后的作用技术栈实现思路部署方式等信息。 --- 格式示例 项目名称 项目地址 xxx 仓库介绍 xxx 是一个用于 xxx 的开源项目。它使用 xxx 技术栈实现采用 xxx 的实现思路。 特点和优势 特点1特点2 部署和使用 可以通过 部署方式如Docker,本地部署云服务器 的方式部署和使用该项目。详细信息可以参考以下链接 执行代码 pip install … --- 当前已有信息如下: 项目名称:{repository_name} 项目地址:{repository_URL} 项目Star:{repository_star} 项目Fork:{repository_fork} 项目语言{repository_language} 项目readme:{repository_readme} .format(repository_namerepo_info[name], repository_URLrepo_info[url], repository_starrepo_info[stars], repository_forkrepo_info[forks], repository_languagerepo_info[language], repository_readmerepo_info[readme])return questionasync def run(self, repo_info_list: Any):repo_summary_list []for repo_info in json.loads(repo_info_list):repository_info self.prompt_format(repo_info)summary await self._aask(repository_info)repo_summary_list.append(summary)return repo_summary_list# 角色设计 class OssWatcher(Role):def __init__(self):super().__init__(namecheems,profileOssWatcher,goal根据我提供给你的资料生成一个有见地的 GitHub 仓库 分析报告。,constraints仅基于提供的 GitHub 仓库 数据进行分析。,)self.set_actions([CrawlAction(), AnalysisOSSRepository()])self._set_react_mode(react_modeby_order)async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})todo self.rc.todomsg self.get_memories(k1)[0]result await todo.run(msg.content)new_msg Message(contentstr(result), roleself.profile, cause_bytype(todo))self.rc.memory.add(new_msg)return result 这里作者并没有将CrawlAction返回的是列表不是字符串因此我们循环遍历的所有仓库因此执行完任务后我们不返回字符串信息而是返回列表给回调函数callback (4)定时触发器Trigger的实现 这里主要有两种实现方法 ①使用 asyncio.sleep 实现简单的定时功能 import asyncio import time from datetime import datetime, timedelta from metagpt.schema import Message from pydantic import BaseModel, Field# 定义一个包含URL和时间戳的基础信息模型 class OssInfo(BaseModel):url: strtimestamp: float Field(default_factorytime.time)# 定义一个异步函数用于定时生成消息 async def oss_trigger(hour: int, minute: int, second: int 0, url: str https://github.com/trending):while True:now datetime.now()next_time datetime(now.year, now.month, now.day, hour, minute, second)if next_time now:next_time timedelta(days1)wait_seconds (next_time - now).total_seconds()print(f等待时间{wait_seconds}秒)await asyncio.sleep(wait_seconds)yield Message(url, OssInfo(urlurl)) 在这个例子中我们使用了 asyncio.sleep 来实现定时功能。yield 语句用于在指定时间生成消息。通过添加时间戳我们确保每次生成的消息都是唯一的从而避免了消息去重的问题。 ②使用 aiocron 实现更灵活的定时功能 import time from aiocron import crontab from typing import Optional from pytz import BaseTzInfo from pydantic import BaseModel, Field from metagpt.schema import Message# 定义一个基于aiocron的定时触发器类 class GithubTrendingCronTrigger:def __init__(self, spec: str, tz: Optional[BaseTzInfo] None, url: str https://github.com/trending):self.crontab crontab(spec, tztz) # 使用cron表达式和时区初始化crontabself.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next() # 等待下一个触发时刻return Message(contentself.url) # 生成消息# 使用UTC时间10:00 AM作为触发时间 cron_trigger GithubTrendingCronTrigger(0 10 * * *) aiocron 提供了一个强大的定时功能可以使用 cron 语法灵活地配置定时规则。这使得代码更简洁功能更全面。 第二种方法更加灵活因此我们这里选择第二种方法完整代码如下 from pytz import timezone from aiocron import crontab from typing import Optional from pytz import BaseTzInfo from pydantic import BaseModel, Field from metagpt.schema import Message# 定义一个基于aiocron的定时触发器类 class GithubTrendingCronTrigger:def __init__(self, spec: str, tz: Optional[BaseTzInfo] None, url: str https://github.com/trending):self.crontab crontab(spec, tztz) # 使用cron表达式和时区初始化crontabself.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next() # 等待下一个触发时刻return Message(contentself.url) # 生成消息# 获取北京时间的时区 beijing_tz timezone(Asia/Shanghai)# 创建定时触发器实例每天北京时间上午9:00触发 cron_trigger GithubTrendingCronTrigger(0 9 * * *, tzbeijing_tz) 在这个例子中我们设置了北京时间上午9:00作为触发时间 ​ (5)Callback设计 callback也就是回调函数当我们爬取到数据并且处理得到结果后我们该怎么做呢我们应该从应用角度出发我们是为了订阅信息那么我们就需要一个平台来发布这些信息下面我们以Discord为例实现Agent自动发布订阅信息 ①Discord配置 Discord[[Discord | 玩耍聊天的地方] 是一个免费的语音和文本聊天平台专门为游戏玩家和社区设计但现在已经发展成为一个更广泛的社区平台。Discord 支持实时语音和视频通话、屏幕分享、文本消息、图片和视频分享等功能。用户可以创建自己的服务器或者加入其他人创建的服务器与其他用户进行交流和讨论。很多开源项目都在Discord上有自己的社区其具有以下特点 高质量的语音和视频通话强大的社区管理工具角色授权、频道分类、_自动化机器人_等丰富的扩展和插件跨平台支持 因此我们选择在该平台上构建一个自动化机器人后端接受我们的Agent服务然后通过调用Discord的API服务来实现信息订阅 注意注册登录Discord并使用本服务需要用户具有科学网络环境如果没有则可以直接跳过进入到下一节考虑邮箱订阅实现 ②构架机器人 下面作者订阅的完整流程参考文章[创建机器人帐户 进入到页面后配置应用的基本信息 接着按照如上内容进行配置注意你的token不要和他人分享如果不小心泄漏或者没有配置Token应该在此界面点击reset Token重新设置现在我们就有了一个机器人帐户可以使用该令牌token登录。 接下来邀请机器人到个人服务器 进入到OAuth界面勾选Bot也就是当前Token选择的服务其他不要勾选) 并且选择该机器人拥有的权限 现在这里生成的 URL 可用于将机器人添加到服务器。将复制的 URL 复制并粘贴到浏览器中选择要邀请机器人访问的服务器然后单击“授权”。 可以看到我们的机器人已经被邀请进入我们的服务器 开始撰写代码 import asyncio import discord# 基于discord机器人发布信息函数 async def send_discord_msg(channel_id: int, msg: str, token: str):intents discord.Intents.default()intents.message_content Trueclient discord.Client(intentsintents)async with client:await client.login(token)channel await client.fetch_channel(channel_id)await channel.send(msg) 因为discord有单条信息的大小限制过长的内容会导致发送不成功并且Agent给我们返回的是一个仓库总结列表因此我们可以遍历该列表作为多条msg发送最终实现的discord_callback函数如下 import asyncio import discordfrom metagpt.config import CONFIGasync def discord_callback(msg_list:List):intents discord.Intents.default()intents.message_content Trueclient discord.Client(intentsintents, proxyCONFIG.global_proxy)token os.environ[DISCORD_TOKEN]channel_id int(os.environ[DISCORD_CHANNEL_ID])async with client:await client.login(token)channel await client.fetch_channel(channel_id)for repo in msg_list:lines []for i in repo.splitlines():if i.startswith((# , ## , ### )):if lines:await channel.send(\n.join(lines))lines []lines.append(i)if lines:await channel.send(\n.join(lines)) DISCORD_TOKEN我们上面截图中有就是我们之前复制的reset token DISCORD_CHANNEL_ID进入到我们选择的服务器后搜索栏URL最有一部分如下图所示框选部分 这里我们再次回顾我们的完整步骤 初始化和配置确保依赖库已安装并正确配置。爬虫动作爬取 GitHub Trending 数据。分析动作对爬取的数据进行分析然后生成总结报告。自动化定时任务设置定时任务每天定时执行爬虫和分析动作。循环处理确保对每个爬取的仓库进行处理并生成报告。 现在我们将所有阶段的代码打包起来得到一个完整的代码文件这里因为测试需要我们将时间每分钟运行一次运行前可以先安装相关库: pip install asyncio aiohttp discord aiocron beautifulsoup4 pydantic pytz PyGithub fire metagpt 完整运行代码如下: import asyncio import os from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional, List from github import Github import aiohttp import discord from aiocron import crontab from bs4 import BeautifulSoup from pydantic import BaseModel, Field from pytz import BaseTzInfo import jsonfrom metagpt.actions.action import Action from metagpt.logs import logger from metagpt.roles import Role from metagpt.schema import Message# 修复 SubscriptionRunner 未完全定义的问题 from metagpt.environment import Environment as _ # noqa: F401 from dotenv import load_dotenv load_dotenv() # 加载我们配置在.env文件中的环境变量# 订阅模块 class SubscriptionRunner(BaseModel):tasks: Dict[Role, asyncio.Task] Field(default_factorydict)class Config:arbitrary_types_allowed Trueasync def subscribe(self,role: Role,trigger: AsyncGenerator[Message, None],callback: Callable[[Message], Awaitable[None]],):# 异步订阅方法loop asyncio.get_running_loop()async def _start_role():async for msg in trigger:resp await role.run(msg)await callback(resp)self.tasks[role] loop.create_task(_start_role(), namefSubscription-{role})async def unsubscribe(self, role: Role):取消订阅角色并取消关联的任务task self.tasks.pop(role)task.cancel()async def run(self, raise_exception: bool True):while True:for role, task in self.tasks.items():if task.done():if task.exception():if raise_exception:raise task.exception()logger.opt(exceptiontask.exception()).error(fTask {task.get_name()} run error)else:logger.warning(fTask {task.get_name()} has completed. If this is unexpected behavior, please check the trigger function.)self.tasks.pop(role)breakelse:await asyncio.sleep(1)# 定义一个爬虫动作类继承自Action类 class CrawlAction(Action):async def run(self, url: str https://github.com/trending/python?sinceweekly):# 定义一个异步方法run用于执行爬虫动作# URL默认为GitHub上按星标排名的Python项目周榜单async with aiohttp.ClientSession() as client:# 使用aiohttp创建一个会话对象clientasync with client.get(url) as response:# 使用client对象发送GET请求并使用代理配置response.raise_for_status()# 如果响应状态码不是200则raise_for_status()会抛出HTTPError异常html await response.text()# 获取响应的HTML文本soup BeautifulSoup(html, html.parser)# 使用BeautifulSoup解析HTML文本生成一个BeautifulSoup对象repositories []# 创建一个空列表用于存储爬取到的仓库信息g Github(os.getenv(GITHUB_ACCESS_TOKEN))# 使用您的 GitHub 访问令牌创建一个 Github 对象for article in soup.select(article.Box-row):# 使用 CSS 选择器选择所有类名为 Box-row 的 article 标签repo_info {}# 创建一个空字典用于存储单个仓库的信息repo_info[name] article.select_one(h2 a).text.strip().replace(\n, ).replace( , )# 获取仓库名称使用 CSS 选择器选择 h2 标签下的 a 标签并提取其文本内容repo_info[url] https://github.comarticle.select_one(h2 a)[href].strip()# 获取仓库 URL使用 CSS 选择器选择 h2 标签下的 a 标签并提取其 href 属性值,并拼接成完整的 GitHub 仓库链接# 获取 README.md 文件try:# 尝试获取仓库的 README.md 文件repo g.get_repo(repo_info[name])contents repo.get_contents(README.md, refmain)repo_info[readme] contents.decoded_content.decode()except Exception as e:# 如果仓库没有 README.md 文件或发生其他错误则打印错误信息并跳过该仓库print(fError getting README.md file for {repo_info[name] }: {e})repo_info[readme] Nonecontinue# Descriptiondescription_element article.select_one(p)repo_info[description] description_element.text.strip() if description_element else None# 获取仓库描述信息如果不存在则设置为None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] language_element.text.strip() if language_element else None# 获取仓库使用的编程语言如果不存在则设置为None# Stars and Forksstars_element article.select(a.Link--muted)[0]forks_element article.select(a.Link--muted)[1]repo_info[stars] stars_element.text.strip()repo_info[forks] forks_element.text.strip()# 获取仓库的星标数和分支数# weeks Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[week_stars] today_stars_element.text.strip() if today_stars_element else None# 获取仓库在本周新增的星标数如果不存在则设置为Noneif repo_info.get(readme) ! None:repositories.append(repo_info)# 将仓库信息添加到列表中return json.dumps(repositories[:2])# 使用json.dumps将字典转换为JSON字符串并存储到字符串中 返回爬取到的仓库信息列表# Actions 的实现 class AnalysisOSSRepository(Action):def prompt_format(self,repo_info):question # 需求 您是一名 GitHub 仓库分析师旨在为用户提供有见地的、个性化的仓库分析。根据上下文填写以下缺失的信息生成吸引人并有信息量的标题确保用户发现与其兴趣相符的仓库。关于仓库的标题 仓库分析深入探索 xxx项目的特点和优势基于基本内容如网页链接了解其背后的作用技术栈实现思路部署方式等信息。 --- 格式示例 项目名称 项目地址 xxx 仓库介绍 xxx 是一个用于 xxx 的开源项目。它使用 xxx 技术栈实现采用 xxx 的实现思路。 特点和优势 特点1特点2 部署和使用 可以通过 部署方式如Docker,本地部署云服务器 的方式部署和使用该项目。详细信息可以参考以下链接 执行代码 pip install … --- 当前已有信息如下: 项目名称:{repository_name} 项目地址:{repository_URL} 项目Star:{repository_star} 项目Fork:{repository_fork} 项目语言{repository_language} 项目readme:{repository_readme} .format(repository_namerepo_info[name], repository_URLrepo_info[url], repository_starrepo_info[stars], repository_forkrepo_info[forks], repository_languagerepo_info[language], repository_readmerepo_info[readme])return questionasync def run(self, repo_info_list: Any):repo_summary_list []for repo_info in json.loads(repo_info_list):repository_info self.prompt_format(repo_info)summary await self._aask(repository_info)repo_summary_list.append(summary)return repo_summary_list# 角色设计 class OssWatcher(Role):def __init__(self):super().__init__(namecheems,profileOssWatcher,goal根据我提供给你的资料生成一个有见地的 GitHub 仓库 分析报告。,constraints仅基于提供的 GitHub 仓库 数据进行分析。,)self.set_actions([CrawlAction(), AnalysisOSSRepository()])self._set_react_mode(react_modeby_order)async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})todo self.rc.todomsg self.get_memories(k1)[0]result await todo.run(msg.content)new_msg Message(contentstr(result), roleself.profile, cause_bytype(todo))self.rc.memory.add(new_msg)return result# 定义一个基于aiocron的定时触发器类 class GithubTrendingCronTrigger:def __init__(self, spec: str, tz: Optional[BaseTzInfo] None, url: str https://github.com/trending/python?sinceweekly) - None:self.crontab crontab(spec, tztz)self.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next()return Message(contentself.url)# discord 回调 async def discord_callback(msg_list: List):intents discord.Intents.default()intents.message_content True# client discord.Client(intentsintents, proxy os.getenv(GLOBAL_PROXYGLOBAL_PROXY)) # 需要代理client discord.Client(intentsintents) # 无需代理token os.getenv(DISCORD_TOKEN)channel_id int(os.getenv(DISCORD_CHANNEL_ID))async with client:await client.login(token)channel await client.fetch_channel(channel_id)for repo in msg_list:lines []for line in repo.splitlines():if line.startswith((# , ## , ### )):if lines:await channel.send(\n.join(lines))lines []lines.append(line)if lines:await channel.send(\n.join(lines))# 运行入口 async def main(spec: str * * * * *, discord: bool True, wxpusher: bool True):callbacks []if discord:callbacks.append(discord_callback)if not callbacks:async def _print(msg: Message):print(msg.content)callbacks.append(_print)async def callback(msg):await asyncio.gather(*(call(msg) for call in callbacks))runner SubscriptionRunner()await runner.subscribe(OssWatcher(), GithubTrendingCronTrigger(spec), callback) # 正式版本await runner.run()if __name__ __main__:import firefire.Fire(main) # 使用 fire 库将 main 函数转换为命令行接口的入口点 当前程序运行后每分钟会将爬虫处理的信息发送到Discord如果需要测试将**main函数中的spec值修改为* * * * ,则会每分钟执行一次* 运行该函数代码效果如下: 费劲千辛万苦我们终于实现了Discord订阅功能给作者点个免费的赞鼓励一下吧 3.基于QQ邮箱实现Huggface论文总结订阅Agent 经过这个OSS订阅项目的完整流程我们学习了MetaGPT订阅模块的使用也了解了Python爬虫的基本知识大家可以将这些模块进行任意组合以及自定义来实现不同的功能下面在作者爬取Huggface论文页面内容(前5篇)并且总结为文档并整理发送到我们QQ邮箱的完整代码完整代码如下: 下面是修改后的完整代码包含爬取Huggingface Papers页面的Action修改后的模板内容以及通过QQ邮箱发送订阅的功能。运行此脚本前请 确保配置好.env文件以下是配置的环境变量内容 GITHUB_ACCESS_TOKENyour_github_access_token DISCORD_TOKENyour_discord_token DISCORD_CHANNEL_IDyour_discord_channel_id QQ_EMAIL_USERyour_qq_email QQ_EMAIL_PASSWORDyour_qq_email_password QQ_EMAIL_TOrecipient_email 确保你已经安装了相关的Python包 pip install aiohttp beautifulsoup4 pydantic python-dotenv discord.py aiocron smtplib fire 修改后的代码如下 import asyncio import os import smtplib from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional, List from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from github import Github import aiohttp import discord from aiocron import crontab from bs4 import BeautifulSoup from pydantic import BaseModel, Field from pytz import BaseTzInfo import jsonfrom metagpt.actions.action import Action from metagpt.logs import logger from metagpt.roles import Role from metagpt.schema import Message# 加载环境变量 from dotenv import load_dotenv load_dotenv()# 订阅模块 class SubscriptionRunner(BaseModel):tasks: Dict[Role, asyncio.Task] Field(default_factorydict)class Config:arbitrary_types_allowed Trueasync def subscribe(self,role: Role,trigger: AsyncGenerator[Message, None],callback: Callable[[Message], Awaitable[None]],):loop asyncio.get_running_loop()async def _start_role():async for msg in trigger:resp await role.run(msg)await callback(resp)self.tasks[role] loop.create_task(_start_role(), namefSubscription-{role})async def unsubscribe(self, role: Role):task self.tasks.pop(role)task.cancel()async def run(self, raise_exception: bool True):while True:for role, task in self.tasks.items():if task.done():if task.exception():if raise_exception:raise task.exception()logger.opt(exceptiontask.exception()).error(fTask {task.get_name()} run error)else:logger.warning(fTask {task.get_name()} has completed. If this is unexpected behavior, please check the trigger function.)self.tasks.pop(role)breakelse:await asyncio.sleep(1)# 定义一个爬虫动作类继承自Action类 class CrawlAction(Action):async def run(self, url: str https://huggingface.co/papers):async with aiohttp.ClientSession() as client:papers await self._fetch_papers(url, client)return json.dumps(papers[:5]) # 获取前5篇Paper信息async def _fetch_papers(self, url: str, client: aiohttp.ClientSession):async with client.get(url) as response:response.raise_for_status()html await response.text()soup BeautifulSoup(html, html.parser)papers []# 只爬取前5页for article in soup.select(h3 a[href^/papers/])[:5]:paper_info {}paper_info[title] article.text.strip()print(article.text.strip())paper_info[url] https://huggingface.co article[href].strip()paper_html await self._fetch_paper_detail(paper_info[url], client)paper_soup BeautifulSoup(paper_html, html.parser)paper_info[abstract] paper_soup.find(section).get_text(separator , stripTrue).strip()papers.append(paper_info)return papersasync def _fetch_paper_detail(self, url: str, client: aiohttp.ClientSession) - str:async with client.get(url) as response:response.raise_for_status()return await response.text()# Actions 的实现 class AnalysisPaper(Action):def prompt_format(self, paper_info):question # 需求 您是一名学术论文分析师旨在为用户提供有见地的、个性化的论文分析。根据上下文填写以下缺失的信息生成吸引人并有信息量的标题确保用户发现与其兴趣相符的论文。关于论文的标题 论文分析深入探索 xxx论文的特点和贡献基于基本内容如网页链接了解其背后的研究背景研究方法实验结果等信息。 --- 格式示例 # 论文标题## 论文链接xxx## 论文摘要xxx## 研究背景xxx## 研究方法xxx## 实验结果xxx## 结论xxx --- 当前已有信息如下: 论文标题:{paper_title} 论文链接:{paper_URL} 论文摘要:{paper_abstract} .format(paper_titlepaper_info[title], paper_URLpaper_info[url], paper_abstractpaper_info[abstract])return questionasync def run(self, paper_info_list: Any):paper_summary_list []for paper_info in json.loads(paper_info_list):paper_info_str self.prompt_format(paper_info)summary await self._aask(paper_info_str)paper_summary_list.append(summary)return paper_summary_list# 角色设计 class PaperWatcher(Role):def __init__(self):super().__init__(namecheems,profilePaperWatcher,goal根据我提供给你的资料生成一个有见地的学术论文分析报告。,constraints仅基于提供的论文数据进行分析。,)self.set_actions([CrawlAction(), AnalysisPaper()])self._set_react_mode(react_modeby_order)async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})todo self.rc.todomsg self.get_memories(k1)[0]result await todo.run(msg.content)new_msg Message(contentstr(result), roleself.profile, cause_bytype(todo))self.rc.memory.add(new_msg)return result# 定义一个基于aiocron的定时触发器类 class HuggingfacePapersCronTrigger:def __init__(self, spec: str, tz: Optional[BaseTzInfo] None, url: str https://huggingface.co/papers) - None:self.crontab crontab(spec, tztz)self.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next()return Message(contentself.url)# 邮件回调 async def email_callback(paper_summary_list: List):gmail_user os.getenv(QQ_EMAIL_USER)gmail_password os.getenv(QQ_EMAIL_PASSWORD)to os.getenv(QQ_EMAIL_TO)subject Huggingface Papers Weekly Digestbody \n\n.join(paper_summary_list)msg MIMEMultipart()msg[Subject] subjectmsg[From] gmail_usermsg[To] tomsg.attach(MIMEText(body, plain))server smtplib.SMTP(smtp.qq.com, 587)server.starttls()server.login(gmail_user, gmail_password)server.send_message(msg)server.quit()print(send success!)# 运行入口 async def main(spec: str * * * * *, email: bool True):callbacks []if email:callbacks.append(email_callback)if not callbacks:async def _print(msg: Message):print(msg.content)callbacks.append(_print)async def callback(msg):await asyncio.gather(*(call(msg) for call in callbacks))runner SubscriptionRunner()await runner.subscribe(PaperWatcher(), HuggingfacePapersCronTrigger(spec), callback)await runner.run()if __name__ __main__:import firefire.Fire(main) 下面是作者的设计思路 爬取ActionCrawlAction类用于爬取Huggingface Papers页面的内容包括每篇论文的标题和摘要这里页面分析方式与之前相同。分析ActionAnalysisPaper类用于格式化爬取的论文信息并生成分析报告。角色设计PaperWatcher角色包含了上述两个Action负责爬取和分析论文。定时触发器HuggingfacePapersCronTrigger类用于定时触发爬取任务。邮件回调email_callback函数用于通过QQ邮箱发送邮件包含爬取和分析的结果。 大家可以cmd启动脚本测试 python main.py 运行效果如下 如何系统的去学习大模型LLM 作为一名热心肠的互联网老兵我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑所以在工作繁忙的情况下还是坚持各种整理和分享。 但苦于知识传播途径有限很多互联网行业朋友无法获得正确的资料得到学习提升故此将并将重要的 AI大模型资料 包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。 有需要的小伙伴可以V扫描下方二维码免费领取 一、全套AGI大模型学习路线 AI大模型时代的学习之旅从基础到前沿掌握人工智能的核心技能 二、640套AI大模型报告合集 这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示。 三、AI大模型经典PDF籍 随着人工智能技术的飞速发展AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型如GPT-3、BERT、XLNet等以其强大的语言理解和生成能力正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。 四、AI大模型商业化落地方案 阶段1AI大模型时代的基础理解 目标了解AI大模型的基本概念、发展历程和核心原理。内容 L1.1 人工智能简述与大模型起源L1.2 大模型与通用人工智能L1.3 GPT模型的发展历程L1.4 模型工程L1.4.1 知识大模型L1.4.2 生产大模型L1.4.3 模型工程方法论L1.4.4 模型工程实践L1.5 GPT应用案例 阶段2AI大模型API应用开发工程 目标掌握AI大模型API的使用和开发以及相关的编程技能。内容 L2.1 API接口L2.1.1 OpenAI API接口L2.1.2 Python接口接入L2.1.3 BOT工具类框架L2.1.4 代码示例L2.2 Prompt框架L2.2.1 什么是PromptL2.2.2 Prompt框架应用现状L2.2.3 基于GPTAS的Prompt框架L2.2.4 Prompt框架与ThoughtL2.2.5 Prompt框架与提示词L2.3 流水线工程L2.3.1 流水线工程的概念L2.3.2 流水线工程的优点L2.3.3 流水线工程的应用L2.4 总结与展望 阶段3AI大模型应用架构实践 目标深入理解AI大模型的应用架构并能够进行私有化部署。内容 L3.1 Agent模型框架L3.1.1 Agent模型框架的设计理念L3.1.2 Agent模型框架的核心组件L3.1.3 Agent模型框架的实现细节L3.2 MetaGPTL3.2.1 MetaGPT的基本概念L3.2.2 MetaGPT的工作原理L3.2.3 MetaGPT的应用场景L3.3 ChatGLML3.3.1 ChatGLM的特点L3.3.2 ChatGLM的开发环境L3.3.3 ChatGLM的使用示例L3.4 LLAMAL3.4.1 LLAMA的特点L3.4.2 LLAMA的开发环境L3.4.3 LLAMA的使用示例L3.5 其他大模型介绍 阶段4AI大模型私有化部署 目标掌握多种AI大模型的私有化部署包括多模态和特定领域模型。内容 L4.1 模型私有化部署概述L4.2 模型私有化部署的关键技术L4.3 模型私有化部署的实施步骤L4.4 模型私有化部署的应用场景 学习计划 阶段11-2个月建立AI大模型的基础知识体系。阶段22-3个月专注于API应用开发能力的提升。阶段33-4个月深入实践AI大模型的应用架构和私有化部署。阶段44-5个月专注于高级模型的应用和部署。 这份完整版的大模型 LLM 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】 有需要的小伙伴可以Vx扫描下方二维码免费领取