☰
飞书机器人接入通义千问:从群聊消息到AI自动回复的完整实现
2026/10/3 8:07:01 网站建设 项目流程

做过企业办公自动化的朋友应该都遇到过这种尴尬:开发群里每天被同样的问题轰炸,明明写好的文档就放在那,新人还是要追着问“这个报错怎么解”“那个配置在哪里改”。我的做法比较直接——既然大家都在用飞书,那就把AI模型直接拉进群里,让通义千问变成群里一个能@的机器人。这个项目我拆了两篇来讲,上篇说了账号准备和应用创建的基础步骤,这篇就是彻底落地的那一半:怎么把通义千问和飞书机器人打通,让它真正在群里回答问题,并顺带解决一个实际痛点——一个机器人怎么切换不同档位的AI模型。

先说清楚这篇博文适合谁看:你已经有了飞书开发者后台的账号,也大概知道通义千问是阿里云百炼平台提供的模型服务,但还没把两者接起来,或者接起来之后发现只能发消息、收不到群里@它的消息。看完这篇,你会拿到一套可以直接复制跑的Python代码,以及踩坑之后沉淀下来的排查清单。整个过程不需要公网服务器,不需要配置HTTPS回调地址,一台能联网的电脑就能跑。

1. 整体架构与方案选型:为什么选择自建应用长连接模式

1.1 飞书机器人的两种形态,选错了后面全是坑

飞书机器人实际开发中基本分两类。一类是群聊里添加的“自定义机器人”(也叫Webhook机器人),只能通过Webhook地址向群里推送消息,是单向的,没法接收群成员的消息,也就做不了“对话”。另一类是“自建应用机器人”,它本质是一个飞书应用,通过事件订阅机制可以接收用户在群里@机器人产生的消息,甚至接收用户与机器人的单聊消息。

我们要做的AI问答机器人,核心交互是“用户@机器人→机器人收到→调用AI模型→群里回复”,所以必须是自建应用机器人。这里有个很多人不解的点:同样是收消息,事件订阅为什么我推荐用长连接而不是Webhook回调?原因很现实——Webhook方式要求飞书服务器能把事件推送到你的服务上,这意味着你的服务必须有一个公网可访问的HTTPS地址,本地开发还得用内网穿透工具把请求转发进来,调试体验很差。而长连接模式是反向的:你的代码主动连接到飞书服务器,建立一条WebSocket连接,事件顺着连接推给你。只要你的电脑能访问公网,就能跑,部署到云服务器或者家里的小主机上也照样行。

我最初用Webhook方式做了一套,踩了证书配置、回调地址验签、断线重推一整套坑,后面换成官方SDK的长连接模式,开发效率直接起飞。所以这个方案选择,不是“哪个更高级”,而是“哪个更适合你当前的场景”。

1.2 调用链路拆解:从群聊消息到AI回复的完整旅程

为了让后面的代码不晕,先把整条调用链路画在脑子里。用户在某飞书群里输入“@我的AI机器人 帮我写一个Python快速排序”,这条消息会先发到飞书服务器,飞书服务器根据应用的订阅配置,把事件数据推送给我们建立的WebSocket长连接。我们的机器人程序从事件中解析出“是哪个人在哪群里发了什么内容”,注意这里要剔除消息文本里的“@机器人”这部分,提取出真正要问AI的问题——“帮我写一个Python快速排序”。接着程序调用通义千问的API接口,传入问题和参数,拿到AI生成的回答文本。最后再调用飞书开放API,以机器人身份往同一个群发一条消息,内容就是AI的回复。

如果画成序列图会更直观,简单说就是一个“飞书——>长连接——>业务代码——>通义千问API——>业务代码——>飞书”的闭环。这个链路里有两个方向相反的集成:一边是飞书的事件订阅体系,另一边是阿里云百炼平台的模型调用体系,我们的业务代码就是中间的翻译官。

技术栈方面我选了Python,依赖分别是飞书官方SDKlark-oapi和OpenAI兼容客户端openai。重点说下为什么用openai这个库去调通义千问,而不是用阿里云官方提供的dashscopeSDK。因为通义千问在百炼平台提供了OpenAI兼容的接口,你用任何语言里熟悉的OpenAI客户端库,改一下base_url和api_key就能调用,代码通用性极强。这带来的额外好处是:以后你想把模型从通义千问换成其他兼容OpenAI协议的服务(包括本地部署的模型),业务代码几乎不用动,只要改配置。我在后面讲到多模型切换和本地模型接入时,你会发现这个选择的价值。

2. 前置准备:环境配置与必要检查

2.1 飞书开放平台侧的三件事

虽然上篇讲过了,但为了这篇文章能独立阅读,我把关键步骤再压一遍。第一,登录飞书开放平台,创建企业自建应用,会得到一个App ID和App Secret,这相当于应用的用户名和密码,后面代码要用。第二,在“添加应用能力”里开启“机器人”能力,这一步决定了应用能不能以机器人身份出现在群聊里。第三,在“事件与回调”里添加事件接收消息 im.message.receive_v1,完成订阅。订阅事件这一步必须手动添加,很多新手开完机器人就直接写代码,结果发现收不到任何消息,十有八九是这里没配。

权限方面,飞书开放平台是权限管控很严的体系。机器人要发消息、读消息,至少要开通这几个权限:im:message(读取消息)、im:message:send_as_bot(以机器人身份发消息)、im:chat(获取群信息,可选)。权限开了之后,还要在“版本管理与发布”里创建一个应用版本并发布,让权限生效。我自己在这块吃了不少亏,开发阶段用的是测试企业,权限改动不发布版本的话,代码跑得再对也没反应。发布版本后等一两分钟,再回群里试,这个顺序别搞反。

2.2 通义千问侧的密钥准备

去阿里云百炼平台开通模型服务,开通后创建一个API Key。这个Key就是sk-开头的字符串,后面调用模型接口时会作为Authorization请求头传到服务端。

通义千问系列模型有几个关键点要注意。模型名要写对,常用的有qwen-turbo、qwen-plus、qwen-max三种,价格和能力从低到高排列。qwen-turbo便宜速度快,适合高频轻量问答;qwen-plus是综合性价比最高的档位,日常用这个就够了;qwen-max是旗舰模型,复杂推理、代码生成、长文本处理更稳,但成本高、响应也慢一些。实际代码里model参数就是直接传这些字符串,大小写敏感。

还有一点,通义千问的API接口地址是https://dashscope.aliyuncs.com/compatible-mode/v1,这个地址和OpenAI官方的https://api.openai.com/v1结构类似,区别只在域名前缀。在openai库的OpenAI构造函数里,把base_url指到上面这个地址即可。我这里再强调一次:很多朋友手滑把base_url配成了https://dashscope.aliyuncs.com/api/v1,然后报404,实际上正确路径是compatible-mode/v1,这个坑我见得太多了。

2.3 本地开发环境与依赖清单

开发机器有Python 3.9以上版本就行,我推荐3.10或3.11,类型推导和异步支持都更好。依赖包非常少,就三个:lark-oapi负责飞书相关,版本用最新稳定版;openai负责调用模型接口;python-dotenv用来读取.env配置文件,方便管理密钥。

建议在项目根目录建一个.env文件,把需要保密的配置集中放进去:

FEISHU_APP_ID=cli_xxxxxxxxxxxxxx FEISHU_APP_SECRET=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 MODEL_NAME=qwen-plus

文件建好后,用pip install -r requirements.txt安装依赖,requirements文件内容就三行:

lark-oapi>=0.4.0 openai>=1.40.0 python-dotenv>=1.0.0

装依赖的时候如果网络波动导致下载失败,换成国内PyPI镜像源pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt就能解决。这块没什么技术含量,但配好之后后面开发会很省心。

3. 核心代码实现:把机器人完整跑起来

3.1 初始化飞书客户端并建立长连接

先写入口逻辑。用lark-oapi初始化飞书客户端,然后注册一个事件处理器,监听P2MessageReceiveV1这个事件(对应飞书开放平台的“接收消息”事件),最后启动长连接。

import os from dotenv import load_dotenv import lark_oapi as lark from lark_oapi.api.im.v1 import * load_dotenv() app_id = os.getenv("FEISHU_APP_ID") app_secret = os.getenv("FEISHU_APP_SECRET") def handle_message(data: P2MessageReceiveV1) -> None: # 在这里处理收到消息的逻辑 pass def main(): # 创建飞书客户端,使用长连接模式 client = lark.Client.builder() \ .app_id(app_id) \ .app_secret(app_secret) \ .log_level(lark.LogLevel.DEBUG) \ .build() # 注册事件处理器 handler = lark.EventDispatcherHandler.builder("", "") \ .register_p2_im_message_receive_v1(handle_message) \ .build() # 建立长连接并阻塞运行 client.start(handler) if __name__ == "__main__": main()

这段代码里,EventDispatcherHandler.builder("", "")的两个空字符串参数在长连接模式下可以留空,Webhook模式下需要填验证相关的配置。client.start(handler)会建立WebSocket长连接并阻塞运行,程序会一直挂在这里,有事件进来就触发handle_message。启动之后,在飞书群里@机器人发一条消息,控制台应该能看到收到事件的日志。

有朋友问为什么不是client.run(),这也是版本差异导致的,新版SDK统一用start,老版本是run,以官方文档为准。我用的lark-oapi0.4.0以上版本,start是标准写法。

3.2 从事件里提取用户真正想问的问题

这是整个对接过程中最容易踩坑的地方。飞书事件里带过来的原始消息内容,不是你看到的纯文本,而是包含了很多标记的JSON结构。比如用户在群里发“@我的AI机器人 帮我写个排序”,事件里的文本大概长这样:<at user_id="xxxx">我的AI机器人</at> 帮我写个排序。所以不能直接把整段文本丢给AI,AI会看到一串奇怪的标签,回答也会莫名其妙。

import json from lark_oapi.api.im.v1 import P2MessageReceiveV1 def handle_message(data: P2MessageReceiveV1) -> None: event = data.event # 消息类型:text、image、post等,只处理文本消息 if event.message.message_type != "text": return # 解析消息内容 message_content = json.loads(event.message.content) # content是一个JSON字符串,text字段才是真正的文本内容 text = message_content.get("text", "") # 去除@机器人的部分 # 飞书文本消息中的@格式是:<at user_id="xxx">名字</at> import re clean_text = re.sub(r'<at[^>]*>.*?</at>', '', text).strip() # 清理后可能就是空消息(只@了机器人但没问问题) if not clean_text: send_message(event.message.chat_id, "你好,我是AI助手,请告诉我你需要什么帮助。") return # 打印收到的问题,便于调试 print(f"收到问题:{clean_text}") # 调AI模型,回复结果 reply = call_qwen(clean_text) send_message(event.message.chat_id, reply)

这里的核心逻辑是:先判断消息类型,只处理text类型,图片和富文本消息先忽略;然后把event.message.content这个字符串解析成JSON,取出text字段;最关键的一步是用正则把<at ...>...</at>这段剔除掉,剩下的才是用户真正输入的问题。如果你用的是Python的re模块,注意正则需要匹配到>结尾的标签头和</at>结尾的标签尾,这在飞书的<at>格式下是稳定的。如果用户是在群里直接输入“帮我看下这个配置”而没有@机器人,飞书默认不会把消息推给机器人,这也是群内@机制的安全策略,不用做什么额外处理。

3.3 调用通义千问API并发送回复

这是真正和AI模型打交道的部分。用openai库创建客户端,调用chat接口,然后把回答发回飞书群。

from openai import OpenAI DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") DASHSCOPE_BASE_URL = os.getenv("DASHSCOPE_BASE_URL") MODEL_NAME = os.getenv("MODEL_NAME", "qwen-plus") client = OpenAI( api_key=DASHSCOPE_API_KEY, base_url=DASHSCOPE_BASE_URL, ) def call_qwen(prompt: str) -> str: try: resp = client.chat.completions.create( model=MODEL_NAME, messages=[ {"role": "system", "content": "你是一位严谨、专业的技术助手,回答问题时尽量简洁、有条理。"}, {"role": "user", "content": prompt}, ], temperature=0.7, timeout=30, ) return resp.choices[0].message.content except Exception as e: print(f"调用通义千问失败:{e}") return f"抱歉,AI服务暂时不可用:{str(e)}"

几个参数的经验值说明一下:temperature=0.7是创造性和稳定性之间的平衡点,如果做代码生成可以降到0.2,做文案创意可以拉到0.9。timeout=30是请求超时时间,通义千问的响应速度一般几秒到十几秒,长文本生成可能超过30秒,如果经常超时可以调到60。system角色消息里我写了“严谨、专业、简洁”的提示词,这个不是必须,但实测加了这个之后回答质量会稳定不少,不会动不动给你写一大段空话。

发消息回飞书群也有一组固定写法:

def send_message(chat_id: str, text: str) -> None: try: # 需要在这里创建im_client im_client = lark.Client.builder() \ .app_id(app_id) \ .app_secret(app_secret) \ .build() request = CreateMessageRequest.builder() \ .receive_id_type("chat_id") \ .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type("text") .content(json.dumps({"text": text})) .build() ) \ .build() response = im_client.im.v1.message.create(request) if not response.success(): print(f"发送消息失败:{response.code} {response.msg}") except Exception as e: print(f"发送消息异常:{e}")

注意这里content字段必须是一个JSON字符串,里面{"text": "内容"}的格式不能错。receive_id_type我传了chat_id,表示receive_id传的是群ID。事件里的event.message.chat_id直接就能拿来做receive_id,不用额外查询。

3.4 完整脚本:拿来就能跑

把上面三块拼在一起,就是一个可以直接运行的完整脚本。我平时为了方便调试,会在文件末尾加一段发布模式检查,确保.env里的配置都读到了。

import os import re import json from dotenv import load_dotenv import lark_oapi as lark from lark_oapi.api.im.v1 import * from openai import OpenAI load_dotenv() APP_ID = os.getenv("FEISHU_APP_ID") APP_SECRET = os.getenv("FEISHU_APP_SECRET") DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") DASHSCOPE_BASE_URL = os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1") MODEL_NAME = os.getenv("MODEL_NAME", "qwen-plus") # 检查必需配置 if not all([APP_ID, APP_SECRET, DASHSCOPE_API_KEY]): raise RuntimeError("请检查.env配置,缺少必要的密钥") ai_client = OpenAI( api_key=DASHSCOPE_API_KEY, base_url=DASHSCOPE_BASE_URL, ) def call_qwen(prompt: str) -> str: try: resp = ai_client.chat.completions.create( model=MODEL_NAME, messages=[ {"role": "system", "content": "你是一位严谨、专业的技术助手,回答问题时尽量简洁、有条理。"}, {"role": "user", "content": prompt}, ], temperature=0.7, timeout=30, ) return resp.choices[0].message.content except Exception as e: print(f"调用通义千问失败:{e}") return f"抱歉,AI服务暂时不可用:{str(e)}" def send_message(chat_id: str, text: str) -> None: client = lark.Client.builder() \ .app_id(APP_ID) \ .app_secret(APP_SECRET) \ .build() request = CreateMessageRequest.builder() \ .receive_id_type("chat_id") \ .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type("text") .content(json.dumps({"text": text})) .build() ) \ .build() response = client.im.v1.message.create(request) if not response.success(): print(f"发送消息失败: code={response.code}, msg={response.msg}") def handle_message(data: P2MessageReceiveV1) -> None: event = data.event if event.message.message_type != "text": return message_content = json.loads(event.message.content) text = message_content.get("text", "") clean_text = re.sub(r'<at[^>]*>.*?</at>', '', text).strip() if not clean_text: send_message(event.message.chat_id, "你好,我是AI助手。请在群里@我并输入你的问题。") return print(f"收到问题:{clean_text}") reply = call_qwen(clean_text) send_message(event.message.chat_id, reply) def main(): client = lark.Client.builder() \ .app_id(APP_ID) \ .app_secret(APP_SECRET) \ .log_level(lark.LogLevel.INFO) \ .build() handler = lark.EventDispatcherHandler.builder("", "") \ .register_p2_im_message_receive_v1(handle_message) \ .build() print("AI机器人已启动,等待飞书消息...") client.start(handler) if __name__ == "__main__": main()

这段代码保存为main.py,终端运行python main.py。看到“AI机器人已启动”的日志后,在飞书群里@机器人发条消息试试。第一次跑通这条链路看到群里出现AI的回复,那种感觉还是有点爽的。

4. 进阶玩法:一个机器人接入多个AI模型

4.1 关键词触发模型切换

群里的需求千奇百怪,有的人问简单问题,有的要跑复杂代码,用单一模型档位总是有人嫌慢或者嫌笨。我给机器人加了一个指令解析层:用户可以在消息里显式指定模型,格式是@机器人 qwen-max 帮我写一个复杂算法,程序解析出第一个词是模型名称就切换,否则走默认模型。

MODEL_ALIAS = { "turbo": "qwen-turbo", "plus": "qwen-plus", "max": "qwen-max", } def parse_command(clean_text: str): parts = clean_text.strip().split(maxsplit=1) if len(parts) == 2 and parts[0].lower() in MODEL_ALIAS: return MODEL_ALIAS[parts[0].lower()], parts[1] return None, clean_text def handle_message(data: P2MessageReceiveV1) -> None: event = data.event if event.message.message_type != "text": return message_content = json.loads(event.message.content) text = message_content.get("text", "") clean_text = re.sub(r'<at[^>]*>.*?</at>', '', text).strip() if not clean_text: send_message(event.message.chat_id, "你好,我是AI助手。请在群里@我并输入你的问题。") return model_alias, prompt = parse_command(clean_text) model_name = model_alias if model_alias else os.getenv("MODEL_NAME", "qwen-plus") print(f"使用模型 {model_name} 处理问题:{prompt}") reply = call_qwen(prompt, model_name) send_message(event.message.chat_id, reply)

parse_command函数做的事情很简单:按空格分词,如果第一个词命中了模型别名映射表,就把这个词摘出来当模型名,剩下的作为问题。这样用户不需要记API模型名,说turbo、plus、max就行。一开始可能觉得这个功能鸡肋,但群里多几个人用起来之后,你会发现这个小小的指令解析大大提升了机器人的实用性。

需要注意一个细节:如果用户的问题本身就以“max”开头,比如“max函数怎么用”,就会被误判成模型切换。我在完整性代码里加了个判断,只有映射表里明确存在的词才触发切模型,所以不会出现这种误伤。

4.2 不想用云API?本地模型的接入方式

聊到模型切模型,那就顺带说一下本地模型。这个月身边好多人都在折腾本地部署,小一点的用Ollama跑量化版模型,机器好的用vLLM部署全量模型。不管是哪种方式,它们普遍暴露的是OpenAI兼容接口,这意味着我们上面写的代码几乎可以无缝接过去。

拿Ollama举例,本地装好Ollama并拉取一个模型后,它会监听http://localhost:11434/v1这个地址,也是OpenAI兼容的。你只需要把.env里配成:

DASHSCOPE_API_KEY=ollama DASHSCOPE_BASE_URL=http://localhost:11434/v1 MODEL_NAME=qwen2.5:7b

然后重新启动机器人,它就能直接调用本地模型了。接口的请求格式完全一致,代码零改动。这就是当初选openai库带来的最大红利。

做一个表格整理一下几种接入方式的差异:

接入方式base_url需要GPU数据私密性单次响应速度
通义千问APIdashscope.aliyuncs.com/compatible-mode/v1不需要数据出公网快(取决于服务端负载)
Ollama本地模型localhost:11434/v1可选,CPU也能跑小模型数据不出本机中(取决于硬件)
vLLM部署模型内网IP:8000/v1需要数据不出内网快(并发能力强)

如果你既要云模型的强能力,又想让部分敏感数据走本地,可以在这套代码基础上加一个“按关键字路由”的规则,比如消息里包含“本地”两个字就走Ollama,否则走通义千问。路由逻辑无非是在handle_message里多几个分支,配合上面已有的parse_command,整体做下来也就二三十行代码的事。

4.3 让机器人发送更丰富的消息

纯文本回复在日常体验上还是有点单薄。飞书机器人支持多种消息类型,我建议至少把post富文本消息学会,这样AI回答里带代码块、带换行时,显示效果会好很多。

def send_post_message(chat_id: str, title: str, content: str) -> None: # 将普通文本按换行拆分成富文本行 lines = content.split("\n") elements = [] for line in lines: elements.append([{"tag": "text", "text": line}]) post_content = { "zh_cn": { "title": title, "content": elements, } } client = lark.Client.builder() \ .app_id(APP_ID) \ .app_secret(APP_SECRET) \ .build() request = CreateMessageRequest.builder() \ .receive_id_type("chat_id") \ .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type("post") .content(json.dumps(post_content)) .build() ) \ .build() response = client.im.v1.message.create(request) if not response.success(): print(f"发送富文本消息失败: code={response.code}, msg={response.msg}")

post消息的JSON结构有两层:zh_cn下的title是消息头,content是一个二维数组,每个数组元素是消息里的一行,行内可以放多个tag段。上面代码就是最简单的“一行放一个文本段”的拆解方式。

富文本方案最多算及格,再进阶一点可以上interactive卡片消息,支持按钮交互、折叠列表、图片等。我尝试做过一个卡片版,每条AI回答下方带一个“重新生成”按钮,点了之后机器人重新调一次模型并更新卡片。这需要处理卡片回调事件card.action.trigger,逻辑也不复杂,就是给按钮加一个value传原始问题,回调时再跑一遍call_qwen。对聊天气氛来说,卡片消息的交互感确实比纯文本强很多,也更像“一个正经的AI产品”,但开发复杂度会上一个台阶,建议先把核心跑通再考虑。

5. 常见问题与排查记录

5.1 机器人完全不回复,怎么定位

遇到“@了机器人没反应”的情况,不要慌,按这个顺序排查。第一步,看程序控制台有没有打印收到事件的日志。如果连“收到问题”都没有,说明事件订阅就有问题,去飞书开放平台检查应用是否已经发布版本、是否添加了im.message.receive_v1事件和对应的权限。第二步,确认你是用“企业自建应用”的方式调用,而不是在聊天窗口直接添加了一个Webhook机器人。自建应用机器人要加到群里,群主或管理员在群设置里找到“群机器人”→“添加机器人”→选中你的应用,这一步遗漏的话,消息根本到不了应用后台。第三步,确认代码里client.start(handler)是否报错。长连接建立失败会有明显的异常日志,把控制台的报错贴到搜索引擎,基本都能找到答案。

还有一次我被困了很久的问题:本地代码跑得好好的,部署到服务器之后不回复了。最后发现是服务器防火墙把出站的WebSocket连接端口拦了,长连接根本没建立成功。如果部署到云服务器,记得放行端口,SDK默认走443端口,多数情况下不会拦,但内网环境要特别注意。

5.2 通义千问API调用报错

API报错是日常最容易碰到的问题。我用一张表把常见错误整理出来:

错误现象可能原因解决办法
401 UnauthorizedAPI Key不正确或已过期检查 .env 里的DASHSCOPE_API_KEY,在百炼平台重新生成
404 Not Foundbase_url路径错误确认是/compatible-mode/v1,不是/api/v1
400 InvalidParameter模型名写错确认是qwen-turbo/qwen-plus/qwen-max,大小写敏感
429 Too Many Requests触发并发限制或额度用尽降低调用频率,或升级服务额度
Request timed out请求超时调大timeout参数,或换更快的模型档位

429这个问题在企业群里比较常见,如果群里好几个人同时@机器人,瞬间就是几十个并发请求,免费额度分分钟打满。我的做法是加了一个简单的信号量并发控制,限制同时最多处理5个请求,超出的排队等待,这样既不会把API打挂,也不会出现恢复之后一瞬间涌入一堆消息的情况。

import threading request_semaphore = threading.Semaphore(5) def call_qwen_safe(prompt: str, model: str = MODEL_NAME) -> str: with request_semaphore: return call_qwen(prompt, model)

5.3 消息被重复处理

飞书事件订阅机制自带重试逻辑,如果我们的程序处理事件时异常退出,飞书会隔一段时间重新推送同一条事件。如果处理逻辑里有副操作(比如发消息),就可能出现“用户发一条,群里回两条”的重复。

解决思路是做幂等。最简单的办法是维护一个已处理消息ID的集合,收到事件时先判断event.message.message_id是否在集合里,在就直接返回。内存集合在进程重启后会清空,但对日常使用足够。

processed_messages = set() def handle_message(data: P2MessageReceiveV1) -> None: event = data.event msg_id = event.message.message_id if msg_id in processed_messages: print(f"重复消息已跳过:{msg_id}") return processed_messages.add(msg_id) # 其他逻辑...

对个人项目或二三十人的团队群来说,内存去重够用了。如果以后做到几千人的大群,可以换成Redis的SETNX做分布式去重,原理一样。

5.4 日志打得好,排查没烦恼

最后分享一个经验:不要小看日志。刚开始调试时,我遇到问题全靠print硬找,后来事件多了根本看不过来。我习惯加一个logger辅助函数,把收到的原始事件、解析出的清理文本、选的模型名、API响应时间都打出来。这样哪怕第二天醒了看到群里有人说机器人没回话,翻一眼日志就能定位是飞书侧的问题还是模型侧的问题。

import time def log_info(msg: str): ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) print(f"[{ts}] {msg}") def handle_message(data: P2MessageReceiveV1) -> None: event = data.event log_info(f"收到事件 message_id={event.message.message_id}, chat_id={event.message.chat_id}") start = time.time() # ...中间逻辑... log_info(f"AI响应耗时 {time.time() - start:.2f}s")

日志格式可以按自己的习惯调整,但有几项信息建议必打:消息ID、群ID、发送者ID、清理后的文本、模型名称、处理耗时。这几项组合起来,95%的问题都能直接定位。

写在最后

整个集成过程做下来,我个人最深的体会是:把通义千问接进飞书机器人,技术难度其实不高,核心就三点——飞书事件订阅机制要搞明白、OpenAI兼容接口的调用方式要灵活运用、消息解析细节要处理干净。反而是需求侧要想清楚:你的群里到底需要AI解决什么问题,是多轮对话、代码生成还是文档问答?这决定了你后续要不要接知识库、要不要做多轮上下文记忆。我目前这套方案已经在我们团队跑了一段时间,日常问题问答基本够用,下一步我准备把飞书云文档里的知识库接进来,让AI能基于团队自己的文档回答,到时候再写一篇分享。如果你在动手过程中卡在哪一步了,欢迎对照着这篇文章的排查清单逐条过一遍,很多时候问题就出在你以为没问题的那一环。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询