# 微信智能机器人开发实录：从零搭建 AI 自动回复系统

> **一句话总结**：通过 wx4py 的 UI 自动化能力 + DeepSeek API 的智能对话 + StyleLearner 的风格学习，打造了一个能模仿真人语气自动回复微信消息的智能机器人。

## 一、项目背景与动机

### 为什么要做这个项目？

我暗恋一个女孩子叫"zwh"，平时我们在微信上聊天。有一天我突发奇想：**能否用 AI 模拟她的语气，让机器人替我回复消息？这样我可以名正言顺的关心“她”，和“她”聊天**

市面上已有的方案（如 ItChat）大多基于 Web 微信协议或者3.9的版本，但 Web 微信早已被腾讯废弃，无法登录。而 PC 版微信 4.x 仍然活跃，且提供了完整的 UI 界面。既然没有现成的轮子，那就自己造一个。

### 开发目标

1. **实时监听**微信消息，自动调用 AI 生成回复
2. **模仿指定人物**的语气风格
3. 支持**上下文对话记忆**
4. 可选的**语音合成**输出
5. **完全本地运行**，无需服务器

### 适用场景

- 个人聊天助手 — 在忙碌时自动回复
- 角色扮演机器人 — 模拟特定人物聊天风格
- 客服机器人原型 — 基于微信的自动应答系统

---

## 二、技术选型与架构设计

### 核心技术栈

| 技术 | 用途 | 选型理由 |
|------|------|----------|
| **wx4py** | 微信 UI 自动化 | 支持微信 4.x，UIA 方式更稳定，比 ItChat 更可靠 |
| **DeepSeek API** | LLM 对话生成 | 性价比极高，1 元可聊约 500 次 |
| **Edge-TTS** | 语音合成 | 微软免费引擎，中文自然，无需 API Key |
| **pywin32** | Windows 接口 | 窗口控制、UI 操作 |
| **Loguru** | 日志记录 | 比标准 logging 简洁 10 倍 |
| **python-dotenv** | 配置管理 | 环境变量安全隔离 |

> **为什么不直接用 ItChat？** ItChat 基于 Web 微信协议，而 Web 微信已在 2023 年被腾讯彻底关闭。wx4py 基于 UIAutomation（UI 自动化）技术，操作的是微信 PC 客户端的界面元素，不涉及协议逆向，更稳定可靠。

### 为什么不用本地模型？

最开始尝试了 Qwen2.5-1.5B 的 LoRA 微调，但实测效果很差——回复碎片化、答非所问。原因在于：训练数据只有简单的聊天记录，模型学到的只是"模仿对方"而不是"作为对方回应"。

最终方案是 **DeepSeek API + StyleLearner 风格学习**：

| 输入 | 本地 LoRA 模型（1.5B） | DeepSeek + StyleLearner |
|------|---------------|----------------------|
| "在干嘛呢" | 语句碎片拼接 | "刚洗完澡躺床上呢～今天下午去图书馆写了会儿线代作业[捂脸]" |
| "明天吃饭？" | 答非所问 | "咦？突然约我吃饭呀～[呲牙] 明天下午没课，几点去呀？" |
| "线代考试好慌" | 回复不相关 | "啊？矩阵那章真的好难啊…[流泪]" |

**核心结论**：在数据量不足（<1000 条）的场景下，提示词工程 + 大模型 API 的效果远超小模型微调。DeepSeek 方案无需 GPU，效果反而更好。

### 整体架构图

```
┌─────────────────────────────────────────────────────────────┐
│                     WeChat AI Bot 系统架构                    │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────┐     ┌──────────────────┐                   │
│  │  微信 PC 4.x │◄────│ wechat_client.py │◄─── 每 2 秒轮询   │
│  │  (UI 窗口)   │     │ (UIAutomation)   │    检测新消息      │
│  └──────┬──────┘     └──────────────────┘                   │
│         │                                                    │
│         ▼                                                    │
│  ┌──────────────────────────────────────────────┐           │
│  │               main.py (主调度器)               │           │
│  │                                               │           │
│  │  ┌────────────┐  ┌──────────┐  ┌──────────┐  │           │
│  │  │ 记忆管理器  │  │风格学习器 │  │LLM 客户端│  │           │
│  │  │(环形缓冲区) │  │(CSV→提示)│  │(DeepSeek)│  │           │
│  │  └────────────┘  └──────────┘  └────┬─────┘  │           │
│  └──────────────────────────────────────┬───────┘           │
│                                         │                    │
│          ┌──────────────────────────────┘                    │
│          ▼                                                   │
│  ┌────────────────┐    ┌──────────────────┐                 │
│  │ TTS 语音合成    │    │ StyleLearner     │                 │
│  │ (Edge-TTS)     │    │ (CSV→提示词)     │                 │
│  └────────────────┘    └──────────────────┘                 │
└─────────────────────────────────────────────────────────────┘
```

---

## 三、核心模块详解（附完整代码）

### 3.1 消息监听引擎 — `wechat_client.py`

这是整个机器人的"耳目"。它通过 wx4py 连接到微信 PC 窗口，每 2 秒轮询一次聊天区域的消息列表。

**核心机制**：
- 通过 UIAutomation 读取聊天窗口中的元素 ClassName
- 使用 `_known` 集合去重（记录已处理消息的唯一标识）
- 同时支持文本和语音两种消息类型的检测

```python
"""
wechat_client.py — 消息监听核心代码
依赖: wx4py, pywin32, win32gui
"""

import time
import logging
from dataclasses import dataclass
from typing import Callable, Optional, Set

logger = logging.getLogger(__name__)

# 微信消息元素的 UIA ClassName
_MESSAGE_CLASSES_TEXT = {
    "mmui::ChatTextItemView",
    "mmui::ChatBubbleItemView",
}

_MESSAGE_CLASSES_VOICE = {
    "mmui::ChatVoiceItemView",
    "mmui::ChatVoiceBubbleView",
}

_MESSAGE_CLASSES = _MESSAGE_CLASSES_TEXT | _MESSAGE_CLASSES_VOICE

@dataclass
class WeChatMessage:
    """微信消息的数据结构"""
    sender: str
    content: str
    wxid: str
    roomid: str = ""
    is_group: bool = False
    is_voice: bool = False  # 标记是否为语音消息

def _element_key(class_name: str, text: str, element) -> str:
    """
    生成消息的唯一标识 key，用于去重。
    
    文本消息 → 用内容去重
    语音消息 → 用元素在屏幕上的位置坐标去重
    
    Debug 经验：
    最初直接用文本内容做 key，发现语音消息始终只处理第一条。
    因为在微信 UI 中，不同时间收到的语音消息可能显示为相同的文本。
    """
    if class_name in _MESSAGE_CLASSES_VOICE:
        try:
            rect = str(element.BoundingRectangle)
            return f"voice:{rect}"
        except Exception:
            return f"voice:{text}"
    return f"text:{text}"
```

**调试技巧**：如果发现消息漏处理，可以开启 UIA 元素调试模式，打印所有控件的 ClassName：

```python
def discover_message_classes(self):
    """扫描消息列表，输出所有 UIA class name（用于调试语音检测）"""
    classes = set()
    for child in self._get_message_children():
        cls = child.ClassName
        if cls:
            classes.add(cls)
    logger.info(f"当前微信版本的消息类名: {sorted(classes)}")
```

### 3.2 风格学习器 — `style_learner.py`

这是项目的核心创新点。传统方案需要 GPU 微调，而我用了一个更巧妙的方法：**将真实聊天记录作为 few-shot 示例注入提示词**。

```python
"""
style_learner.py — 从真实聊天记录学习对话风格
无需 GPU，无需训练，纯提示词工程
"""

import csv
import random
from pathlib import Path
from typing import List

PERSONA_DESCRIPTION = """你是周文慧，19岁，大一女生。

## 性格特征
- 恬静温柔，说话轻声细语，很少发脾气
- 有时也会活泼撒娇，会用[呲牙][耶][强]等微信表情
- 善良体贴，会关心朋友，说话让人感觉舒服
- 学妹类型，带一点可爱的学生气
- 偶尔会抱怨学业，但并不消极
- 说话自然真实，不正式，不说大道理

## 回复原则
1. 模仿周文慧的语气和用词，不要以AI口吻说话
2. 回复自然简短，像普通朋友聊天
3. 根据对方说的内容来回应，不要答非所问
4. 适当使用语气词（啊、啦、呀、呢、哦、嘛）
5. 偶尔可以用颜文字或微信表情，但不要过量
6. 不要说"作为AI"、"作为语言模型"之类的话"""

class StyleLearner:
    """
    从 CSV 聊天记录中提取风格示例，构建提示词。
    
    工作流程：
    1. 读取导出的微信聊天记录 CSV（2582+ 条消息）
    2. 按时间分为前、中、后三段，每段随机抽样
    3. 构建 [人设描述] + [性格特征] + [回复原则] + [真实示例] 提示词
    
    优势：
    - 无需 GPU 训练
    - 换人只需换 CSV 文件
    - 效果远超小模型微调
    """
    
    def __init__(self, csv_dir: str = None):
        self._examples: List[str] = []
        self._csv_dir = csv_dir or self._default_csv_dir()
        self._load_examples()
    
    def _load_examples(self):
        """从指定目录加载所有 CSV 文件中的聊天记录"""
        csv_dir = Path(self._csv_dir)
        if not csv_dir.exists():
            print(f"[WARNING] CSV 目录不存在: {csv_dir}")
            return
        
        messages = []
        for fpath in sorted(csv_dir.glob("*.csv")):
            with open(fpath, "r", encoding="utf-8") as f:
                reader = csv.DictReader(f)
                for row in reader:
                    msg = row.get("msg", "").strip()
                    if msg and len(msg) >= 2:
                        messages.append(msg)
        
        self._examples = messages
        print(f"已加载 {len(self._examples)} 条风格样本")
    
    def get_examples(self, count: int = 15) -> List[str]:
        """
        三段式抽样：保证覆盖整个时间线的聊天风格。
        
        前 1/3 → 早期聊天风格
        中 1/3 → 中期聊天风格
        后 1/3 → 近期聊天风格（最重要）
        """
        n = len(self._examples)
        if n <= count:
            return self._examples.copy()
        
        third = n // 3
        indices = (
            random.sample(range(0, third), min(count // 3, third))
            + random.sample(range(third, 2 * third), min(count // 3, third))
            + random.sample(range(2 * third, n), min(count // 3, n - 2 * third))
        )
        
        random.shuffle(indices)
        return [self._examples[i] for i in indices[:count]]
```

**优化建议**：如果想切换模仿对象，只需修改 `PERSONA_DESCRIPTION` 和更换 CSV 文件即可，StyleLearner 不需要做任何代码改动。

### 3.3 对话记忆系统 — `memory_manager.py`

为了让对话更连贯，记忆系统采用**环形缓冲区** + **JSON 持久化**的轻量级方案。

```python
"""
memory_manager.py — 对话记忆管理
使用 collections.deque 实现环形缓冲区
"""

import json
import time
from pathlib import Path
from typing import List, Dict, Optional
from collections import deque

class ConversationMemory:
    """
    环形缓冲区对话记忆。
    
    设计决策：
    - 用 deque（而非 list）：固定容量，自动淘汰旧数据
    - 用 JSON（而非数据库）：零依赖，开箱即用
    - 多联系人隔离：每个 wxid 独立存储
    """
    
    def __init__(self, max_turns: int = 20, storage_dir: str = "data"):
        self.max_turns = max_turns
        self._history: Dict[str, deque] = {}
        self._storage_dir = Path(storage_dir)
        self._storage_dir.mkdir(parents=True, exist_ok=True)
    
    def add_turn(self, wxid: str, user_msg: str, bot_reply: str):
        """记录一轮对话"""
        if wxid not in self._history:
            self._history[wxid] = deque(maxlen=self.max_turns)
        self._history[wxid].append({
            "timestamp": time.time(),
            "user": user_msg,
            "bot": bot_reply,
        })
    
    def get_context(self, wxid: str, max_turns: int = 5) -> str:
        """
        构建 LLM 可读的上下文文本。
        
        输出格式：
        对方: 今天天气真好
        你: 是啊，要出去走走吗？
        """
        if wxid not in self._history:
            return ""
        
        lines = []
        for turn in list(self._history[wxid])[-max_turns:]:
            lines.append(f"对方: {turn['user']}")
            lines.append(f"你: {turn['bot']}")
        return "\n".join(lines)
```

持久化数据格式：
```json
{
  "wxid_xxx": [
    {
      "timestamp": 1746452800.0,
      "user": "今天天气真好",
      "bot": "是啊，要出去走走吗？"
    }
  ]
}
```

### 3.4 语音合成 — `tts_client.py`

基于 Microsoft Edge-TTS 引擎，完全免费。

```python
"""
tts_client.py — Edge-TTS 语音合成
完全免费，无需 API Key
"""

from edge_tts import Communicate
import hashlib
from pathlib import Path

class TTSClient:
    """
    Edge-TTS 语音合成客户端。
    
    支持发音人：
    - xiaoxiao: 晓晓（女声，亲切自然，推荐）
    - yunyang: 云扬（男声，专业沉稳）
    - xiaoyi: 晓伊（女声，活泼生动）
    - yunxi: 云希（男声，阳光明朗）
    """
    
    def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural"):
        self.voice = voice
        self.cache_dir = Path("data/audio")
        self.cache_dir.mkdir(parents=True, exist_ok=True)
    
    async def speak(self, text: str) -> str:
        """
        异步合成语音。
        
        MD5 缓存：相同文本不重复生成，
        缓存路径: data/audio/{md5(text)}.mp3
        """
        md5 = hashlib.md5(text.encode()).hexdigest()
        path = self.cache_dir / f"{md5}.mp3"
        
        if path.exists():
            return str(path)
        
        communicate = Communicate(text, self.voice)
        await communicate.save(str(path))
        return str(path)
```

> **注意**：第一次运行 Edge-TTS 时，需要联网从微软 CDN 下载语音模型（约 1-5MB），后续可离线使用。

---

## 四、完整消息处理流程

```
微信新消息（文本/语音）
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 1: 消息检测                             │
│ wechat_client 每 2 秒轮询一次               │
│ UIAutomation 读取聊天区域控件列表            │
│ 对比 _known 集合 → 发现新消息                │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 2: 消息过滤                             │
│ main.py 判断:                                │
│   • 群消息 → 过滤（暂不支持）                 │
│   • 语音消息 → 走 ASR 流程                   │
│   • 文本消息 → 正常处理                       │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 3: 构建上下文                           │
│ memory_manager.get_context()                 │
│ → 读取该联系人最近 5 轮对话历史               │
│ → 格式化为 "对方: xxx\n你: xxx" 文本         │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 4: 风格学习                             │
│ style_learner.get_examples()                 │
│ → 三段式抽样 15 条真实聊天示例                │
│ → 拼接完整角色提示词                         │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 5: LLM 生成回复                         │
│ llm_client.chat(message, context)            │
│ → 调用 DeepSeek API (deepseek-chat)          │
│ → 返回自然语言回复                           │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 6: 记忆存储 + 发送                      │
│ memory_manager.add_turn(msg, reply)         │
│ memory_manager.save_to_disk() → JSON        │
│ wechat_client.send_message(reply) → 发送    │
└─────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────┐
│ 步骤 7: [可选] 语音合成                      │
│ tts_client.speak(reply) → MP3               │
│ wechat_client.send_audio_file() → 发送语音  │
└─────────────────────────────────────────────┘
    │
    ▼
等待 2 秒 → 继续下一轮轮询
```

---

## 五、开发历程与踩坑实录

### v1.0 — 基础框架（30 分钟）
- 实现：wx4py 连接微信 + DeepSeek API 调用 + 固定角色提示词
- 最简单的"Hello World"版本

### v2.0 — 本地模型支持（3 小时，踩坑最惨）
- 尝试 Qwen2.5-3B + LoRA fp16 → OOM
- 尝试 Qwen2.5-3B + QLoRA 4-bit → bitsandbytes segfault
- 最终 Qwen2.5-1.5B + LoRA fp16 → 能跑但效果差
- 教训：8GB 显存上限是 1.5B，Windows 上 bitsandbytes 4-bit 有已知兼容问题

### v3.0 — 对话记忆（1 小时）
- 实现：环形缓冲区 + JSON 持久化
- 踩坑：多联系人同时发消息时上下文混淆
- 解决：按 wxid 隔离存储

### v4.0 — 语音合成（30 分钟）
- Edge-TTS 集成，4 种中文发音人
- 踩坑：首次运行需要联网下载语音模型

### v5.0 — 风格学习 + 全面优化（2 小时，关键转折）
- 发现 DeepSeek + StyleLearner 效果远超本地微调
- 废弃本地模型方案，全面转向提示词工程
- 新增 StyleLearner 模块、消息队列、错误处理优化

### 踩坑全记录

| 问题 | 现象 | 解决方案 |
|------|------|----------|
| **消息重复处理** | 同一条消息回复 2-3 次 | 用 `_element_key()` 生成唯一 key + 发送后立即标记 seen |
| **.env 被提交到 GitHub** | API Key 泄露 | `git rm --cached .env` + 添加 .gitignore |
| **微信最小化后无响应** | 机器人不检测新消息 | 不要最小化到托盘，可以最小化到任务栏 |
| **DeepSeek API 超时** | 偶尔返回空回复 | 添加 retry 逻辑 + 超时重试 |
| **Python 3.13 兼容** | wx4py 导入失败 | 使用 Python 3.10/3.11 |

---

## 六、快速上手

### 环境要求

| 依赖项 | 要求 | 说明 |
|--------|------|------|
| 操作系统 | Windows 10 或 11 | 必须，依赖 UIAutomation API |
| 微信客户端 | PC 4.x 版本 | 当前适配 4.1.8+ |
| Python | 3.10 或 3.11 | 推荐 3.10.11 |
| DeepSeek API Key | 需要 | platform.deepseek.com 注册获取 |
| GPU | 不需要 | 所有 AI 处理云端完成 |

### 一键部署

```bash
# 克隆项目
git clone https://github.com/starlight001219/wechat-clone-bot.git
cd wechat-clone-bot

# 安装依赖
pip install -r requirements.txt

# 配置环境变量
copy .env.example .env
# 用记事本打开 .env，填入你的 DeepSeek API Key

# 登录微信（用小号！）

# 启动机器人
py -3.10 bot\main.py
```

### 配置项说明

```ini
# LLM API 配置
LLM_API_KEY=sk-your-deepseek-api-key    # 必填
LLM_API_BASE=https://api.deepseek.com/v1
LLM_MODEL=deepseek-chat

# 人物配置
TARGET_NAME=周文慧                      # 角色名（用于系统提示词）
TARGET_WXID=星夜                        # 要监控的联系人

# 语音合成
TTS_ENABLED=false                       # true=开启语音回复
TTS_VOICE=xiaoxiao                      # 发音人

# 日志
LOG_LEVEL=INFO
```

### 启动确认

正常启动后控制台应看到：

```
2026-05-06 20:00:00 | INFO | WeChat AI Bot starting...
2026-05-06 20:00:00 | INFO | Persona: 周文慧
2026-05-06 20:00:00 | INFO | Contact: 星夜
2026-05-06 20:00:00 | INFO | Mode: TEXT CHAT
```

> **封号风险提醒**：微信自动化操作违反微信用户协议，强烈建议使用小号，不要用主号测试。

---

## 七、项目地址与资源

- **GitHub 仓库**：[github.com/starlight001219/wechat-clone-bot](https://github.com/starlight001219/wechat-clone-bot)
- **相关项目**：
  - 周文慧 AI Web 聊天：[github.com/starlight001219/ai](https://github.com/starlight001219/ai)
  - WeClone（上游项目）：[github.com/xming521/WeClone](https://github.com/xming521/WeClone)

---

## 八、经验总结

1. **不要盲目追"微调"** — 数据量不足的场景下，提示词工程 + 大模型 API 的效果远超小模型微调
2. **UI 自动化是一把双刃剑** — 免去协议逆向风险，但窗口必须在前台可见
3. **去重是消息监听的关键** — 用 `_known` 集合 + 唯一 key 生成策略，避免重复处理和自回复

### 避坑清单

- 使用 Python 3.10/3.11，不要用 3.13（wx4py 兼容问题）
- 用小号测试，大号被封后果自负
- 微信窗口保持可见（可最小化到任务栏，不要到托盘）
- .env 文件不要提交到 Git
- 首次运行 TTS 需要联网
- DeepSeek API 需要绑定手机号

---

> **后记**：这个项目从最初的一个简单 API 调用，逐步演进成了包含消息监听、风格学习、对话记忆、语音合成的完整机器人系统。最有成就感的一刻，是看到机器人用周文慧的语气说出"刚洗完澡躺床上呢～今天下午去图书馆写了会儿线代作业[捂脸]"——那一刻我知道，成了。