1. 项目缘起为什么需要一个更懂上下文的机器人在飞书群里用机器人这事儿大家都不陌生。拉个机器人进来发个通知、查个数据或者触发个自动化流程确实方便。但用久了你肯定会遇到一个让人头疼的问题上下文丢失。想象一下这个场景你在一个技术讨论群里问机器人“上周三我们讨论的那个关于缓存穿透的解决方案文档放哪了” 或者你发了一张错误日志的截图然后问“这个错误一般是什么原因导致的” 对于市面上大多数基于简单关键词匹配或单轮问答的机器人来说这种问题基本等于“天书”。它们要么回复“我不明白你的意思”要么生硬地匹配到某个不相关的指令完全无法理解你这句话和之前群聊里讨论过的内容、分享过的文件之间的关联。这就是“上下文”的魔力也是人类对话如此高效的原因。我们说话是建立在共享的知识背景和连续的对话流之上的。而传统的机器人每一次提问对它来说都是全新的、孤立的。Bub的出现就是为了解决这个核心痛点。它不是另一个简单的“机器人框架”而是一个专为构建能理解长对话、多轮交互的智能体Agent而设计的开发平台。它的目标就是让机器人能像人一样记住并理解一段对话中前后文的关系。所以这个项目的核心价值就非常明确了利用 Bub 的能力为飞书群聊打造一个真正能理解上下文、能进行连续对话、能基于历史信息提供精准回答的智能助手。这不仅仅是“查文档”而是让机器人成为团队讨论的积极参与者能回溯对话、总结要点、关联知识甚至基于讨论内容主动提供建议。2. 核心设计Bub 飞书的架构拆解要实现一个“更懂上下文”的机器人我们不能只靠飞书机器人本身。飞书机器人开放平台提供了消息接收和发送的能力但它不负责也不擅长处理复杂的对话逻辑和上下文管理。我们需要一个“大脑”这就是 Bub 扮演的角色。整个系统的架构可以清晰地分为三层第一层飞书交互层。这是机器人的“五官”和“手脚”。它负责监听飞书群聊中的消息无论是机器人的指令、普通文本还是图片都捕获下来。然后它将这些原始消息进行初步处理比如提取纯文本、下载图片等封装成一个结构化的请求通过 HTTP 调用发送给下一层。同时它也负责接收来自“大脑”的回复并将其转换成飞书支持的消息格式文本、卡片、图片等发送回群里。第二层Bub 智能体层核心大脑。这是项目的核心。我们在 Bub 上创建一个Agent。这个 Agent 定义了机器人的“性格”和“能力”。我们需要为它配置几个关键部分系统提示词System Prompt 这是机器人的“宪法”。在这里我们要明确告诉 Agent“你是一个服务于某某技术团队的飞书群助手。你的核心能力是理解长对话上下文。当用户提问时你必须结合当前问题回顾这个群聊中最近的历史消息比如最近50条来综合理解用户的意图并提供精准回答。你的回答应该专业、简洁。”上下文管理Context Management Bub 的核心优势。我们需要设定上下文窗口的大小例如保留最近20轮对话或者最近8000个Token的文本。Bub 会自动维护这个对话历史并在每次交互时将相关的历史记录连同新问题一起提交给大语言模型。工具Tools与技能Skills 为了让机器人不止于“聊天”还能“做事”。我们可以为 Agent 装备各种工具。例如知识库查询工具 连接团队的 Confluence、Notion 或本地文档库当问题涉及内部知识时自动检索并引用。飞书多维表格工具 让机器人可以查询或更新团队的任务清单、Bug记录表等。代码解释工具 当用户粘贴一段代码时能分析其作用或潜在问题。网络搜索工具 对于未知的公开技术问题可以联网搜索最新信息。第三层大模型与数据层。Bub 本身不产生智能它是智能体的“调度中心”。它需要连接一个大语言模型如 GPT-4, Claude, 或国内的各种大模型 API作为思考引擎。同时如果需要上文提到的知识库、表格等工具还需要连接相应的数据源。整个数据流是这样的飞书群消息 - 飞书机器人服务接收- Bub Agent处理结合历史上下文和工具- 大语言模型思考推理- Bub Agent组织回复可能调用工具- 飞书机器人服务发送- 飞书群。注意这里的一个关键设计点是“状态保持”。飞书的机器人接口本质是无状态的 HTTP 回调。这意味着每次用户发送消息飞书服务器都会向我们预设的“回调地址”发送一个独立的 HTTP 请求。这个请求本身不包含历史对话。因此维护对话上下文即“状态”的责任完全落在了我们自己的服务端也就是 Bub Agent 这一层。Bub 通过为每个“对话会话”通常可以按“飞书群ID 用户ID”组合来唯一标识独立维护上下文窗口完美解决了这个问题。3. 实操搭建从零到一的详细步骤理论讲完我们开始动手。假设你已经有了飞书开发者账号和 Bub 的访问权限。3.1 第一步在飞书开放平台创建机器人登录与创建应用 访问飞书开放平台进入开发者后台。点击“创建企业自建应用”选择“机器人”应用类型填写应用名称和描述。获取凭证 创建成功后在“凭证与基础信息”页面你会得到至关重要的App ID和App Secret。请妥善保存。配置权限 在“权限管理”页面为你的机器人添加必要的权限。对于基础的消息接收和发送你需要im:message接收与发送单聊、群组消息im:message.group_at_msg接收群聊中机器人的消息如果你需要读取群信息还需im:chat获取群组信息等。配置事件订阅 这是让机器人“活”起来的关键。请求网址 URL 填写你即将部署的、用于接收飞书事件回调的服务地址。例如https://your-server.com/feishu/event。在本地开发阶段你需要使用内网穿透工具如 ngrok生成一个公网可访问的临时地址来填写这里。加密密钥 飞书会提供一个Encrypt Key用于验证回调请求的合法性务必保存。订阅事件 在“事件订阅”设置中添加你需要监听的事件。最核心的是“接收消息”事件im.message.receive_v1。发布与启用 将应用版本发布到“企业可用”状态然后在飞书客户端中找到你的应用并启用它。你可以把它拉入任何一个你有管理权限的群聊中。3.2 第二步搭建 Bub Agent 作为“大脑”创建 Bub Agent 登录 Bub 平台创建一个新的 Agent。配置模型与提示词模型选择 在 Agent 设置中连接你的大模型 API如 OpenAI, Anthropic 等。根据预算和需求选择例如gpt-4o-mini在成本、速度和效果上比较平衡。编写系统提示词 这是塑造机器人性格的关键。一个示例你是一个名为“TechPal”的技术助手服务于[你的团队名]的飞书群。你的核心职责是理解连续的对话上下文。用户的问题可能基于之前讨论过的代码、文档或话题。在回答时你必须主动关联最近的历史消息来提供精准、有用的回答。你擅长解释技术概念、总结讨论要点、查找知识库文档。你的语气是专业且乐于助人的。如果用户的问题需要最新信息而你的知识截止日期不够你可以声明这一点。对于不确定的事情不要编造答案。设置上下文与记忆 在 Bub 的 Agent 设置中找到上下文或记忆配置。建议将上下文窗口设置为“会话记忆”模式并设定一个合理的 Token 限制如 8000。这意味着 Bub 会为每个独立的对话会话由我们后端的会话ID决定维护一段对话历史。可选添加工具 如果你需要知识库检索等功能在 Bub 的“工具”或“技能”配置页面添加相应的工具。例如配置一个“文档检索”工具指向你的 Confluence 搜索 API。3.3 第三步编写中间服务粘合层这是整个项目代码量最集中的部分。我们需要一个服务可以用 Python Flask/FastAPI, Node.js Express 等编写它扮演两个角色飞书事件的接收器和Bub Agent 的调用客户端。项目结构概览feishu-bub-bot/ ├── app.py # 主应用入口FastAPI ├── config.py # 配置文件飞书凭证、Bub API Key等 ├── feishu_handler.py # 飞书消息解析与验证 ├── bub_client.py # Bub API 调用封装 ├── session_manager.py # 会话上下文管理 └── requirements.txt # Python依赖核心代码解析飞书请求验证与解析 (feishu_handler.py) 飞书发送的请求是经过加密和验证的。我们必须先验证请求是否合法。# feishu_handler.py import hashlib import base64 import json from typing import Dict, Any import time class FeishuVerifier: def __init__(self, encrypt_key: str): self.encrypt_key encrypt_key def verify_signature(self, timestamp: str, nonce: str, body: str, signature: str) - bool: 验证飞书回调签名 # 飞书签名算法将 timestamp、nonce、encrypt_key、请求体拼接后计算MD5 content f{timestamp}\n{nonce}\n{self.encrypt_key}\n{body}.encode(utf-8) expected_sign hashlib.md5(content).hexdigest() return expected_sign signature def decrypt_event(self, encrypted_data: str) - Dict[str, Any]: 解密飞书事件如果启用了加密 # 这里简化处理实际需按飞书文档进行 AES 解密 # 假设未启用加密或已处理直接解析 return json.loads(encrypted_data) if encrypted_data else {}会话管理 (session_manager.py) 这是实现“上下文理解”的核心逻辑。我们需要为每个独特的对话创建一个会话ID并用它来关联 Bub 的上下文。# session_manager.py class SessionManager: def __init__(self): # 使用内存字典存储生产环境应换为 Redis 等 self.sessions {} # key: session_id, value: 会话相关元数据 def get_session_id(self, event: Dict[str, Any]) - str: 根据飞书事件生成唯一的会话ID。 策略一个群聊为一个会话 (session_id chat_{chat_id}) 或者一个用户在一个群聊为一个会话 (session_id chat_{chat_id}_user_{user_id}) 根据你的需求选择。这里采用“一个群聊一个会话”让机器人能记住群内所有人的对话上下文。 event_body event.get(event, {}) chat_id event_body.get(message, {}).get(chat_id, ) if not chat_id: # 如果是私聊chat_id 可能是 open_id chat_id event_body.get(sender, {}).get(sender_id, {}).get(open_id, private) return fchat_{chat_id} def get_session_context(self, session_id: str) - Dict: 获取会话的上下文信息例如上次交互的Bub会话ID return self.sessions.get(session_id, {}) def update_session_context(self, session_id: str, bub_conversation_id: str): 更新会话的上下文信息存储Bub返回的会话ID self.sessions[session_id] {bub_conversation_id: bub_conversation_id}Bub 客户端 (bub_client.py) 封装与 Bub API 的交互。Bub 通常提供 RESTful API 或 SDK。# bub_client.py import requests import json class BubClient: def __init__(self, api_key: str, agent_id: str, base_url: str https://api.bub.ai): self.api_key api_key self.agent_id agent_id self.base_url base_url self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } def send_message(self, message: str, session_id: str, previous_conversation_id: str None) - Dict: 向 Bub Agent 发送消息。 session_id: 我们自定义的会话ID用于上下文分组。 previous_conversation_id: Bub 返回的上次对话ID用于延续对话。 url f{self.base_url}/v1/agents/{self.agent_id}/conversations if previous_conversation_id: # 如果存在之前的对话ID则向该对话追加消息 url f{self.base_url}/v1/conversations/{previous_conversation_id}/messages payload { message: message, session_id: session_id, # 告诉Bub这是哪个会话 stream: False # 设为True可支持流式响应这里用非流式简化 } response requests.post(url, headersself.headers, jsonpayload) response.raise_for_status() result response.json() # 从响应中提取 Bub 的 conversation_id 和回复内容 bub_conversation_id result.get(conversation_id) reply_text result.get(choices, [{}])[0].get(message, {}).get(content, ) return { bub_conversation_id: bub_conversation_id, reply_text: reply_text }主应用逻辑 (app.py) 将以上模块串联起来。# app.py from fastapi import FastAPI, Request, HTTPException import json from feishu_handler import FeishuVerifier from session_manager import SessionManager from bub_client import BubClient from config import settings app FastAPI() verifier FeishuVerifier(settings.FEISHU_ENCRYPT_KEY) session_mgr SessionManager() bub_client BubClient(settings.BUB_API_KEY, settings.BUB_AGENT_ID) app.post(/feishu/event) async def handle_feishu_event(request: Request): # 1. 获取并验证飞书请求 raw_body await request.body() body_str raw_body.decode(utf-8) body_dict json.loads(body_str) if body_str else {} # 飞书验证令牌URL参数和签名Header 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 verifier.verify_signature(timestamp, nonce, body_str, signature): raise HTTPException(status_code403, detailInvalid signature) # 2. 处理飞书挑战首次配置URL时需要 if body_dict.get(type) url_verification: return {challenge: body_dict.get(challenge)} # 3. 处理消息事件 if body_dict.get(event, {}).get(type) im.message.receive_v1: event body_dict.get(event) message_type event.get(message, {}).get(message_type) # 只处理文本消息图片等需要额外处理 if message_type ! text: return {msg: ok} # 提取纯文本内容飞书消息是JSON格式 text_content json.loads(event[message][content]).get(text, ).strip() # 判断是否是机器人或者直接对话根据机器人设置 # 这里简化处理假设只要了机器人就响应 if not (event.get(mentions) and any(mention[id] settings.BOT_OPEN_ID for mention in event[mentions])): # 如果不是机器人可以选择不响应或响应所有消息根据需求 return {msg: ok} # 4. 管理会话与上下文 session_id session_mgr.get_session_id(body_dict) session_context session_mgr.get_session_context(session_id) previous_bub_conv_id session_context.get(bub_conversation_id) # 5. 调用 Bub Agent bub_response bub_client.send_message( messagetext_content, session_idsession_id, previous_conversation_idprevious_bub_conv_id ) # 6. 更新会话上下文存储新的Bub对话ID session_mgr.update_session_context(session_id, bub_response[bub_conversation_id]) # 7. 调用飞书API将Bub的回复发回群聊 # 这里需要调用飞书发送消息API需要 access_token # 代码略需实现飞书API调用将 bub_response[reply_text] 发送回 event[message][chat_id] send_to_feishu(event[message][chat_id], bub_response[reply_text]) return {msg: ok} def send_to_feishu(chat_id: str, text: str): 调用飞书发送消息API需实现获取tenant_access_token的逻辑 # 伪代码 # access_token get_feishu_token() # requests.post(fhttps://open.feishu.cn/open-apis/im/v1/messages?receive_id_typechat_id, # headers{Authorization: fBearer {access_token}}, # json{receive_id: chat_id, msg_type: text, content: json.dumps({text: text})}) pass3.4 第四步部署与测试本地测试 使用ngrok或localtunnel将你的本地服务如运行在http://localhost:8000暴露到一个公网地址如https://abc123.ngrok.io。将这个地址配置到飞书开放平台的“请求网址”中。部署上线 本地测试无误后将代码部署到云服务器如阿里云ECS、腾讯云CVM或 Serverless 平台如 Vercel, Railway。推荐使用 Docker 容器化部署便于管理依赖和环境。配置环境变量 在部署环境中设置所有敏感信息FEISHU_APP_IDyour_app_id FEISHU_APP_SECRETyour_app_secret FEISHU_ENCRYPT_KEYyour_encrypt_key BUB_API_KEYyour_bub_api_key BUB_AGENT_IDyour_agent_id最终验证 在飞书群中你的机器人进行多轮对话测试。尝试问一些基于上文的问题比如第一轮 “我们项目的技术栈是什么”第二轮 “在机器人回答后那么前端主要用哪个框架” 观察机器人是否能正确理解“前端框架”是承接“技术栈”这个话题的。4. 进阶优化与避坑指南基础功能跑通后我们可以让它变得更强大、更稳定。4.1 功能增强点处理图片与文件 飞书消息可能包含图片、文件。你的feishu_handler需要能识别这些消息类型。对于图片可以调用飞书API下载图片然后通过 Bub 支持的多模态模型如 GPT-4V进行分析或者使用 OCR 提取图中文字。对于文件可以下载后提取文本内容如 PDF, Word再送入 Bub 处理。指令系统 除了自然语言对话可以设计一些快捷指令。例如/summary让机器人总结最近100条消息/find_doc about kubernetes让它在知识库中搜索相关文档。可以在消息处理逻辑中优先匹配这些指令触发特定的工具调用。长期记忆与向量检索 Bub 的上下文窗口有限如 8000 Token。对于更早的、重要的讨论可以引入向量数据库如 Pinecone, Weaviate。每次有价值的对话结束后自动将对话摘要向量化存储。当用户提问时先到向量库中进行语义搜索将相关历史作为“背景信息”插入到本次提问中实现“长期记忆”。多技能编排 利用 Bub 的 Agent 编排能力让机器人根据问题自动判断并调用不同的技能Skill。例如用户问“今天的天气怎么样”触发网络搜索技能问“帮我更新一下Bug状态”触发飞书多维表格更新技能。4.2 常见问题与排查机器人不响应检查点1事件订阅URL。确保你的回调地址是公网可访问的且飞书后台配置的URL末尾没有多余空格或斜杠。使用curl或 Postman 手动向你的 URL 发送一个测试请求看服务是否正常响应。检查点2权限与发布。确认机器人应用已添加im:message等必要权限并且版本已经发布到“企业可用”状态。在飞书客户端检查机器人是否已被添加到测试群。检查点3签名验证。这是最容易出错的地方。仔细核对timestamp,nonce,encrypt_key, 请求体四部分拼接和MD5计算的代码确保与飞书文档完全一致。可以在日志中打印出计算出的签名和收到的签名进行对比。上下文没有保持每次回答都像新对话检查点1session_id生成逻辑。确保你的get_session_id函数为同一场景如同一个群聊生成了相同的 ID。如果每次生成的 ID 都不同Bub 会认为是全新的会话。检查点2previous_conversation_id的传递。确保在调用bub_client.send_message时正确传入了从session_manager中取出的上一次的bub_conversation_id。并且在收到 Bub 响应后及时用新的conversation_id更新会话上下文。检查点3Bub Agent 配置。登录 Bub 平台检查你的 Agent 设置确保“上下文”或“记忆”模式是开启的并且窗口大小设置合理。响应速度慢优化点1异步处理。飞书要求事件回调在5秒内返回否则会重试。对于复杂的处理如图片分析、知识库检索不要在回调函数中同步等待。应该立即返回{msg: ok}然后通过异步任务如 Celery, 或简单的后台线程去处理消息并调用飞书API发送回复。优化点2模型选择。如果对实时性要求高可以尝试使用速度更快的模型如gpt-3.5-turbo或claude-haiku或者在 Bub 中设置响应流式输出让用户能更快看到部分结果。优化点3缓存。对于一些常见问题如“公司地址是什么”可以在你的中间服务层做缓存避免每次都请求 Bub 和大模型。Bub API 调用失败检查点1API Key 和 Agent ID。确认在环境变量或配置文件中配置正确没有过期。检查点2网络与代理。如果你的服务器在国内调用国外的 Bub 或 OpenAI API 可能需要配置网络代理。检查点3请求格式与频率限制。仔细阅读 Bub 的 API 文档确认请求体格式正确。同时注意是否有频率限制必要时加入重试和退避机制。实操心得在开发过程中一定要做好日志记录。将飞书的原始事件、生成的session_id、发送给 Bub 的消息、Bub 的回复、以及发生的任何错误都详细记录下来。这将是你在排查问题时最宝贵的线索。可以使用像structlog或loguru这样的库将日志输出到控制台的同时也写入文件。5. 安全、成本与运维考量将这样一个智能机器人投入生产环境还需要考虑以下几个现实问题安全方面权限最小化 飞书机器人只申请它完成功能所必需的最小权限。例如如果不需要读取用户信息就不要申请contact:user权限。内容审核 大语言模型可能生成不受控的内容。可以在将 Bub 的回复发送到飞书之前加入一层内容安全过滤调用内容审核API或设置关键词黑名单。访问控制 确保你的中间服务回调地址有基本的身份验证或防火墙规则防止被恶意调用。数据隐私 明确告知用户对话数据会用于改善服务如果会并避免在提示词和对话中泄露敏感信息。考虑对流出到外部API如 OpenAI的数据进行脱敏处理。成本控制大模型 Token 消耗 上下文越长消耗的 Token 越多费用越高。需要合理设置上下文窗口大小。对于非常长的历史可以采用“摘要”策略定期将旧对话总结成一段摘要用摘要代替原始长文本放入上下文。工具调用成本 如果接入了需要付费的第三方工具如某些知识库API、搜索API需要监控其调用量和费用。基础设施成本 云服务器或 Serverless 服务的费用。如果用户量不大使用 Serverless 方案按调用次数计费可能比长期运行一台虚拟机更划算。运维监控健康检查 为你的服务设置健康检查端点并配置监控告警如 Uptime Robot。指标监控 监控关键指标消息处理延迟、Bub API 调用错误率、各飞书群的活跃度等。可以使用 Prometheus Grafana 搭建简单的监控面板。错误预警 将应用错误日志接入到告警系统如 Sentry, 钉钉/飞书群机器人告警以便及时响应故障。搭建这样一个机器人初期看起来步骤不少但每一步拆解开来都是清晰的。它的价值在于将一个被动的、工具化的机器人转变为一个主动的、拥有“记忆”和“理解力”的团队协作者。当你的团队成员习惯了向它追问“我们刚才说的那个方案……”并且能得到连贯准确的回答时这个项目的意义就真正体现出来了。