AI百炼大模型接入钉钉，实现在群中免@交互式新闻推送

news/2025/9/28 9:50:58/文章来源:https://www.cnblogs.com/llxg/p/19116170

AI百炼大模型接入钉钉自动化推送新闻

进入百炼大模型创建智能体，或者工作流（我这里采用的智能体更为便捷）

创建百炼大模型

设置大模型选项，打开联网搜索----点击插件---插入对应的插件，如新闻插件

设置大模型提示词

创建API key

创建完API，复制进行保存到一个文档中，用于以后用，------网络搜索-----钉钉开放平台

创建应用机器人

重要，保存凭证信息

设置发布渠道
在百炼放大模型中配置自己创建的智能体，----点击发布渠道----钉钉机器人创建----选择API key----下一步----将之前复制保存的，aPP ID、Client ID和 Client Secret ID输入到对应位置

5.创建钉钉模板
然后到钉钉开放平台----创建卡片平台

6.申请机器人权限

7.创建APP FLow连接流
进入阿里云搜索APPFlow---连接流-----钉钉触发-百炼执行-钉钉发送AI卡片消息的连接流

粘贴百炼大模型的应用ID，钉钉的模板ID

8.发布
#注意，需要点击发布，不然日志信息不显示，连接不上
9.安装python
python直接调度API，在linux服务器中，安装python3
`# 1. 更新系统
sudo apt update && sudo apt upgrade -y

2. 添加官方维护的 deadsnakes PPA

sudo apt install -y software-properties-common
sudo add-apt-repository ppa:deadsnakes/ppa -y

3. 安装 Python 3.9（含 venv/dev 头文件）

sudo apt install -y python3.9 python3.9-venv python3.13-dev

4. 验证

python3.9 --version # 应输出 Python 3.9
`

创建python代码调用API Key
`# 用您的百炼API Key代替YOUR_DASHSCOPE_API_KEY
echo "export DASHSCOPE_API_KEY='YOUR_DASHSCOPE_API_KEY'" >> ~/.bashrc

执行以下命令变更生效
source ~/.bashrc

检查是否生效
echo $DASHSCOPE_API_KEY`

创建python代码调用API Key
`vim robot.py

!/usr/bin/env python3

import os
import time
import re
import hmac
import hashlib
import base64
import urllib.parse
import requests
from http import HTTPStatus
from dashscope import Application

---------- 配置 ----------

DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
APP_ID = "【记得插入哦！】"#机器人的APP ID
WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token= [记得插入哦！]" #机器人的webhook
SECRET = "" # 若机器人开启加签，在此填入密钥；否则留空

---------- 工具函数 ----------

def clean_text(text: str) -> str:
"""去 HTML 标签、压缩空行"""
text = re.sub(r"<[^>]+>", "", text)
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()

def gen_sign(secret: str):
timestamp = str(round(time.time() * 1000))
string_to_sign = f"{timestamp}\n{secret}"
hmac_code = hmac.new(
secret.encode("utf-8"),
string_to_sign.encode("utf-8"),
digestmod=hashlib.sha256,
).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return timestamp, sign

def push_to_ding_md(content: str):
"""推送到钉钉群，Markdown 类型"""
if SECRET:
timestamp, sign = gen_sign(SECRET)
url = f"{WEBHOOK}&timestamp={timestamp}&sign={sign}"
else:
url = WEBHOOK

md_text = f"""![日报](插入图片链接) #更加美观

{content}
!"""

data = {"msgtype": "markdown","markdown": {"title": "今日新闻", "text": md_text},"at": {"isAtAll": False},
}
resp = requests.post(url, json=data, headers={"Content-Type": "application/json"})
print("钉钉推送返回：", resp.json())

---------- 主流程 ----------

if name == "main":
if not DASHSCOPE_API_KEY:
raise RuntimeError("请先 export DASHSCOPE_API_KEY=你的KEY")

response = Application.call(api_key=DASHSCOPE_API_KEY,app_id=APP_ID,prompt="今日最新新闻",
)if response.status_code != HTTPStatus.OK:ans = f"DashScope 调用失败：{response.message}"print(ans)
else:ans = clean_text(response.output.text)print("待推送内容：", ans)
push_to_ding_md(ans)