看到 PearPie 这类“私密 AI 会话 P2P 同步 无账户”方向时很多开发者第一反应是这不就是把聊天记录从中心服务器搬走吗其实没那么简单。传统 AI 聊天应用里“账户”不仅承担登录功能还负责云同步、订阅计费、内容合规审查。一旦去掉账户身份怎么建立、AI 服务如何鉴权、多设备如何保持会话一致全部需要重新设计。这篇文章会从该项目解决的痛点出发拆解免账户 P2P AI 聊天背后的核心架构并给出一套可在本地运行的最小 Python 原型帮助你理解隐私链路和同步机制。1. 为什么私密 AI 聊天需要重新设计数据链路1.1 传统 AI 聊天应用的数据经过哪些环节平时使用 ChatGPT、Claude、各类国产大模型 App 时用户输入问题到看见回复完整的链路大致是客户端把文本发送到厂商 API 网关网关完成鉴权、限流、内容安全过滤再把请求转发给模型推理服务推理结果按原路返回而中间产生的对话记录通常会被保存到厂商数据库中用来做问题排查、模型调优或者用户体验改进。在中心化架构下这种方式效率高、可运营性好但隐私上存在两个很难回避的问题。第一是数据集中存储一旦服务端被攻击、内部人员越权访问或者账号被找回历史聊天内容就容易泄露。第二是用户身份与行为强绑定手机号、邮箱、支付信息与聊天内容在同一个账号体系下长期关联用户很难选择“只使用产品但不留下身份痕迹”。1.2 PearPie 标题中的解题方向是什么PearPie 的定位可以拆成三个关键词Private、AI chat、peer-to-peer no accounts。用更直白的话来说它希望做到用户不需要注册账号就能和 AI 对话聊天记录不只是存在某一家厂商的服务器上而是通过 P2P 方式在用户自己的设备之间同步。“无账户”并不等于“无身份”。没有账号体系后设备本身可以成为身份的最小单位。每台设备在初始化时生成一对属于用户自己的密钥公钥用于标识设备私钥保存在本地用来证明“我是这台设备的主人”。对等设备之间相互交换公钥就建立了一个不依赖注册系统的信任关系这种模型的另外一个好处是迁移性更强只要导出自己的密钥和数据可以随时换到另一套服务端实现中不被厂商锁定。1.3 它能解决什么又不能解决什么这类架构非常适合两类场景一是隐私敏感型的个人知识库、日记式 AI 助手用户不希望 AI 厂商把历史话题和自身真实身份关联二是同一用户拥有多个私人设备希望会话能轻量同步又不想把所有历史发送到云盘或者厂商服务器。不过也要谨慎看待边界。P2P 同步只解决“消息不走中心服务器”这一问题并不天然等同于合规与安全。如果调用的是第三方云端大模型 API用户输入仍然会到达模型服务商如果自建大模型也需要遵守模型开源许可和当地法律法规。同时去中心化也意味着“数据自持”和“丢失自担”并存密钥一旦丢失密文就永远无法恢复。2. 私密 P2P AI 聊天的总体架构设计2.1 四个核心模块拆解从工程实现上看PearPie 这类工具可以划分为四个互相独立、接口清晰的模块。第一是身份模块。设备本地生成长期密钥对并派生出可读 Peer ID。第二是 AI 模块负责把用户消息发送给本地模型、自建推理服务或私有大模型网关。第三是存储模块所有会话内容需要在落盘前加密并保留必要的索引字段用于快速展示会话列表。第四是同步模块在两台或多台设备之间建立连接把增量消息加密传输到对端再通过对端解密后写进本地库。模块化设计有一个明显好处AI 服务商可以替换存储引擎可以替换同步协议也可以替换只要各模块之间的数据格式保持稳定就不会影响整体功能。2.2 没有账号怎么证明“你是你”在中心化系统里密码、验证码、第三方 OAuth 都是账号体系的一部分。在去中心化系统中通常的做法是使用非对称加密密钥作为“根身份”。设备第一次启动时客户端生成一对 Ed25519 或 X25519 密钥。公钥经过哈希和 Base64 编码后可以生成短 ID显示在界面中私钥则被存进系统安全区域或加密数据库中。同步时设备 A 如果要连接设备 B双方可以先交换公钥后续通信通过临时会话密钥加密消息接收方可以验证消息是否由持有对应私钥的设备发出。这解决了“注册”的问题但也引入了更严格的要求用户必须自己对私钥负责。开发者通常会在首次生成身份时给用户一个恢复助记词或导出文件否则设备损坏或应用被卸载所有只能由该密钥解密的会话将永久消失。2.3 同步模块要解决什么问题两台设备各自独立运行还会同时写入新会话如何保证两边数据最终一致是同步的关键。在设计上比较常见的是把数据库中的某条消息当成不可变日志。每次插入新消息时都分配一个全局可比较的逻辑时间戳或向量时钟同步时只交换各自增量通过最新版本比对让两端补齐缺失记录。P2P 同步并不要求两台设备同时在线离线设备重新联网后可以主动向在线节点拉取增量数据。对比传统的服务端同步P2P 方式把“唯一权威数据源”从中心数据库改成了用户设备集群本身。这样数据冗余度可能下降但对隐私保护更有利因为没有第三方长期持有全量副本。2.4 AI 能力接入的三种模式私密 AI 聊天项目对“AI 模型跑在哪里”通常有三种不同选择。第一种是本地推理模式通过 llama.cpp、Ollama、MLC 等方案运行开源模型。这种模式隐私最强所有输入输出都不离开设备或局域网但受机器性能限制适合中低参数量模型。第二种是个人 API 模式由用户自己在配置中填入大模型厂商的 API Key聊天内容从本机直接发送到模型网关产品方无法拦截或存储当然用户的数据是否被模型厂商留存要取决于该厂商的服务条款。第三种是企业代理模式企业内部部署统一网关对内提供 OpenAI 兼容接口既保留审计能力又不让业务数据直接进入公网模型。对 PearPie 这类产品而言“本地模型优先、API Key 可插拔”是比较稳妥的设计。用户有隐私需求时使用本地模型追求高智能时再显式配置远端模型系统必须在 UI 上标注当前请求的最终流向。3. 搭建一个最小可运行原型3.1 适合阅读本节的前提如果你已经理解 P2P 和端到端加密的概念可以直接跳到第 4 节看代码。这一节先从零准备环境方便新手也能完整复现。原型不需要手机 App也不需要公网服务器。我会用 Python 写多个模块演示“设备 A 生成本地身份 - 调用本地大模型 - 把对话加密写入 SQLite - 把同一份会话通过 TCP 推送给同一局域网内的设备 B”这条完整链路。真实生产环境里的 NAT 打洞、中继、协议握手等复杂点我会在对应位置提示扩展方向但不会在这份最小代码里强行实现。3.2 环境准备与版本建议建议使用 Python 3.10 及以上版本因为部分类型注解写法依赖新语法。程序主要用到cryptography库做本地身份和对称加密使用requests库调用本地 Ollama HTTP API用 Python 标准库实现轻量 TCP 同步服务。Ollama 需要在本机单独安装并运行建议先用社区成熟的qwen2.5:7b等开放模型做测试。需要说明的是模型版本变化很快本文不会写死某个具体版本号而是把模型名作为参数传入。下面安装核心 Python 依赖pip install cryptography requests3.3 演示项目目录结构为了让教程清晰我会将模块按职责拆分。你可以先在工作目录里建立如下结构pearpie_demo/ ├── main.py # 设备 A 入口提问、保存、推送 ├── sync_server.py # 设备 B 入口接收并写入加密消息 ├── p2p_chat/ │ ├── __init__.py │ ├── identity.py # 本地身份Ed25519 密钥对 Peer ID │ ├── store.py # 加密会话存储 │ ├── ai_client.py # 本地 Ollama AI 客户端 │ └── sync.py # 发送端 P2P 客户端4. 核心模块实现与解释4.1 身份模块生成本地密钥与 Peer ID第一个模块解决“我是谁”。使用 Ed25519 生成签名密钥对私钥留在设备 A公钥派生为短标识。为了便于演示这里直接把私钥以十六进制文本保存在普通文件里在生产环境中私钥应该写入操作系统 Keychain、Keystore 或安全加密容器。# p2p_chat/identity.py from __future__ import annotations import base64 import hashlib from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) class PeerIdentity: 本地生成的 Peer 身份私钥不离开设备。 def __init__(self, private_raw: bytes, public_raw: bytes): self._private_key Ed25519PrivateKey.from_private_bytes(private_raw) self._public_key Ed25519PublicKey.from_public_bytes(public_raw) classmethod def create(cls) - PeerIdentity: private_key Ed25519PrivateKey.generate() private_raw private_key.private_bytes( encodingserialization.Encoding.Raw, formatserialization.PrivateFormat.Raw, encryption_algorithmserialization.NoEncryption(), ) public_raw private_key.public_key().public_bytes( encodingserialization.Encoding.Raw, formatserialization.PublicFormat.Raw, ) return cls(private_raw, public_raw) property def private_key_hex(self) - str: return self._private_key.private_bytes( encodingserialization.Encoding.Raw, formatserialization.PrivateFormat.Raw, encryption_algorithmserialization.NoEncryption(), ).hex() property def public_key_hex(self) - str: return self._public_key.public_bytes( encodingserialization.Encoding.Raw, formatserialization.PublicFormat.Raw, ).hex() property def peer_id(self) - str: digest hashlib.sha256( bytes.fromhex(self.public_key_hex) ).digest()[:16] return base64.urlsafe_b64encode(digest).rstrip(b).decode()简要说明PeerIdentity.create()可以生成本设备身份private_key_hex和public_key_hex用于保存和传输peer_id是公钥哈希的前 16 字节再做 Base64 编码方便日志与 UI 展示。4.2 本地 AI 客户端对接 Ollama APIAI 模块选择“本地优先”方案通过 Ollama 的/api/chat接口发起对话。它返回的是 OpenAI 兼容结构但我们只需要读取message.content。对于开发者来说这套接口非常适合搭建原型的第二个原因是不需要额外鉴权Ollama 默认绑定在127.0.0.1:11434单机调试非常顺手。# p2p_chat/ai_client.py from __future__ import annotations import requests class LocalAIClient: 调用本地 Ollama 服务的 AI 客户端。 def __init__( self, base_url: str http://127.0.0.1:11434, model: str qwen2.5:7b, timeout: int 180, ): self.base_url base_url self.model model self.timeout timeout def chat( self, user_message: str, history: list[dict] | None None, ) - str: messages [] for item in history or []: messages.append({role: item[role], content: item[content]}) messages.append({role: user, content: user_message}) payload { model: self.model, messages: messages, stream: False, } resp requests.post( f{self.base_url}/api/chat, jsonpayload, timeoutself.timeout, ) resp.raise_for_status() data resp.json() return data[message][content]代码中保留了history参数位便于后续接入多轮对话当前演示只传入单条用户消息。Ollama 的模型名需要提前确认是否已下载可通过ollama pull qwen2.5:7b拉取。4.3 加密存储模块先加密再落盘本地数据库不直接保存明文对话。这里使用 Fernet 对称加密密钥由用户口令加随机盐通过 PBKDF2 派生。这样做能防止其他人直接打开 SQLite 文件就看到聊天内容但数据库结构里仍然保留会话 ID、发送者、时间等必要查询字段。# p2p_chat/store.py from __future__ import annotations import sqlite3 import time from base64 import urlsafe_b64encode from pathlib import Path from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC class SecretBox: 使用口令派生加密密钥用于本地数据库内容字段以及对端同步。 def __init__(self, password: str, salt: bytes): kdf PBKDF2HMAC( algorithmhashes.SHA256(), length32, saltsalt, iterations200_000, ) key urlsafe_b64encode(kdf.derive(password.encode(utf-8))) self._fernet Fernet(key) def encrypt_text(self, text: str) - str: return self._fernet.encrypt(text.encode(utf-8)).decode(utf-8) def decrypt_text(self, token: str) - str: return self._fernet.decrypt(token.encode(utf-8)).decode(utf-8) class ChatStore: def __init__(self, db_path: str | Path, secret_box: SecretBox): self._conn sqlite3.connect(str(db_path)) self._secret_box secret_box self._init_schema() def _init_schema(self) - None: self._conn.execute( CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, sender TEXT NOT NULL, role TEXT NOT NULL, cipher_text TEXT NOT NULL, created_at INTEGER NOT NULL ) ) self._conn.execute( CREATE INDEX IF NOT EXISTS idx_conversation_time ON conversations(conversation_id, created_at) ) self._conn.commit() def append_message( self, conversation_id: str, sender: str, role: str, content: str, ) - None: cipher_text self._secret_box.encrypt_text(content) self._conn.execute( INSERT INTO conversations ( conversation_id, sender, role, cipher_text, created_at ) VALUES (?, ?, ?, ?, ?) , (conversation_id, sender, role, cipher_text, int(time.time())), ) self._conn.commit() def list_messages(self, conversation_id: str) - list[dict]: rows self._conn.execute( SELECT sender, role, cipher_text, created_at FROM conversations WHERE conversation_id ? ORDER BY created_at ASC, id ASC , (conversation_id,), ).fetchall() messages [] for sender, role, cipher_text, created_at in rows: messages.append( { sender: sender, role: role, content: self._secret_box.decrypt_text(cipher_text), created_at: created_at, } ) return messagesSecretBox是两端都能理解的加解密容器。如果两台设备使用相同的口令和盐值那么它们就能交换密文并在本地解密。严格的 P2P 身份认证还需要检查对端公钥而不是只依赖一个共享口令但共享口令模式足以演示最小闭环。4.4 同步模块发送端如何把消息推给对端这一步我们把加密后的 JSON 文本通过 TCP 推送到对端。发送端不关心接收端使用什么数据库只负责把“一条密文记录”发出。# p2p_chat/sync.py from __future__ import annotations import json import socket from p2p_chat.store import SecretBox class PeerSyncClient: 向对端设备推送加密消息的轻量客户端。 def __init__(self, host: str, port: int, secret_box: SecretBox): self.host host self.port port self.secret_box secret_box def push_message( self, conversation_id: str, sender: str, role: str, content: str, ) - None: raw json.dumps( { conversation_id: conversation_id, sender: sender, role: role, content: content, }, ensure_asciiFalse, ) encrypted_line self.secret_box.encrypt_text(raw) self._send_line(encrypted_line) def _send_line(self, line: str) - None: payload line.encode(utf-8) with socket.create_connection((self.host, self.port), timeout10) as sock: sock.sendall(len(payload).to_bytes(4, big)) sock.sendall(payload)同步消息使用 4 字节长度前缀说明后续内容长度避免 TCP 粘包导致接收端无法判断消息边界。如果未来要实现公网 P2P这段传输层可以替换为 WebRTC DataChannel、QUIC 或 libp2p 流但消息格式本身不需要变动。4.5 接收端监听端口并写入本地数据库接收端创建一个小型常驻服务监听 TCP 端口。每次收到密文后用SecretBox解密、解析 JSON、写入自己本地的 SQLite。由于这台接收机也被称为“另一台用户设备”它的身份和数据同样完全归用户所有。# sync_server.py import argparse import json import socketserver from pathlib import Path from p2p_chat.store import ChatStore, SecretBox class SyncHandler(socketserver.BaseRequestHandler): def handle(self): length int.from_bytes(self.request.recv(4), big) data b while len(data) length: chunk self.request.recv(length - len(data)) if not chunk: break data chunk encrypted_line data.decode(utf-8) secret_box self.server.secret_box store self.server.store plaintext secret_box.decrypt_text(encrypted_line) record json.loads(plaintext) store.append_message( conversation_idrecord[conversation_id], senderrecord[sender], rolerecord[role], contentrecord[content], ) print(fsynced: {record[role]} - {record[sender]}) class SyncServer(socketserver.ThreadingTCPServer): allow_reuse_address True def __init__(self, address, store: ChatStore, secret_box: SecretBox): super().__init__(address, SyncHandler) self.store store self.secret_box secret_box def main(): parser argparse.ArgumentParser() parser.add_argument(--host, default0.0.0.0) parser.add_argument(--port, typeint, default9000) parser.add_argument(--db, defaultpeer_b.db) parser.add_argument(--password, requiredTrue) parser.add_argument(--salt, defaultpearpie-demo-salt) args parser.parse_args() secret_box SecretBox(args.password, args.salt.encode(utf-8)) store ChatStore(Path(args.db), secret_box) server SyncServer((args.host, args.port), store, secret_box) print(fsync server listening on {args.host}:{args.port}) server.serve_forever() if __name__ __main__: main()需要注意secret_box和store被挂在server对象上因此SyncHandler内可以通过self.server.store访问到同一个数据库连接。4.6 发送端入口把提问、回答、同步串联起来发送端是设备 A 的主程序它负责加载本地身份、保存用户提问、调用 AI、保存回答并可选地推送给设备 B。# main.py import argparse from pathlib import Path from p2p_chat.ai_client import LocalAIClient from p2p_chat.identity import PeerIdentity from p2p_chat.store import ChatStore, SecretBox from p2p_chat.sync import PeerSyncClient def load_or_create_identity(path: Path) - PeerIdentity: if path.exists(): lines path.read_text(encodingutf-8).strip(). 免账户P2P AI聊天架构解析：Python实现加密同步最小原型
看到 PearPie 这类“私密 AI 会话 P2P 同步 无账户”方向时很多开发者第一反应是这不就是把聊天记录从中心服务器搬走吗其实没那么简单。传统 AI 聊天应用里“账户”不仅承担登录功能还负责云同步、订阅计费、内容合规审查。一旦去掉账户身份怎么建立、AI 服务如何鉴权、多设备如何保持会话一致全部需要重新设计。这篇文章会从该项目解决的痛点出发拆解免账户 P2P AI 聊天背后的核心架构并给出一套可在本地运行的最小 Python 原型帮助你理解隐私链路和同步机制。1. 为什么私密 AI 聊天需要重新设计数据链路1.1 传统 AI 聊天应用的数据经过哪些环节平时使用 ChatGPT、Claude、各类国产大模型 App 时用户输入问题到看见回复完整的链路大致是客户端把文本发送到厂商 API 网关网关完成鉴权、限流、内容安全过滤再把请求转发给模型推理服务推理结果按原路返回而中间产生的对话记录通常会被保存到厂商数据库中用来做问题排查、模型调优或者用户体验改进。在中心化架构下这种方式效率高、可运营性好但隐私上存在两个很难回避的问题。第一是数据集中存储一旦服务端被攻击、内部人员越权访问或者账号被找回历史聊天内容就容易泄露。第二是用户身份与行为强绑定手机号、邮箱、支付信息与聊天内容在同一个账号体系下长期关联用户很难选择“只使用产品但不留下身份痕迹”。1.2 PearPie 标题中的解题方向是什么PearPie 的定位可以拆成三个关键词Private、AI chat、peer-to-peer no accounts。用更直白的话来说它希望做到用户不需要注册账号就能和 AI 对话聊天记录不只是存在某一家厂商的服务器上而是通过 P2P 方式在用户自己的设备之间同步。“无账户”并不等于“无身份”。没有账号体系后设备本身可以成为身份的最小单位。每台设备在初始化时生成一对属于用户自己的密钥公钥用于标识设备私钥保存在本地用来证明“我是这台设备的主人”。对等设备之间相互交换公钥就建立了一个不依赖注册系统的信任关系这种模型的另外一个好处是迁移性更强只要导出自己的密钥和数据可以随时换到另一套服务端实现中不被厂商锁定。1.3 它能解决什么又不能解决什么这类架构非常适合两类场景一是隐私敏感型的个人知识库、日记式 AI 助手用户不希望 AI 厂商把历史话题和自身真实身份关联二是同一用户拥有多个私人设备希望会话能轻量同步又不想把所有历史发送到云盘或者厂商服务器。不过也要谨慎看待边界。P2P 同步只解决“消息不走中心服务器”这一问题并不天然等同于合规与安全。如果调用的是第三方云端大模型 API用户输入仍然会到达模型服务商如果自建大模型也需要遵守模型开源许可和当地法律法规。同时去中心化也意味着“数据自持”和“丢失自担”并存密钥一旦丢失密文就永远无法恢复。2. 私密 P2P AI 聊天的总体架构设计2.1 四个核心模块拆解从工程实现上看PearPie 这类工具可以划分为四个互相独立、接口清晰的模块。第一是身份模块。设备本地生成长期密钥对并派生出可读 Peer ID。第二是 AI 模块负责把用户消息发送给本地模型、自建推理服务或私有大模型网关。第三是存储模块所有会话内容需要在落盘前加密并保留必要的索引字段用于快速展示会话列表。第四是同步模块在两台或多台设备之间建立连接把增量消息加密传输到对端再通过对端解密后写进本地库。模块化设计有一个明显好处AI 服务商可以替换存储引擎可以替换同步协议也可以替换只要各模块之间的数据格式保持稳定就不会影响整体功能。2.2 没有账号怎么证明“你是你”在中心化系统里密码、验证码、第三方 OAuth 都是账号体系的一部分。在去中心化系统中通常的做法是使用非对称加密密钥作为“根身份”。设备第一次启动时客户端生成一对 Ed25519 或 X25519 密钥。公钥经过哈希和 Base64 编码后可以生成短 ID显示在界面中私钥则被存进系统安全区域或加密数据库中。同步时设备 A 如果要连接设备 B双方可以先交换公钥后续通信通过临时会话密钥加密消息接收方可以验证消息是否由持有对应私钥的设备发出。这解决了“注册”的问题但也引入了更严格的要求用户必须自己对私钥负责。开发者通常会在首次生成身份时给用户一个恢复助记词或导出文件否则设备损坏或应用被卸载所有只能由该密钥解密的会话将永久消失。2.3 同步模块要解决什么问题两台设备各自独立运行还会同时写入新会话如何保证两边数据最终一致是同步的关键。在设计上比较常见的是把数据库中的某条消息当成不可变日志。每次插入新消息时都分配一个全局可比较的逻辑时间戳或向量时钟同步时只交换各自增量通过最新版本比对让两端补齐缺失记录。P2P 同步并不要求两台设备同时在线离线设备重新联网后可以主动向在线节点拉取增量数据。对比传统的服务端同步P2P 方式把“唯一权威数据源”从中心数据库改成了用户设备集群本身。这样数据冗余度可能下降但对隐私保护更有利因为没有第三方长期持有全量副本。2.4 AI 能力接入的三种模式私密 AI 聊天项目对“AI 模型跑在哪里”通常有三种不同选择。第一种是本地推理模式通过 llama.cpp、Ollama、MLC 等方案运行开源模型。这种模式隐私最强所有输入输出都不离开设备或局域网但受机器性能限制适合中低参数量模型。第二种是个人 API 模式由用户自己在配置中填入大模型厂商的 API Key聊天内容从本机直接发送到模型网关产品方无法拦截或存储当然用户的数据是否被模型厂商留存要取决于该厂商的服务条款。第三种是企业代理模式企业内部部署统一网关对内提供 OpenAI 兼容接口既保留审计能力又不让业务数据直接进入公网模型。对 PearPie 这类产品而言“本地模型优先、API Key 可插拔”是比较稳妥的设计。用户有隐私需求时使用本地模型追求高智能时再显式配置远端模型系统必须在 UI 上标注当前请求的最终流向。3. 搭建一个最小可运行原型3.1 适合阅读本节的前提如果你已经理解 P2P 和端到端加密的概念可以直接跳到第 4 节看代码。这一节先从零准备环境方便新手也能完整复现。原型不需要手机 App也不需要公网服务器。我会用 Python 写多个模块演示“设备 A 生成本地身份 - 调用本地大模型 - 把对话加密写入 SQLite - 把同一份会话通过 TCP 推送给同一局域网内的设备 B”这条完整链路。真实生产环境里的 NAT 打洞、中继、协议握手等复杂点我会在对应位置提示扩展方向但不会在这份最小代码里强行实现。3.2 环境准备与版本建议建议使用 Python 3.10 及以上版本因为部分类型注解写法依赖新语法。程序主要用到cryptography库做本地身份和对称加密使用requests库调用本地 Ollama HTTP API用 Python 标准库实现轻量 TCP 同步服务。Ollama 需要在本机单独安装并运行建议先用社区成熟的qwen2.5:7b等开放模型做测试。需要说明的是模型版本变化很快本文不会写死某个具体版本号而是把模型名作为参数传入。下面安装核心 Python 依赖pip install cryptography requests3.3 演示项目目录结构为了让教程清晰我会将模块按职责拆分。你可以先在工作目录里建立如下结构pearpie_demo/ ├── main.py # 设备 A 入口提问、保存、推送 ├── sync_server.py # 设备 B 入口接收并写入加密消息 ├── p2p_chat/ │ ├── __init__.py │ ├── identity.py # 本地身份Ed25519 密钥对 Peer ID │ ├── store.py # 加密会话存储 │ ├── ai_client.py # 本地 Ollama AI 客户端 │ └── sync.py # 发送端 P2P 客户端4. 核心模块实现与解释4.1 身份模块生成本地密钥与 Peer ID第一个模块解决“我是谁”。使用 Ed25519 生成签名密钥对私钥留在设备 A公钥派生为短标识。为了便于演示这里直接把私钥以十六进制文本保存在普通文件里在生产环境中私钥应该写入操作系统 Keychain、Keystore 或安全加密容器。# p2p_chat/identity.py from __future__ import annotations import base64 import hashlib from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) class PeerIdentity: 本地生成的 Peer 身份私钥不离开设备。 def __init__(self, private_raw: bytes, public_raw: bytes): self._private_key Ed25519PrivateKey.from_private_bytes(private_raw) self._public_key Ed25519PublicKey.from_public_bytes(public_raw) classmethod def create(cls) - PeerIdentity: private_key Ed25519PrivateKey.generate() private_raw private_key.private_bytes( encodingserialization.Encoding.Raw, formatserialization.PrivateFormat.Raw, encryption_algorithmserialization.NoEncryption(), ) public_raw private_key.public_key().public_bytes( encodingserialization.Encoding.Raw, formatserialization.PublicFormat.Raw, ) return cls(private_raw, public_raw) property def private_key_hex(self) - str: return self._private_key.private_bytes( encodingserialization.Encoding.Raw, formatserialization.PrivateFormat.Raw, encryption_algorithmserialization.NoEncryption(), ).hex() property def public_key_hex(self) - str: return self._public_key.public_bytes( encodingserialization.Encoding.Raw, formatserialization.PublicFormat.Raw, ).hex() property def peer_id(self) - str: digest hashlib.sha256( bytes.fromhex(self.public_key_hex) ).digest()[:16] return base64.urlsafe_b64encode(digest).rstrip(b).decode()简要说明PeerIdentity.create()可以生成本设备身份private_key_hex和public_key_hex用于保存和传输peer_id是公钥哈希的前 16 字节再做 Base64 编码方便日志与 UI 展示。4.2 本地 AI 客户端对接 Ollama APIAI 模块选择“本地优先”方案通过 Ollama 的/api/chat接口发起对话。它返回的是 OpenAI 兼容结构但我们只需要读取message.content。对于开发者来说这套接口非常适合搭建原型的第二个原因是不需要额外鉴权Ollama 默认绑定在127.0.0.1:11434单机调试非常顺手。# p2p_chat/ai_client.py from __future__ import annotations import requests class LocalAIClient: 调用本地 Ollama 服务的 AI 客户端。 def __init__( self, base_url: str http://127.0.0.1:11434, model: str qwen2.5:7b, timeout: int 180, ): self.base_url base_url self.model model self.timeout timeout def chat( self, user_message: str, history: list[dict] | None None, ) - str: messages [] for item in history or []: messages.append({role: item[role], content: item[content]}) messages.append({role: user, content: user_message}) payload { model: self.model, messages: messages, stream: False, } resp requests.post( f{self.base_url}/api/chat, jsonpayload, timeoutself.timeout, ) resp.raise_for_status() data resp.json() return data[message][content]代码中保留了history参数位便于后续接入多轮对话当前演示只传入单条用户消息。Ollama 的模型名需要提前确认是否已下载可通过ollama pull qwen2.5:7b拉取。4.3 加密存储模块先加密再落盘本地数据库不直接保存明文对话。这里使用 Fernet 对称加密密钥由用户口令加随机盐通过 PBKDF2 派生。这样做能防止其他人直接打开 SQLite 文件就看到聊天内容但数据库结构里仍然保留会话 ID、发送者、时间等必要查询字段。# p2p_chat/store.py from __future__ import annotations import sqlite3 import time from base64 import urlsafe_b64encode from pathlib import Path from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC class SecretBox: 使用口令派生加密密钥用于本地数据库内容字段以及对端同步。 def __init__(self, password: str, salt: bytes): kdf PBKDF2HMAC( algorithmhashes.SHA256(), length32, saltsalt, iterations200_000, ) key urlsafe_b64encode(kdf.derive(password.encode(utf-8))) self._fernet Fernet(key) def encrypt_text(self, text: str) - str: return self._fernet.encrypt(text.encode(utf-8)).decode(utf-8) def decrypt_text(self, token: str) - str: return self._fernet.decrypt(token.encode(utf-8)).decode(utf-8) class ChatStore: def __init__(self, db_path: str | Path, secret_box: SecretBox): self._conn sqlite3.connect(str(db_path)) self._secret_box secret_box self._init_schema() def _init_schema(self) - None: self._conn.execute( CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, sender TEXT NOT NULL, role TEXT NOT NULL, cipher_text TEXT NOT NULL, created_at INTEGER NOT NULL ) ) self._conn.execute( CREATE INDEX IF NOT EXISTS idx_conversation_time ON conversations(conversation_id, created_at) ) self._conn.commit() def append_message( self, conversation_id: str, sender: str, role: str, content: str, ) - None: cipher_text self._secret_box.encrypt_text(content) self._conn.execute( INSERT INTO conversations ( conversation_id, sender, role, cipher_text, created_at ) VALUES (?, ?, ?, ?, ?) , (conversation_id, sender, role, cipher_text, int(time.time())), ) self._conn.commit() def list_messages(self, conversation_id: str) - list[dict]: rows self._conn.execute( SELECT sender, role, cipher_text, created_at FROM conversations WHERE conversation_id ? ORDER BY created_at ASC, id ASC , (conversation_id,), ).fetchall() messages [] for sender, role, cipher_text, created_at in rows: messages.append( { sender: sender, role: role, content: self._secret_box.decrypt_text(cipher_text), created_at: created_at, } ) return messagesSecretBox是两端都能理解的加解密容器。如果两台设备使用相同的口令和盐值那么它们就能交换密文并在本地解密。严格的 P2P 身份认证还需要检查对端公钥而不是只依赖一个共享口令但共享口令模式足以演示最小闭环。4.4 同步模块发送端如何把消息推给对端这一步我们把加密后的 JSON 文本通过 TCP 推送到对端。发送端不关心接收端使用什么数据库只负责把“一条密文记录”发出。# p2p_chat/sync.py from __future__ import annotations import json import socket from p2p_chat.store import SecretBox class PeerSyncClient: 向对端设备推送加密消息的轻量客户端。 def __init__(self, host: str, port: int, secret_box: SecretBox): self.host host self.port port self.secret_box secret_box def push_message( self, conversation_id: str, sender: str, role: str, content: str, ) - None: raw json.dumps( { conversation_id: conversation_id, sender: sender, role: role, content: content, }, ensure_asciiFalse, ) encrypted_line self.secret_box.encrypt_text(raw) self._send_line(encrypted_line) def _send_line(self, line: str) - None: payload line.encode(utf-8) with socket.create_connection((self.host, self.port), timeout10) as sock: sock.sendall(len(payload).to_bytes(4, big)) sock.sendall(payload)同步消息使用 4 字节长度前缀说明后续内容长度避免 TCP 粘包导致接收端无法判断消息边界。如果未来要实现公网 P2P这段传输层可以替换为 WebRTC DataChannel、QUIC 或 libp2p 流但消息格式本身不需要变动。4.5 接收端监听端口并写入本地数据库接收端创建一个小型常驻服务监听 TCP 端口。每次收到密文后用SecretBox解密、解析 JSON、写入自己本地的 SQLite。由于这台接收机也被称为“另一台用户设备”它的身份和数据同样完全归用户所有。# sync_server.py import argparse import json import socketserver from pathlib import Path from p2p_chat.store import ChatStore, SecretBox class SyncHandler(socketserver.BaseRequestHandler): def handle(self): length int.from_bytes(self.request.recv(4), big) data b while len(data) length: chunk self.request.recv(length - len(data)) if not chunk: break data chunk encrypted_line data.decode(utf-8) secret_box self.server.secret_box store self.server.store plaintext secret_box.decrypt_text(encrypted_line) record json.loads(plaintext) store.append_message( conversation_idrecord[conversation_id], senderrecord[sender], rolerecord[role], contentrecord[content], ) print(fsynced: {record[role]} - {record[sender]}) class SyncServer(socketserver.ThreadingTCPServer): allow_reuse_address True def __init__(self, address, store: ChatStore, secret_box: SecretBox): super().__init__(address, SyncHandler) self.store store self.secret_box secret_box def main(): parser argparse.ArgumentParser() parser.add_argument(--host, default0.0.0.0) parser.add_argument(--port, typeint, default9000) parser.add_argument(--db, defaultpeer_b.db) parser.add_argument(--password, requiredTrue) parser.add_argument(--salt, defaultpearpie-demo-salt) args parser.parse_args() secret_box SecretBox(args.password, args.salt.encode(utf-8)) store ChatStore(Path(args.db), secret_box) server SyncServer((args.host, args.port), store, secret_box) print(fsync server listening on {args.host}:{args.port}) server.serve_forever() if __name__ __main__: main()需要注意secret_box和store被挂在server对象上因此SyncHandler内可以通过self.server.store访问到同一个数据库连接。4.6 发送端入口把提问、回答、同步串联起来发送端是设备 A 的主程序它负责加载本地身份、保存用户提问、调用 AI、保存回答并可选地推送给设备 B。# main.py import argparse from pathlib import Path from p2p_chat.ai_client import LocalAIClient from p2p_chat.identity import PeerIdentity from p2p_chat.store import ChatStore, SecretBox from p2p_chat.sync import PeerSyncClient def load_or_create_identity(path: Path) - PeerIdentity: if path.exists(): lines path.read_text(encodingutf-8).strip().