基于WebSocket与大模型的本地消息自动化处理架构设计与Notion集成实践

摘要在日常开发与信息管理中开发者常需要将即时通讯工具IM中的碎片化文本、URL链接自动提取并归档至知识库。本文将探讨一种基于PC端内存Hook技术、WebSocket双向通信、以及大语言模型LLM的自动化消息处理架构。该架构旨在实现对本地微信客户端消息的无损监听、异步清洗并最终调用 Notion API 实现结构化存储。整体架构与技术选型针对桌面端IM应用的自动化传统方案通常采用UI自动化如 pyautogui 或 uiautomation但存在阻塞进程、资源占用高、极易被干扰等问题。为保证系统在后台的高可用性与低延迟本系统采用 DLL注入拦截 本地WebSocket服务 异步Python处理端 的解耦架构。底层通信层 (Hook WebSocket)通过C编写DLL注入PC客户端进程拦截 RecvMsg 与 SendMsg 的内存函数。拦截到的数据通过管道传递给本地的 Node.js/Go 守护进程该进程对外暴露 ws://127.0.0.1:8080 服务。路由分发层 (Router)Python 服务端作为 WebSocket Client 持续监听接收到 JSON 格式的消息负载后通过正则提取指令与 URL。业务处理层 (Processor)集成 requests 进行网页 DOM 树拉取集成第三方大模型 API 进行自然语言处理摘要生成最后通过 notion-client 写入 Notion Database。核心处理逻辑与数据流整个消息清洗与流转的实现在 Python 端通过 asyncio 构建异步事件循环确保网络 I/O网页抓取、LLM调用、Notion写入不会阻塞底层的消息接收。核心逻辑流转代码实现import reimport asyncioimport websocketsimport jsonfrom notion_client import ClientNotion API 客户端初始化notion Client(auth“YOUR_NOTION_INTEGRATION_TOKEN”)DATABASE_ID “YOUR_DATABASE_ID”URL 正则提取def extract_url(text):urls re.findall(r’(https?/[^\s]), text)return urls[0] if urls else None异步消息处理流水线async def process_message(msg_data):# 工程规范仅处理特定发件人如文件传输助手的数据隔离无效广播if msg_data.get(‘to_wxid’) ! ‘filehelper’ and msg_data.get(‘from_wxid’) ! ‘my_self_id’:returncontent msg_data.get(content, ) # 路由 1纯文本指令标记 if content.startswith(#todo): task_text content.replace(#todo, ).strip() # 调用写入逻辑 (略) await send_reply(✅ TODO 指令已解析并入库) # 路由 2富文本/URL 解析与大模型摘要 url extract_url(content) if url: await send_reply(⏳ 检测到 URL触发 DOM 解析与摘要生成机制...) # 1. 模拟 I/O 阻塞获取网页元数据 article_text await fetch_webpage_content(url) # 2. 调用 LLM 接口生成结构化摘要 summary await call_llm_summary(article_text) # 3. 构建 Notion Page Payload create_notion_page(url, summary) await send_reply(f 结构化数据构建完成\n提取摘要{summary})WebSocket 消费者挂载async def ws_consumer():async with websockets.connect(“ws://127.0.0.1:8080”) as websocket:while True:message await websocket.recv()data json.loads(message)# 分发至协程池避免队头阻塞asyncio.create_task(process_message(data))ifname “main”:asyncio.run(ws_consumer())工程实践中的难点与解决方案在构建高并发或持续运行的 IM 自动化网关时通常会面临网络延迟、状态死锁等工程级问题。以下是本架构在测试环境中的优化点3.1 消息回声Echo与死锁处理问题描述当本地 API 主动调用 SendMsg 发送一条回复时底层 Hook 同样会将其作为一条新消息拦截并推送至 WebSocket 队列。若业务层未做判断系统会将其视为新输入进而生成新的回复形成无限循环的请求风暴Request Storm。工程解法在路由层的最顶端引入消息发送者身份校验Session Identity Check。必须对 Payload 中的 sender_id 进行断言丢弃由 Bot 自身发出的消息载荷。3.2 异步 I/O 下的媒体资源竞态条件问题描述IM 客户端在接收图片或视频等多媒体文件时底层通知事件往往早于文件 I/O 写入磁盘完成的时间。若立即读取本地缓存路径会发生 FileNotFoundError 或读取到0字节的空文件。工程解法实现一个带指数退避Exponential Backoff的轮询检查器。import osimport timeimport asyncioasync def wait_for_file_ready(file_path, timeout10):start_time time.time()last_size -1while time.time() - start_time timeout:if os.path.exists(file_path):current_size os.path.getsize(file_path)# 文件存在且大小大于0且连续两次检查大小不再变化认为写入完成if current_size 0 and current_size last_size:return Truelast_size current_sizeawait asyncio.sleep(0.5)return False3.3 长文本数据包Payload的传输限流问题描述LLM 生成的摘要或深度分析文本往往超过 1000 字符。若将超大字节流一次性压入底层发送接口容易造成缓冲区溢出Buffer Overflow或因触发 IM 服务器的安全策略而被静默丢弃。工程解法在网络传输层实现基于 Token 长度或特定分隔符的数据分片Chunking。设定单次通信的阈值如 max_length 400并在每个 Chunk 之间加入调度休眠Sleep实现流控Rate Control。结论通过“内存 Hook 拦截 - WebSocket 分发 - 异步协程处理 - LLM与Notion API 联动”的架构可以极其轻量、稳定地将个人电脑的 IM 客户端改造为一个自动化的数据清洗与知识管理网关。该架构解耦了底层协议与上层业务逻辑开发者只需专注于 Python 层的 NLP 与业务代码编写即可实现复杂的自动化流。