看到 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( encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption(), ) public_raw = private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) return cls(private_raw, public_raw) @property def private_key_hex(self) -> str: return self._private_key.private_bytes( encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption(), ).hex() @property def public_key_hex(self) -> str: return self._public_key.public_bytes( encoding=serialization.Encoding.Raw, format=serialization.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 API
AI 模块选择“本地优先”方案，通过 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", json=payload, timeout=self.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( algorithm=hashes.SHA256(), length=32, salt=salt, iterations=200_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_ascii=False, ) 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), timeout=10) 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_id=record["conversation_id"], sender=record["sender"], role=record["role"], content=record["content"], ) print(f"synced: {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", default="0.0.0.0") parser.add_argument("--port", type=int, default=9000) parser.add_argument("--db", default="peer_b.db") parser.add_argument("--password", required=True) parser.add_argument("--salt", default="pearpie-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(f"sync 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(encoding="utf-8").strip().
