在上一篇文章里我们解决了通义千问 API 的鉴权、基础调用和参数调优算是把“大脑”准备好了。这一篇要做的就是给这颗大脑接上“手脚”——把它完整地塞进飞书机器人里让群里的人能像同事一样直接跟 AI 对话。这篇我重点讲三块飞书自定义机器人的创建与安全校验、服务端中转服务怎么搭、以及怎么把富文本表格和流式回复也一并搞定。内容比较干但每一步都是可以直接抄作业的。1. 整体方案设计与架构梳理1.1 为什么选择“Webhook 机器人 服务端中转”这套组合飞书机器人接入通常有两个方向一个是走飞书开放平台的应用机器人需要建应用、配权限、开事件订阅另一个就是今天要用的自定义机器人 Webhook。两者差异很大先搞清楚才能选对路。自定义机器人靠一个 Webhook 地址就能往群里推消息不需要审核、不需要配置权限几秒钟就能建好。但它的短板也很明显只能主动推送收不了用户的消息。而通义千问这种对话式 AI核心是“一问一答”得先拿到用户问了什么才能把答案还回去。怎么破答案就是在中间架一层自己的服务把“群里的提问”和“通义千问的回答”用代码桥接起来。这层的典型工作流长这样用户在群里输入指令并 机器人 → 飞书把这条消息通过事件订阅推送到你的服务端或者你用定时轮询的方式拉取→ 服务端解析消息内容判定确实需要 AI 介入 → 组装 prompt 调用通义千问 API → 拿到结果后再通过 Webhook 往群里发消息。整个过程核心就是三件事收消息、调模型、发结果。我选这套方案的原因很实在它把复杂问题拆解成了两个可以独立测试的环节。服务端和飞书之间的对接用 Webhook 事件订阅服务端和通义千问之间的对接用标准 HTTP API哪边出问题就单独查哪边排查成本低后续扩展也方便。1.2 需要准备的东西和整体目录结构动手之前把家当先列清楚一个飞书群组随便建个测试群就行飞书开放平台后台的权限用来建机器人应用一台能跑 Python 的服务器或本地环境建议 LinuxWindows 也能跑但有些坑通义千问的 API Key阿里云百炼平台申请Python 3.9 环境需要装 flask 或 fastapi、requests 这两个核心库我习惯的项目结构是这样的feishu-qwen-bot/ ├── app.py # 主服务入口处理飞书事件与消息回复 ├── qwen_client.py # 通义千问 API 的封装模块 ├── config.py # 配置文件密钥、Webhook、Bot 信息 ├── utils.py # 加签校验、消息解析等工具函数 └── requirements.txt # 依赖清单为什么要单独拆一个config.py因为密钥、飞书 Webhook 地址、通义千问 API Key 这些信息散落在代码里的话后期维护会非常痛苦。尤其当你把代码丢到 GitHub 或者交给同事时密钥泄露就是安全事故。我用的是环境变量加配置文件双保险默认值写在 config.py 里真实密钥从环境变量读取两者都没有就直接抛异常报错。2. 飞书自定义机器人的创建与安全校验2.1 三步建好机器人并拿到 Webhook进飞书开放平台找到“开发者后台”用你的飞书账号登录。创建一个企业自建应用名字随便起比如就叫“Qwen AI 助手”。创建完成后在应用详情页左侧菜单里找到“添加应用能力”里面有一个“机器人”选项点启用。这一步就是给应用挂上一个能发消息的机器人身份。启用机器人之后还需要在应用里添加一个群组。直接在“版本管理与发布”里创建版本并发布把应用安装到你自己的测试群里。装好之后在群设置里找到“群机器人”→“添加机器人”选择刚才创建的那个应用机器人。这个时候飞书会给你生成一个 Webhook 地址形如https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx这个地址要收好它就是机器人往群里发消息的“钥匙”。另外飞书还支持两种安全设置签名校验和 IP 白名单。我强烈建议你把签名校验打开IP 白名单可以不设因为你服务端的出口 IP 可能是动态的。签名校验的原理是飞书在请求头里带一个X-Lark-Signature字段你用预设的签名密钥对这个字段做校验能有效防止别人拿到 Webhook 地址后往你群里乱发消息。2.2 加签与验签的代码实现飞书的签名校验逻辑是timestamp 换行符 签名密钥拼成一个字符串然后用 SHA256 算出一个 digest再和请求头里的签名做比对。很多新手挂在第一步——不知道这个密钥是从哪来的。注意这个密钥不是你应用的 App Secret而是在机器人安全设置里单独生成的“签名密钥”通常在 Webhook 地址下面有一个“签名校验”的开关打开之后会给你一个以随机字符串形式存在的密钥。import hashlib import base64 import hmac def verify_feishu_signature(timestamp: str, signature: str, secret: str) - bool: string_to_sign f{timestamp}\n{secret} hmac_code hmac.new(string_to_sign.encode(utf-8), digestmodhashlib.sha256).digest() expect_signature base64.b64encode(hmac_code).decode(utf-8) return hmac.compare_digest(signature, expect_signature)这里有两个细节容易踩坑。第一timestamp和signature都是从飞书请求头里取的别混用第二必须用hmac.compare_digest做比较这个函数是常数时间比较能防时序攻击。虽然一个群机器人被攻击的概率不高但安全习惯还是要养成。验签通过后接下来就是解析消息内容。飞书的事件订阅会推送 JSON 数据到你的服务端里面包含消息类型、发送者、群组 ID、消息内容等字段。我遇到的一个坑是飞书的消息事件里msg_type可能是text纯文本、post富文本、image图片等类型初期只处理文本消息就好把其他的直接返回ok应答否则飞书会认为你处理失败然后重试造成重复响应。2.3 事件订阅地址的配置与公网回调飞书要能把消息推给你得有一个公网可访问的 URL。开发阶段我用的办法是内网穿透工具比如 cpolar 或 ngrok把本地 8080 端口映射成一个公网地址然后在飞书开放平台的事件订阅里把这个地址填进去。生产环境的话我建议直接把服务部署到云服务器上用 Nginx 反代到 Python 服务再加一层 SSL因为飞书要求回调地址必须是 HTTPS。配置事件订阅时有一点很关键飞书给你推送事件后如果服务端在 3 秒内没有响应飞书会认为推送失败并重试。所以你的回调接口必须在 3 秒内快速返回。但调用通义千问 API 通常要花好几秒怎么处理我的做法是先用一个线程把任务丢到队列里主线程立刻返回 HTTP 200。后台任务处理完后再通过 Webhook 主动把结果推回群里。也就是说飞书事件订阅的接口永远秒回而真正的 AI 调用和结果发送在异步任务里完成。这个方法简单、可靠不用引入消息队列中间件唯一的注意点是你的服务要支持多线程。3. 服务端对接通义千问的核心实现3.1 API Key 的获取与安全存储这一步相对机械但却是最容易出问题的地方。去阿里云百炼平台bailian.console.aliyun.com开通服务然后在 API-KEY 管理页面生成一个 Key。生成时有一点要注意百炼平台的新版 Key 需要绑定工作空间和业务空间别选错了。拿到 Key 之后存到环境变量里别硬编码在代码里。在config.py里我这样读取import os DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) if not DASHSCOPE_API_KEY: raise ValueError(请设置环境变量 DASHSCOPE_API_KEY)这样设置的好处是代码仓库里永远只留一个占位符真正密钥在部署环境里配置安全性有保障。3.2 调用通义千问的标准接口封装通义千问的 API 走的是 OpenAI 兼容格式这点非常友好。官方 SDK 用起来最省事但如果你想把依赖降到最低直接用requests库也能轻松搞定。这里我给出 requests 版本的实现因为很多人的服务器环境不一定想装额外的 SDK。import requests import json def call_qwen(prompt: str, model: str qwen-plus, max_tokens: int 2000) - str: url https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation headers { Authorization: fBearer {DASHSCOPE_API_KEY}, Content-Type: application/json } payload { model: model, input: { messages: [ {role: user, content: prompt} ] }, parameters: { max_tokens: max_tokens, temperature: 0.8, top_p: 0.8 } } resp requests.post(url, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() return data.get(output, {}).get(text, )注意max_tokens的取值会影响回答长度。很多人在这一步吃亏设置太短回答被截断设置太长计费变高。我实测下来一般问题 1000 足够了但如果你要让它写长篇文章或者代码建议用 4000。这里要提醒一下max_tokens是要计入费用的单位是 token一个汉字大约等于 1.5~2 个 token算钱的时候可别按字数来算。3.3 模型选择qwen-turbo、qwen-plus 还是 qwen-max通义千问的模型家族里不同规格适合不同场景。我实测对比过几款简单说说个人感受模型响应速度理解能力适用场景qwen-turbo快通常在 1~2 秒内一般能处理简单问答高频、低延迟场景qwen-plus中等3~5 秒较好逻辑性明显增强日常对话、知识问答最推荐qwen-max较慢5~8 秒强能处理复杂推理和长文本专业分析、长文创作、代码生成对于飞书机器人这个场景我默认用的就是qwen-plus。理由很直接响应速度在可接受范围回答质量比 turbo 高一个档次价格又不是最贵的。如果你群里消息量特别大可以考虑降级到 turbo 兜底如果是在群里做技术方案评审那就上 max。另外有些场景会用到流式输出enable_streaming: true也就是一个字一个字往外蹦的“打字机效果”。但飞书自定义机器人只支持整条消息推送不支持打字机效果所以我这里不推荐开启流式。就算你开了流式最终也要等它全部生成完再一次性推送到群里反而浪费了服务端资源。流式真正有用的场景是网页端对话不是群机器人。4. 完整集成从群消息到 AI 回复的闭环4.1 主服务的骨架代码把上面的模块拼装起来一个能跑通的最小服务就出来了。这里是app.py的核心逻辑from flask import Flask, request, jsonify import threading import config import qwen_client import utils app Flask(__name__) app.route(/feishu/event, methods[POST]) def handle_feishu_event(): body request.get_json() # 先校验事件是否是 URL 验证请求 if body.get(type) url_verification: return jsonify({challenge: body.get(challenge)}) # 校验签名 timestamp request.headers.get(X-Lark-Request-Timestamp, ) nonce request.headers.get(X-Lark-Request-Nonce, ) signature request.headers.get(X-Lark-Signature, ) if not utils.verify_feishu_signature(timestamp, signature, config.FEISHU_SECRET): return jsonify({code: 1, msg: signature error}), 403 # 只处理 im.message.receive_v1 事件 event body.get(event, {}) if event.get(type) ! im.message.receive_v1: return jsonify({code: 0, msg: ok}) # 丢进线程异步处理主线程快速返回 threading.Thread(targetprocess_message, args(event,)).start() return jsonify({code: 0, msg: ok}) def process_message(event): try: message event.get(message, {}) msg_type message.get(message_type, ) if msg_type ! text: return content json.loads(message.get(content, {})) text content.get(text, ) chat_id event.get(message, {}).get(chat_id, ) user_id event.get(sender, {}).get(sender_id, {}).get(open_id, ) # 可选判断是否机器人或者是否包含特定前缀 if _user_ not in text: # 这里简化处理实际要判断是否为机器人ID return # 去掉 信息提取真正的 prompt prompt utils.strip_mention(text) # 调用通义千问 answer qwen_client.call_qwen(prompt) # 通过 Webhook 发回群里 utils.send_feishu_message(chat_id, answer) except Exception as e: utils.send_feishu_message(chat_id, f出错了{str(e)})这段代码里最容易被忽略的是第 34 行判断用户有没有 机器人。如果群里有别的人说话你的机器人也会回复那就成“话痨”了。实际判断逻辑要拿到机器人自身的 open_id然后去消息内容里查是否有对应字符串。为了简便我这里用了strip_mention函数把文本里的_user_xxx标签去掉只保留真正的问题内容。4.2 往飞书群里推送消息的实现推送消息的代码比较标准用requests直接打 Webhook 就能搞定。这里我给一个支持不同消息类型的版本def send_feishu_message(chat_id: str, content: str, msg_type: str text): url config.FEISHU_WEBHOOK_URL if msg_type text: payload { msg_type: text, content: { text: content } } elif msg_type post: payload { msg_type: post, content: { post: { zh_cn: { title: Qwen AI 回复, content: [ [{tag: text, text: content}] ] } } } } resp requests.post(url, jsonpayload, timeout10) result resp.json() if result.get(code) ! 0: raise Exception(f飞书推送失败: {result.get(msg)})这里有个实战细节飞书 Webhook 对消息长度有限制text 消息最长不能超过 15000 字节超过了会报错。通义千问如果生成了超长回答建议先做个截断处理或者在推送前判断长度超了就分段推但分段推送顺序不好保证更简单的是让 AI 在回答时就控制篇幅。我的处理方式是在 prompt 后面追加一句“请将回答控制在 800 字以内”实测效果很好回答基本都在限制范围内。4.3 消息去重与幂等处理在群机器人实际使用中最烦人的问题之一就是重复回复。原因是飞书事件订阅是“至少一次”投递服务端可能收到同一条消息多次。如果不做幂等处理群里就会看到同样的回答出现两三次。这个问题我在早期接入时栽过跟头。解决方式其实很简单在 Redis 里存一个消息 ID 到处理状态的映射收到一次处理一次。如果你不想引入 Redis用 Python 的dict加超时时间也能应付低频场景processed_msg {} def is_duplicate(msg_id: str) - bool: if msg_id in processed_msg: return True processed_msg[msg_id] time.time() return False但注意dict存久了会内存泄漏所以要加个定时清理机制或者用一个固定大小加“过期时间”的字典。如果群规模不大、消息量少这个方案完全够用如果消息量大还是老老实实上 Redis。5. 飞书机器人发送富文本与表格消息5.1 为什么需要富文本和表格群里聊 AI最常见的回答是纯文本。但你让 AI 输出一个对比表格的时候比如“帮我对比一下 Python 和 Java 的优劣”纯文本格式就完全不可读了五行八列的 ASCII 字符混在一起手机上看到直接崩溃。飞书的post富文本消息支持段落、加粗、链接而交互卡片interactive card更强大能渲染真正的表格、按钮、图片。我建议日常问答用 text 消息需要结构化输出时用 post 富文本真正要展示多行多列数据时用 interactive card。这里的成本递增也很明显text 最简单post 中等card 要按飞书的消息卡片 JSON 规范来组织数据。5.2 用交互卡片发送表格交互卡片是飞书消息体系里能力最强的一种它其实是一个 JSON 结构飞书客户端收到后会按声明式语法渲染。下面这段代码演示了如何把 AI 返回的 Markdown 表格转成飞书卡片的表格def send_feishu_table(chat_id: str, headers: list, rows: list): columns [] for header in headers: columns.append({ name: header, display_name: header, width: auto }) table_rows [] for row in rows: cells [] for value in row: cells.append({text: value}) table_rows.append({cells: cells}) card { msg_type: interactive, card: { config: {wide_screen_mode: True}, header: {title: {tag: plain_text, content: AI 对比结果}}, elements: [ { tag: table, columns: columns, rows: table_rows } ] } } resp requests.post(config.FEISHU_WEBHOOK_URL, jsoncard, timeout10) print(resp.json())表格卡片的 JSON 结构比较严格columns数组里的name和display_name不能为空rows数组里每个单元格必须是{text: xxx}形式缺一个字段整个卡片可能渲染失败。而且rows最多只能有 20 行如果 AI 返回的表格超过 20 行就得拆成多张卡片或者只取前 20 行。5.3 让 AI 输出结构化数据并解析把 Markdown 表格转成卡片表格关键一步是让通义千问以固定格式输出。我用的技巧是在 prompt 里写清楚请以下面这种 JSON 数组的格式输出结果不要输出其他内容 {headers: [列1, 列2], rows: [[值1, 值2], [值3, 值4]]}这样 AI 返回的就是一个标准 JSON 字符串解析非常方便。但这里有个坑通义千问有时会在 JSON 前后加 Markdown 代码块标记json ...直接json.loads会报错。我的解决方案是先用正则把 JSON 部分提取出来再解析import re def extract_json(text: str): match re.search(r\{.*\}, text, re.S) if match: return json.loads(match.group()) raise ValueError(未找到合法的 JSON 内容)这个技巧在处理 AI 结构化输出时非常实用通用性很强不只是飞书场景任何对接 LLM 的编程场景都会用到。6. 常见问题与排查技巧实录6.1 事件订阅验证失败或回调超时飞书在配置事件订阅地址时会先发一个url_verification请求来验证你的服务是否可用。这个请求里带一个challenge字段你的服务必须原样返回它。很多新手在这里遇到“验证失败”原因就是只处理了im.message.receive_v1事件没处理url_verification。我上面的代码里专门加了这个分支就是在提醒这个分支不能省。回调超时的问题也很常见。飞书要求 3 秒内响应但你的服务可能因为某段代码卡住了。解决方案我在前面说过了异步处理。这里再补充一个细节Flask 默认是单进程多线程模式threading.Thread虽然能用但如果消息量大了线程数会失控。更好的做法是用任务队列比如 Celery 或者最简单的 Redis 队列。如果只是个人群使用线程方案足够。6.2 Webhook 地址泄露或签名校验失败Webhook 地址一旦泄露别人就拿到了你群里的发言权限可以随便往你群里灌垃圾消息。飞书的安全设置里“签名校验”一定要打开。但开了签名校验之后自己调用 Webhook 时也要在请求头里带上对应的签名参数否则连你自己都推不进去。这里有个容易忽略的点签名密钥要跟 Webhook 地址一起生成而且它是独立于应用 App Secret 的。我之前对接过一个项目同事把应用的 App Secret 当成签名密钥传了进来结果死活验签失败排查了半天才发现是密钥用错了。6.3 通义千问超时或返回空值通义千问的接口在高峰期偶尔会变慢特别是qwen-max模型。我设置的是 60 秒超时但在实际场景中如果 60 秒才返回飞书那边早就没有交互感了。我建议你对调用做分层处理服务端调用通义千问设 30 秒超时超时后机器人回复“思考时间太长了请换个简单点的问题再试试”。另外如果返回的内容为空字符串多半是max_tokens设置太小模型“思考到一半发现字数额度满了”输出被截断成了空白。这种情况可以从两个角度排查调大max_tokens或者让 AI 在 prompt 里就收到“回答请简短”的指令。6.4 机器人被频繁触发导致限流飞书机器人对 Webhook 的调用频率有限制超过了会返回 429 错误码。通义千问自己也有 QPS每秒请求数限制。群里如果有多个人同时问问题很容易触发限流。我的应对策略很简单在服务端做一个简单的互斥锁同一时间只处理一个 AI 请求其余排队等待。import threading ai_lock threading.Lock() def process_message(event): with ai_lock: answer qwen_client.call_qwen(prompt) utils.send_feishu_message(chat_id, answer)这个方案对个人群组完全够用缺点是回答是串行的一个慢了后面全慢。如果你追求并发体验可以升级成信号量机制同时放行 2~3 个请求既能保证不被限流也保留了并发能力。6.5 常见问题速查表现象可能原因排查与解决事件订阅验证失败没有处理 url_verification检查服务代码是否返回 challenge 字段签名校验一直失败密钥用错用了 App Secret确认使用的是机器人安全设置里的签名密钥消息收到了但没回复事件类型判断错误检查是否处理的是im.message.receive_v1其他类型直接忽略回复出现重复飞书重复推送消息服务端做消息 ID 幂等处理AI 回复内容为空max_tokens 设置太小调大 max_tokens或提示 AI 控制回复长度推送消息报 429 错误触发飞书限流服务端做并发控制限制同一时间的推送数量7. 一些补充的自动化玩法机器人能跑通之后玩法其实比想象的多。我自己后来做了一个小改进在 prompt 前面加一个系统指令告诉通义千问“你现在是群里的 AI 助手请用简洁的 markdown 格式回复回答末尾不要啰嗦”。就这么一句话回答质量瞬间提升一个档次因为它会让模型更清楚自己在什么场景下工作。另外可以给机器人加“记忆”能力。把历史对话存到本地文件或数据库里每次提问就把最近的几轮对话拼进 prompt这样 AI 就能记住上下文而不是“一问三不知”。但要注意 token 消耗会明显增加我的经验是保留最近 5 轮对话就够用了再多性价比就不高了。还有一个小技巧如果你觉得每次都要 机器人很麻烦可以在服务端做一个简单的白名单 / 前缀设定——比如只要群里有人发“!q 问题”这个格式机器人就自动回应。这样可以省掉 操作在手机上打字更方便。不过要注意这会增加机器人误触发的概率建议只在信任的小群里开启。最后再分享一个我在实际使用中的发现通义千问对中文的理解和对代码的理解都不错但如果你在群里问的是一些时效性很强的话题比如“今天有什么新闻”它的回答可能不够新。这种情况可以在 prompt 里明确告诉它“根据你的知识截止日期回答如果不知道就说不知道”避免它硬编造内容。加上这一句之后回答的可靠性会高不少。这个搭好的飞书 AI 机器人往后还能继续扩展的方向不少比如接入飞书多维表格让 AI 能写入数据或者接上审批流实现用自然语言发起审批。核心的服务架构不用大改套一层业务逻辑就行。动手练起来吧有什么问题可以在评论区一起交流。