# wwlbotdemo

**Repository Path**: hanphycai/wwlbotdemo

## Basic Information

- **Project Name**: wwlbotdemo
- **Description**: 私有化企业微信的智能机器人接口对接示例。
- **Primary Language**: Python
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 1
- **Created**: 2026-03-06
- **Last Updated**: 2026-04-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 企业微信智能机器人 Demo

基于企业微信智能机器人 API 和 DeepSeek 大模型，支持 Agent Skill 可扩展技能的智能机器人。

## 功能特性

- 🤖 **智能对话**：基于 DeepSeek 大模型的智能回复
- 🔌 **流式回复**：支持企业微信流式消息协议，实时展示回复内容
- 🛠️ **Skill 扩展**：支持通过 SKILL.md 文件动态扩展机器人能力
- 💾 **历史记录**：SQLite 持久化存储，支持多轮对话上下文
- 🔐 **安全加密**：完整的企业微信消息加解密支持
- 🌐 **双协议支持**：同时支持 HTTP 回调和 WebSocket 长连接两种协议模式

## 协议模式

本项目支持两种企业微信协议模式：

| 模式 | 说明 | 适用场景 |
|------|------|----------|
| **callback** (默认) | HTTP 回调协议，企业微信主动推送消息到服务端 | 需要公网可访问的服务器 |
| **websocket** | WebSocket 长连接协议，客户端主动连接企业微信服务器 | 无公网 IP 或内网环境 |

通过环境变量 `PROTOCOL_MODE` 切换协议模式。

## 架构概览

```
wwlbotdemo/
├── src/
│   ├── adapters/                  # 协议适配器
│   │   ├── base.py                # 适配器基类
│   │   └── websocket_adapter.py   # WebSocket 长连接适配器
│   ├── api/                       # API 路由层
│   │   └── router.py              # HTTP 路由，URL验证、消息回调
│   ├── core/                      # 核心配置
│   │   └── config.py              # 应用配置（环境变量）
│   ├── llm/                       # LLM 客户端
│   │   └── client.py              # DeepSeek API 封装
│   ├── models/                    # 数据模型
│   │   ├── database.py            # 数据库连接管理
│   │   └── message.py             # 消息数据模型
│   ├── services/                  # 业务服务
│   │   ├── message_handler.py     # 消息处理核心
│   │   ├── skill_loader.py        # Skill 加载服务
│   │   └── stream_manager.py      # 流式状态管理
│   ├── utils/                     # 工具模块
│   │   └── crypto.py              # 企业微信加解密
│   └── main.py                    # 应用入口
├── skills/                        # Skill 目录
├── tests/                         # 测试用例
├── system_prompt.md               # 系统提示词
├── requirements.txt               # Python 依赖
└── pytest.ini                     # 测试配置
```

## 系统架构图

```mermaid
flowchart TB
    subgraph 企业微信
        WX[企业微信后台]
    end
    
    subgraph 协议层
        API[HTTP Router<br/>callback 模式]
        WS[WebSocket Adapter<br/>websocket 模式]
    end
    
    subgraph 业务层
        MH[Message Handler]
        SS[Stream Manager]
        LLM[LLM Client]
        DB[(SQLite)]
        SK[Skill Loader]
        CR[Crypto Utils]
    end
    
    WX -->|POST 消息回调| API
    WX -->|GET URL验证| API
    WX -->|POST stream刷新| API
    WX <-->|WebSocket 长连接| WS
    
    API --> CR
    API --> MH
    API --> SS
    
    WS --> MH
    WS --> SS
    
    MH --> SK
    MH --> LLM
    MH --> DB
    
    SS --> DB
    
    SK -->|启动加载| SKILL[skills/*.md]
```

## 快速开始

### 环境要求

- Python 3.10+
- pip 或 uv 包管理器

### 安装依赖

```bash
# 使用 pip
pip install -r requirements.txt

# 或使用 uv
uv pip install -r requirements.txt
```

### 配置环境变量

创建 `.env` 文件或设置环境变量：

```bash
# ===============================================
# 协议模式配置
# ===============================================
# 协议模式：callback（HTTP 回调，默认）或 websocket（WebSocket 长连接）
PROTOCOL_MODE=callback

# ===============================================
# HTTP 回调模式配置（PROTOCOL_MODE=callback 时使用）
# ===============================================
# 企业微信配置（必填）
PLATFORM_API_TOKEN=your_token
PLATFORM_ENCODING_AES_KEY=your_aes_key
PLATFORM_CORP_ID=your_corp_id

# ===============================================
# WebSocket 模式配置（PROTOCOL_MODE=websocket 时使用）
# ===============================================
# 从企业微信后台「智能机器人」页面获取（必填）
WECHAT_BOT_ID=your_bot_id
WECHAT_BOT_SECRET=your_bot_secret
# WebSocket 服务地址（可选，SDK 有默认值）
# WS_URL=wss://bot.weixin.qq.com/api/aibot

# ===============================================
# 通用配置
# ===============================================
# DeepSeek LLM 配置（必填）
LLM_API_KEY=your_deepseek_api_key
LLM_API_BASE=https://api.deepseek.com/v1
LLM_MODEL=deepseek-chat

# 应用配置（可选）
APP_NAME=智能小秘书
HOST=0.0.0.0
PORT=8000
LOGLEVEL=info

# 数据库和 Skill 配置（可选）
DATABASE_PATH=data/bot.db
SKILLS_PATH=skills
SYSTEM_PROMPT_FILE=system_prompt.md
CHAT_ROUNDS=5

# Skill 脚本执行超时（可选，单位：秒，默认 120）
SKILL_SCRIPT_TIMEOUT=120

# 流式状态清理配置（可选）
STREAM_MAX_AGE=300          # 未完成流的最大存活时间（秒，默认 300）
STREAM_CLEANUP_INTERVAL=60  # 清理任务运行间隔（秒，默认 60）
```

### 启动应用

#### HTTP 回调模式（默认）

```bash
# 设置协议模式为 callback（或不设置，默认即为 callback）
export PROTOCOL_MODE=callback

# 开发模式（热重载）
python -m src.main

# 或直接使用 uvicorn
uvicorn src.main:app --host 0.0.0.0 --port 8000 --reload
```

#### WebSocket 长连接模式

```bash
# 设置协议模式为 websocket
export PROTOCOL_MODE=websocket

# 设置 WebSocket 模式所需的凭证
export WECHAT_BOT_ID=your_bot_id
export WECHAT_BOT_SECRET=your_bot_secret

# 启动应用
python -m src.main
```

> **注意**：WebSocket 模式不会启动 HTTP 服务，而是直接建立与企业微信服务器的长连接。

### 配置企业微信回调

在企业微信管理后台配置智能机器人的回调 URL：
- URL: `http://your-server:8000/`
- Token: 与 `PLATFORM_API_TOKEN` 一致
- EncodingAESKey: 与 `PLATFORM_ENCODING_AES_KEY` 一致

## Skill 扩展

### 创建 Skill

在 `skills/` 目录下创建子目录，添加 `SKILL.md` 文件：

```
skills/
└── example-skill/
    └── SKILL.md
```

### SKILL.md 格式

```markdown
---
name: example-skill
description: 这是一个示例技能，用于演示 Skill 机制。
---

# Example Skill

这里是 Skill 的详细说明和使用指南。
```

### Skill 注入机制

系统在每次 LLM 调用时，会将所有 Skill 的描述动态注入到上下文中：

```
## Available Skills

The following skills are available:

- **skill-name**: Skill description

When a task matches a skill's purpose, follow its instructions.
```

## 运行测试

```bash
# 运行所有测试
pytest tests/ -v

# 运行测试并生成覆盖率报告
pytest tests/ --cov=src --cov-report=html

# 运行特定测试文件
pytest tests/test_api_integration.py -v

# 运行 WebSocket 适配器测试
pytest tests/test_websocket_adapter.py -v
```

## 项目结构说明

### 协议层 (`src/adapters/`)

- **base.py**: 协议适配器基类，定义通用接口
- **websocket_adapter.py**: WebSocket 长连接协议适配器
  - 封装 wecom-aibot-python-sdk
  - 消息格式转换（frame -> MessageHandler 格式）
  - 流式回复适配（StreamManager -> reply_stream）
  - 事件处理（on_text, on_enter_chat 等）

### API 层 (`src/api/`)

- **router.py**: 处理企业微信回调请求
  - `GET /`: URL 验证接口
  - `POST /`: 消息回调和流式刷新接口
  - `GET /health`: 健康检查接口

### 服务层 (`src/services/`)

- **message_handler.py**: 消息处理核心服务
  - 解析消息类型（text、stream 等）
  - 协调 LLM 调用、数据库存储
  - 构建含 Skill 描述的上下文
  - Skill 脚本执行超时保护（`SKILL_SCRIPT_TIMEOUT`），超时自动终止子进程

- **stream_manager.py**: 流式状态管理
  - 使用 asyncio.Lock 保证并发安全
  - 管理 stream_id 到内容的映射
  - 后台定时清理超时未完成的幽灵流和已完成的流，防止内存泄漏（`STREAM_MAX_AGE` / `STREAM_CLEANUP_INTERVAL`）

- **skill_loader.py**: Skill 加载服务
  - 启动时扫描 skills 目录
  - 解析 YAML frontmatter
  - 格式化 Skill 描述供 LLM 使用

### 数据层 (`src/models/`)

- **database.py**: 异步 SQLite 数据库管理
- **message.py**: 消息数据模型和 CRUD 操作

### LLM 客户端 (`src/llm/`)

- **client.py**: DeepSeek API 调用封装
  - 支持流式和非流式模式
  - 带回调的流式调用

## 日志配置

日志级别可通过 `LOGLEVEL` 环境变量配置：

- `debug`: 详细调试信息
- `info`: 一般运行信息（推荐生产环境）
- `warning`: 警告信息
- `error`: 错误信息

可通过 `LOGFILE` 配置日志文件路径。

## 开发指南

### TDD 开发模式

本项目遵循测试驱动开发（TDD）模式：

1. 先编写测试用例
2. 运行测试确认失败
3. 实现功能代码
4. 运行测试确认通过
5. 重构优化

### 代码风格

- 使用 Python 类型注解
- 遵循 PEP 8 代码风格
- 使用 async/await 进行异步编程
- 敏感信息不输出到日志

## License

MIT License