1. 项目概述为什么我们需要一个“懂上下文”的机器人在飞书、钉钉这类协作工具的群聊里机器人已经不是什么新鲜玩意儿了。无论是自动推送天气、提醒待办还是根据关键词回复固定话术它们大多遵循一个简单的“刺激-反应”模式你输入一个特定的命令或关键词它给你一个预设好的答案。这种模式在处理简单、独立的查询时很高效但一旦对话变得稍微复杂一点比如需要结合之前的聊天记录来理解你的意图它就立刻“失忆”了。想象一个典型的场景你在项目群里问“上周三我们讨论的那个关于用户登录流程的优化方案最终结论是什么” 一个传统的机器人即使它接入了知识库也可能因为无法关联“上周三”、“讨论”、“用户登录流程”这几个分散在历史消息中的要素而无法给出准确回答。它缺乏的正是“上下文理解”能力。这就是“用 Bub 和飞书搭一个更懂群聊上下文的小机器人”这个项目的核心价值。它不是一个简单的应答机而是一个能记住并理解一段对话脉络的智能助手。Bub 在这里扮演了关键角色——它是一个轻量级、开发友好的框架能帮助我们快速构建具备上下文感知能力的对话代理Agent。结合飞书开放平台强大的消息接收与发送能力我们可以打造出一个真正能融入群聊讨论、提供连续、精准支持的伙伴。这个机器人适合谁如果你是团队的技术负责人希望提升群内信息检索和决策回溯的效率如果你是开发者对构建智能对话应用感兴趣想从简单的关键词匹配升级到更智能的上下文处理或者你只是一个被海量群消息淹没渴望有一个“记忆外挂”的普通用户那么这个项目都值得你深入了解。接下来我将拆解如何一步步实现它分享其中关键的技术选型、实现细节以及我踩过的坑。2. 核心架构与工具选型解析构建一个上下文感知的机器人核心在于设计一个能有效处理、存储和利用对话历史的系统。整个架构可以清晰地分为三层交互层、逻辑处理层和记忆与智能层。2.1 交互层为什么选择飞书飞书机器人作为交互入口是一个成熟且稳定的选择。其优势在于生态集成好与日历、文档、云表格深度打通未来扩展功能如根据会议纪要回答疑问非常方便。消息类型丰富支持文本、图片、富文本、卡片消息等交互形式灵活。权限与安全可控可以精细控制机器人能访问哪些群、能看到哪些消息类型符合企业安全要求。开发文档完善飞书开放平台提供了详尽的API文档和SDK事件订阅机制清晰能可靠地接收群聊消息。在飞书侧我们需要创建一个“自定义机器人”应用并配置两个关键能力获取群消息开通im:message权限中的接收消息事件。这允许机器人监听指定群内的所有消息需将机器人添加到群中。发送消息开通im:message权限中的发送消息、发送群消息等。这是机器人回复的基础。一个关键的配置点是事件订阅。你需要提供一个公网可访问的URL开发初期可以用内网穿透工具如ngrok或localtunnel快速暴露本地服务供飞书服务器在事件发生时推送过来。飞书会对这个URL进行有效性验证你需要正确处理它发来的challenge参数。2.2 逻辑处理层Bub 框架的核心作用Bub 是一个新兴的 Python 框架它简化了构建基于大语言模型LLM的智能体Agent的过程。在这个项目中Bub 的核心价值在于它提供了管理“对话上下文”的高级抽象。传统的做法可能是自己维护一个列表把用户和机器人的对话记录往里塞然后在每次提问时把最近N条记录连同问题一起扔给LLM。但Bub做得更多会话Session管理Bub 天然支持会话概念能自动将同一聊天窗口如飞书群内的多次交互关联到一个会话中。上下文窗口与摘要当对话历史很长超过了LLM单次处理的上下文长度限制例如GPT-4 Turbo的128K或更小模型的4K、8KBub 可以集成策略来自动处理。例如它可以将遥远的、不重要的历史消息进行摘要Summarization只保留关键结论从而在有限的上下文窗口内塞进更长的历史记忆。工具Tools调用Bub 让为Agent定义和调用工具如搜索知识库、查询数据库、执行计算变得非常简单。这对于扩展机器人能力至关重要。选择Bub而非直接调用OpenAI API或使用LangChain是看中了它的轻量和“开箱即用”特性。对于这样一个聚焦于上下文对话的机器人Bub提供了恰到好处的封装避免了过度设计。2.3 记忆与智能层上下文存储与LLM驱动这是机器人的“大脑”由两部分构成向量数据库Vector Database用于长期、大规模的“事实记忆”。例如你可以将公司的项目文档、产品手册、历史会议纪要等文本资料切片、编码成向量Embeddings后存入。当用户问题涉及这些知识时机器人可以快速检索出最相关的片段作为上下文提供给LLM。这对于回答基于固定知识的问题至关重要。常见的轻量级选择有ChromaDB、FAISS本地库或Qdrant可服务化部署。大语言模型LLM作为推理和生成的核心。它负责理解用户意图、结合对话历史和检索到的知识生成自然、准确的回复。你可以根据需求选择OpenAI的GPT系列、Anthropic的Claude或开源的本地模型如Qwen、ChatGLM等。考虑到飞书机器人的响应速度要求需要权衡模型的性能效果与延迟、成本。这三层如何协作用户在飞书群发送消息。飞书服务器将消息事件推送到我们部署的服务逻辑处理层。服务收到消息提取出文本、发送者、群ID等信息。根据群ID从存储中加载或创建对应的Bub会话Session。将本次用户消息添加到该会话的历史记录中。可选如果问题可能涉及外部知识则用问题文本作为查询条件去向量数据库中检索相关文档片段。将当前的对话历史可能经过摘要压缩和检索到的知识片段共同构造成一个清晰的提示词Prompt发送给LLM。LLM生成回复。服务将回复通过飞书API发送回原群聊。将本次交互的完整记录用户消息机器人回复保存到该会话的历史中完成一个循环。3. 详细实现步骤与核心代码拆解接下来我们进入实操环节。假设你已经有了Python环境并安装了pip。我们将一步步搭建这个机器人。3.1 第一步飞书机器人创建与配置进入飞书开放平台访问开发者后台创建一个新的“企业自建应用”。配置权限在“权限管理”中找到“消息与群组”权限开通im:message下的接收群消息、发送消息、发送群消息等必要权限。配置事件订阅在“事件订阅”页面填写请求网址URL。例如你本地服务运行在http://localhost:8000用ngrok暴露后得到https://abc123.ngrok.io那么请求网址就填https://abc123.ngrok.io/feishu/event。添加事件选择接收消息版本选2.0。飞书会向该URL发送一个带有challenge参数的GET请求进行验证。你的服务必须能正确解析并返回这个challenge值。发布与启用在“版本管理与发布”中创建一个版本并申请发布。审核通过或直接在企业内自用后在“凭证与基础信息”页面拿到App ID和App Secret用于后续API调用鉴权。添加机器人到群在飞书客户端进入目标群点击群设置-添加机器人-找到你刚创建的应用添加即可。3.2 第二步搭建后端服务核心FastAPI Bub我们使用 FastAPI 作为 web 框架因为它异步性能好编写简单。# 创建项目目录并安装核心依赖 mkdir feishu_context_bot cd feishu_context_bot python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install fastapi uvicorn httpx python-multipart pip install bub # 根据你选的LLM安装对应SDK例如OpenAI pip install openai # 如果要用向量数据库例如Chroma pip install chromadb首先我们创建处理飞书事件验证和消息的核心路由# main.py import hashlib import json import time from typing import Dict, Any, Optional import httpx from fastapi import FastAPI, Request, HTTPException, Header, Body from fastapi.responses import JSONResponse from pydantic import BaseModel, Field import asyncio # 配置信息应从环境变量读取 FEISHU_APP_ID your_app_id FEISHU_APP_SECRET your_app_secret FEISHU_VERIFICATION_TOKEN your_verification_token # 在事件订阅页面 ENCRYPT_KEY None # 如果启用了加密需要配置 app FastAPI(titleFeishu Context Bot) class FeishuEvent(BaseModel): schema_: str Field(aliasschema) header: Dict[str, Any] event: Dict[str, Any] app.get(/feishu/event) async def handle_feishu_verification(challenge: str, token: str, type: str): 处理飞书事件订阅的URL验证请求 if token ! FEISHU_VERIFICATION_TOKEN: raise HTTPException(status_code403, detailInvalid token) if type url_verification: return JSONResponse(content{challenge: challenge}) return {msg: ok} app.post(/feishu/event) async def handle_feishu_event( request: Request, x_feishu_request_timestamp: str Header(None), x_feishu_request_nonce: str Header(None), x_feishu_signature: str Header(None), ): 处理飞书推送过来的所有事件 # 1. 验证签名略生产环境必须做 # raw_body await request.body() # 计算签名并与 x_feishu_signature 比对 # 2. 解析事件体 body await request.json() event_wrapper FeishuEvent(**body) # 3. 判断事件类型并异步处理避免超时 if event_wrapper.header.event_type im.message.receive_v1: # 异步处理消息立即返回成功给飞书 asyncio.create_task(process_message(event_wrapper.event)) return {msg: ok} return {msg: event not handled} async def process_message(event: Dict[str, Any]): 异步处理接收到的消息 message_type event.get(message, {}).get(message_type) if message_type ! text: return # 本例只处理文本消息 chat_id event.get(message, {}).get(chat_id) user_id event.get(sender, {}).get(sender_id, {}).get(user_id) text_content json.loads(event.get(message, {}).get(content, {})).get(text, ) if not text_content: return # 这里调用我们的核心对话逻辑 reply_text await generate_reply_with_context(chat_id, user_id, text_content) # 调用飞书API发送回复 await send_feishu_message(chat_id, reply_text) async def send_feishu_message(chat_id: str, text: str): 调用飞书发送消息API # 1. 获取 tenant_access_token token_url https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal async with httpx.AsyncClient() as client: token_resp await client.post( token_url, json{app_id: FEISHU_APP_ID, app_secret: FEISHU_APP_SECRET} ) token_data token_resp.json() if token_data.get(code) ! 0: print(fFailed to get token: {token_data}) return access_token token_data[tenant_access_token] # 2. 发送消息 send_url https://open.feishu.cn/open-apis/im/v1/messages headers {Authorization: fBearer {access_token}, Content-Type: application/json} body { receive_id: chat_id, msg_type: text, content: json.dumps({text: text}, ensure_asciiFalse) } async with httpx.AsyncClient() as client: resp await client.post(send_url, headersheaders, jsonbody, params{receive_id_type: chat_id}) print(fSend message result: {resp.json()})3.3 第三步集成 Bub 管理上下文会话这是项目的核心。我们需要为每个飞书群chat_id维护一个独立的 Bub 会话。# context_manager.py from typing import Dict from bub import Agent, Session, ModelSettings from bub.memory import SummaryMemory # 一种可以自动摘要长历史的记忆类型 import openai import os # 设置LLM例如OpenAI openai.api_key os.getenv(OPENAI_API_KEY) class ChatSessionManager: def __init__(self): # 存储每个群聊的会话 self.sessions: Dict[str, Session] {} # 初始化一个通用的Agent self.agent Agent( modelgpt-4-turbo-preview, # 或 gpt-3.5-turbo model_settingsModelSettings(temperature0.7, max_tokens500), memorySummaryMemory(llmopenai, max_context_length4000), # 使用摘要记忆 system_prompt你是一个专业的飞书群助手负责回答群成员的问题。请根据对话历史和上下文提供准确、简洁、有帮助的回答。如果信息不足可以礼貌地询问更多细节。 ) async def get_or_create_session(self, chat_id: str) - Session: 获取或创建一个属于特定群聊的会话 if chat_id not in self.sessions: # 为每个群创建一个新的会话实例 # 注意这里我们复用同一个Agent但Session会隔离记忆 session self.agent.new_session(session_idchat_id) self.sessions[chat_id] session return self.sessions[chat_id] async def generate_reply(self, chat_id: str, user_input: str) - str: 基于上下文生成回复 session await self.get_or_create_session(chat_id) # 将用户输入交给Agent处理它会自动管理历史并调用LLM response await session.run(user_input) return response.content # 全局管理器实例 session_manager ChatSessionManager()然后在主逻辑中调用它# 在 main.py 中补充 from context_manager import session_manager async def generate_reply_with_context(chat_id: str, user_id: str, user_input: str) - str: 整合上下文管理生成回复 try: # 可以在用户输入前加上发送者信息让上下文更清晰 formatted_input f[用户 {user_id[:8]}...] 说{user_input} reply await session_manager.generate_reply(chat_id, formatted_input) return reply except Exception as e: print(fError generating reply: {e}) return 抱歉我暂时无法处理您的请求。3.4 第四步增强记忆——集成向量知识库为了让机器人能回答超出当前对话历史的问题例如公司制度、产品文档我们需要引入向量检索。# knowledge_base.py import chromadb from chromadb.config import Settings from openai import OpenAI import os from typing import List import hashlib client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) chroma_client chromadb.PersistentClient(path./chroma_db) # 获取或创建知识库集合 collection chroma_client.get_or_create_collection(namecompany_docs) def get_embedding(text: str) - List[float]: 调用OpenAI Embedding API获取文本向量 response client.embeddings.create( modeltext-embedding-3-small, inputtext ) return response.data[0].embedding def add_to_knowledge_base(docs: List[str], metadatas: List[dict] None): 将文档切片添加到知识库 ids [] embeddings [] for i, doc in enumerate(docs): # 生成一个稳定的ID例如基于内容哈希 doc_id hashlib.md5(doc.encode()).hexdigest()[:16] ids.append(doc_id) embeddings.append(get_embedding(doc)) collection.add( documentsdocs, embeddingsembeddings, idsids, metadatasmetadatas ) def query_knowledge_base(query: str, n_results: int 3) - List[str]: 查询知识库返回最相关的文档片段 query_embedding get_embedding(query) results collection.query( query_embeddings[query_embedding], n_resultsn_results ) if results and results[documents]: return results[documents][0] # 返回最相关的几个文档文本 return [] # 初始化时可以加载你的文档例如从Markdown、PDF解析出的文本块 # init_docs [文档片段1..., 文档片段2...] # add_to_knowledge_base(init_docs)现在修改我们的回复生成逻辑将知识库检索结果作为上下文的一部分# 更新 context_manager.py 中的 generate_reply 方法 from knowledge_base import query_knowledge_base async def generate_reply(self, chat_id: str, user_input: str) - str: session await self.get_or_create_session(chat_id) # 1. 检索相关知识 relevant_knowledge query_knowledge_base(user_input, n_results2) knowledge_context if relevant_knowledge: knowledge_context \n\n【相关参考信息】\n \n---\n.join(relevant_knowledge) # 2. 构建增强的提示词也可以通过在Agent中定义Tool来实现更优雅的集成 # 这里采用简单直接的方式将知识拼接到用户问题前 enhanced_input user_input if knowledge_context: enhanced_input f以下是一些可能相关的信息{knowledge_context}\n\n基于以上信息和我们的对话历史请回答{user_input} # 3. 交给Agent处理 response await session.run(enhanced_input) return response.content4. 部署、优化与避坑指南将代码跑起来只是第一步要让机器人稳定、智能地服务还需要考虑部署、性能优化和解决实际问题。4.1 服务部署与运行本地开发与测试使用uvicorn main:app --reload --host 0.0.0.0 --port 8000运行服务。同时使用ngrok http 8000获取公网地址配置到飞书事件订阅。生产环境部署服务器选择一台有公网IP的云服务器如阿里云ECS、腾讯云CVM。进程管理使用systemd或supervisord来管理uvicorn进程确保服务崩溃后能自动重启。反向代理使用Nginx作为反向代理处理SSL/TLS加密HTTPS这是飞书事件回调的强制要求。同时Nginx还可以做负载均衡和静态文件服务。环境变量将所有敏感信息APP_ID,APP_SECRET,OPENAI_API_KEY等通过环境变量或配置文件管理切勿硬编码在代码中。4.2 性能与成本优化策略上下文长度管理这是成本Token消耗和效果平衡的关键。SummaryMemory会自动将较早的对话压缩成摘要。你可以调整max_context_length参数控制保留多少token的原始历史。对于非关键闲聊可以设置更激进的摘要策略。LLM模型选型对响应速度要求高、问题相对简单的场景可以使用gpt-3.5-turbo。对回答质量要求高、逻辑复杂的场景使用gpt-4-turbo。也可以考虑部署开源模型如Qwen-7B-Chat虽然初期部署复杂但长期成本可控且数据隐私有保障。异步处理确保消息处理尤其是调用LLM和向量检索是异步的就像我们上面用asyncio.create_task做的避免阻塞飞书的回调请求飞书有超时限制。向量检索优化文档分块Chunking策略不要将整篇文档扔进去。使用滑动窗口或按语义段落进行分块例如用langchain的RecursiveCharacterTextSplitter块大小在256-512字左右效果较好。元数据过滤在存储文档时附加元数据如“部门技术部”、“文档类型API手册”。查询时可以结合元数据过滤缩小检索范围提升准确率和速度。缓存对常见问题或相同的查询结果可以加入缓存如redis短期内相同问题直接返回缓存答案减少LLM调用和向量检索次数。4.3 常见问题与排查技巧实录在实际开发和运营中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方案问题1飞书事件订阅URL验证失败表现在飞书后台保存事件订阅URL时提示“验证失败”。排查检查你的服务是否真的在运行且公网可访问。用curl或浏览器直接访问你的https://your-url.com/feishu/event?challenge123tokenxxxtypeurl_verification看是否能返回{challenge:123}。检查FEISHU_VERIFICATION_TOKEN是否填写正确。检查你的代码处理验证请求的逻辑是否正确是否是从GET请求的查询参数中获取challenge并以JSON格式返回。心得一定要先处理好这个验证这是后续所有消息接收的基础。本地开发务必用好ngrok注意每次重启ngrokURL都会变需要去飞书后台更新。问题2机器人收不到群消息表现机器人已在群里但用户机器人或发送消息后端服务没收到事件推送。排查检查权限确保应用已开通接收消息权限并且已经发布/申请了可用版本。检查事件订阅确保“接收消息”事件已成功添加到事件订阅列表并且状态是“已启用”。检查加密如果你在飞书后台开启了“事件加密”那么你的服务端必须实现解密逻辑使用ENCRYPT_KEY否则收到的是加密数据无法解析。建议开发初期先关闭加密。查看日志在飞书开放后台的“事件分析”或“日志”中可以看到事件推送的历史记录和状态码如果推送失败这里会有错误信息。问题3LLM回复慢或超时表现用户提问后很久才收到回复甚至飞书提示发送失败。解决方案异步化确保process_message函数是异步的并且消息处理尤其是调用LLM没有阻塞主线程。使用asyncio.create_task是标准做法。设置超时在调用OpenAI API或其它LLM API时使用httpx或aiohttp设置合理的超时时间如30秒并做好异常处理超时后给用户一个友好的提示。流式响应高级对于生成内容较长的回复可以考虑使用飞书的“消息卡片”或分条发送先发送一个“正在思考”的提示生成完再更新或发送完整内容提升用户体验。问题4上下文混乱或“失忆”表现机器人似乎忘记了刚才的对话或者把不同用户的对话混在了一起。排查与解决会话隔离确保你的session_manager是以chat_id群ID为键进行存储的。每个群应该有独立的会话历史。检查get_or_create_session逻辑。记忆持久化上面的示例中会话内存放在进程字典里服务重启就丢失了。生产环境需要将会话历史持久化到数据库如SQLite、PostgreSQL或Redis中。Bub的Memory类可以扩展将其load和save方法对接你的数据库。上下文过长检查LLM的上下文窗口限制。如果历史对话太长即使使用了SummaryMemory也可能触及上限。需要监控会话长度并在接近上限时进行更激进的摘要或清理老旧消息。问题5向量检索结果不相关表现机器人根据检索到的知识回答但答案明显答非所问。解决方案优化分块尝试不同的分块大小和重叠overlap参数。太大会引入无关信息太小会割裂语义。优化查询不要直接用原始用户问题去检索。可以尝试用LLM将用户问题重写Query Rewriting成更利于检索的关键词或陈述句。混合检索Hybrid Search除了向量相似度搜索可以结合关键词BM25搜索综合两者得分来排序结果这在某些场景下效果更好。一些向量数据库如Qdrant和Weaviate支持此功能。人工评估与迭代定期查看“检索-回答”的日志对于回答不好的案例分析是检索错了还是LLM理解错了针对性调整分块策略或提示词。5. 进阶扩展思路当基础版本稳定运行后你可以考虑以下方向进行扩展让机器人更强大工具调用能力利用Bub的Tool机制让机器人不仅能聊天还能“做事”。例如查询日历工具当用户问“我今天下午有什么会”机器人可以调用飞书日历API查询并返回结果。创建待办工具用户说“提醒我明天提交周报”机器人可以创建一个飞书待办。搜索知识库工具将我们之前实现的向量检索封装成一个标准的Tool让Agent自主决定何时调用。多模态支持飞书消息支持图片。可以集成视觉模型如GPT-4V让机器人能“看懂”群里的截图并回答问题例如“这张图表反映了什么趋势”工作流集成将机器人作为自动化工作流的触发器或交互节点。例如当群里有人说“我们需要申请一台测试服务器”机器人可以自动发起一个预填好的审批流程卡片。个性化与用户记忆目前会话记忆是基于群的。你可以进一步引入用户级记忆记住不同用户的偏好和过往交互提供更个性化的服务需注意隐私。评估与监控建立一套简单的评估体系随机采样机器人的回答进行人工或自动化的质量评估。监控Token消耗、API调用延迟、错误率等指标为优化提供数据支持。构建一个“更懂上下文”的机器人是一个持续迭代的过程。从最简单的关键词匹配到具备短期记忆再到融合长期知识库和外部工具每一步升级都能显著提升其在真实协作场景中的价值。希望这份详细的指南能帮助你顺利启动项目并避开我曾经遇到的那些陷阱。记住从最简单的版本开始快速让它在群里跑起来收集真实反馈再逐步完善这是最有效的开发路径。