公司网站建设维护管理办法,给女友惊喜做网站,海外推广面试问题,手机优化电池充电要开吗传送门#xff1a;
《2024.5组队学习——MetaGPT#xff08;0.8.1#xff09;智能体理论与实战#xff08;上#xff09;#xff1a;MetaGPT安装、单智能体开发》《2024.5组队学习——MetaGPT#xff08;0.8.1#xff09;智能体理论与实战#xff08;下#xff09;
《2024.5组队学习——MetaGPT0.8.1智能体理论与实战上MetaGPT安装、单智能体开发》《2024.5组队学习——MetaGPT0.8.1智能体理论与实战下多智能体开发》 文章目录 订阅智能体OSS Open Source Software一、 SubscriptionRunner1.1 源码解析1.2 官方文档示例1.3 Trigger异步生成器1.4 设置aiohttp代理 二、 OSSWatcher Role实现2.1 爬取代码2.2 GitHub Trending总结 三、 OSSWatcher Role实现四、 Trigger实现五、 Callback设计5.1 Discord5.2 Wechat 六、 运行示例七、 开发作业 学习资料项目地址——hugging-multi-agent、在线阅读、MetaGPT项目、MetaGPT中文文档优秀作业链接《MetaGPT环境搭建和尝试》、Destory的《MetaGPT教程笔记》、乐正萌的《MetaGPT教程Task3 》、 GISer Liu的《基于MetaGPT构建单智能体》、《MetaGPT课程3作业》 订阅智能体OSS Open Source Software 当我们持续关注某些感兴趣的信息时Agent可以实时获取这些信息并进行处理然后通过一些如邮件、微信、discord等通知渠道将处理后的信息发送给我们我们将这类Agent称为订阅智能体。此时Agent的Role可称为“资讯订阅员”的其包含的Action则主要有两种
从外界信息源中搜集信息对搜集得到的信息进行总结
一、 SubscriptionRunner
1.1 源码解析 我们还可以为开发一些额外功能比如定时运行和发送通知。在MetaGPT中metagpt.subscription模块提供了SubscriptionRunner类它是一个简单的包装器用于使用asyncio管理不同角色的订阅任务。其主要作用是定时触发运行一个Role然后将运行结果通知给用户。
class SubscriptionRunner(BaseModel):model_config ConfigDict(arbitrary_types_allowedTrue)# tasks字典,用于存储每个角色对应的异步任务tasks: dict[Role, asyncio.Task] Field(default_factorydict)async 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})subscribe 方法用于订阅一个角色传入角色、触发器(trigger)和回调函数。它会创建一个异步任务——从触发器获取消息并将角色处理后的响应传递给回调函数。
async def run(self, raise_exception: bool True):Runs all subscribed tasks and handles their completion or exception.Args:raise_exception: _description_. Defaults to True.Raises:task.exception: _description_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)使用一个无限的 while True 循环来持续检查所有任务的状态。对于 self.tasks 字典中的每个 (role, task) 项: 如果该任务已完成 (task.done() 为真): 检查是否有异常 (task.exception()) 如果有异常,并且 raise_exception 参数为真,则抛出该异常如果有异常,并且 raise_exception 参数为假,则使用 logger 记录错误日志 如果没有异常,则使用 logger 记录警告日志,提示任务已完成(可能是不期望的行为) 从 self.tasks 字典中移除该任务使用 break 语句退出循环,等待下一次循环 如果在当前循环中没有任何任务完成则使用 await asyncio.sleep(1) 等待1秒继续下一次循环。 总的来说run函数的作用是持续监视所有已订阅的任务一旦有任务完成(无论是正常完成还是异常),就进行相应的处理包括抛出异常、记录日志或移除已完成的任务。如果所有任务都还在运行中它会等待一段时间后继续检查。这确保了所有已订阅的任务都可以持续运行直到它们完成为止。
1.2 官方文档示例
import asyncio
from metagpt.subscription import SubscriptionRunner
from metagpt.roles import Searcher
from metagpt.schema import Messageasync def trigger():while True:yield Message(contentthe 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()await main()从例子可以知道订阅智能体的实现主要有3个要素分别是Role、Trigger、Callback即智能体本身、触发器、数据回调。其中trigger 是一个无限循环的异步函数
在循环中,它首先 yield 一个 Message 对象其 content 属性设置为 “the latest news about OpenAI”。使用 await asyncio.sleep(3600 * 24) 暂停一天执行循环以上过程 所以trigger 函数的作用是每隔24小时产生一个包含 “the latest news about OpenAI” 内容的 Message 对象。在 main 函数中trigger() 被传递给 subscribe 方法作为 Searcher 角色的触发器。每当 trigger 生成一个新的 MessageSearcher 角色就会处理该消息并将响应传递给 callback 函数打印出来。 要注意的是我们虽然不对订阅智能体的Role做限制但是不是所有的Role都适合用来做订阅智能体比如MetaGPT软件公司中的几个角色例如产品经理、架构师、工程师等因为当给这些Role一个需求输入时它们的产出往往是类似的并没有定时执行然后发送给我们能的必要。 从应用的角度出发订阅智能体的输出应该具有实时性相同的一个需求描述输入输出的内容一般是会随着时间的变化而不同例如新闻资讯、技术前沿进展、热门的开源项目等。 1.3 Trigger异步生成器 Trigger是个异步生成器Asynchronous Generators。在Python中生成器(Generators)是一种特殊的迭代器区别于list、dict等类型它是在需要时才生成数据而不是一次性把所有数据加载到内存中。这种按需生成的方式可以提高内存使用效率特别是在处理大量数据或无限数据流时非常有用。
def generate_data(n):生成包含n个整数的生成器for i in range(n):yield idef process_data(data):对每个数据进行平方运算for item in data:print(item**2)# 生成一百万个整数的生成器
data_generator generate_data(1000000)# 处理生成器中的数据
process_data(data_generator)在上面的代码中generate_data函数是一个生成器它每次被调用时只生成一个整数而不是一次性将所有数据生成到内存中这样就大大减少了内存使用量。 传统生成器是在同步模式下使用yield关键字来产生值当生成器函数遇到yield语句时,它会暂停执行并将值产出下次再次调用next()方法时,生成器会从上次暂停的地方继续执行。 而异步生成器则是在异步的上下文中工作的。它使用async和await关键字,可以在yield语句处暂停执行并等待一个潜在的耗时异步操作完成后再继续。这使得它特别适合于处理异步数据流,比如从网络套接字接收数据、从异步API获取数据等。
import asyncioasync def counter(start, stop):n startwhile n stop:yield nn 1await asyncio.sleep(0.5) # 模拟异步操作async def main():async for i in counter(3, 8):print(i)asyncio.run(main())# 输出:
# 3
# (暂停0.5秒)
# 4
# (暂停0.5秒)
# 5
# ...在这个异步生成器中每次产出一个数值后,它会通过await asyncio.sleep(0.5)模拟一个耗时0.5秒的异步操作然后再继续执行。这样的异步方式可以避免阻塞主线程提高程序的响应能力。
1.4 设置aiohttp代理 aiohttp是一个用于异步 HTTP 客户端/服务器的非常流行的Python库它基于 asyncio 库构建可以让你方便地编写单线程并发代码。 作为一个异步HTTP客户端aiohttp 允许你发送对服务器的请求而不阻塞事件循环。这意味着你的脚本可以高效地发送多个HTTP请求无需为每个请求启动新线程。这在需要发起大量HTTP请求时特别有用比如爬虫、API客户端等。如果你需要编写高度可扩展、高并发的网络应用程序,使用 aiohttp 将是一个不错的选择。
aiohttp 的一些主要特性包括:
Client/Server模式: 可以作为客户端或服务器使用Web框架集成: 可以与现有的web框架(如Django、Flask等)集成中间件: 支持中间件功能,方便插入自定义行为WebSocket支持: 支持WebSocket协议Gunicorn工作: 可以与Gunicorn集成作为ASGI服务器运行URLs构建: 方便构造查询参数或URL编码 教程涉中需要访问到一些国外的网站可能会遇到网络问题因为 aiohttp 默认不走系统代理所以需要做下代理配置。MetaGPT中已经提供了GLOBAL_PROXY参数用来表示全局代理配置教程中遇到使用aiohttp进行请求的地方都会将代理设置为GLOBAL_PROXY的值所以可以通过在config/key.yaml配置文件中添加自己代理服务器的配置以解决网络问题
GLOBAL_PROXY: http://127.0.0.1:8118 # 改成自己的代理服务器地址我是在AutoDL上跑的项目代理设置可参考帖子《AutoDL 使用代理加速》讲解了如何在一台服务器命令行中启用 clash 代理。
mkdir mihomo
cd mihomo# 下载lash 二进制文件并解压
# 最原始的 clash 项目已经删库这个是目前找到的规模比较大的继任 fork ,二进制文件也更名为 mihomo
wget https://github.com/MetaCubeX/mihomo/releases/download/v1.18.1/mihomo-linux-amd64-v1.18.1.gz
gzip -d mihomo-linux-amd64-v1.18.1.gz# 下载Country.mmdb文件
wget https://github.com/Dreamacro/maxmind-geoip/releases/download/20240512/Country.mmdb
vim config.yaml # 配置config文件# 授予执行权限
chmod x ./mihomo-linux-amd64-v1.18.1config文件需要订阅clash服务商获取比如v2net。 打开Clash Verge v1.3.8下载clash-verge_1.3.8_amd64.AppImage然后运行sudo apt install fuse安装FUSE。之后运行chmod x clash-verge_1.3.8_amd64.AppImage修改此文件权限最后运行./clash-verge_1.3.8_amd64.AppImage直接启动clash-verge软件。然后参考Set Up V2NET on Linux就可以配置正确的config.yaml文章中是apt安装clash-verge_x.x.x_amd64.deb文件但是我都失败了.AppImage文件无需安装可以直接启动。。 最后直接执行
./mihomo-linux-amd64-v1.18.1 -d .
#这里的 -d 选项很重要用于设置工作目录为当前所在目录否则找不到 config.yaml看到类似如下输出就成功了 保留这个终端以使得 mihomo 能持续运行并且监听服务端口。然后新开其他终端并在新开终端中配置环境变量
export https_proxyhttp://127.0.0.1:7890/
export http_proxyhttp://127.0.0.1:7890/到这一步就能顺利访问到目标网址。测试
# 如果返回结果中包含HTTP状态码如200 OK则表示通过代理访问成功。
curl -I https://www.google.com也可以使用aiohttp库检测代理
import aiohttp
import asyncioasync def fetch(url):proxy http://127.0.0.1:7890async with aiohttp.ClientSession() as session:async with session.get(url, proxyproxy) as response:return await response.text()url https://api.github.com
result await.fetch(url) # notebook中运行
# result asyncio.run(fetch(url)) # .py文件运行
print(result)二、 OSSWatcher Role实现
总结起来 OSSWatcher Role 需要以下两个主要 Action
爬取热门开源项目信息这里以GitHub Trending为目标网站可选筛选条件包括编程语言 (language)、时间范围 (daily/weekly/monthly)、所用语言 (spoken language)。使用LLM分析热门开源项目分析角度包括 编程语言趋势观察Trending列表中使用的编程语言了解当前哪些编程语言在开发者社区中更受欢迎项目类型和用途分析Trending列表中的项目看看它们是属于哪些类别以及它们的具体用途是什么社区活跃度查看项目的星标数量、贡献者数量新兴技术和工具注意新项目和涌现的技术以便了解当前的技术趋势
2.1 爬取代码
不熟悉爬虫的话可以参考对话用ChatGPT帮我们写爬取分析Github Trending。以下是最终的代码
import aiohttp
import asyncio
from bs4 import BeautifulSoupasync def fetch_html(url):async with aiohttp.ClientSession() as session:async with session.get(url) as response:return await response.text()async def parse_github_trending(html):soup BeautifulSoup(html, html.parser)repositories []for article in soup.select(article.Box-row):repo_info {}repo_info[name] article.select_one(h2 a).text.strip()repo_info[url] article.select_one(h2 a)[href].strip()# Descriptiondescription_element article.select_one(p)repo_info[description] description_element.text.strip() if description_element else None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] language_element.text.strip() if language_element else 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()# Todays Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[today_stars] today_stars_element.text.strip() if today_stars_element else Nonerepositories.append(repo_info)return repositoriesasync def main():url https://github.com/trendinghtml await fetch_html(url)repositories await parse_github_trending(html)for repo in repositories:print(fName: {repo[name]})print(fURL: https://github.com{repo[url]})print(fDescription: {repo[description]})print(fLanguage: {repo[language]})print(fStars: {repo[stars]})print(fForks: {repo[forks]})print(fTodays Stars: {repo[today_stars]})print()稍微修改一下把它作为一个Action类。
导入
from metagpt.actions.action import Action
from metagpt.config2 import Config设计 action的异步run方法 创建 aiohttp.ClientSession 对象: 使用 async with 语句创建 aiohttp.ClientSession 对象用于发送 HTTP 请求。发送 GET 请求: 使用 client.get() 方法发送 GET 请求并指定要访问的 URL。设置代理: 通过 proxyConfig.default().proxy 参数设置全局代理服务器。检查响应状态: 使用 response.raise_for_status() 方法检查响应状态如果状态码不为 200则会抛出异常。读取响应内容: 使用 await response.text() 方法读取响应内容并将其存储在 html 变量中。返回 HTML 内容: 将获取到的 HTML 内容返回给调用者。
async def run(self, url: str https://github.com/trending):async with aiohttp.ClientSession() as client:async with client.get(url, proxyCONFIG.global_proxy) as response:response.raise_for_status()html await response.text()将先前的parse_github_trending(html)方法搬运到action中用repositories做返回值
完整代码如下
import aiohttp
from bs4 import BeautifulSoup
from metagpt.actions.action import Action
from metagpt.config2 import Configclass CrawlOSSTrending(Action):async def run(self, url: str https://github.com/trending):async with aiohttp.ClientSession() as client:async with client.get(url, proxyConfig.default().proxy) as response:response.raise_for_status()html await response.text()soup BeautifulSoup(html, html.parser)repositories []for article in soup.select(article.Box-row):repo_info {}repo_info[name] article.select_one(h2 a).text.strip().replace(\n, ).replace( , )repo_info[url] https://github.com article.select_one(h2 a)[href].strip()# Descriptiondescription_element article.select_one(p)repo_info[description] description_element.text.strip() if description_element else None# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] language_element.text.strip() if language_element else 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()# Todays Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[today_stars] today_stars_element.text.strip() if today_stars_element else Nonerepositories.append(repo_info)return repositories2.2 GitHub Trending总结 总结Action的实现比较简单主要是写提示词。在提示词中我们可以要求LLM从几个角度进行分析并按照一定的格式进行输出例如
今天榜单的整体趋势例如哪几个编程语言比较热门、最热门的项目是哪些、主要集中在哪些领域榜单的仓库分类推荐进一步关注哪些仓库推荐原因是什么
from typing import Any
from metagpt.actions.action import ActionTRENDING_ANALYSIS_PROMPT # Requirements
You are a GitHub Trending Analyst, aiming to provide users with insightful and personalized recommendations based on the latest
GitHub Trends. Based on the context, fill in the following missing information, generate engaging and informative titles,
ensuring users discover repositories aligned with their interests.# The title about Todays GitHub Trending
## Todays Trends: Uncover the Hottest GitHub Projects Today! Explore the trending programming languages and discover key domains capturing developers attention. From ** to **, witness the top projects like never before.
## The Trends Categories: Dive into Todays GitHub Trending Domains! Explore featured projects in domains such as ** and **. Get a quick overview of each project, including programming languages, stars, and more.
## Highlights of the List: Spotlight noteworthy projects on GitHub Trending, including new tools, innovative projects, and rapidly gaining popularity, focusing on delivering distinctive and attention-grabbing content for users.
---
# Format Example# [Title]## Todays Trends
Today, ** and ** continue to dominate as the most popular programming languages. Key areas of interest include **, ** and **.
The top popular projects are Project1 and Project2.## The Trends Categories
1. Generative AI- [Project1](https://github/xx/project1): [detail of the project, such as star total and today, language, ...]- [Project2](https://github/xx/project2): ...
...## Highlights of the List
1. [Project1](https://github/xx/project1): [provide specific reasons why this project is recommended].
...---
# Github Trending
{trending}
class AnalysisOSSTrending(Action):async def run(self,trending: Any):return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trendingtrending))三、 OSSWatcher Role实现 以上Action都实现了把它们都写到metagpt/actions/oss_trending.py文件中然后新建文件metagpt/roles/oss_watcher.py就可以写Role的代码了
from metagpt.actions.oss_trending import CrawlOSSTrending, AnalysisOSSTrending
from metagpt.roles import Roleclass OssWatcher(Role):def __init__(self,nameCodey,profileOssWatcher,goalGenerate an insightful GitHub Trending analysis report.,constraintsOnly analyze based on the provided GitHub Trending data.,):super().__init__(name, profile, goal, constraints)self._init_actions([CrawlOSSTrending, AnalysisOSSTrending])self._set_react_mode(react_modeby_order)async def _act(self) - Message:logger.info(f{self._setting}: ready to {self._rc.todo})# By choosing the Action by order under the hood# todo will be first SimpleWriteCode() then SimpleRunCode()todo self.rc.todomsg self.get_memories(k1)[0] # find the most k recent messagesresult await todo.run(msg.content)msg Message(contentstr(result), roleself.profile, cause_bytype(todo))self._rc.memory.add(msg)return msg四、 Trigger实现 最简单的触发方式即定时触发Github Trending 大约是在10:00 AM UTC更新代价可以选取一个比较适合自己的推送时间即可比如每天早上9点。
import asyncio
import timefrom datetime import datetime, timedelta
from metagpt.schema import Message
from pydantic import BaseModel, Fieldclass 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 next_time timedelta(1)wait next_time - nowprint(wait.total_seconds())await asyncio.sleep(wait.total_seconds())yield Message(url, OssInfo(urlurl))yield 语句被用于异步函数oss_trigger中用于生成消息。每当调用这个异步函数时它会在指定的时间间隔内生成一个消息并在下一次调用时继续执行。此处我们预定义了OssInfo的结构加入了时间戳的信息并将其实例作为trigger生成的Message的instruct_content属性作用是在早期的版本中角色在接收Message会有一个去重操作如果我们每次生成的Message只有url信息那么第2次运行时角色将不会接收新的Message但是加入时间戳后trigger生成的每个Message就不再相等角色也会接收对应的Message。 上述的简单例子可以实现简单的按天定时触发的能力不过如果需要更精细的控制这个函数还需要继续优化。但我们可以借助一些第三方包实现这个功能使用crontab实现定时触发是非常常见的一个做法而且python也有一个异步的cron工具即aiocron。使用aiocron我们可以直接使用cron的语法制定定时任务。上面我们使用了函数的方式来实现了定时Trigger异步生成器接下来我们结合aiocron使用类的方式来实现定时Trigger
import time
from aiocron import crontab
from typing import Optional
from pytz import BaseTzInfo
from pydantic import BaseModel, Field
from metagpt.schema import Messageclass GithubTrendingCronTrigger:def __init__(self,spec: str,tz: Optional[BaseTzInfo] None,url: str https://github.com/trending,) - None:self.crontab crontab(spec, tztz)self.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next()return Message(contentself.url)基于aiocron我们可以少写很多代码功能也更加强大可以用cron语法非常灵活地配置定时规则。如果您想指定UTC 时间 10:00 AM 触发
# 创建 GithubTrendingCronTrigger 实例指定每天 UTC 时间 10:00 AM 触发
cron_trigger GithubTrendingCronTrigger(0 10 * * *)如果您想指定北京时间上午8:00来触发这个任务您需要做两件事
设置正确的 cron 表达式。确保时区设置正确。 北京时间是东八区UTC8所以您应该在 tz 参数中设置相应的时区。而 cron 表达式遵循特定的格式通常是分钟、小时、日、月、星期几。 对于每天上午8:00cron 表达式应该是 “0 8 * * *”这表示每天的第8小时的第0分钟触发。因此您的 GithubTrendingCronTrigger 类的初始化代码应该类似于以下形式
from pytz import timezone
beijing_tz timezone(Asia/Shanghai) 获取北京时间的时区
cron_trigger GithubTrendingCronTrigger(0 8 * * *, tzbeijing_tz)思考1如果需要榜单更新再推送可以如何实现思考2Crontab的定时方式可能不是很方便进行调试有什么方便调试的方法吗 五、 Callback设计 Callback就是定义了如何处理智能体生成的信息它本身没有过多难点但是如果想将信息发送到我们日常使用的一些应用可能会有一些成本。下面提供将智能体产生的数据发送到discord/微信的示例供大家参考。
5.1 Discord Discord是一款免费的通讯软件让你可以与你的好友社群以及开发者们进行语音视频及文字聊天。目前MetaGPT的海外社区就是在Discord上维护的在国内MetaGPT也有庞大的微信社区所以本文档选取目前MetaGPT比较活跃的两个社区工具作为示例。
账号注册在discord的开发者面板添加BOT并将BOT添加到某个服务器中详见参考教程。使用discord发送消息的示例如下
import asyncio
import discordasync 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单条消息有大小限制过长的内容会导致发送不成功我们可以按章节分多条msg发送最终实现的discord_callback函数如下
import asyncio
import discordfrom metagpt.config import CONFIGasync def discord_callback(msg: Message):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)lines []for i in msg.content.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参考官方文档discord readthedocsCreating a Bot Account章节的第7步 DISCORD_CHANNEL_ID即希望Bot发送消息的频道
5.2 Wechat wxpusher开发文档 由于我们的内容是markdown格式的直接发送微信消息阅读体验较差所以我们需要寻找合适的微信消息发送方式。 公众号可以发送富本文消息比较符合我们的场景但是为了个推送的功能开发个公众号的成本也是比较大。目前已有许多的第三方公众号提供了消息推送的功能例如server酱、wxpusher、Pushplus等我们可以选择其中之一例如wxpusher它的代码是开源的还有详细的文开发文档。
wxpusher的python客户端是同步的根据API文档可以快速简单地实现一个异步的客户端
import os
from typing import Optional
import aiohttpclass WxPusherClient:def __init__(self, token: Optional[str] None, base_url: str http://wxpusher.zjiecode.com):self.base_url base_urlself.token token or os.environ[WXPUSHER_TOKEN]async def send_message(self,content,summary: Optional[str] None,content_type: int 1,topic_ids: Optional[list[int]] None,uids: Optional[list[int]] None,verify: bool False,url: Optional[str] None,):payload {appToken: self.token,content: content,summary: summary,contentType: content_type,topicIds: topic_ids or [],uids: uids or os.environ[WXPUSHER_UIDS].split(,),verifyPay: verify,url: url,}url f{self.base_url}/api/send/messagereturn await self._request(POST, url, jsonpayload)async def _request(self, method, url, **kwargs):async with aiohttp.ClientSession() as session:async with session.request(method, url, **kwargs) as response:response.raise_for_status()return await response.json()然后实现callback
async def wxpusher_callback(msg: Message):client WxPusherClient()await client.send_message(msg.content, content_type3)WXPUSHER_TOKEN获取见官方文档 WXPUSHER_UIDS可以从应用管理页的”用户管理-用户列表“获取用户的UID如果要发送给多个用户可以用逗号将不同用户UID隔开 六、 运行示例
我们将上述代码都写在一个main.py文件里完整代码文件见main.py或main.ipynb。
import asyncio
import os
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optionalimport aiohttp
import discord
from aiocron import crontab
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
from pytz import BaseTzInfofrom metagpt.actions.action import Action
from metagpt.config import CONFIG
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message# fix SubscriptionRunner not fully defined
from metagpt.environment import Environment as _ # noqa: F401# 订阅模块可以from metagpt.subscription import SubscriptionRunner导入这里贴上代码供参考
class SubscriptionRunner(BaseModel):A simple wrapper to manage subscription tasks for different roles using asyncio.Example: import asyncio from metagpt.subscription import SubscriptionRunner from metagpt.roles import Searcher from metagpt.schema import 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())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],],):Subscribes a role to a trigger and sets up a callback to be called with the roles response.Args:role: The role to subscribe.trigger: An asynchronous generator that yields Messages to be processed by the role.callback: An asynchronous function to be called with the response from the role.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):Unsubscribes a role from its trigger and cancels the associated task.Args:role: The role to unsubscribe.task self.tasks.pop(role)task.cancel()async def run(self, raise_exception: bool True):Runs all subscribed tasks and handles their completion or exception.Args:raise_exception: _description_. Defaults to True.Raises:task.exception: _description_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)# Actions 的实现
TRENDING_ANALYSIS_PROMPT # Requirements
You are a GitHub Trending Analyst, aiming to provide users with insightful and personalized recommendations based on the latest
GitHub Trends. Based on the context, fill in the following missing information, generate engaging and informative titles,
ensuring users discover repositories aligned with their interests.# The title about Todays GitHub Trending
## Todays Trends: Uncover the Hottest GitHub Projects Today! Explore the trending programming languages and discover key domains capturing developers attention. From ** to **, witness the top projects like never before.
## The Trends Categories: Dive into Todays GitHub Trending Domains! Explore featured projects in domains such as ** and **. Get a quick overview of each project, including programming languages, stars, and more.
## Highlights of the List: Spotlight noteworthy projects on GitHub Trending, including new tools, innovative projects, and rapidly gaining popularity, focusing on delivering distinctive and attention-grabbing content for users.
---
# Format Example
# [Title]## Todays Trends
Today, ** and ** continue to dominate as the most popular programming languages. Key areas of interest include **, ** and **.
The top popular projects are Project1 and Project2.## The Trends Categories
1. Generative AI- [Project1](https://github/xx/project1): [detail of the project, such as star total and today, language, ...]- [Project2](https://github/xx/project2): ...
...## Highlights of the List
1. [Project1](https://github/xx/project1): [provide specific reasons why this project is recommended].
...
---
# Github Trending
{trending}
class CrawlOSSTrending(Action):async def run(self, url: str https://github.com/trending):async with aiohttp.ClientSession() as client:async with client.get(url, proxyCONFIG.global_proxy) as response:response.raise_for_status()html await response.text()soup BeautifulSoup(html, html.parser)repositories []for article in soup.select(article.Box-row):repo_info {}repo_info[name] (article.select_one(h2 a).text.strip().replace(\n, ).replace( , ))repo_info[url] (https://github.com article.select_one(h2 a)[href].strip())# Descriptiondescription_element article.select_one(p)repo_info[description] (description_element.text.strip() if description_element else None)# Languagelanguage_element article.select_one(span[itempropprogrammingLanguage])repo_info[language] (language_element.text.strip() if language_element else 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()# Todays Starstoday_stars_element article.select_one(span.d-inline-block.float-sm-right)repo_info[today_stars] (today_stars_element.text.strip() if today_stars_element else None)repositories.append(repo_info)return repositoriesclass AnalysisOSSTrending(Action):async def run(self, trending: Any):return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trendingtrending))# Role实现
class OssWatcher(Role):def __init__(self,nameCodey,profileOssWatcher,goalGenerate an insightful GitHub Trending analysis report.,constraintsOnly analyze based on the provided GitHub Trending data.,):super().__init__(namename, profileprofile, goalgoal, constraintsconstraints)self._init_actions([CrawlOSSTrending, AnalysisOSSTrending])self._set_react_mode(react_modeby_order)async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})# By choosing the Action by order under the hood# todo will be first SimpleWriteCode() then SimpleRunCode()todo self.rc.todomsg self.get_memories(k1)[0] # find the most k recent messagesresult await todo.run(msg.content)msg Message(contentstr(result), roleself.profile, cause_bytype(todo))self.rc.memory.add(msg)return msg# Trigger
class GithubTrendingCronTrigger:def __init__(self,spec: str,tz: Optional[BaseTzInfo] None,url: str https://github.com/trending,) - None:self.crontab crontab(spec, tztz)self.url urldef __aiter__(self):return selfasync def __anext__(self):await self.crontab.next()return Message(contentself.url)# callback
async def discord_callback(msg: Message):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)lines []for i in msg.content.splitlines():if i.startswith((# , ## , ### )):if lines:await channel.send(\n.join(lines))lines []lines.append(i)if lines:await channel.send(\n.join(lines))class WxPusherClient:def __init__(self,token: Optional[str] None,base_url: str http://wxpusher.zjiecode.com,):self.base_url base_urlself.token token or os.environ[WXPUSHER_TOKEN]async def send_message(self,content,summary: Optional[str] None,content_type: int 1,topic_ids: Optional[list[int]] None,uids: Optional[list[int]] None,verify: bool False,url: Optional[str] None,):payload {appToken: self.token,content: content,summary: summary,contentType: content_type,topicIds: topic_ids or [],uids: uids or os.environ[WXPUSHER_UIDS].split(,),verifyPay: verify,url: url,}url f{self.base_url}/api/send/messagereturn await self._request(POST, url, jsonpayload)async def _request(self, method, url, **kwargs):async with aiohttp.ClientSession() as session:async with session.request(method, url, **kwargs) as response:response.raise_for_status()return await response.json()async def wxpusher_callback(msg: Message):client WxPusherClient()await client.send_message(msg.content, content_type3)# 运行入口
async def main(spec: str 0 9 * * *, discord: bool True, wxpusher: bool True):callbacks []if discord:callbacks.append(discord_callback)if wxpusher:callbacks.append(wxpusher_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)七、 开发作业
尝试着完成一个能订阅自己感兴趣的资讯的Agent
根据前面你所学习的爬虫基本知识如果你对写爬虫代码感到不熟练使用GPT帮助你为你的Agent自定义两个获取资讯的Action类 Action 1独立实现对Github Trending页面的爬取并获取每一个项目的名称、URL链接、描述Action 2独立完成对Huggingface Papers页面的爬取先获取到每一篇Paper的链接提示标题元素中的href标签并通过链接访问标题的描述页面例如https://huggingface.co/papers/2312.03818在页面中获取一篇Paper的 标题、摘要 重写有关方法使你的Agent能自动生成总结内容的目录然后根据二级标题进行分块每块内容做出对应的总结形成一篇资讯文档自定义Agent的SubscriptionRunner类独立实现Trigger、Callback的功能让你的Agent定时为通知渠道发送以上总结的资讯文档尝试实现邮箱发送的功能这是加分项
思考作业- 目前为止我们设计的所有思考模式都可以总结为是链式的思考chain of thought能否利用 MetaGPT 框架实现树结构的思考tree of thought图结构的思考graph of thought 文章转载自： http://www.morning.pwppk.cn.gov.cn.pwppk.cn http://www.morning.yxyyp.cn.gov.cn.yxyyp.cn http://www.morning.jgnst.cn.gov.cn.jgnst.cn http://www.morning.mpwgs.cn.gov.cn.mpwgs.cn http://www.morning.qqnjr.cn.gov.cn.qqnjr.cn http://www.morning.pyncx.cn.gov.cn.pyncx.cn http://www.morning.rkfwr.cn.gov.cn.rkfwr.cn http://www.morning.gwsfq.cn.gov.cn.gwsfq.cn http://www.morning.skkln.cn.gov.cn.skkln.cn http://www.morning.mdtfh.cn.gov.cn.mdtfh.cn http://www.morning.pkdng.cn.gov.cn.pkdng.cn http://www.morning.nfmtl.cn.gov.cn.nfmtl.cn http://www.morning.mtrfz.cn.gov.cn.mtrfz.cn http://www.morning.lrzst.cn.gov.cn.lrzst.cn http://www.morning.yrcxg.cn.gov.cn.yrcxg.cn http://www.morning.nxbsq.cn.gov.cn.nxbsq.cn http://www.morning.mqxzh.cn.gov.cn.mqxzh.cn http://www.morning.qnxtz.cn.gov.cn.qnxtz.cn http://www.morning.fnkcg.cn.gov.cn.fnkcg.cn http://www.morning.hqwcd.cn.gov.cn.hqwcd.cn http://www.morning.btqqh.cn.gov.cn.btqqh.cn http://www.morning.qykxj.cn.gov.cn.qykxj.cn http://www.morning.mkzdp.cn.gov.cn.mkzdp.cn http://www.morning.gpfuxiu.cn.gov.cn.gpfuxiu.cn http://www.morning.bdkhl.cn.gov.cn.bdkhl.cn http://www.morning.gtdf.cn.gov.cn.gtdf.cn http://www.morning.ygztf.cn.gov.cn.ygztf.cn http://www.morning.lwtld.cn.gov.cn.lwtld.cn http://www.morning.mngyb.cn.gov.cn.mngyb.cn http://www.morning.wdwfm.cn.gov.cn.wdwfm.cn http://www.morning.qnsmk.cn.gov.cn.qnsmk.cn http://www.morning.rpms.cn.gov.cn.rpms.cn http://www.morning.benqc.com.gov.cn.benqc.com http://www.morning.dpppx.cn.gov.cn.dpppx.cn http://www.morning.jklns.cn.gov.cn.jklns.cn http://www.morning.jbtwq.cn.gov.cn.jbtwq.cn http://www.morning.hptbp.cn.gov.cn.hptbp.cn http://www.morning.gfmpk.cn.gov.cn.gfmpk.cn http://www.morning.mwlxk.cn.gov.cn.mwlxk.cn http://www.morning.mpyry.cn.gov.cn.mpyry.cn http://www.morning.pslzp.cn.gov.cn.pslzp.cn http://www.morning.mxnrl.cn.gov.cn.mxnrl.cn http://www.morning.jcbjy.cn.gov.cn.jcbjy.cn http://www.morning.rjmd.cn.gov.cn.rjmd.cn http://www.morning.hphrz.cn.gov.cn.hphrz.cn http://www.morning.mqfhy.cn.gov.cn.mqfhy.cn http://www.morning.bqnhh.cn.gov.cn.bqnhh.cn http://www.morning.xphcg.cn.gov.cn.xphcg.cn http://www.morning.sjsfw.cn.gov.cn.sjsfw.cn http://www.morning.ynbyk.cn.gov.cn.ynbyk.cn http://www.morning.rcrnw.cn.gov.cn.rcrnw.cn http://www.morning.qbdqc.cn.gov.cn.qbdqc.cn http://www.morning.cfocyfa.cn.gov.cn.cfocyfa.cn http://www.morning.zwxfj.cn.gov.cn.zwxfj.cn http://www.morning.gqtzb.cn.gov.cn.gqtzb.cn http://www.morning.trjdr.cn.gov.cn.trjdr.cn http://www.morning.lrnfn.cn.gov.cn.lrnfn.cn http://www.morning.rbffj.cn.gov.cn.rbffj.cn http://www.morning.nlglm.cn.gov.cn.nlglm.cn http://www.morning.rlqqy.cn.gov.cn.rlqqy.cn http://www.morning.pqwjh.cn.gov.cn.pqwjh.cn http://www.morning.kqpq.cn.gov.cn.kqpq.cn http://www.morning.rngyq.cn.gov.cn.rngyq.cn http://www.morning.ydgzj.cn.gov.cn.ydgzj.cn http://www.morning.bklkt.cn.gov.cn.bklkt.cn http://www.morning.fjscr.cn.gov.cn.fjscr.cn http://www.morning.kycxb.cn.gov.cn.kycxb.cn http://www.morning.kszkm.cn.gov.cn.kszkm.cn http://www.morning.qmzwl.cn.gov.cn.qmzwl.cn http://www.morning.ftnhr.cn.gov.cn.ftnhr.cn http://www.morning.jjhng.cn.gov.cn.jjhng.cn http://www.morning.tpmnq.cn.gov.cn.tpmnq.cn http://www.morning.nrcbx.cn.gov.cn.nrcbx.cn http://www.morning.krbjb.cn.gov.cn.krbjb.cn http://www.morning.kmrgl.cn.gov.cn.kmrgl.cn http://www.morning.dhmll.cn.gov.cn.dhmll.cn http://www.morning.bynf.cn.gov.cn.bynf.cn http://www.morning.gjcdr.cn.gov.cn.gjcdr.cn http://www.morning.cwyrp.cn.gov.cn.cwyrp.cn http://www.morning.lkbdy.cn.gov.cn.lkbdy.cn
