豆包 1.5 Pro 接入微信完整指南：企业微信群机器人 + 公众号双路径实测

1✅ 营销文案、种草笔记、短视频脚本
2✅ 客服话术生成与改写
3✅ 社群运营内容（欢迎语、活动公告）
4✅ 口语化、有人味的表达
5✅ 已经在用火山引擎、能复用账单的团队
6❌ 长文档摘要（32K 上下文不够）
7❌ 超长内容一次生成（输出上限 4K）
8❌ 需要严格 JSON Schema 输出的关键链路
9❌ 对内容安全宽松度要求高的场景（容易被拦）

1export ARK_API_KEY=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
2export ARK_ENDPOINT_ID=ep-20260513-8x9k2m
3
4curl https://ark.cn-beijing.volces.com/api/v3/chat/completions \
5  -H "Authorization: Bearer $ARK_API_KEY" \
6  -H "Content-Type: application/json" \
7  -d "{
8        \"model\": \"$ARK_ENDPOINT_ID\",
9        \"messages\": [{\"role\":\"user\",\"content\":\"用一句话介绍你自己\"}],
10        \"max_tokens\": 200
11      }"

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

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, hmac, hashlib, base64, urllib.parse
3import requests
4from flask import Flask, request, jsonify
5
6app = Flask(__name__)
7
8ARK_KEY     = os.environ["ARK_API_KEY"]
9ARK_URL     = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
10ENDPOINT_ID = os.environ["ARK_ENDPOINT_ID"]     # 形如 ep-20260513-8x9k2m
11
12WEBHOOK_KEY    = os.environ["WECOM_WEBHOOK_KEY"]
13WEBHOOK_SECRET = os.environ["WECOM_WEBHOOK_SECRET"]
14
15# 豆包对绝对化用语敏感，写进 system prompt 能显著降低拦截率
16SYSTEM_PROMPT = """你是企业微信工作群的 AI 助手，基于豆包 1.5 Pro。
17- 回答控制在 300 字以内
18- 写营销文案时：口语化、有场景感，不使用「最好」「第一」「顶级」「彻底」「根治」等绝对化用语
19- 不承诺医疗功效、投资回报、法律效力
20- 不确定就说不确定，不要编造数据、功效或用户评价
21- 输出用 markdown，标题 ##，重点 ** 加粗 **
22"""
23
24BANNED_WORDS = ["最好", "第一", "顶级", "彻底", "根治", "国家级", "绝对", "100%"]
25
26def sanitize(text: str) -> str:
27    """推送前把绝对化用语替换掉，降低被内容安全拦截的概率。"""
28    for w in BANNED_WORDS:
29        text = text.replace(w, "")
30    return text
31
32def ask_doubao(question: str, retry: int = 2) -> str:
33    """带一次改写重试：被内容安全拦截时，加上更强的约束再生成。"""
34    prompt = question
35    for attempt in range(retry + 1):
36        resp = requests.post(
37            ARK_URL,
38            headers={"Authorization": f"Bearer {ARK_KEY}"},
39            json={
40                "model": ENDPOINT_ID,          # 👈 注意：填接入点 ID，不是模型名
41                "messages": [
42                    {"role": "system", "content": SYSTEM_PROMPT},
43                    {"role": "user", "content": prompt},
44                ],
45                "temperature": 0.7 if "文案" in question else 0.3,   # 文案要创意，问答要稳
46                "max_tokens": 1500,
47            },
48            timeout=60,
49        )
50        data = resp.json()
51        if "choices" in data:
52            content = data["choices"][0]["message"]["content"]
53            if content and content.strip():
54                return content
55        # 被拦截或返回空 → 加强约束重试
56        prompt = (f"{question}\n\n（上一次生成未通过审核，请务必避免使用绝对化用语"
57                  f"和功效承诺，改用更克制的表达重试）")
58    return "内容生成未通过审核，请换个说法再试。"
59
60def gen_sign(secret, ts):
61    raw = f"{ts}\n{secret}".encode()
62    return urllib.parse.quote_plus(
63        base64.b64encode(hmac.new(secret.encode(), raw, hashlib.sha256).digest()).decode()
64    )
65
66def push_markdown(md: str):
67    ts = str(int(time.time()))
68    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
69           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
70    return requests.post(
71        url,
72        json={"msgtype": "markdown", "markdown": {"content": sanitize(md)[:4096]}},
73        timeout=10,
74    ).json()
75
76@app.post("/ask")
77def ask():
78    q = ((request.json or {}).get("q") or "").strip()
79    if not q:
80        return jsonify(err="empty question"), 400
81    answer = ask_doubao(q)
82    push_markdown(f"**Q**: {q}\n\n{answer}")
83    return jsonify(ok=True, answer=answer)
84
85if __name__ == "__main__":
86    app.run(host="0.0.0.0", port=8000)

1{
2  "msgtype": "news",
3  "news": {
4    "articles": [
5      {
6        "title": "本周社群活动：晒单抽奖",
7        "description": "5 月 20 日前晒单即可参与，10 个名额",
8        "url": "https://example.com/activity/2026-spring",
9        "picurl": "https://example.com/cover.png"
10      }
11    ]
12  }
13}

1import hashlib
2from flask import Flask, request
3
4app = Flask(__name__)
5TOKEN = "DoubaoBot2026"
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   → ① 丢进队列（线程池 / Celery / Redis）
3   → ② 立刻返回 "success"（3ms 内）
4   → ③ 后台调豆包（含改写重试，慢慢来）
5   → ④ 客服消息接口推送结果

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    if msg.type != "text":
11        return "success"
12    threading.Thread(target=async_answer, args=(msg.source, msg.content), daemon=True).start()
13    return "success"          # 立刻返回，绝不让微信等
14
15def async_answer(openid, question):
16    answer = ask_doubao(question)        # 慢调用，含改写重试
17    push_cs_msg(openid, answer)          # 客服消息补发
18
19def push_cs_msg(openid, content):
20    token = get_wx_token()               # 必须缓存，见下节
21    requests.post(
22        f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={token}",
23        json={"touser": openid, "msgtype": "text", "text": {"content": sanitize(content)[:2048]}},
24        timeout=10,
25    )

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    """公众号 access_token：2 小时过期，重新获取会让上一个立刻失效。"""
7    tok = r.get("wx:token")
8    if tok and r.ttl("wx:token") > 300:
9        return tok.decode()
10    with _lock:                                  # 双重检查 + 锁
11        tok = r.get("wx:token")
12        if tok and r.ttl("wx:token") > 300:
13            return tok.decode()
14        resp = requests.get(
15            "https://api.weixin.qq.com/cgi-bin/token",
16            params={"grant_type": "client_credential", "appid": APPID, "secret": APPSECRET},
17            timeout=10,
18        ).json()
19        if "access_token" not in resp:
20            raise RuntimeError(f"微信 token 获取失败: {resp}")
21        r.setex("wx:token", int(resp.get("expires_in", 7200)), resp["access_token"])
22        return resp["access_token"]

1请为下面的产品写 3 版小红书种草文案，每版 60 字以内：
2
3产品：便携式电动咖啡机
4卖点：30 秒出杯、USB-C 充电、重量 380g、可拆洗
5人群：通勤族和露营爱好者
6
7三版风格分别是：
81. 通勤场景，语气疲惫打工人自救
92. 露营场景，语气松弛治愈
103. 参数党，语气理性克制
11
12约束：
13- 每版独立成段，开头标风格
14- 不使用「最好」「第一」「顶级」「彻底」等绝对化用语
15- 不承诺功效，不写「包治百病」这类
16- 口语化，可以有「谁懂啊」「真的」这类语气词，但每版最多 1 个
17- 不要 emoji 堆砌，每版最多 2 个

1下面是用户的一句抱怨，请帮我改写客服回复：
2
3用户原话：「你们这个快递也太慢了吧，下单五天了还没到，我要退货！」
4
5请输出三个版本：
61. **共情版**：先认可情绪，再给解决方案
72. **效率版**：直奔解决方案，不绕弯
83. **补偿版**：给出明确补偿方案
9
10约束：
11- 每版 50 字以内
12- 不使用「非常抱歉给您带来不便」这类模板套话
13- 不承诺「立刻」「马上」这类无法保证的时间
14- 不出现「亲」字
15- 如果涉及退款，明确写「为您登记退款申请」而不是「马上退给您」

1请为下面这个产品写一条 30 秒短视频脚本：
2
3产品：便携咖啡机
4目标：让观众在 3 秒内知道「这玩意儿能随身带」
5平台：抖音
6
7输出格式（每行一个分镜）：
8| 时间 | 画面 | 口播 | 字幕 |
9
10约束：
11- 一共 5-7 个分镜，总时长 30 秒
12- 前 3 秒必须有视觉钩子（不是产品特写，而是冲突或使用场景）
13- 口播用口语，每句不超过 15 字
14- 不使用绝对化用语
15- 结尾要有明确的行动指令（如「评论区告诉我你想带它去哪」）

1阶段 1：企业微信群机器人 + 加签，跑通文案推送        ← 你现在在这里
2阶段 2：公众号服务器配置 + 异步客服消息，实现双向问答
3阶段 3：双档模型路由（Pro / Lite），按任务分流省成本
4阶段 4：批量推理 + 模板文案库，做成运营基础设施