上周末我把这个机器人从本地测试群推到部门大群之后，半小时内被同事@了二十几次。有人问它能不能写周报，有人让它解释一段线上日志里的报错，还有人直接扔了个需求文档链接过来。那一刻我才觉得，这个"通义千问对接飞书机器人"的项目算真正做完了。
这篇是系列的第二篇（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秒超时限制，而且它有格式限制、不能发卡片。所以我在实际项目里全部采用"主动消息"方案：收到消息事件后，解析出会话ID（chat_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( model="qwen-turbo", # 按需求选qwen-max/qwen-plus messages=messages, result_format="message" ) 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_ascii=False)) .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让机器人写SQL，B让机器人讲笑话，如果共享同一份上下文，模型会把两个人的话题混在一起，对话逻辑一塌糊涂。
正确做法是：单聊场景按用户维度记忆，群聊场景按（群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(maxlen=20)) # session_id -> last_active_time active_time_store = {} def get_session_id(event): if event.message.chat_type == "p2p": return f"p2p:{event.message.chat_id}" else: sender_id = event.message.sender.sender_id.open_id return f"group:{event.message.chat_id}:{sender_id}"内存方案的好处是零依赖，本机就能跑通。坏处很明显：重启服务上下文全丢，多实例部署时上下文不同步。但作为一个MVP阶段的机器人，我完全接受这种折中。我同时给每条上下文带上了时间戳，超过TIMEOUT_MINUTES=30的会话会被清理掉，避免内存无限增长。
4.3 上下文的Token裁剪策略
通义千问有上下文长度限制（不同模型不同，qwen-turbo是32K，qwen-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), timeout=30 ) except asyncio.TimeoutError: return "抱歉，模型响应超时了，请稍后再试。"5. 部署上线与持续集成：我是怎么把服务跑稳的
代码写完，本地跑通，这只是完成了40%。一个机器人能不能长期稳定服务，部署架构和运维策略占了剩下的60%。这个项目里我踩的坑主要集中在部署这块。
5.1 进程守护与重启策略
长连接模式的客户端进程，最怕的就是SDK底层的连接断开后没有自动重连。实际上lark-oapi的SDK已经有重连机制，但保险起见，我还是用systemd做了一个守护进程，万一整个进程挂掉，自动拉起来。
[Unit] Description=Qwen Feishu Bot After=network.target [Service] User=ubuntu WorkingDirectory=/opt/qwen-bot EnvironmentFile=/etc/qwen-bot.env ExecStart=/opt/qwen-bot/venv/bin/python main.py Restart=always RestartSec=5 [Install] WantedBy=multi-user.target有几个点值得注意：
EnvironmentFile指定了外部环境变量文件，这样APP_ID、APP_SECRET这些密钥不落地到仓库，也不出现在systemd启动命令里。Restart=always意味着进程异常退出时系统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/checkout@v4 - name: Set up Python uses: actions/setup-python@v5 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-action@v1.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和表格组件）。这样同事在手机上的阅读体验，就比纯文本好太多了。
这里顺便说一个我的经验：让模型输出结构化JSON，Prompt里一定要给"极端示例"作为示范。比如明确告诉模型："如果某个单元格没有数据，输出'无'而不是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时，把多余的空格也粘进去了。
问题二：消息事件重复推送
我在本地日志里发现同一条消息被处理了两次。原因是飞书的事件推送有"至少一次"语义，消息重试机制会重复投递。
解决方案：引入简单的幂等机制，用消息ID（message_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——都是在这个基础上按需叠加的。先解决"通"，再追求"智"，这个顺序能让你少走许多弯路。
