通义千问 Qwen3.8-Omni-Flash 接入 钉钉 完整指南

1❌ 纯文字群聊（贵 50%，用 qwen3.8-plus）
2❌ 需要识别超长代码截图的场景（代码图像识别准确率只有 72%，不如让用户直接发文字）
3❌ 高并发批量图像处理（Omni 的并发额度比纯文本模型紧）

1import time, hmac, hashlib, base64, urllib.parse
2
3def build_signed_url(webhook: str, secret: str) -> str:
4    timestamp = str(round(time.time() * 1000))        # 毫秒级，13 位
5    string_to_sign = f'{timestamp}\n{secret}'          # 换行符是 \n
6    hmac_code = hmac.new(
7        secret.encode('utf-8'),
8        string_to_sign.encode('utf-8'),
9        digestmod=hashlib.sha256
10    ).digest()
11    sign = base64.b64encode(hmac_code).decode('utf-8')
12    return f'{webhook}&timestamp={timestamp}&sign={urllib.parse.quote_plus(sign)}'

1curl https://api.ipify.org

1curl 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN' \
2  -H 'Content-Type: application/json' \
3  -d '{"msgtype":"text","text":{"content":"通义助手已上线，这是连通性测试"}}'

1{
2  "msgtype": "text",
3  "text": { "content": "通义助手：多模态服务已就绪，支持图片与语音" },
4  "at": { "atMobiles": ["13800138000"], "isAtAll": false }
5}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "title": "报错截图解析结果",
5    "text": "### 通义助手 · 截图诊断\n\n- **错误类型**：<font color=#FF0000>NullPointerException</font>\n- **出错位置**：`OrderService.java:142`\n- **直接原因**：`getCoupon()` 返回 null 后未判空\n\n> 建议：第 142 行前加空值判断，或改用 Optional\n\n[查看完整堆栈](https://log.example.com/e/881)"
6  }
7}

1{
2  "msgtype": "link",
3  "link": {
4    "title": "通义助手：会议纪要已生成",
5    "text": "25 分钟语音已转写，识别出 6 项待办",
6    "picUrl": "https://example.com/meeting.png",
7    "messageUrl": "https://doc.example.com/m/881"
8  }
9}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "通义助手：检测到 1 个高危报错",
5    "text": "### 线上异常\n\n- `OrderService.java:142` 空指针，近 1 小时 37 次",
6    "hideAvatar": "0",
7    "btnOrientation": "0",
8    "btns": [
9      { "title": "查看日志", "actionURL": "https://log.example.com/e/881" },
10      { "title": "建工单", "actionURL": "https://oa.example.com/new?t=npe" }
11    ]
12  }
13}

1{
2  "msgtype": "feedCard",
3  "feedCard": {
4    "links": [
5      { "title": "通义日报：今日处理 43 张截图", "messageURL": "https://ops.example.com/today", "picURL": "https://example.com/1.png" },
6      { "title": "3 段会议语音已转纪要", "messageURL": "https://doc.example.com/m", "picURL": "https://example.com/2.png" }
7    ]
8  }
9}

1import time, threading
2from collections import deque
3
4class RateLimiter:
5    """钉钉每个机器人每分钟最多 20 条"""
6    def __init__(self, max_calls=20, period=60):
7        self.max_calls, self.period = max_calls, period
8        self.calls, self.lock = deque(), threading.Lock()
9
10    def acquire(self):
11        with self.lock:
12            now = time.time()
13            while self.calls and now - self.calls[0] > self.period:
14                self.calls.popleft()
15            if len(self.calls) >= self.max_calls:
16                time.sleep(self.period - (now - self.calls[0]) + 0.5)
17                return self.acquire()
18            self.calls.append(now)
19
20limiter = RateLimiter()

1def send_long_answer(text: str, title: str = '通义助手', limit: int = 4000):
2    chunks, buf = [], ''
3    for para in text.split('\n\n'):
4        if len(buf) + len(para) + 2 > limit:
5            chunks.append(buf); buf = para
6        else:
7            buf = f'{buf}\n\n{para}' if buf else para
8    if buf:
9        chunks.append(buf)
10    for i, chunk in enumerate(chunks, 1):
11        limiter.acquire()
12        send_markdown(chunk, title=f'{title}（{i}/{len(chunks)}）')

1CorpId:     ding7a8b9c0d1e2f3a4b
2AppKey:     dingoaqxxxxxxxxxxxxxxx
3AppSecret:  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

1import time, threading, requests
2
3class DingToken:
4    def __init__(self, appkey, appsecret, early=300):
5        self.appkey, self.appsecret, self.early = appkey, appsecret, early
6        self._token, self._expire_at, self._lock = None, 0, threading.Lock()
7
8    def get(self):
9        with self._lock:                       # 并发下只刷新一次
10            if self._token and time.time() < self._expire_at:
11                return self._token
12            r = requests.get('https://oapi.dingtalk.com/gettoken',
13                             params={'appkey': self.appkey,      # 全小写
14                                     'appsecret': self.appsecret},
15                             timeout=10)
16            data = r.json()
17            if data.get('errcode') != 0:
18                raise RuntimeError(f'token 获取失败：{data}')
19            self._token = data['access_token']
20            self._expire_at = time.time() + data['expires_in'] - self.early
21            return self._token

1import os, requests
2
3def download_media(download_code: str, ding_token: str) -> bytes:
4    """用 downloadCode 换真实文件内容"""
5    r = requests.post(
6        'https://oapi.dingtalk.com/media/downloadFile',
7        params={'access_token': ding_token},
8        json={'downloadCode': download_code, 'robotCode': os.environ['DINGTALK_APPKEY']},
9        timeout=30
10    )
11    body = r.json()
12    if body.get('errcode') not in (0, None):
13        raise RuntimeError(f'获取下载链接失败：{body}')
14    download_url = body['result']['downloadUri']     # 带时效签名的真实地址
15
16    # 立刻下载，别存 URL 慢慢处理
17    return requests.get(download_url, timeout=60).content

1import base64, io
2from PIL import Image
3
4def image_to_data_uri(raw: bytes, max_side: int = 1280) -> str:
5    """压缩长边到 1280 再转 base64：省 40% token 且准确率几乎不掉"""
6    img = Image.open(io.BytesIO(raw)).convert('RGB')
7    w, h = img.size
8    scale = min(1.0, max_side / max(w, h))
9    if scale < 1.0:
10        img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
11    buf = io.BytesIO()
12    img.save(buf, format='JPEG', quality=85)          # JPEG 比 PNG 省一半
13    b64 = base64.b64encode(buf.getvalue()).decode()
14    return f'data:image/jpeg;base64,{b64}'

1from openai import OpenAI
2
3qwen = OpenAI(
4    api_key=os.environ['DASHSCOPE_API_KEY'],
5    base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
6)
7
8def ask_with_image(data_uri: str, question: str) -> str:
9    resp = qwen.chat.completions.create(
10        model='qwen3.8-omni-flash',
11        messages=[{
12            'role': 'user',
13            'content': [
14                # 图片和文本放在同一个 content 数组里，顺序会影响理解
15                {'type': 'image_url', 'image_url': {'url': data_uri}},
16                {'type': 'text', 'text': question},
17            ]
18        }],
19        temperature=0.2,
20        max_tokens=800,
21        # Omni 支持同时输出文本和语音，钉钉场景默认关掉省带宽
22        modalities=['text'],
23    )
24    return resp.choices[0].message.content

1ffmpeg -i input.amr -ar 16000 -ac 1 output.mp3

1def ask_with_audio(mp3_path: str, question: str) -> str:
2    with open(mp3_path, 'rb') as f:
3        b64 = base64.b64encode(f.read()).decode()
4    resp = qwen.chat.completions.create(
5        model='qwen3.8-omni-flash',
6        messages=[{
7            'role': 'user',
8            'content': [
9                {'type': 'input_audio',
10                 'input_audio': {'data': f'data:audio/mp3;base64,{b64}'}},
11                {'type': 'text', 'text': question},
12            ]
13        }],
14        temperature=0.2, max_tokens=1200, modalities=['text'],
15    )
16    return resp.choices[0].message.content

1pip install dingtalk-stream

1import os
2import dingtalk_stream
3from dingtalk_stream import AckMessage
4
5class QwenCardBot(dingtalk_stream.ChatbotHandler):
6    def __init__(self, token_mgr):
7        super().__init__()
8        self.replier = dingtalk_stream.AICardReplier()
9        self.token_mgr = token_mgr
10
11    async def process(self, callback: dingtalk_stream.CallbackMessage):
12        incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
13        content = incoming.message_content
14        question = (content.get('text') or {}).get('content', '').strip()
15
16        # 1. 先建卡片占位：用户立刻看到标题栏，回调不超时
17        card = self.replier.create_and_send_card(incoming, title='通义正在看图…')
18
19        # 2. 有图片就走多模态，没有就走纯文本
20        if 'richText' in content or 'imageContent' in content:
21            code = content['imageContent']['downloadCode']
22            raw = download_media(code, self.token_mgr.get())
23            data_uri = image_to_data_uri(raw)
24            stream = stream_qwen_image(data_uri, question or '请描述并分析这张图')
25        else:
26            stream = stream_qwen_text(question)
27
28        # 3. 流式推送，攒够 20 字再刷
29        buf = ''
30        for chunk in stream:
31            buf += chunk
32            if len(buf) >= 20:
33                self.replier.replay_streaming(card.card_id, buf, finished=False)
34
35        # 4. 结束帧（漏了卡片一直转圈）
36        self.replier.replay_streaming(card.card_id, buf, finished=True)
37        return AckMessage.STATUS_OK, 'ok'
38
39def stream_qwen_image(data_uri: str, question: str):
40    resp = qwen.chat.completions.create(
41        model='qwen3.8-omni-flash',
42        messages=[{'role': 'user', 'content': [
43            {'type': 'image_url', 'image_url': {'url': data_uri}},
44            {'type': 'text', 'text': question}]}],
45        stream=True, temperature=0.2, max_tokens=800, modalities=['text']
46    )
47    for chunk in resp:
48        if chunk.choices and chunk.choices[0].delta.content:
49            yield chunk.choices[0].delta.content
50
51def stream_qwen_text(question: str):
52    resp = qwen.chat.completions.create(
53        model='qwen3.8-plus',            # 纯文字切 plus，省 33%
54        messages=[{'role': 'user', 'content': question}],
55        stream=True, temperature=0.3, max_tokens=800
56    )
57    for chunk in resp:
58        if chunk.choices and chunk.choices[0].delta.content:
59            yield chunk.choices[0].delta.content
60
61credential = dingtalk_stream.Credential(os.environ['DINGTALK_APPKEY'],
62                                        os.environ['DINGTALK_APPSECRET'])
63client = dingtalk_stream.DingTalkStreamClient(credential)
64client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
65                                 QwenCardBot(DingToken(os.environ['DINGTALK_APPKEY'],
66                                                       os.environ['DINGTALK_APPSECRET'])))
67client.start_forever()

1@通义助手
2
3[图片：IDE 报错截图]
4
5这是什么错误？请给出：
61. 错误类型和直接原因（一句话）
72. 出错的代码位置（文件名 + 行号）
83. 修复建议（给出修改后的代码片段）
94. 如果是配置问题，指出具体配置项
10
11300 字以内，代码用代码块。

1@通义助手
2
3[语音：25 分钟项目周会录音]
4
5请输出：
61. 会议主题与参会人（从内容推断，不确定标「待确认」）
72. 已确认的决策（每条一行，「动词 + 对象」句式）
83. 待办事项：负责人 + 事项 + 截止时间（缺失标「未指定」）
94. 有争议未决的问题（单独列出）
10
11500 字以内，不要复述讨论过程。

1@通义助手
2
3[图片：Grafana 监控面板]
4
5请做例行巡检：
61. 哪些面板的曲线有明显异常（尖刺、断崖、持续高位）？给出面板名和大致时间点
72. 有没有面板数值突破了常识阈值（如错误率 > 5%、RT > 2s）？
83. 按严重程度排序，最需要立刻处理的排第一
94. 如果整体正常，输出「巡检通过」
10
11300 字以内。

1@通义助手
2
3[图片：App 首页设计稿]
4
5请从可用性角度评审：
61. 主操作按钮是否足够突出（位置、尺寸、颜色对比）
72. 信息层级是否清晰（标题/正文/辅助文字的区分度）
83. 有没有明显的视觉拥挤或留白失衡区域
94. 文字是否存在过小（< 12px 等效）或对比度不足
10
11每条给出「问题 + 位置描述 + 修改建议」，最多 5 条。
12没问题输出「设计通过」。

1# 招式一：图片压缩（省 50%）
2data_uri = image_to_data_uri(raw, max_side=1280)
3
4# 招式二：纯文字自动降级到 plus（省 33%）
5model = 'qwen3.8-omni-flash' if has_image_or_audio else 'qwen3.8-plus'
6
7# 招式三：长语音先做 VAD 切静音（再省 18%）
8# ffmpeg -i in.mp3 -af silenceremove=1:0:-50dB out.mp3

1阶段 1（入门）：群机器人 Webhook + markdown 推送                  ← 你现在在这里
2阶段 2（进阶）：加签 + 令牌桶限流 + 五种消息类型
3阶段 3（高级）：企业内部应用 + 媒体下载 + 多模态消息体 + 流式卡片
4阶段 4（专家）：多模态 RAG + Function Calling + 语音播报