上周末我把这个机器人从本地测试群推到部门大群之后半小时内被同事了二十几次。有人问它能不能写周报有人让它解释一段线上日志里的报错还有人直接扔了个需求文档链接过来。那一刻我才觉得这个通义千问对接飞书机器人的项目算真正做完了。这篇是系列的第二篇2-2上一篇已经把账号准备、飞书应用的创建、通义千问API Key申请这些地基工作讲完了。这篇我直接进入正题飞书机器人和通义千问之间的完整对接方案包括架构选型、关键代码实现、上下文管理、流式体验处理以及最后部署上线时踩过的坑。核心就一个目标——让飞书里的机器人能像一个人一样把通义千问的能力用起来。1. 整体架构与接入模式选型先分清三种对接路子飞书机器人接AI模型网上的教程大多只给一种解法但实际你在选型时至少要面对三个岔路口机器人形态、消息链路、以及回复方式。这三个选择直接决定了后面代码怎么写、部署在哪里、能扛多大并发。1.1 自定义机器人还是企业自建应用飞书的机器人分两种自定义机器人和企业自建应用。自定义机器人像群里的一个webhook投递员它的核心能力是往群里发消息也能通过关键词或者触发被动的命令但它的交互能力很弱——没有事件订阅、没有用户身份识别、没有主动发消息的权限更没有读写云文档的授权能力。企业自建应用则是飞书开放平台里的正规军能启用机器人能力、订阅消息事件、发送主动消息、操作卡片交互、访问通讯录和云文档。如果你只想做一个群里发个指令、机器人回一段话的玩具自定义机器人也能跑通。但我的项目明确需要一个带上下文、能并发、后续还要接知识库和卡片的机器人所以我选了企业自建应用。这是我踩完两种方案之后最直接的建议回归到你的真实需求如果交互方案里包含判断用户身份多轮对话主动通知任何一个词不用犹豫直接走自建应用。1.2 消息链路的几种连接方式自建应用的消息链路本质上解决的是一个互联网通用问题飞书收到用户消息怎么把这条消息给你的后端服务飞书官方给了三种方式Webhook事件订阅飞书把用户消息以HTTP POST请求推送到你指定的公网URL上。消息是飞书主动推给你的。长连接WebSocket模式飞书开放平台提供了长连接的方式你的服务主动去飞书服务器建立一条长连接事件沿着这个连接推下来。轮询新版本已淘汰已经不推荐就不多说了。做这个项目的时候我一开始用了Webhook方式把服务部署在一台有公网IP的云服务器上用Nginx反代到Flask服务这套链路本身没问题但后来我发现一个巨大的坑在Webhook模式下飞书服务器要求你的服务在3秒内返回200响应超过则判定回调超时触发重试。而这个AI机器人调用通义千问的API快则一两秒慢则十几秒3秒根本扛不住。解决办法有两个一是收到事件后立即返回200把消息丢进队列异步处理再通过主动发消息接口把结果发回去二是我后来实际采用的方案——切换成长连接模式。长连接模式下飞书不再要求快速的HTTP响应事件是推送给我本地常驻的客户端我可以慢慢处理处理完了再调API把结果发出去。这样一来整个项目对公网依赖几乎为零部署在个人电脑或者内网服务器也能跑。这个选型带来的收益在后来的开发中多次体现出来我强烈建议你先看长连接模式的官方文档别急着配Webhook。1.3 回复方式被动回复还是主动消息飞书的事件回调里可以直接返回一个被动回复的响应。听起来方便但这个被动回复同样受3秒超时限制而且它有格式限制、不能发卡片。所以我在实际项目里全部采用主动消息方案收到消息事件后解析出会话IDchat_id然后调用im/v1/messages接口把结果主动推送到这个会话里。消息链路完整跑起来之后就一句话飞书事件长连接或Webhook→ 后端解析校验 → 调通义千问API → 组装消息 → 主动推送回会话。2. 飞书侧配置细节权限、事件订阅与密钥管理其实对接这个活三分之二的难度在飞书开放平台的控制台里而不是在代码里。配置错了代码写得再漂亮都白搭。我把配置过程中最容易出问题的点拆开说。2.1 创建应用并开通机器人能力在飞书开放平台后台创建企业自建应用名字我起的就是通义千问助手创建后进入应用详情页。第一步就是找到应用能力→机器人点击启用。这一步不启用后面所有事件订阅都无从谈起。然后是权限管理这一步很多人会漏。机器人要读取用户发给它的消息内容必须开通以下权限im:message读取消息im:message:send_as_bot以机器人身份发消息im:chat读取群信息用于判断群聊和单聊权限申请之后要创建版本并发布发布后管理员审核通过才算生效。我这里卡过一次我在控制台以为权限配好了但没走发布流程代码里一直报权限不足。2.2 事件订阅长连接配置与请求地址进入事件与回调页面订阅方式选使用长连接接收事件。然后添加事件这里必须勾选这两个im.message.receive_v1接收消息im.message.message_read_v1消息已读可选长连接的好处前面说过了但使用长连接有个前提官方推荐使用飞书开放平台的SDK来建立连接。我项目里用的是Python直接装lark-oapi这个SDK里封装了长连接客户端几行代码就能连上。2.3 加解密与Verification Token的用处飞书事件订阅里有一个Encrypt Key和Verification Token。加密密钥用于对推送过来的事件body做AES解密Verification Token用来做基础的事件来源校验。配置好之后飞书会提供一个验证challenge的流程在你保存配置的瞬间飞书会向你的服务发送一个带有challenge字段的请求长连接模式下这个验证发生在SDK内部你的服务需要原样返回这个字段。用SDK时这段逻辑已经封装好了但如果你手写Webhook需要特别处理否则连保存配置那一步都过不去。密钥管理上我的习惯是所有密钥存环境变量不写进代码仓库。包括APP_ID、APP_SECRET、VERIFICATION_TOKEN、ENCRYPT_KEY以及通义千问的DASHSCOPE_API_KEY。后来上CI/CD流水线时这套环境变量的设计帮我省了很大麻烦——仓库里一份配置模板真正密钥只在服务器的环境变量里。3. Python后端核心实现事件处理、加解密与调用通义千问配置和选型都定下来之后进入代码阶段。我用的是Python 3.10 FastAPI调用通义千问用的是阿里云百炼平台的dashscopeSDK。这部分我只贴核心代码并解释关键逻辑。3.1 长连接事件接收服务用lark-oapi的SDK长连接模式的起服务非常简单import lark_oapi as lark from lark_oapi.api.im.v1 import * def on_message(event: lark.MessageReceiveEvent) - None: # 处理消息事件 pass # 创建client client lark.Client.builder() \ .app_id(os.environ[APP_ID]) \ .app_secret(os.environ[APP_SECRET]) \ .log_level(lark.LogLevel.INFO) \ .build() # 注册事件处理器 client.event.handler.register(lark.EventType.MESSAGE_RECEIVE, on_message) # 建立长连接 client.ws.start()核心逻辑就三行注册事件处理器、注册回调函数、启动长连接。SDK会自动处理加密解密、自动重连、心跳保活省掉了一个大工程。这里有一个细节值得注意SDK的MessageReceiveEvent里已经帮你解好了消息体和发送者信息你拿到的event.message.content是JSON字符串里面是消息的具体内容。3.2 消息内容的解析与触发词判断消息进来后我做的第一件事不是调用AI而是先判断这条消息到底该不该回。我的触发规则是单聊消息必回群聊消息必须机器人本人mention里有机器人的ID。import json def should_reply(event: lark.MessageReceiveEvent) - bool: message event.message # 单聊 if message.chat_type p2p: return True # 群聊必须机器人 if message.chat_type group: mentions json.loads(message.mentions) if message.mentions else [] bot_id event.request_id # 简化示意实际用机器人的open_id判断 for m in mentions: if m.get(id, {}).get(open_id) bot_id: return True return False这样做不只是为了体验更是为了成本。通义千问每次调用都消耗token如果群里每句话都触发一天几千次调用账单会涨得让你肉疼。这也是我给所有做AI机器人的朋友的第一条建议触发策略一定要写得足够保守。消息内容解析飞书的消息content格式是JSON字符串比如文本消息长这样{text: 你好帮我写个Python脚本}解析出来就是原始文本。我把文本拿出来去掉机器人的那段_user_1占位符飞书在群聊文本里会把对象渲染成类似_user_1的形式得到用户真正想问的内容。3.3 调用通义千问非流式与流式两种方式调用通义千问我用的是dashscopeSDK。最基础的非流式调用import dashscope from dashscope import Generation dashscope.api_key os.environ[DASHSCOPE_API_KEY] def call_qwen(prompt: str, history: list None) - str: messages [] if history: messages.extend(history) messages.append({role: user, content: prompt}) response Generation.call( modelqwen-turbo, # 按需求选qwen-max/qwen-plus messagesmessages, result_formatmessage ) if response.status_code 200: return response.output.choices[0].message.content else: # 降级处理 return f调用失败{response.code} {response.message}但实际项目里我用的不是这种方式而是流式调用因为通义千问的完整回复可能需要几秒到十几秒非流式接口会让用户感觉卡死了。用流式我可以先把回复的前几个字推给用户后面边生成边推。不过这里有个飞书的现实问题飞书消息接口是整条发送的不支持真正的打字机流式推送。一次只能发出去一条完整消息想要打字机效果通常的手段是发一张更新卡片通过反复调用patch接口更新卡片内容。但那样做会消耗大量API调用次数和触发更复杂的消息状态管理。经过实测我最后采取的方案是折中收到消息后先立刻给用户发一条占位消息收到正在思考中...然后调用通义千问的非流式接口或者流式接口我取完整结果得到完整回复后把之前的占位消息编辑掉替换成真正的回复编辑消息用的是im/v1/messages/{message_id}的PATCH接口SDK里封装的MessageService直接能调。这样用户既能看到机器人已响应又不用等十几秒的沉默。3.4 发送消息回飞书回消息的代码from lark_oapi.api.im.v1 import * def send_text(chat_id: str, text: str): request CreateMessageRequest.builder() \ .receive_id_type(chat_id) \ .request_body(CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type(text) .content(json.dumps({text: text}, ensure_asciiFalse)) .build()) \ .build() response client.im.v1.message.create(request) return response就这么简单一个chat_id和一个消息体。但注意content必须是一个JSON字符串而且消息体里的文本不能超过飞书的限制。实测下来飞书文本消息的content长度上限大约在150KB但出于阅读体验考虑我会在代码里对回复内容做个截断逻辑超过一定长度就拆成多条或者提示用户内容太长已为你生成文件。4. 让机器人聊得下去上下文管理与用户状态处理通义千问本身是支持多轮对话的但API接口是无状态的——你不把历史消息带过去它就是一次性问答。要让机器人有记忆就得自己在服务端维护上下文。4.1 会话维度的选择我最初很天真地用chat_id也就是群维度作为会话维度。但很快就发现问题一个20人的群里A让机器人写SQLB让机器人讲笑话如果共享同一份上下文模型会把两个人的话题混在一起对话逻辑一塌糊涂。正确做法是单聊场景按用户维度记忆群聊场景按群ID 用户ID组合记忆。也就是说在群里每个人有自己独立的对话历史互不干扰。4.2 上下文存储先从内存方案起步存储这块我第一阶段用的是最简单的方案内存字典。from collections import defaultdict, deque from datetime import datetime, timedelta # session_id - deque of {role, content} context_store defaultdict(lambda: deque(maxlen20)) # session_id - last_active_time active_time_store {} def get_session_id(event): if event.message.chat_type p2p: return fp2p:{event.message.chat_id} else: sender_id event.message.sender.sender_id.open_id return fgroup:{event.message.chat_id}:{sender_id}内存方案的好处是零依赖本机就能跑通。坏处很明显重启服务上下文全丢多实例部署时上下文不同步。但作为一个MVP阶段的机器人我完全接受这种折中。我同时给每条上下文带上了时间戳超过TIMEOUT_MINUTES30的会话会被清理掉避免内存无限增长。4.3 上下文的Token裁剪策略通义千问有上下文长度限制不同模型不同qwen-turbo是32Kqwen-max是32K新版本支持更长。但不可能把所有历史消息都塞给模型。我的策略是只保留最近10轮用户助手各算一轮的对话每轮消息按500字截断超长的直接裁掉尾部如果当前会话内容总长度超过模型token限制的70%就丢弃最久远的历史消息只保留当前这条问题这个裁剪策略在原项目里被验证非常管用。特别是群聊中粘贴代码块时一条消息就可能占几千个token不做裁剪后续对话直接报上下文超限。4.4 超时、并发与降级我之前用的通义千问接口默认的并发限制是几十个QPS个人项目完全够用。但要注意一个隐藏问题如果多个用户同时机器人每个请求都要排队调通义千问通义千问在段时间内大量请求时会有随机超时。我加了一个简单的信号量来控制并发同时给调用设置了超时时间我设的是30秒。一旦超时就返回友好的降级文案而不是让用户一直等。import asyncio from aiostream import stream semaphore asyncio.Semaphore(5) async def safe_call_qwen(prompt, history): async with semaphore: try: return await asyncio.wait_for( call_qwen_async(prompt, history), timeout30 ) except asyncio.TimeoutError: return 抱歉模型响应超时了请稍后再试。5. 部署上线与持续集成我是怎么把服务跑稳的代码写完本地跑通这只是完成了40%。一个机器人能不能长期稳定服务部署架构和运维策略占了剩下的60%。这个项目里我踩的坑主要集中在部署这块。5.1 进程守护与重启策略长连接模式的客户端进程最怕的就是SDK底层的连接断开后没有自动重连。实际上lark-oapi的SDK已经有重连机制但保险起见我还是用systemd做了一个守护进程万一整个进程挂掉自动拉起来。[Unit] DescriptionQwen Feishu Bot Afternetwork.target [Service] Userubuntu WorkingDirectory/opt/qwen-bot EnvironmentFile/etc/qwen-bot.env ExecStart/opt/qwen-bot/venv/bin/python main.py Restartalways RestartSec5 [Install] WantedBymulti-user.target有几个点值得注意EnvironmentFile指定了外部环境变量文件这样APP_ID、APP_SECRET这些密钥不落地到仓库也不出现在systemd启动命令里。Restartalways意味着进程异常退出时系统5秒后自动拉起。实测SDK偶发断线时这一招能让机器人秒恢复。WorkingDirectory要写对否则相对路径导入模块会挂。5.2 持续集成用GitHub Actions自动部署项目上了GitHub仓库之后我配了一个简单的CI流水线。每次推送到main分支触发测试、构建和部署name: Deploy Qwen Bot on: push: branches: [main] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - name: Set up Python uses: actions/setup-pythonv5 with: python-version: 3.10 - name: Install dependencies run: pip install -r requirements.txt - name: Run tests run: pytest - name: Deploy via SSH uses: appleboy/ssh-actionv1.0.3 with: host: ${{ secrets.SERVER_HOST }} username: ${{ secrets.SERVER_USER }} key: ${{ secrets.SSH_PRIVATE_KEY }} script: | cd /opt/qwen-bot git pull origin main /opt/qwen-bot/venv/bin/pip install -r requirements.txt sudo systemctl restart qwen-bot这个流水线做得很朴素但已经能把本地改代码→测试→服务器拉取→重启服务这一整套串起来了省去的重复劳动非常可观。对个人项目来说CI/CD不一定要用K8s或者Docker先用systemd SSH跑通最小闭环价值就已经很大了。5.3 日志与监控AI机器人比普通机器人更依赖日志因为模型输出的不确定性导致很多问题只有跑起来才能看到。我的日志策略是每个请求打一条INFO日志时间、会话ID、用户输入摘要、模型回复长度、耗时模型调用失败打ERROR日志包含完整入参和出参用tail -f /var/log/qwen-bot.log实时观察另外我在飞书里建了一个告警群通过自定义机器人把ERROR日志同步推送过去。这样服务出问题时我能第一时间知道不用等用户抱怨。6. 进阶玩法从文本回复到表格卡片与知识库文本回复只是基础真正让这个机器人产生更大价值的是把通义千问的输出变成飞书原生的结构化内容。6.1 让机器人发表格飞书机器人可以发送富文本消息和消息卡片卡片里支持表格样式。我用消息卡片做了一次升级让通义千问按固定JSON格式输出然后我把JSON解析成卡片的表格组件发出去。比如用户问帮我整理上周发布的三个版本的功能对比我提示模型返回如下结构{ title: 版本功能对比, table: { header: [版本号, 发布日期, 核心功能, 风险], rows: [ [v1.2.0, 2025-01-06, 新增登录流程, 低], [v1.2.1, 2025-01-10, 修复支付回调, 中], [v1.3.0, 2025-01-15, 重构消息模块, 高] ] } }然后调用飞书消息卡片的interactive类型把表格数据渲染成卡片里的table组件飞书卡片支持lark_md和表格组件。这样同事在手机上的阅读体验就比纯文本好太多了。这里顺便说一个我的经验让模型输出结构化JSONPrompt里一定要给极端示例作为示范。比如明确告诉模型如果某个单元格没有数据输出无而不是null如果列表为空返回N/A。否则模型偶尔输出不符合预期的JSON解析就直接报错。6.2 交互卡片让回复可操作卡片不止能展示还能接收点击事件。我给机器人加了一个重新生成按钮卡片下方放一个按钮用户点击后触发一个新事件后端收到这个事件后重新调一次通义千问换个随机temperature把新结果更新到同一张卡片里。实现要点卡片里定义按钮的value字段比如{action: regenerate, session_id: xxx}飞书把按钮点击事件推送到后端事件类型是card.action.trigger在事件处理里根据value分发处理用PATCH接口更新原卡片这个玩法虽然简单但它是从工具到产品的一个分水岭。用户不再只是被动接收结果而是能和结果交互。6.3 私有知识库让模型懂你的业务最后聊一下知识库的方向。纯通义千问是通用大模型不懂你的业务细节。要让它懂目前个人项目里性价比最高的路子是RAG检索增强生成——把内部文档向量化存起来收到问题时先检索最相关的文档片段再和用户问题一起拼进Prompt发给模型。结合之前的热搜词ai知识库向量模型spring-ai集成rag我在这个项目里做过一个简化版用通义千问的embedding接口把文档切成chunk向量化存到本地向量数据库我用的是Chroma用户提问时先检索Top5相关段落和问题一起组装成Prompt调通义千问这个方向上要注意的点是chunk的切分策略按语义切还是按固定字符切对检索效果影响很大。我后来是把Markdown标题作为天然的分段标志来切效果比固定长度好很多。7. 避坑清单我从这次对接中总结的七个现实问题最后这部分是我这次开发中真正让人抓狂的问题集合。每一个都是真实踩过、翻阅了大量文档和源码才解决的写在这里希望能帮你省掉一部分排查时间。问题一事件订阅的challenge校验失败现象飞书后台保存事件订阅配置时提示URL验证失败。原因排查链路如果是Webhook模式检查服务端是否正确返回challenge字段如果是长连接模式检查SDK启动的日志里有没有handshake成功的记录。我遇到过的情况是Encrypt Key配置错误导致SDK解密失败——因为我在飞书后台复制Encrypt Key时把多余的空格也粘进去了。问题二消息事件重复推送我在本地日志里发现同一条消息被处理了两次。原因是飞书的事件推送有至少一次语义消息重试机制会重复投递。解决方案引入简单的幂等机制用消息IDmessage_id做去重。代码里维护一个最近处理过的message_id集合如果一条消息的ID已经处理过直接丢弃。问题三群聊里机器人被所有消息刷屏这个前面提过了触发策略写得不保守就会爆。我把群聊必须机器人才回复作为一个明确的断言写进了代码开头。另外注意即使你的触发条件是机器人飞书事件里仍然会推送群里所有消息你必须自己判断mention列表里有没有机器人自己。问题四长文本被截断通义千问的回答经常超过2000字飞书的文本消息接口对超长文本的渲染很不友好。我的处理方式是检测到消息长度超过阈值后改用发文方式把内容作为云文档内容创建出来调用飞书的云文档API然后在聊天里只发一条文档链接给用户。这个体验反而比纯文本更好用户可以点开稍后阅读。问题五通义千问的模型名写错qwen-max、qwen-plus、qwen-turbo这几个模型名看着都差不多但它们的能力、限流、价格都不一样。在某个时间段内部分模型的名称会有带日期后缀的版本比如qwen-max-2025-01-25。如果直接复制网上的代码用的模型名不对会报404模型不存在。建议去百炼平台控制台看一下当前可用的模型列表确定你实际开通的是哪个。问题六Session Key的键冲突我给会话ID设计的格式是p2p:{chat_id}和group:{chat_id}:{user_id}这个格式有个坑如果chat_id本身包含冒号拼接出来的key可能和另一个会话冲突。实际上飞书的open_id格式里包含下划线不包含冒号但为了安全我还是用了JSON序列化来组key彻底避免拼字符串的歧义。问题七权限范围与发布版本过期飞书的权限申请之后如果应用有新版发布某些旧的权限范围可能需要重新审核。我曾经遇到一个情况代码里调用的API要求im:chat权限但应用的发布版本里忘了带上这个权限导致生产环境报permission denied。排查半天才意识到是这个原因。现在这个机器人已经在部门里跑了大半个月了稳定性和体验基本达到了生产可用的程度。要说当初有没有走了弯路确实有——比如一开始选Webhook方案绕了一大圈最后切了长连接才真正舒坦。如果让我给后来者一个建议先把消息链路跑通再追求智能。机器人能收到消息、能发消息哪怕只是一个echo机器人整个对接的基础就已经搭好了。剩下的事情——上下文、卡片、知识库、CI/CD——都是在这个基础上按需叠加的。先解决通再追求智这个顺序能让你少走许多弯路。 
长河君 · 2026/9/16 4:08:23 · 阅读 1.2k 通义千问接入飞书机器人：从长连接到上下文管理的实战指南
上周末我把这个机器人从本地测试群推到部门大群之后半小时内被同事了二十几次。有人问它能不能写周报有人让它解释一段线上日志里的报错还有人直接扔了个需求文档链接过来。那一刻我才觉得这个通义千问对接飞书机器人的项目算真正做完了。这篇是系列的第二篇2-2上一篇已经把账号准备、飞书应用的创建、通义千问API Key申请这些地基工作讲完了。这篇我直接进入正题飞书机器人和通义千问之间的完整对接方案包括架构选型、关键代码实现、上下文管理、流式体验处理以及最后部署上线时踩过的坑。核心就一个目标——让飞书里的机器人能像一个人一样把通义千问的能力用起来。1. 整体架构与接入模式选型先分清三种对接路子飞书机器人接AI模型网上的教程大多只给一种解法但实际你在选型时至少要面对三个岔路口机器人形态、消息链路、以及回复方式。这三个选择直接决定了后面代码怎么写、部署在哪里、能扛多大并发。1.1 自定义机器人还是企业自建应用飞书的机器人分两种自定义机器人和企业自建应用。自定义机器人像群里的一个webhook投递员它的核心能力是往群里发消息也能通过关键词或者触发被动的命令但它的交互能力很弱——没有事件订阅、没有用户身份识别、没有主动发消息的权限更没有读写云文档的授权能力。企业自建应用则是飞书开放平台里的正规军能启用机器人能力、订阅消息事件、发送主动消息、操作卡片交互、访问通讯录和云文档。如果你只想做一个群里发个指令、机器人回一段话的玩具自定义机器人也能跑通。但我的项目明确需要一个带上下文、能并发、后续还要接知识库和卡片的机器人所以我选了企业自建应用。这是我踩完两种方案之后最直接的建议回归到你的真实需求如果交互方案里包含判断用户身份多轮对话主动通知任何一个词不用犹豫直接走自建应用。1.2 消息链路的几种连接方式自建应用的消息链路本质上解决的是一个互联网通用问题飞书收到用户消息怎么把这条消息给你的后端服务飞书官方给了三种方式Webhook事件订阅飞书把用户消息以HTTP POST请求推送到你指定的公网URL上。消息是飞书主动推给你的。长连接WebSocket模式飞书开放平台提供了长连接的方式你的服务主动去飞书服务器建立一条长连接事件沿着这个连接推下来。轮询新版本已淘汰已经不推荐就不多说了。做这个项目的时候我一开始用了Webhook方式把服务部署在一台有公网IP的云服务器上用Nginx反代到Flask服务这套链路本身没问题但后来我发现一个巨大的坑在Webhook模式下飞书服务器要求你的服务在3秒内返回200响应超过则判定回调超时触发重试。而这个AI机器人调用通义千问的API快则一两秒慢则十几秒3秒根本扛不住。解决办法有两个一是收到事件后立即返回200把消息丢进队列异步处理再通过主动发消息接口把结果发回去二是我后来实际采用的方案——切换成长连接模式。长连接模式下飞书不再要求快速的HTTP响应事件是推送给我本地常驻的客户端我可以慢慢处理处理完了再调API把结果发出去。这样一来整个项目对公网依赖几乎为零部署在个人电脑或者内网服务器也能跑。这个选型带来的收益在后来的开发中多次体现出来我强烈建议你先看长连接模式的官方文档别急着配Webhook。1.3 回复方式被动回复还是主动消息飞书的事件回调里可以直接返回一个被动回复的响应。听起来方便但这个被动回复同样受3秒超时限制而且它有格式限制、不能发卡片。所以我在实际项目里全部采用主动消息方案收到消息事件后解析出会话IDchat_id然后调用im/v1/messages接口把结果主动推送到这个会话里。消息链路完整跑起来之后就一句话飞书事件长连接或Webhook→ 后端解析校验 → 调通义千问API → 组装消息 → 主动推送回会话。2. 飞书侧配置细节权限、事件订阅与密钥管理其实对接这个活三分之二的难度在飞书开放平台的控制台里而不是在代码里。配置错了代码写得再漂亮都白搭。我把配置过程中最容易出问题的点拆开说。2.1 创建应用并开通机器人能力在飞书开放平台后台创建企业自建应用名字我起的就是通义千问助手创建后进入应用详情页。第一步就是找到应用能力→机器人点击启用。这一步不启用后面所有事件订阅都无从谈起。然后是权限管理这一步很多人会漏。机器人要读取用户发给它的消息内容必须开通以下权限im:message读取消息im:message:send_as_bot以机器人身份发消息im:chat读取群信息用于判断群聊和单聊权限申请之后要创建版本并发布发布后管理员审核通过才算生效。我这里卡过一次我在控制台以为权限配好了但没走发布流程代码里一直报权限不足。2.2 事件订阅长连接配置与请求地址进入事件与回调页面订阅方式选使用长连接接收事件。然后添加事件这里必须勾选这两个im.message.receive_v1接收消息im.message.message_read_v1消息已读可选长连接的好处前面说过了但使用长连接有个前提官方推荐使用飞书开放平台的SDK来建立连接。我项目里用的是Python直接装lark-oapi这个SDK里封装了长连接客户端几行代码就能连上。2.3 加解密与Verification Token的用处飞书事件订阅里有一个Encrypt Key和Verification Token。加密密钥用于对推送过来的事件body做AES解密Verification Token用来做基础的事件来源校验。配置好之后飞书会提供一个验证challenge的流程在你保存配置的瞬间飞书会向你的服务发送一个带有challenge字段的请求长连接模式下这个验证发生在SDK内部你的服务需要原样返回这个字段。用SDK时这段逻辑已经封装好了但如果你手写Webhook需要特别处理否则连保存配置那一步都过不去。密钥管理上我的习惯是所有密钥存环境变量不写进代码仓库。包括APP_ID、APP_SECRET、VERIFICATION_TOKEN、ENCRYPT_KEY以及通义千问的DASHSCOPE_API_KEY。后来上CI/CD流水线时这套环境变量的设计帮我省了很大麻烦——仓库里一份配置模板真正密钥只在服务器的环境变量里。3. Python后端核心实现事件处理、加解密与调用通义千问配置和选型都定下来之后进入代码阶段。我用的是Python 3.10 FastAPI调用通义千问用的是阿里云百炼平台的dashscopeSDK。这部分我只贴核心代码并解释关键逻辑。3.1 长连接事件接收服务用lark-oapi的SDK长连接模式的起服务非常简单import lark_oapi as lark from lark_oapi.api.im.v1 import * def on_message(event: lark.MessageReceiveEvent) - None: # 处理消息事件 pass # 创建client client lark.Client.builder() \ .app_id(os.environ[APP_ID]) \ .app_secret(os.environ[APP_SECRET]) \ .log_level(lark.LogLevel.INFO) \ .build() # 注册事件处理器 client.event.handler.register(lark.EventType.MESSAGE_RECEIVE, on_message) # 建立长连接 client.ws.start()核心逻辑就三行注册事件处理器、注册回调函数、启动长连接。SDK会自动处理加密解密、自动重连、心跳保活省掉了一个大工程。这里有一个细节值得注意SDK的MessageReceiveEvent里已经帮你解好了消息体和发送者信息你拿到的event.message.content是JSON字符串里面是消息的具体内容。3.2 消息内容的解析与触发词判断消息进来后我做的第一件事不是调用AI而是先判断这条消息到底该不该回。我的触发规则是单聊消息必回群聊消息必须机器人本人mention里有机器人的ID。import json def should_reply(event: lark.MessageReceiveEvent) - bool: message event.message # 单聊 if message.chat_type p2p: return True # 群聊必须机器人 if message.chat_type group: mentions json.loads(message.mentions) if message.mentions else [] bot_id event.request_id # 简化示意实际用机器人的open_id判断 for m in mentions: if m.get(id, {}).get(open_id) bot_id: return True return False这样做不只是为了体验更是为了成本。通义千问每次调用都消耗token如果群里每句话都触发一天几千次调用账单会涨得让你肉疼。这也是我给所有做AI机器人的朋友的第一条建议触发策略一定要写得足够保守。消息内容解析飞书的消息content格式是JSON字符串比如文本消息长这样{text: 你好帮我写个Python脚本}解析出来就是原始文本。我把文本拿出来去掉机器人的那段_user_1占位符飞书在群聊文本里会把对象渲染成类似_user_1的形式得到用户真正想问的内容。3.3 调用通义千问非流式与流式两种方式调用通义千问我用的是dashscopeSDK。最基础的非流式调用import dashscope from dashscope import Generation dashscope.api_key os.environ[DASHSCOPE_API_KEY] def call_qwen(prompt: str, history: list None) - str: messages [] if history: messages.extend(history) messages.append({role: user, content: prompt}) response Generation.call( modelqwen-turbo, # 按需求选qwen-max/qwen-plus messagesmessages, result_formatmessage ) if response.status_code 200: return response.output.choices[0].message.content else: # 降级处理 return f调用失败{response.code} {response.message}但实际项目里我用的不是这种方式而是流式调用因为通义千问的完整回复可能需要几秒到十几秒非流式接口会让用户感觉卡死了。用流式我可以先把回复的前几个字推给用户后面边生成边推。不过这里有个飞书的现实问题飞书消息接口是整条发送的不支持真正的打字机流式推送。一次只能发出去一条完整消息想要打字机效果通常的手段是发一张更新卡片通过反复调用patch接口更新卡片内容。但那样做会消耗大量API调用次数和触发更复杂的消息状态管理。经过实测我最后采取的方案是折中收到消息后先立刻给用户发一条占位消息收到正在思考中...然后调用通义千问的非流式接口或者流式接口我取完整结果得到完整回复后把之前的占位消息编辑掉替换成真正的回复编辑消息用的是im/v1/messages/{message_id}的PATCH接口SDK里封装的MessageService直接能调。这样用户既能看到机器人已响应又不用等十几秒的沉默。3.4 发送消息回飞书回消息的代码from lark_oapi.api.im.v1 import * def send_text(chat_id: str, text: str): request CreateMessageRequest.builder() \ .receive_id_type(chat_id) \ .request_body(CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type(text) .content(json.dumps({text: text}, ensure_asciiFalse)) .build()) \ .build() response client.im.v1.message.create(request) return response就这么简单一个chat_id和一个消息体。但注意content必须是一个JSON字符串而且消息体里的文本不能超过飞书的限制。实测下来飞书文本消息的content长度上限大约在150KB但出于阅读体验考虑我会在代码里对回复内容做个截断逻辑超过一定长度就拆成多条或者提示用户内容太长已为你生成文件。4. 让机器人聊得下去上下文管理与用户状态处理通义千问本身是支持多轮对话的但API接口是无状态的——你不把历史消息带过去它就是一次性问答。要让机器人有记忆就得自己在服务端维护上下文。4.1 会话维度的选择我最初很天真地用chat_id也就是群维度作为会话维度。但很快就发现问题一个20人的群里A让机器人写SQLB让机器人讲笑话如果共享同一份上下文模型会把两个人的话题混在一起对话逻辑一塌糊涂。正确做法是单聊场景按用户维度记忆群聊场景按群ID 用户ID组合记忆。也就是说在群里每个人有自己独立的对话历史互不干扰。4.2 上下文存储先从内存方案起步存储这块我第一阶段用的是最简单的方案内存字典。from collections import defaultdict, deque from datetime import datetime, timedelta # session_id - deque of {role, content} context_store defaultdict(lambda: deque(maxlen20)) # session_id - last_active_time active_time_store {} def get_session_id(event): if event.message.chat_type p2p: return fp2p:{event.message.chat_id} else: sender_id event.message.sender.sender_id.open_id return fgroup:{event.message.chat_id}:{sender_id}内存方案的好处是零依赖本机就能跑通。坏处很明显重启服务上下文全丢多实例部署时上下文不同步。但作为一个MVP阶段的机器人我完全接受这种折中。我同时给每条上下文带上了时间戳超过TIMEOUT_MINUTES30的会话会被清理掉避免内存无限增长。4.3 上下文的Token裁剪策略通义千问有上下文长度限制不同模型不同qwen-turbo是32Kqwen-max是32K新版本支持更长。但不可能把所有历史消息都塞给模型。我的策略是只保留最近10轮用户助手各算一轮的对话每轮消息按500字截断超长的直接裁掉尾部如果当前会话内容总长度超过模型token限制的70%就丢弃最久远的历史消息只保留当前这条问题这个裁剪策略在原项目里被验证非常管用。特别是群聊中粘贴代码块时一条消息就可能占几千个token不做裁剪后续对话直接报上下文超限。4.4 超时、并发与降级我之前用的通义千问接口默认的并发限制是几十个QPS个人项目完全够用。但要注意一个隐藏问题如果多个用户同时机器人每个请求都要排队调通义千问通义千问在段时间内大量请求时会有随机超时。我加了一个简单的信号量来控制并发同时给调用设置了超时时间我设的是30秒。一旦超时就返回友好的降级文案而不是让用户一直等。import asyncio from aiostream import stream semaphore asyncio.Semaphore(5) async def safe_call_qwen(prompt, history): async with semaphore: try: return await asyncio.wait_for( call_qwen_async(prompt, history), timeout30 ) except asyncio.TimeoutError: return 抱歉模型响应超时了请稍后再试。5. 部署上线与持续集成我是怎么把服务跑稳的代码写完本地跑通这只是完成了40%。一个机器人能不能长期稳定服务部署架构和运维策略占了剩下的60%。这个项目里我踩的坑主要集中在部署这块。5.1 进程守护与重启策略长连接模式的客户端进程最怕的就是SDK底层的连接断开后没有自动重连。实际上lark-oapi的SDK已经有重连机制但保险起见我还是用systemd做了一个守护进程万一整个进程挂掉自动拉起来。[Unit] DescriptionQwen Feishu Bot Afternetwork.target [Service] Userubuntu WorkingDirectory/opt/qwen-bot EnvironmentFile/etc/qwen-bot.env ExecStart/opt/qwen-bot/venv/bin/python main.py Restartalways RestartSec5 [Install] WantedBymulti-user.target有几个点值得注意EnvironmentFile指定了外部环境变量文件这样APP_ID、APP_SECRET这些密钥不落地到仓库也不出现在systemd启动命令里。Restartalways意味着进程异常退出时系统5秒后自动拉起。实测SDK偶发断线时这一招能让机器人秒恢复。WorkingDirectory要写对否则相对路径导入模块会挂。5.2 持续集成用GitHub Actions自动部署项目上了GitHub仓库之后我配了一个简单的CI流水线。每次推送到main分支触发测试、构建和部署name: Deploy Qwen Bot on: push: branches: [main] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - name: Set up Python uses: actions/setup-pythonv5 with: python-version: 3.10 - name: Install dependencies run: pip install -r requirements.txt - name: Run tests run: pytest - name: Deploy via SSH uses: appleboy/ssh-actionv1.0.3 with: host: ${{ secrets.SERVER_HOST }} username: ${{ secrets.SERVER_USER }} key: ${{ secrets.SSH_PRIVATE_KEY }} script: | cd /opt/qwen-bot git pull origin main /opt/qwen-bot/venv/bin/pip install -r requirements.txt sudo systemctl restart qwen-bot这个流水线做得很朴素但已经能把本地改代码→测试→服务器拉取→重启服务这一整套串起来了省去的重复劳动非常可观。对个人项目来说CI/CD不一定要用K8s或者Docker先用systemd SSH跑通最小闭环价值就已经很大了。5.3 日志与监控AI机器人比普通机器人更依赖日志因为模型输出的不确定性导致很多问题只有跑起来才能看到。我的日志策略是每个请求打一条INFO日志时间、会话ID、用户输入摘要、模型回复长度、耗时模型调用失败打ERROR日志包含完整入参和出参用tail -f /var/log/qwen-bot.log实时观察另外我在飞书里建了一个告警群通过自定义机器人把ERROR日志同步推送过去。这样服务出问题时我能第一时间知道不用等用户抱怨。6. 进阶玩法从文本回复到表格卡片与知识库文本回复只是基础真正让这个机器人产生更大价值的是把通义千问的输出变成飞书原生的结构化内容。6.1 让机器人发表格飞书机器人可以发送富文本消息和消息卡片卡片里支持表格样式。我用消息卡片做了一次升级让通义千问按固定JSON格式输出然后我把JSON解析成卡片的表格组件发出去。比如用户问帮我整理上周发布的三个版本的功能对比我提示模型返回如下结构{ title: 版本功能对比, table: { header: [版本号, 发布日期, 核心功能, 风险], rows: [ [v1.2.0, 2025-01-06, 新增登录流程, 低], [v1.2.1, 2025-01-10, 修复支付回调, 中], [v1.3.0, 2025-01-15, 重构消息模块, 高] ] } }然后调用飞书消息卡片的interactive类型把表格数据渲染成卡片里的table组件飞书卡片支持lark_md和表格组件。这样同事在手机上的阅读体验就比纯文本好太多了。这里顺便说一个我的经验让模型输出结构化JSONPrompt里一定要给极端示例作为示范。比如明确告诉模型如果某个单元格没有数据输出无而不是null如果列表为空返回N/A。否则模型偶尔输出不符合预期的JSON解析就直接报错。6.2 交互卡片让回复可操作卡片不止能展示还能接收点击事件。我给机器人加了一个重新生成按钮卡片下方放一个按钮用户点击后触发一个新事件后端收到这个事件后重新调一次通义千问换个随机temperature把新结果更新到同一张卡片里。实现要点卡片里定义按钮的value字段比如{action: regenerate, session_id: xxx}飞书把按钮点击事件推送到后端事件类型是card.action.trigger在事件处理里根据value分发处理用PATCH接口更新原卡片这个玩法虽然简单但它是从工具到产品的一个分水岭。用户不再只是被动接收结果而是能和结果交互。6.3 私有知识库让模型懂你的业务最后聊一下知识库的方向。纯通义千问是通用大模型不懂你的业务细节。要让它懂目前个人项目里性价比最高的路子是RAG检索增强生成——把内部文档向量化存起来收到问题时先检索最相关的文档片段再和用户问题一起拼进Prompt发给模型。结合之前的热搜词ai知识库向量模型spring-ai集成rag我在这个项目里做过一个简化版用通义千问的embedding接口把文档切成chunk向量化存到本地向量数据库我用的是Chroma用户提问时先检索Top5相关段落和问题一起组装成Prompt调通义千问这个方向上要注意的点是chunk的切分策略按语义切还是按固定字符切对检索效果影响很大。我后来是把Markdown标题作为天然的分段标志来切效果比固定长度好很多。7. 避坑清单我从这次对接中总结的七个现实问题最后这部分是我这次开发中真正让人抓狂的问题集合。每一个都是真实踩过、翻阅了大量文档和源码才解决的写在这里希望能帮你省掉一部分排查时间。问题一事件订阅的challenge校验失败现象飞书后台保存事件订阅配置时提示URL验证失败。原因排查链路如果是Webhook模式检查服务端是否正确返回challenge字段如果是长连接模式检查SDK启动的日志里有没有handshake成功的记录。我遇到过的情况是Encrypt Key配置错误导致SDK解密失败——因为我在飞书后台复制Encrypt Key时把多余的空格也粘进去了。问题二消息事件重复推送我在本地日志里发现同一条消息被处理了两次。原因是飞书的事件推送有至少一次语义消息重试机制会重复投递。解决方案引入简单的幂等机制用消息IDmessage_id做去重。代码里维护一个最近处理过的message_id集合如果一条消息的ID已经处理过直接丢弃。问题三群聊里机器人被所有消息刷屏这个前面提过了触发策略写得不保守就会爆。我把群聊必须机器人才回复作为一个明确的断言写进了代码开头。另外注意即使你的触发条件是机器人飞书事件里仍然会推送群里所有消息你必须自己判断mention列表里有没有机器人自己。问题四长文本被截断通义千问的回答经常超过2000字飞书的文本消息接口对超长文本的渲染很不友好。我的处理方式是检测到消息长度超过阈值后改用发文方式把内容作为云文档内容创建出来调用飞书的云文档API然后在聊天里只发一条文档链接给用户。这个体验反而比纯文本更好用户可以点开稍后阅读。问题五通义千问的模型名写错qwen-max、qwen-plus、qwen-turbo这几个模型名看着都差不多但它们的能力、限流、价格都不一样。在某个时间段内部分模型的名称会有带日期后缀的版本比如qwen-max-2025-01-25。如果直接复制网上的代码用的模型名不对会报404模型不存在。建议去百炼平台控制台看一下当前可用的模型列表确定你实际开通的是哪个。问题六Session Key的键冲突我给会话ID设计的格式是p2p:{chat_id}和group:{chat_id}:{user_id}这个格式有个坑如果chat_id本身包含冒号拼接出来的key可能和另一个会话冲突。实际上飞书的open_id格式里包含下划线不包含冒号但为了安全我还是用了JSON序列化来组key彻底避免拼字符串的歧义。问题七权限范围与发布版本过期飞书的权限申请之后如果应用有新版发布某些旧的权限范围可能需要重新审核。我曾经遇到一个情况代码里调用的API要求im:chat权限但应用的发布版本里忘了带上这个权限导致生产环境报permission denied。排查半天才意识到是这个原因。现在这个机器人已经在部门里跑了大半个月了稳定性和体验基本达到了生产可用的程度。要说当初有没有走了弯路确实有——比如一开始选Webhook方案绕了一大圈最后切了长连接才真正舒坦。如果让我给后来者一个建议先把消息链路跑通再追求智能。机器人能收到消息、能发消息哪怕只是一个echo机器人整个对接的基础就已经搭好了。剩下的事情——上下文、卡片、知识库、CI/CD——都是在这个基础上按需叠加的。先解决通再追求智这个顺序能让你少走许多弯路。