1. 项目缘起：为什么需要一个更懂上下文的机器人？

在飞书群里用机器人，这事儿大家都不陌生。拉个机器人进来，发个通知、查个数据，或者触发个自动化流程，确实方便。但用久了，你肯定会遇到一个让人头疼的问题：上下文丢失。

想象一下这个场景：你在一个技术讨论群里，问机器人：“上周三我们讨论的那个关于缓存穿透的解决方案文档放哪了？” 或者，你发了一张错误日志的截图，然后问：“这个错误一般是什么原因导致的？” 对于市面上大多数基于简单关键词匹配或单轮问答的机器人来说，这种问题基本等于“天书”。它们要么回复“我不明白你的意思”，要么生硬地匹配到某个不相关的指令，完全无法理解你这句话和之前群聊里讨论过的内容、分享过的文件之间的关联。

这就是“上下文”的魔力，也是人类对话如此高效的原因。我们说话是建立在共享的知识背景和连续的对话流之上的。而传统的机器人，每一次提问对它来说都是全新的、孤立的。Bub的出现，就是为了解决这个核心痛点。它不是另一个简单的“机器人框架”，而是一个专为构建能理解长对话、多轮交互的智能体（Agent）而设计的开发平台。它的目标就是让机器人能像人一样，记住并理解一段对话中前后文的关系。

所以，这个项目的核心价值就非常明确了：利用 Bub 的能力，为飞书群聊打造一个真正能理解上下文、能进行连续对话、能基于历史信息提供精准回答的智能助手。这不仅仅是“查文档”，而是让机器人成为团队讨论的积极参与者，能回溯对话、总结要点、关联知识，甚至基于讨论内容主动提供建议。

2. 核心设计：Bub + 飞书的架构拆解

要实现一个“更懂上下文”的机器人，我们不能只靠飞书机器人本身。飞书机器人开放平台提供了消息接收和发送的能力，但它不负责，也不擅长处理复杂的对话逻辑和上下文管理。我们需要一个“大脑”，这就是 Bub 扮演的角色。

整个系统的架构可以清晰地分为三层：

第一层：飞书交互层。这是机器人的“五官”和“手脚”。它负责监听飞书群聊中的消息，无论是@机器人的指令、普通文本还是图片，都捕获下来。然后，它将这些原始消息进行初步处理（比如提取纯文本、下载图片等），封装成一个结构化的请求，通过 HTTP 调用发送给下一层。同时，它也负责接收来自“大脑”的回复，并将其转换成飞书支持的消息格式（文本、卡片、图片等）发送回群里。

第二层：Bub 智能体层（核心大脑）。这是项目的核心。我们在 Bub 上创建一个Agent。这个 Agent 定义了机器人的“性格”和“能力”。我们需要为它配置几个关键部分：

1. 系统提示词（System Prompt）： 这是机器人的“宪法”。在这里，我们要明确告诉 Agent：“你是一个服务于某某技术团队的飞书群助手。你的核心能力是理解长对话上下文。当用户提问时，你必须结合当前问题，回顾这个群聊中最近的历史消息（比如最近50条），来综合理解用户的意图，并提供精准回答。你的回答应该专业、简洁。”
2. 上下文管理（Context Management）： Bub 的核心优势。我们需要设定上下文窗口的大小（例如，保留最近20轮对话，或者最近8000个Token的文本）。Bub 会自动维护这个对话历史，并在每次交互时，将相关的历史记录连同新问题一起提交给大语言模型。
3. 工具（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 第一步：在飞书开放平台创建机器人

1. 登录与创建应用： 访问飞书开放平台，进入开发者后台。点击“创建企业自建应用”，选择“机器人”应用类型，填写应用名称和描述。
2. 获取凭证： 创建成功后，在“凭证与基础信息”页面，你会得到至关重要的App ID和App Secret。请妥善保存。
3. 配置权限： 在“权限管理”页面，为你的机器人添加必要的权限。对于基础的消息接收和发送，你需要：
   • im:message（接收与发送单聊、群组消息）
   • im:message.group_at_msg（接收群聊中@机器人的消息）
   • 如果你需要读取群信息，还需im:chat（获取群组信息）等。
4. 配置事件订阅： 这是让机器人“活”起来的关键。
   • 请求网址 URL： 填写你即将部署的、用于接收飞书事件回调的服务地址。例如https://your-server.com/feishu/event。在本地开发阶段，你需要使用内网穿透工具（如 ngrok）生成一个公网可访问的临时地址来填写这里。
   • 加密密钥： 飞书会提供一个Encrypt Key，用于验证回调请求的合法性，务必保存。
   • 订阅事件： 在“事件订阅”设置中，添加你需要监听的事件。最核心的是“接收消息”事件（im.message.receive_v1）。
5. 发布与启用： 将应用版本发布到“企业可用”状态，然后在飞书客户端中，找到你的应用并启用它。你可以把它拉入任何一个你有管理权限的群聊中。

3.2 第二步：搭建 Bub Agent 作为“大脑”

1. 创建 Bub Agent： 登录 Bub 平台，创建一个新的 Agent。
2. 配置模型与提示词：
   • 模型选择： 在 Agent 设置中，连接你的大模型 API（如 OpenAI, Anthropic 等）。根据预算和需求选择，例如gpt-4o-mini在成本、速度和效果上比较平衡。
   • 编写系统提示词： 这是塑造机器人性格的关键。一个示例：

     你是一个名为“TechPal”的技术助手，服务于[你的团队名]的飞书群。你的核心职责是理解连续的对话上下文。用户的问题可能基于之前讨论过的代码、文档或话题。在回答时，你必须主动关联最近的历史消息来提供精准、有用的回答。你擅长解释技术概念、总结讨论要点、查找知识库文档。你的语气是专业且乐于助人的。如果用户的问题需要最新信息而你的知识截止日期不够，你可以声明这一点。对于不确定的事情，不要编造答案。

3. 设置上下文与记忆： 在 Bub 的 Agent 设置中，找到上下文或记忆配置。建议将上下文窗口设置为“会话记忆”模式，并设定一个合理的 Token 限制（如 8000）。这意味着 Bub 会为每个独立的对话会话（由我们后端的会话ID决定）维护一段对话历史。
4. （可选）添加工具： 如果你需要知识库检索等功能，在 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依赖

核心代码解析：

1. 飞书请求验证与解析 (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 {}

2. 会话管理 (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 f"chat_{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}

3. 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': f'Bearer {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, headers=self.headers, json=payload) 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 }

4. 主应用逻辑 (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_code=403, detail="Invalid 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( message=text_content, session_id=session_id, previous_conversation_id=previous_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(f'https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id', # headers={'Authorization': f'Bearer {access_token}'}, # json={"receive_id": chat_id, "msg_type": "text", "content": json.dumps({"text": text})}) pass

3.4 第四步：部署与测试

1. 本地测试： 使用ngrok或localtunnel将你的本地服务（如运行在http://localhost:8000）暴露到一个公网地址（如https://abc123.ngrok.io）。将这个地址配置到飞书开放平台的“请求网址”中。
2. 部署上线： 本地测试无误后，将代码部署到云服务器（如阿里云ECS、腾讯云CVM）或 Serverless 平台（如 Vercel, Railway）。推荐使用 Docker 容器化部署，便于管理依赖和环境。
3. 配置环境变量： 在部署环境中设置所有敏感信息：

   FEISHU_APP_ID=your_app_id FEISHU_APP_SECRET=your_app_secret FEISHU_ENCRYPT_KEY=your_encrypt_key BUB_API_KEY=your_bub_api_key BUB_AGENT_ID=your_agent_id

4. 最终验证： 在飞书群中@你的机器人，进行多轮对话测试。尝试问一些基于上文的问题，比如：
   • 第一轮： “我们项目的技术栈是什么？”
   • 第二轮： “（在机器人回答后）那么前端主要用哪个框架？” 观察机器人是否能正确理解“前端框架”是承接“技术栈”这个话题的。

4. 进阶优化与避坑指南

基础功能跑通后，我们可以让它变得更强大、更稳定。

4.1 功能增强点

1. 处理图片与文件： 飞书消息可能包含图片、文件。你的feishu_handler需要能识别这些消息类型。对于图片，可以调用飞书API下载图片，然后通过 Bub 支持的多模态模型（如 GPT-4V）进行分析，或者使用 OCR 提取图中文字。对于文件，可以下载后提取文本内容（如 PDF, Word）再送入 Bub 处理。
2. 指令系统： 除了自然语言对话，可以设计一些快捷指令。例如，/summary让机器人总结最近100条消息；/find_doc about kubernetes让它在知识库中搜索相关文档。可以在消息处理逻辑中优先匹配这些指令，触发特定的工具调用。
3. 长期记忆与向量检索： Bub 的上下文窗口有限（如 8000 Token）。对于更早的、重要的讨论，可以引入向量数据库（如 Pinecone, Weaviate）。每次有价值的对话结束后，自动将对话摘要向量化存储。当用户提问时，先到向量库中进行语义搜索，将相关历史作为“背景信息”插入到本次提问中，实现“长期记忆”。
4. 多技能编排： 利用 Bub 的 Agent 编排能力，让机器人根据问题自动判断并调用不同的技能（Skill）。例如，用户问“今天的天气怎么样？”触发网络搜索技能；问“帮我更新一下Bug状态”触发飞书多维表格更新技能。

4.2 常见问题与排查

1. 机器人不响应：

   • 检查点1：事件订阅URL。确保你的回调地址是公网可访问的，且飞书后台配置的URL末尾没有多余空格或斜杠。使用curl或 Postman 手动向你的 URL 发送一个测试请求，看服务是否正常响应。
   • 检查点2：权限与发布。确认机器人应用已添加im:message等必要权限，并且版本已经发布到“企业可用”状态。在飞书客户端检查机器人是否已被添加到测试群。
   • 检查点3：签名验证。这是最容易出错的地方。仔细核对timestamp,nonce,encrypt_key, 请求体四部分拼接和MD5计算的代码，确保与飞书文档完全一致。可以在日志中打印出计算出的签名和收到的签名进行对比。

2. 上下文没有保持（每次回答都像新对话）：

   • 检查点1：session_id生成逻辑。确保你的get_session_id函数为同一场景（如同一个群聊）生成了相同的 ID。如果每次生成的 ID 都不同，Bub 会认为是全新的会话。
   • 检查点2：previous_conversation_id的传递。确保在调用bub_client.send_message时，正确传入了从session_manager中取出的上一次的bub_conversation_id。并且，在收到 Bub 响应后，及时用新的conversation_id更新会话上下文。
   • 检查点3：Bub Agent 配置。登录 Bub 平台，检查你的 Agent 设置，确保“上下文”或“记忆”模式是开启的，并且窗口大小设置合理。

3. 响应速度慢：

   • 优化点1：异步处理。飞书要求事件回调在5秒内返回，否则会重试。对于复杂的处理（如图片分析、知识库检索），不要在回调函数中同步等待。应该立即返回{'msg': 'ok'}，然后通过异步任务（如 Celery, 或简单的后台线程）去处理消息并调用飞书API发送回复。
   • 优化点2：模型选择。如果对实时性要求高，可以尝试使用速度更快的模型（如gpt-3.5-turbo或claude-haiku），或者在 Bub 中设置响应流式输出，让用户能更快看到部分结果。
   • 优化点3：缓存。对于一些常见问题（如“公司地址是什么？”），可以在你的中间服务层做缓存，避免每次都请求 Bub 和大模型。

4. Bub API 调用失败：

   • 检查点1：API 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, 钉钉/飞书群机器人告警），以便及时响应故障。

搭建这样一个机器人，初期看起来步骤不少，但每一步拆解开来都是清晰的。它的价值在于，将一个被动的、工具化的机器人，转变为一个主动的、拥有“记忆”和“理解力”的团队协作者。当你的团队成员习惯了向它追问“我们刚才说的那个方案……”，并且能得到连贯准确的回答时，这个项目的意义就真正体现出来了。