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

1✅ 群里语音提问（值班同学开车时发语音问问题）
2✅ 截图、白板照片、仪表盘图片解读
3✅ 需要机器人「开口说话」的场景（播报、语音客服）
4✅ 短视频片段理解（产品演示、故障录屏）
5✅ 对成本敏感的高频交互（输入 ¥0.6 属国产最低档）
6❌ 纯文本的深度推理任务（同门 Qwen3.8-Plus 更稳）
7❌ 超长代码库检索（256K 不如 V4.1 Flash 的 1M）
8❌ 需要严格 JSON Schema 输出的关键业务（多模态模型结构化稳定性略弱）

1export DASHSCOPE_API_KEY=sk-xxxx
2
3curl https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions \
4  -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
5  -H "Content-Type: application/json" \
6  -d '{
7        "model": "qwen3.8-omni-flash",
8        "messages": [{"role":"user","content":"用一句话介绍你自己"}],
9        "max_tokens": 200
10      }'

1curl 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key' \
2   -H 'Content-Type: application/json' \
3   -d '{"msgtype":"text","text":{"content":"Qwen3.8-Omni-Flash 已就位"}}'

1import time, hmac, hashlib, base64, urllib.parse
2import requests
3
4WEBHOOK_KEY = "693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa"
5SECRET      = "SEC-xxxxxxxxxxxx"
6
7def gen_sign(secret: str, timestamp: str) -> str:
8    """秒级 timestamp + 单个 LF 换行 + secret → HMAC-SHA256 → base64 → urlencode。"""
9    raw = f"{timestamp}\n{secret}"
10    digest = hmac.new(secret.encode(), raw.encode(), hashlib.sha256).digest()
11    return urllib.parse.quote_plus(base64.b64encode(digest).decode())
12
13def push(payload: dict):
14    ts = str(int(time.time()))
15    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
16           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(SECRET, ts)}")
17    r = requests.post(url, json=payload, timeout=10).json()
18    if r.get("errcode") != 0:
19        raise RuntimeError(f"推送失败: {r}")
20    return r

1const crypto = require('crypto');
2const https = require('https');
3
4const WEBHOOK_KEY = '693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa';
5const SECRET = 'SEC-xxxxxxxxxxxx';
6
7function genSign(secret, timestamp) {
8  const raw = `${timestamp}\n${secret}`;
9  return encodeURIComponent(crypto.createHmac('sha256', secret).update(raw).digest('base64'));
10}
11
12function push(payload) {
13  const ts = Math.floor(Date.now() / 1000).toString();
14  const url = `https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=${WEBHOOK_KEY}&timestamp=${ts}&sign=${genSign(SECRET, ts)}`;
15  const body = JSON.stringify(payload);
16  return new Promise((resolve, reject) => {
17    const req = https.request(url, {
18      method: 'POST',
19      headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(body) }
20    }, (res) => {
21      let d = ''; res.on('data', (c) => (d += c));
22      res.on('end', () => resolve(JSON.parse(d)));
23    });
24    req.on('error', reject); req.write(body); req.end();
25  });
26}

1# app.py —— pip install flask requests
2import os, time, json, hmac, hashlib, base64, urllib.parse, subprocess, uuid
3import requests
4from flask import Flask, request, jsonify
5
6app = Flask(__name__)
7
8DASHSCOPE_KEY = os.environ["DASHSCOPE_API_KEY"]
9DASHSCOPE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
10MODEL_ID = "qwen3.8-omni-flash"          # 型号
11MAX_AUDIO_SECONDS = 60                    # 语音上限，超了拒绝并提示
12
13WEBHOOK_KEY = os.environ["WECOM_WEBHOOK_KEY"]
14WEBHOOK_SECRET = os.environ["WECOM_WEBHOOK_SECRET"]
15
16SYSTEM_PROMPT = """你是企业微信工作群的 AI 助手，基于通义千问 Qwen3.8-Omni-Flash。
17- 回答控制在 300 字以内
18- 用户可能用语音提问，语音转写可能有错字，请结合上下文推断
19- 不确定就说不确定，不要编造接口名、订单号或人名
20- 输出用 markdown，标题 ##，重点 ** 加粗 **
21"""
22
23def amr_to_mp3(src: str, dst: str):
24    """微信语音是 AMR/SILK 编码，Qwen Omni 不支持，必须先转 mp3。"""
25    subprocess.run(
26        ["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1", dst],
27        check=True, capture_output=True
28    )
29
30def build_audio_content(question: str, mp3_path: str):
31    """Omni 模型的多模态 content 是一个数组，文本和音频分别是一个 item。"""
32    import base64 as b64
33    with open(mp3_path, "rb") as f:
34        data = b64.b64encode(f.read()).decode()
35    return [
36        {"type": "input_audio", "input_audio": {"data": f"data:;base64,{data}", "format": "mp3"}},
37        {"type": "text", "text": question},
38    ]
39
40def ask_qwen(question: str, image_url=None, mp3_path=None) -> str:
41    if mp3_path:
42        content = build_audio_content(question, mp3_path)
43    elif image_url:
44        content = [
45            {"type": "image_url", "image_url": {"url": image_url}},
46            {"type": "text", "text": question},
47        ]
48    else:
49        content = question
50
51    resp = requests.post(
52        DASHSCOPE_URL,
53        headers={"Authorization": f"Bearer {DASHSCOPE_KEY}"},
54        json={
55            "model": MODEL_ID,
56            "messages": [
57                {"role": "system", "content": SYSTEM_PROMPT},
58                {"role": "user", "content": content},
59            ],
60            "temperature": 0.3,
61            "max_tokens": 1500,
62        },
63        timeout=90,          # 多模态推理比纯文本慢，别设 30
64    )
65    resp.raise_for_status()
66    return resp.json()["choices"][0]["message"]["content"]
67
68def gen_sign(secret, ts):
69    raw = f"{ts}\n{secret}".encode()
70    return urllib.parse.quote_plus(
71        base64.b64encode(hmac.new(secret.encode(), raw, hashlib.sha256).digest()).decode()
72    )
73
74def push_markdown(md: str):
75    ts = str(int(time.time()))
76    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
77           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
78    return requests.post(url, json={"msgtype": "markdown", "markdown": {"content": md[:4096]}}, timeout=10).json()
79
80@app.post("/ask")
81def ask():
82    body = request.json or {}
83    q = (body.get("q") or "").strip()
84    if not q and not body.get("image_url") and not body.get("audio_path"):
85        return jsonify(err="empty input"), 400
86
87    if body.get("audio_path"):
88        mp3 = f"/tmp/{uuid.uuid4().hex}.mp3"
89        amr_to_mp3(body["audio_path"], mp3)       # AMR → MP3
90        answer = ask_qwen(q or "请回答这段语音里的问题", mp3_path=mp3)
91    else:
92        answer = ask_qwen(q, image_url=body.get("image_url"))
93
94    push_markdown(f"**Q**: {q or '[图片/语音]'}\n\n{answer}")
95    return jsonify(ok=True, answer=answer)
96
97if __name__ == "__main__":
98    app.run(host="0.0.0.0", port=8000)

1def upload_media(file_path: str, media_type: str = "file") -> str:
2    """第一步：上传临时素材拿 media_id（有效期 3 天）。
3    注意这是 upload_media 接口，不是 send 接口。"""
4    ts = str(int(time.time()))
5    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media"
6           f"?key={WEBHOOK_KEY}&type={media_type}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
7    with open(file_path, "rb") as f:
8        r = requests.post(url, files={"media": f}, timeout=30).json()
9    if r.get("errcode") != 0:
10        raise RuntimeError(f"上传失败: {r}")
11    return r["media_id"]
12
13def push_voice(mp3_path: str):
14    """第二步：用 media_id 发 file 消息。"""
15    media_id = upload_media(mp3_path)
16    ts = str(int(time.time()))
17    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
18           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
19    return requests.post(url, json={"msgtype": "file", "file": {"media_id": media_id}}, timeout=10).json()

1def qwen_tts(text: str, voice: str = "Chelsie", out: str = "/tmp/out.mp3"):
2    resp = requests.post(
3        DASHSCOPE_URL,
4        headers={"Authorization": f"Bearer {DASHSCOPE_KEY}"},
5        json={
6            "model": MODEL_ID,
7            "messages": [{"role": "user", "content": f"请朗读以下内容：{text}"}],
8            "modalities": ["text", "audio"],        # 同时要文字和语音
9            "audio": {"voice": voice, "format": "mp3"},
10            "stream": True,                          # 流式合成，边生成边落盘
11        },
12        stream=True, timeout=120,
13    )
14    import base64 as b64
15    audio = b""
16    for line in resp.iter_lines():
17        if not line or not line.startswith(b"data:"):
18            continue
19        chunk = json.loads(line[5:])
20        for choice in chunk.get("choices", []):
21            a = choice.get("message", {}).get("audio", {})
22            if a.get("data"):
23                audio += b64.b64decode(a["data"])
24    with open(out, "wb") as f:
25        f.write(audio)
26    return out

1import hashlib
2from flask import Flask, request
3
4app = Flask(__name__)
5TOKEN = "QwenBot2026"
6
7@app.get("/wechat")
8def verify():
9    signature = request.args.get("signature", "")
10    timestamp = request.args.get("timestamp", "")
11    nonce     = request.args.get("nonce", "")
12    echostr   = request.args.get("echostr", "")
13    tmp = "".join(sorted([TOKEN, timestamp, nonce]))   # 字典序排序！
14    if hashlib.sha1(tmp.encode()).hexdigest() == signature:
15        return echostr                                  # 原样返回，不多不少
16    return "fail", 403

1pip install wechatpy      # 封装了 WXBizMsgCrypt，别自己抄官方 demo

1from wechatpy.crypto import WeChatCrypto
2from wechatpy.exceptions import InvalidSignatureException
3
4crypto = WeChatCrypto(TOKEN, ENCODING_AES_KEY, APP_ID)
5try:
6    plain_xml = crypto.decrypt_message(body, msg_signature, timestamp, nonce)
7except InvalidSignatureException:
8    return "fail", 403

1用户发语音/图片 → 微信 POST /wechat
2   → ① 下载素材（语音 amr / 图片 jpg）
3   → ② 丢进队列
4   → ③ 立刻返回 "success"（3ms 内）
5   → ④ 后台转码 + 调 Qwen Omni
6   → ⑤ 客服消息接口推送（文本或语音）

1import threading, requests
2from flask import Flask, request
3from wechatpy import parse_message
4
5app = Flask(__name__)
6
7@app.post("/wechat")
8def handle():
9    msg = parse_message(request.data)
10
11    if msg.type == "text":
12        payload = ("text", msg.content)
13    elif msg.type == "voice":
14        payload = ("voice", msg.media_id)     # 语音给的是 media_id，需再下载
15    elif msg.type == "image":
16        payload = ("image", msg.media_id)
17    else:
18        return "success"
19
20    threading.Thread(target=async_answer, args=(msg.source, *payload), daemon=True).start()
21    return "success"          # 立刻返回
22
23def async_answer(openid, kind, content):
24    if kind == "voice":
25        amr = download_media(content)          # 用 access_token 下载 amr
26        mp3 = amr.replace(".amr", ".mp3")
27        amr_to_mp3(amr, mp3)
28        answer = ask_qwen("请回答这段语音里的问题", mp3_path=mp3)
29    elif kind == "image":
30        url = download_media(content)
31        answer = ask_qwen("请解读这张图", image_url=url)
32    else:
33        answer = ask_qwen(content)
34    push_cs_msg(openid, answer)

1import threading, time, requests, redis
2r = redis.Redis(host="localhost", port=6379, db=0)
3_lock = threading.Lock()
4
5def get_wx_token() -> str:
6    tok = r.get("wx:token")
7    if tok and r.ttl("wx:token") > 300:
8        return tok.decode()
9    with _lock:
10        tok = r.get("wx:token")
11        if tok and r.ttl("wx:token") > 300:
12            return tok.decode()
13        resp = requests.get(
14            "https://api.weixin.qq.com/cgi-bin/token",
15            params={"grant_type": "client_credential", "appid": APPID, "secret": APPSECRET},
16            timeout=10,
17        ).json()
18        r.setex("wx:token", int(resp.get("expires_in", 7200)), resp["access_token"])
19        return resp["access_token"]
20
21def download_media(media_id: str) -> str:
22    """下载用户发的语音/图片。语音是 amr，图片是 jpg。"""
23    token = get_wx_token()
24    resp = requests.get(
25        f"https://api.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}",
26        timeout=30,
27    )
28    path = f"/tmp/{media_id}"
29    with open(path, "wb") as f:
30        f.write(resp.content)
31    return path

1下面是一段语音（已转码）。用户可能说的是运维问题或业务咨询。
2
3请：
41. 先判断用户在问什么（一句话概括）
52. 给出直接答案，不要复述问题
63. 如果需要用户补充信息，只问最关键的一个问题
7
8约束：
9- 语音转写可能有同音错字（如「十点二」「四点二」），结合上下文推断
10- 如果语音里是多个不相关的问题，分点回答
11- 答案 250 字以内

1这是一张监控仪表盘截图。请解读：
2
31. **指标现状**：列出你能读到的关键指标（名称 + 当前值）
42. **异常判断**：哪些指标明显异常？给出判断依据
53. **可能原因**：2-3 条，按可能性排序
64. **建议动作**：一句话
7
8如果图中数值无法辨认，直说「无法辨认」，不要编数字。

1你是播报员。下面是要播报的内容：
2
3【粘贴当日数据摘要】
4
5请将它改写成一段口语化播报稿：
6- 时长控制在 40 秒以内（约 120 字）
7- 数字读法要口语化：如「92%」读作「百分之九十二」，不要用符号
8- 不要出现 markdown 符号、括号、英文缩写（把 QPS 读成「每秒请求数」）
9- 开头说「早上好，这是今天的播报」，结尾说「播报完毕」

1阶段 1：企业微信群机器人 + 加签，跑通文本播报        ← 你现在在这里
2阶段 2：公众号服务器配置 + 异步客服消息，实现双向问答
3阶段 3：接入语音输入（AMR 转码）与图片理解
4阶段 4：语音合成播报 + 多模态 Function Calling