MetaGPT实践——用ActionNode与SubscriptionRunner实现真正的“订阅智能体”

1. 从“软件体”到订阅智能体：MetaGPT 里 ActionNode 与 SubscriptionRunner 到底怎么配合

很多人第一次看 MetaGPT 的订阅智能体示例，都会卡在同一个地方：GitHubTrendingWatcher明明继承了Role，外面却又套了一个SubscriptionRunner，那到底谁才是智能体？更别扭的是，翻遍代码发现真正用到 LLM 的只有一个AnalyzeGitHubTrending，其余全是硬编码的爬取和推送逻辑，说它是智能体，总觉得名不副实，叫它“软件体”反而更贴切。

我理解这种疑惑。真正的订阅智能体，应该是你丢一句自然语言过去，比如“帮我抓取某站点今天的文章，总结主要资讯，每天下午 5 点 30 分推送到微信”，然后它自己拆解需求、写爬虫、定时执行、生成报告、推送消息，全程不用你再插手。这中间涉及两个关键能力：一是把模糊的自然语言需求结构化成可执行的字段，二是把周期任务编排起来并挂上回调。MetaGPT 里对应的就是ActionNode和SubscriptionRunner。

ActionNode可以理解成一棵动作树，父节点能访问所有子节点，定义好之后按树的结构顺序执行每个子动作，天然带一层 Chain-of-Thought 效果。它还能做格式校验和结构化输出，让 LLM 返回的内容更规整，这对后续代码生成特别重要。SubscriptionRunner则负责驱动周期性任务，接收role、callback、trigger三个入口参数，trigger用 cron 表达式控制触发时机，每次触发就跑一遍SubAction并调用回调。

这篇就围绕这两个组件，把订阅智能体的落地路径走一遍：用ActionNode定义需求解析节点，用SubscriptionRunner驱动定时任务，面向资讯聚合、定时推送这类场景，给出可复制的配置片段和启动脚本，最后演示一次端到端验证。适合已经跑过 MetaGPT 基础示例、想把它用到真实订阅场景的开发者。

2. 前置准备：TaoToken 接入 MetaGPT 的模型配置与依赖安装

在写订阅智能体之前，得先把模型调用这条链路打通。MetaGPT 默认走 OpenAI 兼容接口，所以只要有一个兼容 OpenAI 协议的 Base URL 和 Key，就能直接接上。我这边用的是 TaoToken 的接口，它提供 OpenAI 兼容的调用方式，配置起来比较省事。

先装依赖。MetaGPT 本身依赖比较多，建议单独建虚拟环境：

python -m venv metagpt-sub source metagpt-sub/bin/activate # Windows 用 metagpt-sub\Scripts\activate pip install metagpt pip install aiocron aiohttp pytz beautifulsoup4

aiocron用来做 cron 触发，aiohttp给微信推送用，beautifulsoup4是爬虫解析的基础。装完之后配置模型。MetaGPT 支持通过环境变量或配置文件指定 LLM，最直接的方式是写一个config2.yaml：

llm: api_type: "openai" base_url: "https://taotoken.net/api" api_key: "sk-你的TaoToken密钥" model: "gpt-4o-mini" temperature: 0.3 max_token: 4096

这里base_url填 TaoToken 的 API 地址，api_key去控制台生成。模型 ID 按你实际订阅的填，gpt-4o-mini这类性价比高的模型适合订阅场景，因为需求解析和报告生成都是短文本任务，没必要上大模型。如果你更习惯用 Claude 系列做代码生成，也可以把model换成对应的 ID，接口协议是一样的。

配置好之后，写个最小验证脚本确认能通：

import asyncio from metagpt.llm import LLM async def main(): llm = LLM() resp = await llm.aask("用一句话说明什么是订阅智能体") print(resp) asyncio.run(main())

能正常打印出内容，说明模型链路通了。如果报 401，多半是 Key 没填对或者环境变量没生效；如果报连接超时，检查base_url有没有写全。这一步过了，再往下写ActionNode才有意义，否则后面调试会分不清是模型问题还是代码问题。

3. 可复制配置：用 ActionNode 定义订阅需求解析节点

订阅智能体的第一步，是把用户那句自然语言拆成结构化字段。这里用ActionNode定义六个子节点：语言、cron 表达式、爬取 URL 列表、页面内容提取要求、爬取后处理要求、信息补充提示。这六个字段基本覆盖了订阅任务需要的全部输入。

import datetime from typing import Optional from uuid import uuid4 from aiocron import crontab from metagpt.actions import UserRequirement from metagpt.actions.action import Action from metagpt.actions.action_node import ActionNode from metagpt.roles import Role from metagpt.schema import Message from metagpt.tools.web_browser_engine import WebBrowserEngine from metagpt.utils.common import CodeParser, any_to_str from metagpt.utils.parse_html import _get_soup from pytz import BaseTzInfo from metagpt.logs import logger LANGUAGE = ActionNode( key="Language", expected_type=str, instruction="Provide the language used in the project, typically matching the user's requirement language.", example="zh_cn", ) 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="35 15 * * *", ) 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://example1.com", "https://example2.com"], ) 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.", ) 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_SUBSCRIPTION_REQUIREMENTS_NODE = ActionNode.from_children("ParseSubscriptionReq", NODES) PARSE_SUBSCRIPTION_REQUIREMENT_TEMPLATE = """ ### User Requirement {requirements} """ class ParseSubscriptionRequirement(Action): async def run(self, requirements): requirements = "\n".join(i.content for i in requirements) context = PARSE_SUBSCRIPTION_REQUIREMENT_TEMPLATE.format(requirements=requirements) node = await PARSE_SUBSCRIPTION_REQUIREMENTS_NODE.fill(context=context, llm=self.llm) return node

这里有几个细节值得说。CRON_EXPRESSION的example我特意填了一个真实的 5 字段表达式，而不是留空。实测下来，给一个具体例子能显著提高 cron 生成的正确率，否则模型容易返回带秒的 6 字段格式，或者干脆返回自然语言描述。CRAWLER_URL_LIST的expected_type是list[str]，ActionNode会按这个类型做校验，返回的不是列表就会触发重试。

ActionNode.from_children把六个子节点组装成一棵串行的动作树，fill方法会依次执行每个子节点的simple_fill，把 LLM 返回的内容按 schema 解析成结构化对象。如果某个字段格式不对，_aask_v1里带了重试装饰器，会重新调用 LLM 生成，最多六次。这就是ActionNode相比裸调 LLM 的价值：格式约束和自动修复都内置了。

把这段存成subscriptionassistant.py，先单独测一下需求解析。定义一个只有单个 action 的SubscriptionAssistant：

class SubscriptionAssistant(Role): name: str = "Grace" 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.set_actions([ParseSubscriptionRequirement]) async def _act(self) -> Message: logger.info(f"{self._setting}: ready to {self.rc.todo}") response = await self.rc.todo.run(self.rc.history) msg = Message( content=response.content, instruct_content=response.instruct_content, role=self.profile, cause_by=self.rc.todo, sent_from=self, ) self.rc.memory.add(msg) return msg

跑一句完整需求：

import asyncio from metagpt.team import Team team = Team() team.hire([SubscriptionAssistant()]) team.run_project("从36kr创投平台https://pitchhub.36kr.com/financing-flash爬取所有初创企业融资的信息，获取标题，链接，时间，总结今天的融资新闻，然后在14:55发送给我") asyncio.run(team.run())

返回的instruct_content里，Cron Expression是55 14 * * *，Crawler URL List是完整的融资快讯地址，Page Content Extraction和Crawl Post Processing也都拆得比较准。这里有个坑要注意：如果用户需求里没给完整 URL，比如只说“从36kr创投平台爬取”，模型会自作主张把 URL 填成https://36kr.com，后续爬虫大概率抓不到目标数据。所以INFORMATION_SUPPLEMENT这个节点很关键，它负责在信息不全时提示用户补充，实际用的时候最好在解析后检查这个字段是否为空，不为空就把提示返回给用户。

4. 验证请求：SubscriptionRunner 驱动定时任务与端到端跑通

需求解析通了之后，接下来是爬虫代码生成和定时任务编排。先定义爬虫工程师角色，它监听ParseSubscriptionRequirement的输出，拿到 URL 和提取要求后生成解析函数。

CRAWLER_CODE_PROMPT_TEMPLATE = """Please complete the web page crawler parse function to achieve the User Requirement. The parse \ function should take a BeautifulSoup object as input, which corresponds to the HTML outline provided in the Context. ```python from bs4 import BeautifulSoup # only complete the parse function def parse(soup: BeautifulSoup): ... # Return the object that the user wants to retrieve, don't use print

User Requirement

{requirement}

Context

The outline of html page to scrabe is show like below:

{outline}

"""

class WriteCrawlerCode(Action): async def run(self, requirement): requirement: Message = requirement[-1] data = requirement.instruct_content.dict() urls = data["Crawler URL List"] query = data["Page Content Extraction"] codes = {} for url in urls: codes[url] = await self._write_code(url, query) return "\n".join(f"# {url}\n{code}" for url, code in codes.items())

async def _write_code(self, url, query): page = await WebBrowserEngine().run(url) outline = get_outline(page) outline = "\n".join( f"{' '*i['depth']}{'.'.join([i['name'], *i.get('class', [])])}: {i['text'] if i['text'] else ''}" for i in outline ) code_rsp = await self._aask(CRAWLER_CODE_PROMPT_TEMPLATE.format(outline=outline, requirement=query)) code = CodeParser.parse_code(block="", text=code_rsp) return code

class CrawlerEngineer(Role): name: str = "John" profile: str = "Crawling Engineer" goal: str = "Write elegant, readable, extensible, efficient code" constraints: str = "The code should conform to standards like PEP8 and be modular and maintainable"

def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self.set_actions([WriteCrawlerCode]) self._watch([ParseSubscriptionRequirement])

`_watch` 是角色间的消息传递机制。`SubscriptionAssistant` 执行完 `ParseSubscriptionRequirement` 后，`CrawlerEngineer` 检测到这个动作产生的消息，就触发自己的 `WriteCrawlerCode`。`WriteCrawlerCode.run` 收到的 `requirement` 是 Memory 里的 Message 列表，取最后一个就是上游刚生成的结构化需求，`instruct_content.dict()` 转成字典后就能拿到 URL 和提取要求。 然后是核心的 `RunSubscription`，它负责把爬虫代码、cron 表达式、后处理要求组装成一个 `SubAction`，交给 `SubscriptionRunner` 定时执行： ```python class CronTrigger: def __init__(self, spec: str, tz: Optional[BaseTzInfo] = None) -> None: segs = spec.split(" ") if len(segs) == 6: spec = " ".join(segs[1:]) self.crontab = crontab(spec, tz=tz) def __aiter__(self): return self async def __anext__(self): await self.crontab.next() return Message(datetime.datetime.now().isoformat()) class RunSubscription(Action): async def run(self, msgs): from metagpt.roles.role import Role from metagpt.subscription import SubscriptionRunner import asyncio code = msgs[-1].content req = msgs[-2].instruct_content.dict() urls = req["Crawler URL List"] process = req["Crawl Post Processing"] spec = req["Cron Expression"] SubAction = self.create_sub_action_cls(urls, code, process) SubRole = type("SubRole", (Role,), {}) role = SubRole() role.set_actions([SubAction]) runner = SubscriptionRunner() callbacks = [] callbacks.append(wxpusher_callback) async def callback(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], code: str, process: str): modules = {} for url in urls[::-1]: code, current = code.rsplit(f"# {url}", maxsplit=1) name = uuid4().hex module = type(sys)(name) exec(current, module.__dict__) modules[url] = module class SubAction(Action): async def run(self, *args, **kwargs): pages = await WebBrowserEngine().run(*urls) if len(urls) == 1: pages = [pages] data = [] for url, page in zip(urls, pages): try: data.append(getattr(modules[url], "parse")(page.soup)) except AttributeError: print("Error: 'NoneType' object has no attribute 'text'") return await self.llm.aask(SUB_ACTION_TEMPLATE.format(process=process, data=data)) return SubAction

SubscriptionRunner的三个入口参数在这里都齐了：role是动态创建的SubRole，只挂一个SubAction；callback是wxpusher_callback，把结果推到微信；trigger是CronTrigger，接收 cron 表达式，每次触发就跑一遍SubAction并调用回调。create_sub_action_cls把生成的爬虫代码按 URL 切分、动态执行成模块，SubAction运行时先爬页面、调parse提取数据，再把数据交给 LLM 按后处理要求生成报告。

把SubscriptionAssistant补全，让它同时挂ParseSubscriptionRequirement和RunSubscription，并重写_think处理不同上游动作的响应：

class SubscriptionAssistant(Role): name: str = "Grace" 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: from crawlerengineer import WriteCrawlerCode super().__init__(**kwargs) self.set_actions([ParseSubscriptionRequirement, RunSubscription]) self._watch([UserRequirement, WriteCrawlerCode]) async def _think(self) -> bool: from crawlerengineer import WriteCrawlerCode cause_by = self.rc.history[-1].cause_by if cause_by == any_to_str(UserRequirement): state = 0 elif cause_by == any_to_str(WriteCrawlerCode): state = 1 if self.rc.state == state: self.rc.todo = None return False self._set_state(state) return True async def _act(self) -> Message: logger.info(f"{self._setting}: ready to {self.rc.todo}") response = await self.rc.todo.run(self.rc.history) msg = Message( content=response.content, instruct_content=response.instruct_content, role=self.profile, cause_by=self.rc.todo, sent_from=self, ) self.rc.memory.add(msg) return msg

端到端跑一次：

import asyncio from metagpt.team import Team from crawlerengineer import CrawlerEngineer from subscriptionassistant import SubscriptionAssistant team = Team() team.hire([SubscriptionAssistant(), CrawlerEngineer()]) team.run_project("从 https://www.jiqizhixin.com/ 获取机器之心今天推送的文章，总结今天的主要资讯，然后在每天下午3点35分发送给我") asyncio.run(team.run())

运行日志里能看到完整链路：ParseSubscriptionRequirement把需求拆成35 15 * * *和机器之心首页；WriteCrawlerCode生成解析article-item__container的 BeautifulSoup 函数；RunSubscription启动SubscriptionRunner，到点触发SubAction，爬取、解析、总结，最后通过回调推送。一次需求解析约 550 tokens，写爬虫代码约 7900 tokens，生成报告约 960 tokens。如果把整个 HTML 直接丢给 LLM 生成报告，每天大概要 9000 tokens，现在每天不到 1000 tokens，成本降到十分之一左右。

5. 本篇常见错排查：401、JSON 解析失败与 cron 不触发

跑订阅智能体时，报错基本集中在几个地方，这里按真实日志对照着说。

401 或鉴权失败。日志里出现AuthenticationError或401 Unauthorized，先检查config2.yaml里的api_key和base_url。TaoToken 的 API 地址是https://taotoken.net/api，注意不要多加路径。如果 Key 是从环境变量读的，确认变量名和代码里一致。还有一种情况是模型 ID 写错，比如订阅里没有gpt-4o却填了它，也会返回鉴权类错误，换成实际可用的模型 ID 即可。

JSON 解析失败。日志里出现Expecting property name enclosed in double quotes或parse json from content inside [CONTENT][/CONTENT] failed，这是ActionNode在解析 LLM 返回时遇到的格式问题。常见原因是模型在 JSON 里加了注释，比如"Cron Expression": "", # 请根据需要提供定时触发的cron表达式。ActionNode内置了repair_invalid_json会尝试修复，但修复失败就会重试。降低这类报错的办法是在instruction里明确要求返回纯 JSON，不要加注释，同时把temperature调低到 0.3 以下。

cron 表达式不触发。如果SubscriptionRunner启动后一直没动静，先看Cron Expression字段是不是空的。用户需求里没提时间时，模型会留空，CronTrigger拿到空字符串会报错。这时候要检查INFORMATION_SUPPLEMENT字段，把提示返回给用户补充时间。另外注意 cron 是 5 字段格式，如果模型返回了 6 字段（带秒），CronTrigger里做了截断处理，但最好在ActionNode的example里给 5 字段样例，从源头减少错误。

爬虫解析返回 None。日志里出现Error: 'NoneType' object has no attribute 'text'，说明生成的parse函数里选择器没匹配到元素。这通常是页面结构变了，或者get_outline提取的大纲不够准确。可以在WriteCrawlerCode里把outline打印出来，对照实际 HTML 检查 class 名。如果页面是动态渲染的，WebBrowserEngine可能拿不到完整 DOM，需要确认它是否支持等待渲染完成。

OAuth 或本地代理相关报错。如果日志里出现local proxy failed或OAuth字样，先确认没有配置额外的网络代理环境变量，比如HTTP_PROXY、HTTPS_PROXY。这些变量会干扰aiohttp和WebBrowserEngine的请求。清掉之后重启脚本再试。如果用的是需要 OAuth 的模型服务，确认 token 没过期，必要时重新生成。

6. 把订阅智能体接到 Coding Plan 上长期跑

订阅智能体的价值在于长期运行，每天定时触发、生成报告、推送消息。如果只是本地跑一次，意义不大。要让它稳定跑下去，模型调用的稳定性和成本控制就很重要。我现在的做法是把需求解析和报告生成这类高频短任务，走 TaoToken 的 Coding Plan，模型 ID 按任务类型分开配：需求解析用轻量模型，爬虫代码生成用代码能力强的模型，报告总结用性价比高的模型。这样既保证效果，又把每天的成本压住。

接入方式还是 OpenAI 兼容协议，Base URL 填https://taotoken.net/api，Key 在控制台生成后填进配置。如果你想让订阅智能体跑在服务器上，建议用systemd或supervisor托管进程，配合日志轮转，避免长时间运行后日志撑爆磁盘。cron 触发的时间点尽量避开整点，减少和其他任务撞车的概率。

需要生成 Key 的话，可以去控制台创建：https://taotoken.net/console/api-keys 。接入文档在 https://taotoken.net/doc ，里面有不同语言的调用示例。如果只是想先验证模型能不能通，用模型对话页面试一句就行：https://taotoken.net/chat 。长期跑编码和 Agent 类任务，Coding Plan 会更划算：https://taotoken.net/coding-plan 。

订阅智能体这个方向，核心不是把代码写得多复杂，而是把需求解析、代码生成、定时编排这三段串稳。ActionNode解决结构化，SubscriptionRunner解决周期驱动，剩下的就是按场景调 prompt 和模型。跑通一次之后，换个 URL 和总结要求，就是一个新的订阅任务。