WechatApi首页价格个人微信API开发文档博客立即注册
首页 / 博客 / AI·大模型接入

用 Dify / Coze 工作流驱动微信机器人（低代码）

分类：AI·大模型接入　·　标签：微信机器人、Dify、Coze

前言

大语言模型的能力越来越强，但很多开发者面临同一个问题：怎么把 GPT/Claude/通义千问这些模型的能力，低成本地接入到微信这个最高频的沟通工具里？

传统做法是自己写一套后端——监听微信消息、解析内容、调用模型、再回复。这套链路不复杂，但样板代码多、调试费时，对非专职后端的同学来说门槛不低。

Dify 和 Coze 的出现改变了这个局面。两者都提供可视化的工作流编排能力：你在画布上拖拽节点、配置 Prompt、接入知识库，保存即可得到一个可被 HTTP 调用的工作流端点。把这个端点挂到微信机器人的消息回调里，一个拥有 RAG 检索、多步推理、工具调用能力的智能客服就搭好了，几乎不用写业务逻辑代码。

本文以一个"知识库问答 + 定时播报"的微信机器人为例，完整介绍架构设计、Dify/Coze 工作流配置要点、与微信 HTTP 接口对接的代码逻辑，以及上线后的注意事项。

一、整体架构

先把链路说清楚，后面每节对应其中一个环节。

微信用户发消息
      │
      ▼
微信 HTTP 接口（回调 Webhook）
      │  收到消息后 POST 到你的服务
      ▼
中间层服务（Python/Node 均可）
      │  解析消息 → 调用 Dify 或 Coze 工作流
      ▼
Dify 工作流 / Coze Bot
      │  LLM 推理 + 知识库检索 + 工具调用
      ▼
中间层服务
      │  拿到回答 → 调用发消息接口
      ▼
微信用户收到回复

这条链路里，微信 HTTP 接口负责消息的收发；中间层服务是粘合剂，代码量极少；Dify/Coze 工作流是真正的 AI 大脑，在可视化画布里配置。

二、Dify 工作流配置要点

2.1 创建工作流应用

登录 Dify，选择「工作流」类型新建应用。工作流与普通聊天应用的区别在于：它有明确的输入输出节点，适合被外部服务以 API 方式调用，每次调用是无状态的独立请求（多轮对话的上下文需要你自己在外部维护）。

2.2 基础节点结构

一个典型的问答工作流节点结构如下：

节点类型作用
开始节点定义输入变量，例如 user_question（用户原始问题）、history（历史对话）
知识检索节点在知识库中向量检索，返回相关段落
LLM 节点把用户问题 + 检索结果 + 历史对话拼入 Prompt，调用大模型生成答案
结束节点定义输出变量，例如 answer

Prompt 模板示例（LLM 节点内填写）：

你是一个专业客服，只基于以下参考资料回答问题，不要编造。

参考资料：
{{#knowledge.content#}}

历史对话：
{{history}}

用户问题：{{user_question}}

请用简洁中文回答，若参考资料无相关内容，回复"抱歉，我暂时没有找到相关信息"。

其中 {{#knowledge.content#}} 是 Dify 知识检索节点的内置变量语法，选中知识库节点的输出即可自动填入。

2.3 获取 API 调用地址

工作流保存并发布后，在「API 访问」页面可以拿到：

调用示例：

pythonimport requests

DIFY_API_URL = "https://api.dify.ai/v1/workflows/run"
DIFY_API_KEY = "你的_Dify_API_Key"    # 在 Dify 应用 API 页面获取

def call_dify_workflow(user_question: str, history: str = "") -> str:
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "inputs": {
            "user_question": user_question,
            "history": history
        },
        "response_mode": "blocking",   # 同步等待结果
        "user": "wechat-bot"
    }
    resp = requests.post(DIFY_API_URL, json=payload, headers=headers, timeout=30)
    resp.raise_for_status()
    data = resp.json()
    # 取工作流 outputs 中的 answer 字段
    return data.get("data", {}).get("outputs", {}).get("answer", "处理中，请稍候")

response_mode 设为 blocking 时，接口会等工作流跑完再返回，适合机器人这种需要立即回复的场景；若工作流耗时超过 60 秒，建议改用 streaming 模式分段接收。

三、Coze 工作流配置要点

3.1 Bot 还是工作流？

Coze 提供两种模式：

接入微信机器人时，两种方式都可行。Bot 模式更简单，发布后直接得到 API 端点；工作流模式更灵活，可以加条件分支、循环、代码节点。本节以 Bot 模式为主介绍，工作流模式节点逻辑与 Dify 类似。

3.2 创建 Bot 并发布 API

  1. 在 Coze 创建 Bot，配置人设与 Prompt
  2. 添加知识库（文档上传或 URL 抓取）
  3. 按需添加插件（天气、联网搜索、代码执行等）
  4. 发布时选择「API」渠道，获得 Bot ID

3.3 调用 Coze API

Coze 开放平台提供 /v3/chat 接口（以官方最新文档为准）：

pythonimport requests

COZE_API_URL = "https://api.coze.cn/v3/chat"
COZE_API_KEY = "你的_Coze_API_Key"    # 在 Coze 开发者后台获取
COZE_BOT_ID = "你的_Bot_ID"

def call_coze_bot(user_message: str, conversation_id: str = "") -> str:
    headers = {
        "Authorization": f"Bearer {COZE_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "bot_id": COZE_BOT_ID,
        "user_id": "wechat-user",
        "stream": False,
        "additional_messages": [
            {"role": "user", "content": user_message, "content_type": "text"}
        ]
    }
    if conversation_id:
        payload["conversation_id"] = conversation_id

    resp = requests.post(COZE_API_URL, json=payload, headers=headers, timeout=60)
    resp.raise_for_status()
    result = resp.json()
    # 从 messages 列表中取 assistant 的最后一条回复
    messages = result.get("data", {}).get("messages", [])
    for msg in reversed(messages):
        if msg.get("role") == "assistant" and msg.get("type") == "answer":
            return msg.get("content", "")
    return "暂无回复"

四、中间层服务：对接微信消息收发

4.1 消息接收（回调 Webhook）

微信 HTTP 接口会把用户消息 POST 到你预先注册的回调地址。回调数据结构示例（具体字段以官方文档为准）：

json{
  "appId": "你的设备ID",
  "fromWxid": "发消息人的wxid",
  "toWxid": "接收方wxid（机器人自身或群ID）",
  "type": 1,
  "content": "你好，我想问一下退款流程",
  "msgId": "消息唯一ID",
  "createTime": 1718000000
}

type 字段区分消息类型：1=文本，3=图片，43=视频，49=链接/小程序，等等。对于 AI 问答场景，优先处理 type==1 的文本消息。

4.2 Flask 回调服务示例

pythonfrom flask import Flask, request, jsonify
import requests, threading

app = Flask(__name__)

BASE  = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"
HEADERS = {"token": TOKEN}       # 鉴权字段名以官方文档为准

def send_text(to_wxid: str, content: str):
    """调用发消息接口（示例，具体接口/字段以官方文档为准）"""
    url = f"{BASE}/message/postText"
    payload = {"appId": APPID, "toWxid": to_wxid, "content": content}
    requests.post(url, json=payload, headers=HEADERS, timeout=10)

def handle_message(data: dict):
    msg_type = data.get("type")
    from_wxid = data.get("fromWxid", "")
    content = data.get("content", "").strip()

    if msg_type != 1 or not content:
        return  # 只处理文本消息

    # 调用 Dify 工作流（替换为 call_coze_bot 也可以）
    answer = call_dify_workflow(user_question=content)
    send_text(from_wxid, answer)

@app.route("/callback", methods=["POST"])
def callback():
    data = request.json or {}
    # 异步处理，防止回调超时
    threading.Thread(target=handle_message, args=(data,), daemon=True).start()
    return jsonify({"code": 200})   # 必须及时返回 200，否则平台会重试

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

关键点：回调必须快速返回 200，AI 调用（可能耗时几秒）要放到异步线程里，否则超时会导致重复推送。

4.3 注册回调地址

服务部署到公网后，调用 setCallback 接口注册地址：

pythondef set_callback(callback_url: str):
    url = f"{BASE}/login/setCallback"
    payload = {"appId": APPID, "callbackUrl": callback_url}
    resp = requests.post(url, json=payload, headers=HEADERS)
    print(resp.json())

set_callback("https://你的公网域名/callback")

4.4 主动推送：定时播报

除了被动回复，还可以让机器人定时主动推送内容。用 APScheduler 或 cron 定时任务，在指定时间生成内容并发送到群/个人：

pythonfrom apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

def morning_broadcast():
    """每天早上 8 点推送早报"""
    # 可以调用 Dify 工作流生成内容，或直接拼固定模板
    content = call_dify_workflow(user_question="请生成今天的早报摘要，100字以内")
    group_wxid = "群的wxid"
    send_text(group_wxid, content)

scheduler.add_job(morning_broadcast, "cron", hour=8, minute=0)
scheduler.start()

五、多轮对话：上下文维护

Dify 工作流默认无状态，每次调用是独立的。要实现多轮对话，需要在中间层维护对话历史：

pythonfrom collections import deque

# 用字典存每个用户的最近 N 条历史，防止无限增长
history_store: dict[str, deque] = {}

def get_history_str(wxid: str, max_turns: int = 5) -> str:
    dq = history_store.get(wxid, deque(maxlen=max_turns * 2))
    lines = []
    msgs = list(dq)
    for i in range(0, len(msgs), 2):
        if i + 1 < len(msgs):
            lines.append(f"用户：{msgs[i]}\n助手：{msgs[i+1]}")
    return "\n".join(lines)

def update_history(wxid: str, user_msg: str, bot_reply: str, max_turns: int = 5):
    if wxid not in history_store:
        history_store[wxid] = deque(maxlen=max_turns * 2)
    history_store[wxid].append(user_msg)
    history_store[wxid].append(bot_reply)

# 修改 handle_message 中的调用
def handle_message(data: dict):
    from_wxid = data.get("fromWxid", "")
    content = data.get("content", "").strip()
    if data.get("type") != 1 or not content:
        return

    history = get_history_str(from_wxid)
    answer = call_dify_workflow(user_question=content, history=history)
    update_history(from_wxid, content, answer)
    send_text(from_wxid, answer)

如果用 Coze Bot，它自带 conversation_id 机制管理多轮上下文，把首次请求返回的 conversation_id 存下来，后续调用传入即可，无需自己维护历史。

六、接入托管 HTTP 接口

上述示例中，微信消息收发部分依赖一套能以 HTTP 方式读写个人微信的接口服务。WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可，详情以官方文档为准。

扫码登录流程（简述）：

python# 1. 获取登录二维码（示例，具体接口以官方文档为准）
resp = requests.post(f"{BASE}/login/getLoginQrCode", json={}, headers=HEADERS)
qr_url = resp.json()["data"]["qrCodeUrl"]  # 展示或下载二维码

# 2. 轮询检查登录状态
import time
for _ in range(60):
    check = requests.post(f"{BASE}/login/checkLogin",
                          json={"appId": APPID}, headers=HEADERS)
    status = check.json().get("data", {}).get("status")
    if status == 2:   # 已扫码确认
        print("登录成功")
        break
    time.sleep(3)

登录成功后，appId 即为该账号的设备标识，后续所有接口调用都带上它。

七、上线注意事项

7.1 发送频率控制

微信对自动化行为有风控。批量发消息时务必加随机延迟：

pythonimport time, random

def send_batch(wxid_list: list, content: str):
    for wxid in wxid_list:
        send_text(wxid, content)
        time.sleep(random.uniform(3, 8))   # 每条间隔 3-8 秒

群发、加好友、建群等操作均有频率上限，新账号建议在线稳定几天后再开启批量操作。此外，机器人回复内容应避免带有广告性质的话术，纯工具类、客服类场景风险相对较低。

7.2 异常处理与降级

pythondef handle_message_safe(data: dict):
    try:
        handle_message(data)
    except requests.Timeout:
        from_wxid = data.get("fromWxid", "")
        send_text(from_wxid, "AI 服务响应超时，请稍后重试")
    except Exception as e:
        print(f"处理消息异常: {e}")
        # 记录日志，不向用户暴露错误详情

生产环境中建议将日志写入文件或接入日志平台（如 Loki、CloudWatch），同时对 AI 接口调用增加重试机制，避免因偶发超时导致用户无响应。

7.3 关键词过滤

在调用 AI 之前，可以加一层关键词路由：匹配到"退款""投诉"等词转人工；匹配到"价格""规格"等词走知识库工作流；其余走通用对话工作流。这样既降低 AI 调用成本，也提高响应准确率。

pythondef route_message(content: str) -> str:
    """根据内容路由到不同处理分支"""
    if any(kw in content for kw in ["退款", "投诉", "人工"]):
        return "transfer_human"
    if any(kw in content for kw in ["价格", "规格", "参数"]):
        return "product_qa"
    return "general_chat"

关键词列表建议做成配置文件，上线后可以随时调整，不必重启服务。

7.4 Dify vs Coze 选型参考

维度DifyCoze
部署方式支持私有化部署（开源）仅云端
知识库支持多种向量数据库内置，配置简单
工作流复杂度节点类型丰富，支持代码节点相对简单，插件生态好
API 调用RESTful，文档完善RESTful，有 SDK
免费额度开源自部署无限制有免费 Token 额度
适合场景需要高度定制、私有数据快速上线、模型切换灵活

两者并不互斥——可以把 Dify 用于知识库问答（数据留在自己服务器），把 Coze 用于需要联网搜索、插件丰富的场景，中间层根据消息内容路由到不同工作流。

总结

用 Dify 或 Coze 工作流驱动微信机器人，核心思路是：把 AI 能力封装成一个 HTTP 端点，中间层服务只做消息解析和接口转发，业务逻辑全部在可视化画布里编排。这种架构让非专职后端的开发者也能快速迭代 AI 能力，而不必每次改 Prompt 都重新部署代码。

落地时有几点需要特别注意：①回调接口必须毫秒级返回，AI 推理要异步执行；②多轮对话上下文在 Dify 侧需要自己维护，Coze Bot 模式则原生支持；③发消息频率要做好限速，避免触发平台风控；④关键词路由和异常降级是提升用户体验的必要手段，上线前务必测试覆盖。整体而言，这套低代码方案适合快速验证产品思路，等业务逻辑稳定后再视需求决定是否迁移到自研模型调用层。

想动手试试？

WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，注册后几分钟跑通。

立即免费注册查看开发文档

相关产品页

🔗 微信机器人开发（产品页）🔗 微信客服机器人（产品页）🔗 微信群管理机器人（产品页）

相关文章

微信接入通义千问做智能客服（国产大模型）微信机器人接入知识库 RAG，做企业专属智能问答微信 AI 机器人多轮对话与上下文管理实战微信 AI 客服意图识别与智能转人工方案
© 2025 WechatApi · 企业级微信智能机器人接入平台
官网价格帮助文档博客
苏ICP备2024128799号 · 苏ICP备2023038368号