基于工作流的定时科技新闻推送小助手

2026-08-17   访问量:1009


本文以搭建定时科技新闻推送小助手为例,详细介绍如何基于智能体开发平台创建单工作流应用并使用 云函数 实现定时任务功能。

应用目标:

使用智能体开发平台搭建“科技新闻推送小助手”应用,实现科技新闻收集及新闻推送功能。

使用云函数定时调用“科技新闻推送小助手”应用,实现定时任务功能。

第一部分:搭建“科技新闻推送小助手”应用

第一步:新建应用

前往 智能体开发平台控制台。

新建应用。

在左侧菜单栏,单击应用开发。



单击新建应用,输入应用名称“科技新闻推送小助手”,输入应用简介,应用模式选择 单工作流模式。




单击新建,自动跳转至应用设置页面。

第二步:配置工作流

新建工作流。

单击新建-手动录入,输入工作流名称“科技新闻收集”,输入工作流描述。




1.2. 单击确定,自动跳转至工作流画布页面。

在画布中配置工作流。

工作流需要实现“今日日期和Prompt获取”、“科技新闻搜索总结”、“文章整理”、“发送到指定渠道”步骤。整体工作流如下:




2.2. 配置开始节点。

2.2.1. 我们设置两个工作流级变量,mss用于控制科技文章发送的渠道,便于后续调试使用,默认值中“1”表示在调试窗输出,“2”表示推送到邮箱,“3”表示推送到短信;mailAddress表示邮箱列表,发送到邮箱时使用该变量;phoneList表示手机号列表,发送到短信时使用该变量。详细配置如下图所示:




2.3. 今日日期和Prompt获取。

2.3.1. 从左侧的节点列表中拖拽代码节点,获取今天的日期,以及调用AI搜索时使用的Prompt。代码及详细配置如下所示:

from datetime import datetime



def main(params: dict) -> dict:

   prompt_str = []

   today = datetime.now()

   date_str = today.strftime("%Y年%m月%d日")

   prompt_str.append(f"请帮我查询{date_str}的科技新闻,内容要来源于主流且权威的媒体,必须包含人工智能、云计算、开源生态、国内外大型科技公司相关新闻,不要有其他主题新闻。")

   prompt = ' '.join(prompt_str)



   return {

       'date': date_str,

       'prompt': prompt

   }



Running Environment

Operating System: Ubuntu 24.04.2 LTS / x86_64Runtime Version: Python 3.11.1




2.4. 科技新闻搜索总结。

2.4.1. 本教程中我们使用“混元AI搜索”插件实现科技新闻的搜索总结,这里也可以通过其他AI搜索插件或自行搭建AI搜索流程。

2.4.2. 从左侧的节点列表中拖拽插件节点,搜索“混元AI搜索”,添加“HunyuanSearchSummary”工具。




2.4.3. 在工具的“body”中,“引用”前序“今日日期和Prompt”节点输出变量中的“prompt”变量。




2.5. 文章整理。

2.5.1. 从左侧的节点列表中拖拽代码节点,整理文章内容及参考来源。代码如下:

import json



def format_references(answer, references):

   # 按Index字段排序参考文献

   sorted_refs = sorted(references, key=lambda x: x['Index'])

   # 构建参考文献部分

   ref_section = " 参考资料: "

   for i, ref in enumerate(sorted_refs, 1):

       title = ref.get('Title', '无标题')

       url = ref.get('Url', '')

       ref_section += f"{i}. {title}"

       if url:

           ref_section += f" ({url})"

       ref_section += " "

   # 将参考文献拼接到回答后面

   full_answer = answer + ref_section

   return full_answer



def main(params: dict) -> dict:

   answer = params.get('answer', '')

   refs = params.get('refs', [])

   full_answer = format_references(answer, refs)

   templateData = json.dumps({"name": full_answer}, ensure_ascii=False)



   return {

       'answer': full_answer,

       'templateData': templateData

   }

Running Environment

Operating System: Ubuntu 24.04.2 LTS / x86_64Runtime Version: Python 3.11.1




2.6. 发送到指定渠道。

2.6.1. 首先使用条件判断节点,判断mss变量值并运行到特定的发送渠道。整体配置如下图:




2.6.2. 调试时可以令mss值为1,此时接入回复节点直接将文章整理的结果输出到调试框。

2.6.3. 发送到邮箱渠道,我们从节点列表中拖入插件节点,搜索“邮件推送”插件,并添加其中的“SendEmail”工具。




说明:

使用“邮件推送”插件前,请在 腾讯云邮件推送 SES控制台 申请密钥并开通服务,以及设定邮件模板。并将密钥、邮件模板ID等信息填写到“SendEmail”工具对应的字段中。详细操作手册请参见 邮件发送。

2.6.3. 发送到短信渠道,我们从节点列表中拖入插件节点,搜索“短信”插件,并添加其中的“SendSms”工具。




说明:

使用“短信”插件前,请在 腾讯云短信 SMS控制台 申请密钥并开通服务,以及设定短信模板。并将密钥、邮件模板ID等信息填写到“SendSms”工具对应的字段中。详细操作手册请参见 快速入门。

第三步:调试工作流

单击页面右上角的调试,进入调试页面,任意输出query调试刚刚配置的工作流,运行结果示意如下:




第四步:发布应用

工作流调试无误后,返回应用配置页面,右上角单击发布,即可发布应用。




应用发布成功后,进入应用发布-服务状态,即可在API管理中获取当前应用的appkey,后续配置云函数时需要使用。




第二部分:基于云函数配置定时任务

第一步:新建函数服务

前往 云函数控制台。

新建函数服务。建议通过“模板创建”,选择“定时播测”(标签是python3.6)的模板进行创建。




第二步:配置函数服务

进入函数服务页面,单击函数管理-函数代码,在编辑框中编辑代码。这里实现通过API调用智能体开发平台的对话接口(对话端接口文档(HTTP SSE)),触发第一步配置的单工作流应用。


示例代码如下,请替换其中的session_id、bot_app_key、visitor_biz_id字段。

import requests

import json



def main_handler(event, context):

   url = 'https://wss.lke.cloud.tencent.com/v1/qbot/chat/sse'



   headers = {

       'Content-Type': 'application/json'

   }



   data = {

       "session_id": "aaaaaaaa",      # 请替换成您自己的session_id

       "bot_app_key": "aaaaaaaa",     # 请替换成应用发布成功后,生成的appkey

       "visitor_biz_id": "aaaaaaa",   # 请替换成您自己的visitor_biz_id

       "content": "触发定时任务",

       "incremental": True,

       "streaming_throttle": 10,

       "visitor_labels": [],

       "custom_variables": {},

       "search_network": "disable",

       "stream": "enable",

       "workflow_status": "disable",

       "tcadp_user_id": ""

   }



   # 使用流式请求处理SSE响应

   response = requests.post(url, headers=headers, data=json.dumps(data), stream=True)



   # 处理SSE事件流

   for line in response.iter_lines():

       if line:

           decoded_line = line.decode('utf-8')

           print(decoded_line)

Running Environment

Operating System: Ubuntu 24.04.2 LTS / x86_64Runtime Version: Python 3.11.1

第三步:配置触发器

单击触发管理-创建触发器,在触发器配置页面根据需求调整“触发方式”、“触发周期”、“立即启用”等配置。




单击提交完成触发器的配置。

至此,定时任务已经配置完成。您可以在设定的时间,通过邮箱或短信接收到当天的科技新闻汇总文章。


热门文章
更多>