乐于分享
好东西不私藏
当前位置：夜雨聆风 > 技术教程 > 软件教程 > 飞书智能自动回复系统 – 技术文档

飞书智能自动回复系统 – 技术文档

当前时间： 2026-03-01 21:10:09 分类：软件教程 评论(0)

飞书智能自动回复系统 – 技术文档

1. 系统概述

1.1 项目背景

本系统是一个基于飞书开放平台的智能自动回复系统，通过集成本地大语言模型（LM Studio），实现飞书群聊消息的自动获取、智能回复和状态管理。

1.2 核心功能

  • 消息同步
    ：自动从飞书群聊获取历史消息
  • 数据存储
    ：将消息持久化到 SQLite 数据库
  • 智能回复
    ：调用本地大语言模型生成回复
  • 状态管理
    ：跟踪消息处理状态（未回复/正在回复/已回复）
  • 指令控制
    ：支持通过飞书消息控制回复格式

1.3 技术栈

  • 后端语言
    ：Python 3.x
  • 数据库
    ：SQLite3
  • API 框架
    ：Requests（HTTP 客户端）
  • 大模型
    ：LM Studio（本地部署）
  • 开放平台
    ：飞书开放平台

2. 系统架构

2.1 整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                        飞书智能自动回复系统                        │
├─────────────────────────────────────────────────────────────────┤
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐      │
│  │   飞书API    │◄──►│   核心业务    │◄──►│  本地模型    │      │
│  │   客户端     │    │   逻辑层      │    │   客户端     │      │
│  └──────────────┘    └──────────────┘    └──────────────┘      │
│           │                 │                    │              │
│           ▼                 ▼                    ▼              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                    SQLite 数据库                         │   │
│  │         (消息存储、状态管理、历史记录)                      │   │
│  └─────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

2.2 核心组件

2.2.1 FeishuAPI 类

职责：封装飞书开放平台 API 调用

核心方法：

  • get_access_token()
    ：获取应用访问令牌
  • get_chat_messages()
    ：分页获取群聊消息
  • send_message()
    ：发送消息到群聊

分页逻辑：

while has_more and page_count < max_pages:
# 请求当前页数据
    response = requests.get(url, params=params, headers=headers)
# 处理响应数据
    all_messages.extend(messages)
# 更新分页参数
    params["page_token"] = data.get("data", {}).get("page_token", "")
    page_count += 1

2.2.2 DatabaseManager 类

职责：SQLite 数据库操作和状态管理

数据表结构：

CREATE TABLE feishu_messages_full (
    id INTEGERPRIMARY KEY AUTOINCREMENT,
    message_id TEXT UNIQUENOT NULL,      -- 消息唯一标识
    root_id TEXT,                          -- 根消息ID（用于回复链）
    parent_id TEXT,                        -- 父消息ID
    chat_id TEXT,                          -- 群聊ID
    create_time INTEGER,                   -- 创建时间戳（毫秒）
    create_time_str TEXT,                  -- 创建时间（可读格式）
    update_time INTEGER,                   -- 更新时间戳
    update_time_str TEXT,                  -- 更新时间（可读格式）
    msg_type TEXT,                         -- 消息类型
    deleted INTEGERDEFAULT0,             -- 是否删除
    content TEXT,                          -- 解析后的消息内容
    body_content TEXT,                     -- 原始消息内容
    sender_id TEXT,                        -- 发送者ID
    sender_type TEXT,                      -- 发送者类型（user/app）
    sender_id_type TEXT,                   -- ID类型
    tenant_key TEXT,                       -- 租户标识
    mentions TEXT,                         -- @提及信息（JSON）
    reactions TEXT,                        -- 表情回复（JSON）
    is_replying INTEGERDEFAULT0,         -- 是否正在回复
    is_replied INTEGERDEFAULT0,          -- 是否已回复
    reply_content TEXT,                    -- 回复内容
    reply_time TIMESTAMP,                  -- 回复时间
    reply_message_id TEXT,                 -- 回复消息ID
    inserted_at TIMESTAMP,                 -- 插入时间
    updated_at TIMESTAMP-- 更新时间
);

状态流转：

未回复 (is_replying=0, is_replied=0)
    ↓
正在回复 (is_replying=1, is_replied=0)
    ↓
已回复 (is_replying=0, is_replied=1, reply_content=内容)

2.2.3 LocalModelClient 类

职责：本地大语言模型客户端

API 调用：

POST /v1/chat/completions
{
"model": "qwen/qwen3-8b",
"messages": [
        {"role": "system", "content": "系统提示"},
        {"role": "user", "content": "用户消息"}
    ],
"temperature": 0.7,
"max_tokens": 1000
}

重试机制：

max_retries = 3
for attempt inrange(max_retries):
    reply = self.model_client.chat_completion(messages)
if reply:
break
if attempt < max_retries - 1:
        time.sleep(2)  # 等待2秒后重试

3. 核心流程

3.1 主循环流程

┌─────────────┐
│   启动系统   │
└──────┬──────┘
       ▼
┌─────────────┐
│ 初始化配置   │
│ - 加载config │
│ - 初始化DB   │
└──────┬──────┘
       ▼
┌─────────────┐     ┌──────────────────┐
│ 同步飞书消息 │◄────│  每30秒循环一次   │
│ - 获取token  │     └──────────────────┘
│ - 分页获取   │              ▲
│ - 写入数据库 │              │
└──────┬──────┘              │
       ▼                     │
┌─────────────┐              │
│ 处理未回复   │              │
│ - 查询10分钟内│              │
│   未回复消息 │              │
│ - 排除机器人 │              │
└──────┬──────┘              │
       ▼                     │
┌─────────────┐              │
│ 生成智能回复 │              │
│ - 调用模型   │              │
│ - 重试3次   │              │
│ - 过滤标签   │              │
└──────┬──────┘              │
       ▼                     │
┌─────────────┐              │
│ 发送回复     │              │
│ - 发送到飞书 │              │
│ - 更新数据库 │──────────────┘
└─────────────┘

3.2 消息处理详细流程

3.2.1 消息获取与过滤

# 1. 分页获取消息（最多10页，每页50条）
messages = feishu_api.get_chat_messages(chat_id, page_size=50, max_pages=10)

# 2. 反转消息顺序（最新的在前）
messages.reverse()

# 3. 过滤条件
- 排除 sender_type = "app" 的消息（机器人消息）
- 检查 message_id 是否已存在于数据库

3.2.2 未回复消息查询

# 查询条件
SELECT * FROM feishu_messages_full 
WHERE (reply_content IS NULL OR reply_content = '')
    AND sender_type = 'user'# 只处理用户消息
    AND create_time >= ?                  # 10分钟内的消息
    AND is_replying = 0# 不在处理中
ORDER BY create_time ASC                  # 按时间先后顺序
LIMIT 5# 最多5条

3.2.3 智能回复生成

# 1. 构建对话历史
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": "用户: {content}"}
]

# 2. 调用模型（带重试）
for attempt inrange(3):
    reply = model_client.chat_completion(messages)
if reply:
break
    time.sleep(2)

# 3. 标签过滤（根据 show_think 标志）
ifnot show_think:
    reply = re.sub(r'<think>.*?</think>', '', reply, flags=re.DOTALL)

4. 配置说明

4.1 配置文件结构

{
"feishu":{
"app_id":"cli_xxxxx",// 飞书应用ID
"app_secret":"xxxxx",// 应用密钥
"chat_id":"oc_xxxxx"// 群聊ID
},
"model":{
"url":"http://127.0.0.1:1234",// LM Studio 服务地址
"api_key":"sk-local",// API密钥
"name":"qwen/qwen3-8b"// 模型名称
},
"chat":{
"poll_interval":30,// 轮询间隔（秒）
"max_history":10,// 最大历史消息数
"system_prompt":"...",// 系统提示词
"message_time_window":35// 消息时间窗口（分钟）
}
}

4.2 环境要求

  • Python 3.8+
  • SQLite3
  • LM Studio（本地运行）
  • 飞书应用（已开通消息读取和发送权限）

5. 指令系统

5.1 控制指令

| 指令 | 功能 | 权限 | |——|——|——| | /showthink | 显示 <think> 标签内容 | 所有用户 | | /hidethink | 隐藏 <think> 标签内容 | 所有用户 | | /thinkstatus | 查询当前标签显示状态 | 所有用户 |

5.2 指令处理逻辑

content_lower = content.lower().strip()

if content_lower == "/showthink":
    show_think = True
    send_message("✅ 已开启 <think> 标签显示")
elif content_lower == "/hidethink":
    show_think = False
    send_message("✅ 已关闭 <think> 标签显示")
elif content_lower == "/thinkstatus":
    status = "开启"if show_think else"关闭"
    send_message(f"📊 当前 <think> 标签显示状态: {status}")

5.3 默认行为

  • 默认状态
    ：/hidethink（隐藏思考标签）
  • 正常消息默认按照 /hidethink 方式处理

6. 错误处理

6.1 模型调用失败

try:
    reply = model_client.chat_completion(messages)
ifnot reply:
# 重试3次
for attempt inrange(3):
            reply = model_client.chat_completion(messages)
if reply:
break
            time.sleep(2)

ifnot reply:
# 发送失败提示
            send_message("抱歉，我暂时无法生成回复，请稍后再试 😔")
except Exception as e:
    logger.error(f"模型调用异常: {e}")

6.2 数据库锁定

retry_count = 0
while retry_count < max_retries:
try:
        conn = sqlite3.connect(db_name, timeout=10.0)
# 执行操作
        conn.commit()
returnTrue
except sqlite3.Error as e:
if"locked"instr(e).lower():
            retry_count += 1
            time.sleep(0.5 * retry_count)  # 指数退避
else:
raise

6.3 API 令牌失效

defget_access_token(self):
# 检查token是否有效（提前5分钟刷新）
ifself.access_token andself.token_expire_time:
if datetime.now().timestamp() < self.token_expire_time - 300:
returnself.access_token

# 重新获取token
    response = requests.post(url, json=payload)
# 更新token和过期时间

7. 数据库设计

7.1 索引设计

-- 主键索引
CREATE INDEX idx_message_id ON feishu_messages_full(message_id);

-- 查询优化索引
CREATE INDEX idx_sender_id ON feishu_messages_full(sender_id);
CREATE INDEX idx_create_time ON feishu_messages_full(create_time);
CREATE INDEX idx_is_replied ON feishu_messages_full(is_replied);
CREATE INDEX idx_is_replying ON feishu_messages_full(is_replying);

7.2 数据保留策略

  • 消息数据：永久保留
  • 回复内容：永久保留
  • 状态标记：实时更新

8. 性能优化

8.1 分页策略

  • 每页获取 50 条消息
  • 最多获取 10 页（500 条消息）
  • 页间延迟 0.5 秒，避免请求过快

8.2 数据库优化

  • 使用连接池（上下文管理器）
  • 批量插入替代单条插入
  • 合理使用索引加速查询

8.3 缓存策略

  • Token 缓存（提前 5 分钟刷新）
  • 消息去重检查（基于 message_id）

9. 安全考虑

9.1 数据脱敏

  • 配置文件中的敏感信息（app_id, app_secret）需要妥善保管
  • 日志中不输出完整的消息内容（仅输出前 100 字符）
  • 数据库文件设置适当权限

9.2 访问控制

  • 飞书应用需要配置 IP 白名单
  • API 调用频率限制
  • Token 定期刷新机制

10. 部署与运维

10.1 启动方式

python feishu_auto_reply.py

10.2 停止方式

  • 按 Ctrl+C 优雅停止
  • 系统会自动完成当前轮询后退出

10.3 日志监控

  • 日志文件：feishu_auto_reply.log
  • 日志级别：INFO
  • 关键指标：
    • 消息同步数量
    • 模型调用成功率
    • 回复发送成功率

10.4 健康检查

# 系统状态统计
{
"total_messages": 54,      # 总消息数
"unreplied": 34,           # 未回复数
"replying": 1,             # 正在回复数
"replied": 19# 已回复数
}

11. 扩展性设计

11.1 多群聊支持

可通过修改配置支持多个群聊：

CHAT_IDS = ["chat_id_1", "chat_id_2", "chat_id_3"]
for chat_id in CHAT_IDS:
    process_chat(chat_id)

11.2 多模型支持

可通过配置切换不同模型：

{
"model":{
"provider":"lmstudio",// 或 "openai", "anthropic"
"name":"qwen/qwen3-8b"
}
}

11.3 插件系统

预留扩展接口：

defprocess_with_plugin(message, plugin_name):
    plugin = load_plugin(plugin_name)
return plugin.process(message)

12. 附录

12.1 术语表

| 术语 | 说明 | |——|——| | message_id | 飞书消息唯一标识 | | sender_type | 发送者类型（user/app） | | is_replying | 消息是否正在处理中 | | is_replied | 消息是否已回复 | | reply_content | 模型生成的回复内容 | | show_think | 是否显示思考标签 |

12.2 相关文档

  • 飞书开放平台文档
  • LM Studio API 文档
  • SQLite 文档
AD：【资源若失效或者不可用，请留言评论或加老夜微信 al_tmen】>>点此前往<<
本站文章均为手工撰写未经允许谢绝转载：夜雨聆风 » 飞书智能自动回复系统 – 技术文档
xiaoqiang, wang

wang

猜你喜欢

评论 抢沙发

取消

3 + 8 =
  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
×
订阅图标按钮