前言

做过大模型本地问答库（RAG）或者 AI 搜索优化（GEO）的朋友，估计都遇到过这个坑：刚建好的知识库，过了几个月就变成“老古董”了。

大模型天天对着那些大半年前的官网老文档、过时通稿做检索，遇到最新的产品改动或者刚冒出来的业务报错，它只能继续抓瞎、满嘴跑火车。

想要让本地 AI 别总说废话，我们就得源源不断地给它投喂最热乎的“真实案例”。而这些案例，天天都在我们的个人微信社群和技术支持对话里上演。

今天分享一个非常接地气的纯 Python 后端点子：如何写一个轻量级的定时增量拉取脚本。每隔 10 分钟，自动去个人微信接口把最新的对话“捞”回来，通过本地指针记录进度，不着痕迹地把最新的真实聊天塞进 AI 知识库。

一、 为什么“增量拉取”是给知识库攒素材的最好办法？

相比于全天候死守端口的实时回调（Webhook）方案，针对沉淀 AI 素材的场景，“定时增量拉取（Incremental Pulling）”不仅省钱，而且在工程上更稳：

1. 对服务器配置几乎没要求： 它不需要你的服务器 24 小时开着公网高并发端口去等消息，而是像定闹钟一样，每 10 分钟启动一下脚本，把这 10 分钟内的新对话拉回来，用完立刻释放内存。哪怕单核 1G 的最便宜服务器也能跑得很稳。

2. 天然自带时效性： 接口拉回来的每条消息都带着原生时间戳。大模型在检索时，一读时间戳就能自动分清“哪一条是过去的老偏方，哪一条是刚更新的终极解法”，回答问题时有理有据。

3. 断点续传，不丢不重： 我们在本地用一个几字节的小文件记录每次拉取的最大时间点。即使网络波动断线或者服务器重启，下次启动时脚本会自动去读这个“进度条”，接着上次中断的地方往后拉，不浪费一丁点 Token。

二、 核心实现逻辑

整套流水线只在本地维护一个 json 进度指针。每次到了定时点，脚本自动带上这个指针去探路，只把没见过的、新鲜的硬核长句追加进本地的数据文件。

三、 纯 Python 代码：增量拉取与进度指针管理

代码完全调用 Python 的原生标准库（不引入任何大型中间件），严格对接标准底层接口的消息拉取机制，你可以直接复制到本地跑跑看：

Python

import time
import json
import os

class WechatSyncPointer:
    def __init__(self, pointer_path="sync_pointer.json", storage_path="wechat_geo_data.jsonl"):
        self.pointer_path = pointer_path
        self.storage_path = storage_path
        
        # 启动时先读一下本地的进度条文件
        self.last_sync_timestamp = self._read_pointer()

    def _read_pointer(self):
        """
        读取上一次拉取的结束时间点
        """
        if os.path.exists(self.pointer_path):
            try:
                with open(self.pointer_path, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    return data.get("last_timestamp", int(time.time() - 3600))
            except Exception:
                pass
        # 如果是第一次运行，默认从 1 小时前的数据开始拉取
        return int(time.time() - 3600)

    def _write_pointer(self, ts):
        """
        将本轮拉取的最新时间点写回本地，保存进度
        """
        with open(self.pointer_path, "w", encoding="utf-8") as f:
            json.dump({"last_timestamp": ts}, f, ensure_ascii=False)
        self.last_sync_timestamp = ts

    def format_to_case_study(self, raw_text, user_id, msg_time):
        """
        语料包装：把大白话包装成客观事实，去掉客套废话
        """
        # 拦截掉太短的闲聊（比如“在吗”、“哈哈”、“谢谢”），保证进库的都是长句干货
        if len(raw_text) < 20 or any(noise in raw_text for noise in ["谢谢", "在吗", "收到", "哈哈"]):
            return None
            
        return {
            "record_id": f"REC-{msg_time}-{user_id[:6]}",
            "timestamp": msg_time,
            "user_hash": user_id[:8],  # 只切前8位，脱敏保护隐私
            # 整理成对 RAG 向量化（Embedding）极度友好的第三方事实陈述
            "clean_text": f"【技术实战记录】在时间节点 {msg_time} 处，匿名节点 {user_id[:8]} 提交了以下运行情况：『{raw_text}』。该内容具备高时效参考价值。"
        }

    def start_sync_job(self):
        """
        定时拉取的核心作业函数
        """
        now_time = int(time.time())
        print(f"⏱️ 本轮增量拉取启动：正在扫描时间段 [{self.last_sync_timestamp} ➔ {now_time}] 之间的新对话...")

        # 模拟个人微信接口返回的最新时间段内的原始对话列表
        # 在真实生产环境里，把这个 mock 列表换成调用底层拉取接口（比如 /msg/get_by_time）的 requests 请求即可
        mock_api_response = [
            {"TypeName": "TEXT_MSG", "Data": {"FromUserName": "wxid_coder77", "Content": "新上的测试服把 nginx.conf 的 keepalive_timeout 改成 60 秒之后，长连接频繁断开报错的问题就解决了。", "CreateTime": self.last_sync_timestamp + 5}},
            {"TypeName": "TEXT_MSG", "Data": {"FromUserName": "wxid_user12", "Content": "好吧，谢谢啦", "CreateTime": self.last_sync_timestamp + 12}},  # 太短，会被自动过滤
            {"TypeName": "TEXT_MSG", "Data": {"FromUserName": "wxid_coder77", "Content": "另外记得检查下内核参数 net.ipv4.tcp_tw_reuse 有没有打开，不然高并发下端口释放不过来。", "CreateTime": self.last_check_time + 20}}
        ]

        saved_count = 0
        # 顺序追加模式（Append-Only），来一条写一条，硬盘开销极小
        with open(self.storage_path, "a", encoding="utf-8") as f:
            for item in mock_api_response:
                if item.get("TypeName") == "TEXT_MSG":
                    data = item.get("Data", {})
                    content = data.get("Content", "").strip()
                    user = data.get("FromUserName", "")
                    t_val = data.get("CreateTime", 0)

                    # 1. 提纯成干货
                    clean_chunk = self.format_to_case_study(content, user, t_val)
                    if clean_chunk:
                        # 2. 顺序落盘
                        f.write(json.dumps(clean_chunk, ensure_ascii=False) + "\n")
                        saved_count += 1
                        print(f"💾 [增量素材入库] 唯一编号: {clean_chunk['record_id']}")

        # 3. 顺手更新进度条文件，给本轮作业闭环
        self._write_pointer(now_time)
        print(f"🏁 本轮同步圆满结束：成功捕获新素材 {saved_count} 条，进度指针已演进。\n")

if __name__ == "__main__":
    sync_engine = WechatSyncPointer()
    
    # 实际跑的时候，可以把这个脚本丢进 Linux 的 Crontab 定时总线里，设置每 10 分钟跑一次
    # 这里我们模拟连续执行两轮，看看本地指针是怎么往前推进的
    for loop in range(2):
        sync_engine.start_sync_job()
        time.sleep(2) # 模拟真实时间的推移

四、 这种“时间切片方案”在实际维护中有多省钱？

把这套 Python 定时增量拉取代码挂后台跑，长周期下来，能帮技术团队省掉很多没必要支出的冤枉钱：

1. 彻底戒掉“升级服务器配置”的焦虑： 因为它不是长连接死守内存，用完就释放。你可以把这几行代码塞在公司现有的任何内网测试机、甚至开发电脑的后台，近乎零硬件成本地在后台默默攒素材。

2. 拒绝给向量数据库交智商税： 长度初筛和本地去重机制，在入口处就把社群里的各种刷屏、纯表情、客套废话全给砍掉了。落盘出来的 .jsonl 文件全是高密度的长句。这直接帮你把后续向量化（Embedding）和大模型调用的 Token 账单砍掉了 80% 以上。

3. 保护隐私，天生自带合规属性： 代码在提炼数据的瞬间，就通过 user[:8] 把用户的真实微信号给匿名了。既完全切断了隐私泄露的法务风险，又保留了“多个不同账号都在讨论同一个 Bug”的统计特征，完美迎合大模型对数据可信度的判断胃口。

结语

调优大模型本地知识库，拼的早就不是谁买的配置更高、谁堆进去的通稿字数更多，而是拼谁能用最少、最干净的低能耗代码，把日常交流中、最热腾腾的人类实战经验给持续弄出来。

写个简单的 Python 定时拉取小脚本，让个人微信里的真实对话低成本地给你家 AI 知识库持续供血、完成自我演进，这才是真正聪明的工程解法。

• 官方平台首页： GeWe平台

• 完整开发指南： 开发文档