文档中心>腾讯云智能体开发平台>实践教程>基于工作流的定时科技新闻推送小助手

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

最近更新时间:2025-09-25 11:36:21

我的收藏
本文以搭建定时科技新闻推送小助手为例,详细介绍如何基于智能体开发平台创建单工作流应用并使用 云函数 实现定时任务功能。
应用目标
1. 使用智能体开发平台搭建“科技新闻推送小助手”应用,实现科技新闻收集及新闻推送功能。
2. 使用云函数定时调用“科技新闻推送小助手”应用,实现定时任务功能。

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

第一步:新建应用

2. 新建应用。
2.1 在左侧菜单栏,单击应用开发



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



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

第二步:配置工作流

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



1.2. 单击确定,自动跳转至工作流画布页面。
2. 在画布中配置工作流
2.1 工作流需要实现“今日日期和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 = '\\n'.join(prompt_str)

return {
'date': date_str,
'prompt': prompt
}




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 = "\\n\\n参考资料:\\n"
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 += "\\n"
# 将参考文献拼接到回答后面
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
}



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”工具对应的字段中。详细操作手册请参见 快速入门

第三步:调试工作流

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




第四步:发布应用

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



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




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

第一步:新建函数服务

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




第二步:配置函数服务

1. 进入函数服务页面,单击函数管理-函数代码,在编辑框中编辑代码。这里实现通过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)

第三步:配置触发器

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



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