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

1图片：按分辨率折算，约 1 张 720P ≈ 400-800 tokens，1080P ≈ 800-1500 tokens
2音频：按秒折算，约 1 秒 ≈ 25-40 tokens，一分钟会议录音约 1500-2400 tokens
3视频：按抽帧折算，8 帧 ≈ 3200-6400 tokens，超过 16 帧性价比骤降

1纯文本 35 次：35 × (500/1M × 0.5 + 400/1M × 2) = 35 × 0.00105 = ¥0.037
2带图   25 次：25 × ((500+900)/1M × 0.5 + 400/1M × 2) = 25 × 0.0015 = ¥0.037
3合计：约 ¥0.074/天，约 ¥2.2/月

1❌ 需要深度多步推理的复杂分析（Flash 档位，切 qwen3.8-max 或推理型模型）
2❌ 视频理解要逐帧精确（超过 16 帧性价比骤降，建议抽关键帧 + 人工确认）
3❌ 要求严格 JSON Schema 给下游消费（建议加后处理校验 + 失败重试）
4❌ 超长文档问答（闪档上下文有限，长文档请做切片检索）

1Webhook 地址：
2https://open.feishu.cn/open-apis/bot/v2/hook/5e4d3c2b-1a09-8f7e-6d5c-4b3a29180f7e
3
4密钥（secret）：
5Rn5Tq8Wz3Xc6Yb2

1import hmac, hashlib, time
2
3def gen_sign(secret: str) -> tuple[str, str]:
4    """飞书自定义机器人加签。三个易错点：
5    1) timestamp 用秒，不要 int(time.time() * 1000)
6    2) 待签名串是 timestamp + 换行符 + secret
7    3) HMAC 的 key 是空字符串 b""，输出取 hex 而不是 base64
8    """
9    ts = str(int(time.time()))
10    sign = hmac.new(b"", f"{ts}\n{secret}".encode("utf-8"), hashlib.sha256).hexdigest()
11    return ts, sign

1import httpx
2
3WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/5e4d3c2b-1a09-8f7e-6d5c-4b3a29180f7e"
4SECRET = "Rn5Tq8Wz3Xc6Yb2"
5
6def push(msg_type: str, content: dict) -> dict:
7    ts, sign = gen_sign(SECRET)
8    body = {"timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content}
9    return httpx.post(WEBHOOK, json=body, timeout=10).json()
10    # 成功 {"code":0,"msg":"success"}；失败 {"code":19021,"msg":"sign match fail..."}
11
12print(push("text", {"text": "通义千问接入测试"}))

1import crypto from 'node:crypto';
2
3const WEBHOOK = 'https://open.feishu.cn/open-apis/bot/v2/hook/5e4d3c2b-1a09-8f7e-6d5c-4b3a29180f7e';
4const SECRET = 'Rn5Tq8Wz3Xc6Yb2';
5
6function genSign(secret) {
7  const ts = String(Math.floor(Date.now() / 1000));   // 秒级！不要 Date.now()
8  const sign = crypto
9    .createHmac('sha256', Buffer.alloc(0))             // 空 Buffer 作密钥
10    .update(`${ts}\n${secret}`)
11    .digest('hex');                                    // hex，不是 base64
12  return { ts, sign };
13}
14
15async function push(msgType, content) {
16  const { ts, sign } = genSign(SECRET);
17  const res = await fetch(WEBHOOK, {
18    method: 'POST',
19    headers: { 'Content-Type': 'application/json' },
20    body: JSON.stringify({ timestamp: ts, sign, msg_type: msgType, content }),
21  });
22  return res.json();
23}

1# mac / linux
2echo 'export DASHSCOPE_API_KEY="sk-ab12cd34ef56gh78ij90kl12mn34op56"' >> ~/.bashrc && source ~/.bashrc
3
4# windows powershell
5setx DASHSCOPE_API_KEY "sk-ab12cd34ef56gh78ij90kl12mn34op56"

1Base URL（OpenAI 兼容）：https://dashscope.aliyuncs.com/compatible-mode/v1
2Model：                  qwen3.8-omni-flash

1feishu-qwen-bot/
2├── .env          # 密钥，务必加进 .gitignore
3├── .gitignore    # 至少一行：.env
4├── bot.py
5└── card.py

1import os, json, time, hmac, hashlib, httpx
2
3WEBHOOK = os.environ["FEISHU_WEBHOOK"]
4SECRET  = os.environ["FEISHU_SECRET"]
5DS_KEY  = os.environ["DASHSCOPE_API_KEY"]
6DS_URL  = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
7
8SYSTEM_PROMPT = """你是飞书群里的 AI 助手，底层模型是通义千问 Qwen3.8-Omni-Flash。
9回答要求：
101. 中文，简洁，控制在 300 字以内，能用列表就用列表
112. 用户发来图片时，先说明你看到了什么，再回答他的问题
123. 图片里的文字请原样引用，不要改写
134. 看不清或不确定就直说，不要猜测图片内容
14"""
15
16def ask_qwen(messages: list, model: str = "qwen3.8-omni-flash",
17             timeout: int = 30) -> tuple[str, int]:
18    """messages 支持多模态 content 数组。返回 (回答, 总 token 消耗)"""
19    r = httpx.post(
20        DS_URL,
21        headers={"Authorization": f"Bearer {DS_KEY}"},
22        json={"model": model, "messages": messages,
23              "temperature": 0.3, "max_tokens": 1024},
24        timeout=timeout,
25    )
26    data = r.json()
27    if "choices" not in data:
28        raise RuntimeError(f"通义千问调用失败：{data}")
29    return data["choices"][0]["message"]["content"], data["usage"]["total_tokens"]
30
31def text_messages(question: str) -> list:
32    return [
33        {"role": "system", "content": SYSTEM_PROMPT},
34        {"role": "user", "content": question},
35    ]
36
37def gen_sign(secret: str) -> tuple[str, str]:
38    ts = str(int(time.time()))
39    return ts, hmac.new(b"", f"{ts}\n{secret}".encode(), hashlib.sha256).hexdigest()
40
41def push(msg_type: str, content: dict) -> dict:
42    ts, sign = gen_sign(SECRET)
43    return httpx.post(WEBHOOK, json={
44        "timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content
45    }, timeout=10).json()
46
47if __name__ == "__main__":
48    q = "用一句话说明什么是全模态模型"
49    answer, tokens = ask_qwen(text_messages(q))
50    print(push("text", {"text": f"{answer}\n\n（qwen3.8-omni-flash · {tokens} tokens）"}))

1pip install httpx python-dotenv
2python bot.py
3# 控制台打印 {'code': 0, 'msg': 'success', 'data': {}} 即链路打通

1import base64
2from PIL import Image
3import io
4
5def compress_image(path: str, max_side: int = 768) -> bytes:
6    """把图片短边压到 768px 以内。
7    这是省钱的关键：1080P 相比 720P 准确率只高 1 个点，成本却高 27%。
8    """
9    img = Image.open(path).convert("RGB")
10    w, h = img.size
11    scale = min(1.0, max_side / max(w, h))
12    if scale < 1.0:
13        img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
14    buf = io.BytesIO()
15    img.save(buf, format="JPEG", quality=85)
16    return buf.getvalue()
17
18def image_messages(question: str, image_bytes: bytes) -> list:
19    b64 = base64.b64encode(image_bytes).decode()
20    return [
21        {"role": "system", "content": SYSTEM_PROMPT},
22        {"role": "user", "content": [
23            # data URI 前缀必须带，写成裸 base64 会返回 400
24            {"type": "image_url",
25             "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
26            {"type": "text", "text": question},
27        ]},
28    ]
29
30raw = compress_image("./error_screenshot.png")
31answer, tokens = ask_qwen(image_messages("这张截图报的是什么错？怎么处理？", raw))

1{"type": "image_url",
2 "image_url": {"url": "https://your-bucket.oss-cn-hangzhou.aliyuncs.com/shot.jpg"}}

1# 音频：type 为 audio_url，支持 data URI 或公网 URL
2{"type": "audio_url", "audio_url": {"url": f"data:audio/mp3;base64,{b64}"}}
3
4# 视频：type 为 video_url，建议先抽帧，超过 16 帧性价比骤降
5{"type": "video_url", "video_url": {"url": "https://xxx.oss.../clip.mp4", "fps": 2}}

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": { "wide_screen_mode": true, "enable_forward": true },
5    "header": {
6      "template": "green",
7      "title": { "tag": "plain_text", "content": "通义千问 Qwen3.8-Omni-Flash · 回答" },
8      "subtitle": { "tag": "plain_text", "content": "图片理解 · 1.9s · 1,204 tokens" }
9    },
10    "elements": [
11      { "tag": "div", "text": { "tag": "lark_md", "content": "**你发了一张图**\n错误截图 · 720P" } },
12      { "tag": "hr" },
13      { "tag": "div", "text": { "tag": "lark_md", "content": "**我看到了**\n一个红色的错误弹窗，首行是 `ConnectionPoolTimeout`" } },
14      { "tag": "div", "text": { "tag": "lark_md", "content": "**处理建议**\n1. 检查数据库连接池配置\n2. 查看慢查询日志" } },
15      { "tag": "note", "elements": [
16        { "tag": "plain_text", "content": "AI 生成内容，涉及生产操作请人工确认" }
17      ]}
18    ],
19    "actions": [
20      { "tag": "action", "actions": [
21        { "tag": "button", "text": { "tag": "plain_text", "content": "再放大一点看" },
22          "type": "primary", "value": { "action": "zoom", "img": "oss://shot.jpg" } },
23        { "tag": "button", "text": { "tag": "plain_text", "content": "转成工单" },
24          "type": "default", "value": { "action": "ticket" } },
25        { "tag": "button", "text": { "tag": "plain_text", "content": "解释专业术语" },
26          "type": "default", "value": { "action": "explain" } }
27      ]}
28    ]
29  }
30}

1def build_vision_card(question: str, seen: str, answer: str,
2                      elapsed: float, tokens: int) -> dict:
3    return {
4        "config": {"wide_screen_mode": True, "enable_forward": True},
5        "header": {
6            "template": "green",
7            "title": {"tag": "plain_text", "content": "通义千问 Qwen3.8-Omni-Flash · 回答"},
8            "subtitle": {"tag": "plain_text",
9                         "content": f"图片理解 · {elapsed:.1f}s · {tokens} tokens"},
10        },
11        "elements": [
12            {"tag": "div", "text": {"tag": "lark_md", "content": f"**你的问题**\n{question}"}},
13            {"tag": "hr"},
14            {"tag": "div", "text": {"tag": "lark_md", "content": f"**我看到了**\n{seen}"}},
15            {"tag": "div", "text": {"tag": "lark_md", "content": f"**建议**\n{answer}"}},
16            {"tag": "note", "elements": [
17                {"tag": "plain_text", "content": "AI 生成内容，涉及生产操作请人工确认"}
18            ]},
19        ],
20        "actions": [{
21            "tag": "action",
22            "actions": [
23                {"tag": "button", "text": {"tag": "plain_text", "content": "再放大一点看"},
24                 "type": "primary", "value": {"action": "zoom"}},
25                {"tag": "button", "text": {"tag": "plain_text", "content": "转成工单"},
26                 "type": "default", "value": {"action": "ticket"}},
27            ],
28        }],
29    }
30
31push("interactive", build_vision_card(q, seen, answer, 1.9, 1204))

1App ID：     cli_d4e5f6g7h8i9j0k1
2App Secret： Rn5Tq8Wz3Xc6Yb2A3b4C5d6E7f8G9h0

1import os, time, threading, httpx
2
3APP_ID = os.environ["FEISHU_APP_ID"]
4APP_SECRET = os.environ["FEISHU_APP_SECRET"]
5FS_TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
6
7_fs = {"token": None, "expire_at": 0}
8_fs_lock = threading.Lock()
9
10def get_tenant_access_token() -> str:
11    """有效期 2 小时，过期前 5 分钟刷新。
12    为什么必须缓存：飞书对鉴权接口有独立限流，高频换取会被拒绝，
13    进而导致所有发消息请求失败。
14    """
15    now = time.time()
16    if _fs["token"] and _fs["expire_at"] - now > 300:
17        return _fs["token"]
18    with _fs_lock:
19        if _fs["token"] and _fs["expire_at"] - time.time() > 300:
20            return _fs["token"]
21        r = httpx.post(FS_TOKEN_URL, json={
22            "app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10)
23        data = r.json()
24        if data.get("code") != 0:
25            raise RuntimeError(f"获取 tenant_access_token 失败：{data}")
26        _fs["token"] = data["tenant_access_token"]
27        _fs["expire_at"] = now + data["expire"]     # 通常 7200 秒
28        return _fs["token"]

1SEND_URL = "https://open.feishu.cn/open-apis/im/v1/messages"
2
3def download_feishu_image(message_id: str, image_key: str) -> bytes:
4    """用飞书接口下载用户发来的图片"""
5    r = httpx.get(
6        f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/resources/{image_key}",
7        params={"type": "image"},
8        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
9        timeout=30,
10    )
11    return r.content
12
13def send_to_chat(chat_id: str, card: dict, receive_id_type: str = "chat_id") -> str:
14    r = httpx.post(SEND_URL, params={"receive_id_type": receive_id_type},
15        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
16        json={"receive_id": chat_id, "msg_type": "interactive",
17              "content": json.dumps(card, ensure_ascii=False)},   # 必须字符串化
18        timeout=10).json()
19    return r["data"]["message_id"]
20
21def update_card(message_id: str, card: dict):
22    return httpx.patch(f"{SEND_URL}/{message_id}",
23        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
24        json={"content": json.dumps(card, ensure_ascii=False)}, timeout=10).json()

1import os, json as _json
2import lark_oapi as lark
3from lark_oapi.api.im.v1 import *
4
5def on_receive(data: P2ImMessageReceiveV1) -> None:
6    msg = data.event.message
7    content = _json.loads(msg.content)
8
9    if msg.message_type == "text":
10        question = content.get("text", "").strip()
11        if not question:
12            return
13        answer, tokens = ask_qwen(text_messages(question))
14        card = build_vision_card(question, "（纯文本提问）", answer, 0, tokens)
15
16    elif msg.message_type == "image":
17        image_key = content["image_key"]
18        raw = download_feishu_image(msg.message_id, image_key)
19        raw = compress_bytes(raw)                     # 压到 768px，省 40% tokens
20        question = "这张图说了什么？如果里面有报错，请告诉我怎么处理。"
21        answer, tokens = ask_qwen(image_messages(question, raw))
22        card = build_vision_card(question, "一张截图", answer, 0, tokens)
23
24    else:
25        return                                        # 忽略表情、系统消息等
26
27    client.im.v1.message.reply(ReplyMessageRequest.builder()
28        .message_id(msg.message_id)
29        .request_body(ReplyMessageRequestBody.builder()
30            .msg_type("interactive")
31            .content(_json.dumps(card, ensure_ascii=False)).build())
32        .build())
33
34handler = lark.EventDispatcherHandler.builder("", "") \
35    .register_p2_im_message_receive_v1(on_receive).build()
36client = lark.ws.Client(os.environ["FEISHU_APP_ID"], os.environ["FEISHU_APP_SECRET"],
37                        event_handler=handler, log_level=lark.LogLevel.INFO)
38client.start()

1@app.post("/feishu/event")
2async def feishu_event(req: Request):
3    body = await req.json()
4    if "challenge" in body:                      # 1. URL 校验，1 秒内返回
5        return {"challenge": body["challenge"]}
6    if "encrypt" in body:                        # 2. 配了 Encrypt Key 时是密文
7        body = decrypt(body["encrypt"], os.environ["FEISHU_ENCRYPT_KEY"])
8    if body.get("header", {}).get("event_type") == "im.message.receive_v1":
9        asyncio.create_task(handle_message(body["event"]))   # 异步，别阻塞
10    return {"code": 0}

1import base64, hashlib
2from Crypto.Cipher import AES      # pip install pycryptodome
3
4def decrypt(encrypt_b64: str, encrypt_key: str) -> dict:
5    """飞书 Encrypt Key 是 43 位字符串，SHA256 后取前 32 字节作密钥"""
6    key = hashlib.sha256(encrypt_key.encode()).digest()[:32]
7    raw = base64.b64decode(encrypt_b64)
8    iv, cipher_text = raw[:16], raw[16:]           # 前 16 字节是 IV
9    plain = AES.new(key, AES.MODE_CBC, iv).decrypt(cipher_text)
10    return _json.loads(plain[:-plain[-1]].decode("utf-8"))    # 去 PKCS7 填充

1第 1 天：可用范围只放自己，单聊自测 + 发一张报错截图
2第 2 天：拉 3 人测试群，验证 @、私聊、图片、卡片按钮
3第 3 天：放开到 20 人小群，观察 token 消耗与响应时长
4第 5 天：全部门放开，群里置顶「机器人使用说明」

1- [x] 密钥全在环境变量里，git status 看不到 .env
2- [x] 连续发 100 条测试消息，没有触发飞书限流
3- [x] 图片已做压缩（短边 ≤ 768px），单图 tokens 在预期内
4- [x] 断网 10 秒再恢复，长连接能自动重连
5- [x] 卡片按钮点下去都有正确回传
6- [x] 百炼控制台设了用量告警

1@通义 Qwen3.8 Omni 助手
2（发一张截图）
3这是什么错？最可能的原因是什么？我下一步该点哪里？

1@通义 Qwen3.8 Omni 助手
2（发一张设计稿）
3按移动端可用性检查这份设计稿：
41）有没有点击区域小于 44px 的元素；2）文字对比度够不够；
53）信息层级是否清晰。按严重程度排序输出。

1@通义 Qwen3.8 Omni 助手
2（发一段录音）
3转成文字并整理成会议纪要：结论 3 条 + 待办清单（负责人 + 时间），
4口语词去掉，不要逐字稿。

1@通义 Qwen3.8 Omni 助手
2（发一张看板截图）
3这个看板里哪个指标最值得关注？请用「数字 + 变化 + 可能的业务原因」的格式，
4给出 3 条观察。

1阶段 1（入门）：群自定义机器人 + 加签，能把通义千问的回答推进群里      ← 你现在在这里
2阶段 2（进阶）：多模态消息体，让机器人看懂群里发的截图
3阶段 3（高级）：交互卡片 + 占位卡 + 异步更新，把感知延迟压到 0.3 秒
4阶段 4（专家）：自建应用收图 + 长连接事件订阅 + 音频转写与抽帧
5阶段 5（架构）：Webhook 回调 + 多副本 + KMS 密钥管理 + 敏感图片脱敏