MSG Chain AI Agent 数据持久化与状态管理指南
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
适用链:
msg-chain-1| 地址前缀:msg| 版本: 1.0.0
目录
1. 概述
1.1 为什么状态管理对 AI Agent 至关重要
AI Agent 在 MSG Chain 上运行时，其行为、记忆和资产都依赖于持久化的状态数据。与无状态的传统智能合约不同，AI Agent 需要跨越多个会话、多次执行、甚至多个生命周期维护一致的内部状态。状态管理的可靠性直接决定了 Agent 的行为一致性、用户体验和资产安全。
如果没有健壮的状态管理，Agent 会在重启后丢失对话记忆、忘记用户偏好、无法从中断的工作流中恢复，甚至可能导致资产错乱。在 MSG Chain 上使用 msg 前缀地址的 Agent 尤其需要精心设计状态持久化方案。
1.2 状态类型分类
| 状态类型 | 描述 | 存储位置 | 更新频率 | 数据量 |
|---|---|---|---|---|
| Memory (记忆) | 对话历史、用户交互记录 | IPFS / 本地DB | 高 | 大 |
| Config (配置) | Agent 参数、权限、偏好设置 | 链上 / 本地 | 低 | 小 |
| Session (会话) | 活跃会话上下文、临时数据 | 内存 / Redis | 极高 | 中 |
| Results (结果) | 任务执行结果、推理输出 | IPFS / 本地 | 中 | 大 |
1.3 存储金字塔
AI Agent 的数据存储在多层架构中，根据访问频率、数据量大小和持久性需求分层存放：
┌──────────────────────┐
│ Redis / 内存 │ ← 热数据: 活跃会话、实时状态
│ (μs 级访问) │
├──────────────────────┤
│ SQLite / LevelDB │ ← 温数据: Agent 配置、本地缓存
│ (ms 级访问) │
├──────────────────────┤
│ IPFS / Arweave │ ← 冷数据: 对话归档、检查点
│ (s 级访问) │
├──────────────────────┤
│ MSG Chain 链上 │ ← 核心数据: 所有权、信誉、锚点
│ (tx 级延迟) │
└──────────────────────┘
设计原则:
- 链上只存最精简的核心状态和验证信息
- 大数据体通过 IPFS CID 引用，链上只存哈希
- 热数据优先走缓存层，减少链上查询
- 所有存储层通过校验和保证完整性
1.4 状态管理的关键指标
- 一致性: 跨存储层的数据不冲突，最终一致性能被接受
- 可恢复性: 从任一检查点可完整恢复 Agent 状态
- 版本兼容: 状态 schema 支持向前/向后迁移
- 成本控制: 链上存储成本高，需严格控制存储量
2. 链上状态管理模式
2.1 MSG Chain 智能合约状态结构
在 MSG Chain 上，每个 AI Agent 的核心状态存储在 CosmWasm 智能合约中。以下是在 msg 前缀地址上运行的标准状态定义：
use cosmwasm_std::{
Addr, DepsMut, Deps, StdResult, StdError, Storage,
Uint128, Response, MessageInfo, Env,
};
use cw_storage_plus::{Item, Map, SnapshotMap};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
/// Agent 核心链上状态
/// 链上只存储最小必要字段，大量数据通过 metadata_uri 引用
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct AgentState {
/// Agent 所有者地址 (msg 前缀)
pub owner: Addr,
/// Agent 宪法/行为准则哈希 (IPFS CID)
pub constitution_hash: String,
/// Agent 能力列表
pub capabilities: Vec<String>,
/// Agent 信誉分数
pub reputation: u64,
/// Agent 累计收入 (MSG 原生代币)
pub total_earnings: Uint128,
/// 最后心跳时间戳 (Unix 秒)
pub last_heartbeat: u64,
/// Agent 详细状态元数据 URI (IPFS 链接)
pub metadata_uri: String,
/// 当前状态 schema 版本号
pub version: u8,
/// Agent 是否激活
pub active: bool,
/// Agent 创建时间
pub created_at: u64,
/// 最后状态更新时间
pub updated_at: u64,
}
/// Agent 配置参数
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct AgentConfig {
pub name: String,
pub description: String,
pub model_endpoint: String,
pub model_name: String,
pub temperature: f64,
pub max_tokens: u32,
pub session_timeout: u64,
pub allowed_tools: Vec<String>,
}
/// Agent 权限控制
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct AgentPermissions {
pub admins: Vec<Addr>,
pub operators: Vec<Addr>,
pub public_allowed: bool,
pub gas_limit: Uint128,
pub max_fee_per_call: Uint128,
}
/// 链上存储键定义
pub const STATE_KEY: Item<AgentState> = Item::new("agent_state");
pub const CONFIG_KEY: Item<AgentConfig> = Item::new("agent_config");
pub const PERMISSIONS_KEY: Item<AgentPermissions> = Item::new("agent_permissions");
pub const SESSION_CHECKPOINTS: Map<&str, String> = Map::new("session_cp");
pub const BACKUP_RECORDS: Map<&str, String> = Map::new("backup_records");
pub const REPUTATION_HISTORY: Map<u64, u64> = Map::new("reputation_hist");
2.2 链上/链下状态拆分策略
决定哪些数据上链、哪些数据存储在链下，是状态管理架构设计的最关键决策。
必须上链的数据:
- 所有权和权限控制
- 资产余额和代币信息
- 信誉分数和声誉记录
- 关键状态哈希和 IPFS CID 锚点
- 合约配置和版本信息
应留在链下的数据:
- 对话历史和消息记录
- 大型知识库和训练数据
- 日志和调试信息
- 完整的检查点快照
- 临时会话上下文
/// 链上/链下拆分决策辅助函数
pub fn should_store_on_chain(field_name: &str, data_size: usize) -> bool {
match field_name {
"owner" | "admins" | "operators" => true,
"total_earnings" | "gas_limit" | "max_fee_per_call" => true,
"reputation" | "last_heartbeat" => true,
"name" | "description" if data_size < 256 => true,
"metadata_uri" | "constitution_hash" => true,
_ => data_size < 1024,
}
}
2.3 状态 Schema 版本化
随着 Agent 功能的演进，状态结构需要升级。Schema 版本化机制确保新旧代码兼容：
pub const STATE_VERSION_V1: u8 = 1;
pub const STATE_VERSION_V2: u8 = 2;
pub const STATE_VERSION_V3: u8 = 3;
pub const CURRENT_STATE_VERSION: u8 = STATE_VERSION_V3;
/// 通用状态迁移入口
pub fn migrate_state(
deps: DepsMut,
from_version: u8,
to_version: u8,
) -> StdResult<()> {
if from_version >= to_version {
return Err(StdError::generic_err(
format!("invalid migration: {} -> {}", from_version, to_version)
));
}
match (from_version, to_version) {
(1, 2) => migrate_v1_to_v2(deps),
(2, 3) => migrate_v2_to_v3(deps),
(1, 3) => {
migrate_v1_to_v2(deps)?;
migrate_v2_to_v3(deps)
}
_ => Err(StdError::generic_err(
format!("unsupported migration path: {} -> {}", from_version, to_version)
)),
}
}
/// V1 -> V2 迁移: 添加 active 和 reputation 字段
fn migrate_v1_to_v2(deps: DepsMut) -> StdResult<()> {
let mut state = STATE_KEY.load(deps.storage)?;
state.active = true;
state.reputation = 0;
state.updated_at = deps.block.time.seconds();
state.version = STATE_VERSION_V2;
STATE_KEY.save(deps.storage, &state)?;
Ok(())
}
/// V2 -> V3 迁移: 添加权限分离
fn migrate_v2_to_v3(deps: DepsMut) -> StdResult<()> {
let state = STATE_KEY.load(deps.storage)?;
let permissions = AgentPermissions {
admins: vec![state.owner.clone()],
operators: vec![],
public_allowed: false,
gas_limit: Uint128::from(100_000u128),
max_fee_per_call: Uint128::from(1_000u128),
};
PERMISSIONS_KEY.save(deps.storage, &permissions)?;
let mut new_state = state;
new_state.version = STATE_VERSION_V3;
new_state.updated_at = deps.block.time.seconds();
STATE_KEY.save(deps.storage, &new_state)?;
Ok(())
}
/// 检查是否需要迁移并执行
pub fn ensure_state_migrated(deps: DepsMut) -> StdResult<()> {
let state = STATE_KEY.load(deps.storage)?;
if state.version < CURRENT_STATE_VERSION {
migrate_state(deps, state.version, CURRENT_STATE_VERSION)?;
}
Ok(())
}
2.4 链上状态读写操作
/// 初始化新 Agent 状态
pub fn instantiate_agent(
deps: DepsMut,
env: Env,
info: MessageInfo,
config: AgentConfig,
constitution_cid: String,
capabilities: Vec<String>,
) -> StdResult<Response> {
let state = AgentState {
owner: info.sender.clone(),
constitution_hash: constitution_cid,
capabilities,
reputation: 0,
total_earnings: Uint128::zero(),
last_heartbeat: env.block.time.seconds(),
metadata_uri: String::new(),
version: CURRENT_STATE_VERSION,
active: true,
created_at: env.block.time.seconds(),
updated_at: env.block.time.seconds(),
};
STATE_KEY.save(deps.storage, &state)?;
CONFIG_KEY.save(deps.storage, &config)?;
let permissions = AgentPermissions {
admins: vec![info.sender.clone()],
operators: vec![],
public_allowed: false,
gas_limit: Uint128::from(100_000u128),
max_fee_per_call: Uint128::from(1_000u128),
};
PERMISSIONS_KEY.save(deps.storage, &permissions)?;
Ok(Response::new()
.add_attribute("action", "instantiate")
.add_attribute("agent", info.sender.to_string())
.add_attribute("version", CURRENT_STATE_VERSION.to_string()))
}
/// 更新 Agent 心跳
pub fn update_heartbeat(deps: DepsMut, env: Env) -> StdResult<Response> {
STATE_KEY.update(deps.storage, |mut state| -> StdResult<AgentState> {
state.last_heartbeat = env.block.time.seconds();
state.updated_at = env.block.time.seconds();
Ok(state)
})?;
Ok(Response::new()
.add_attribute("action", "heartbeat")
.add_attribute("time", env.block.time.seconds().to_string()))
}
/// 更新 metadata_uri (IPFS CID)
pub fn update_metadata_uri(
deps: DepsMut,
env: Env,
info: MessageInfo,
new_uri: String,
) -> StdResult<Response> {
let permissions = PERMISSIONS_KEY.load(deps.storage)?;
if !permissions.admins.contains(&info.sender) {
return Err(StdError::generic_err("unauthorized: only admins can update metadata"));
}
STATE_KEY.update(deps.storage, |mut state| -> StdResult<AgentState> {
state.metadata_uri = new_uri.clone();
state.updated_at = env.block.time.seconds();
Ok(state)
})?;
Ok(Response::new()
.add_attribute("action", "update_metadata")
.add_attribute("uri", new_uri))
}
/// 增加信誉分数
pub fn add_reputation(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: u64,
) -> StdResult<Response> {
let permissions = PERMISSIONS_KEY.load(deps.storage)?;
if !permissions.admins.contains(&info.sender) {
return Err(StdError::generic_err("unauthorized"));
}
STATE_KEY.update(deps.storage, |mut state| -> StdResult<AgentState> {
state.reputation = state.reputation.saturating_add(amount);
state.updated_at = env.block.time.seconds();
Ok(state)
})?;
let current_rep = STATE_KEY.load(deps.storage)?.reputation;
REPUTATION_HISTORY.save(deps.storage, env.block.time.seconds(), ¤t_rep)?;
Ok(Response::new()
.add_attribute("action", "add_reputation")
.add_attribute("amount", amount.to_string())
.add_attribute("total", current_rep.to_string()))
}
2.5 链上状态查询
pub fn query_agent_state(deps: Deps) -> StdResult<AgentState> {
STATE_KEY.load(deps.storage)
}
pub fn query_agent_config(deps: Deps) -> StdResult<AgentConfig> {
CONFIG_KEY.load(deps.storage)
}
pub fn query_agent_permissions(deps: Deps) -> StdResult<AgentPermissions> {
PERMISSIONS_KEY.load(deps.storage)
}
pub fn query_reputation_history(
deps: Deps,
start_after: Option<u64>,
limit: usize,
) -> StdResult<Vec<(u64, u64)>> {
let limit = limit.min(30).max(1);
REPUTATION_HISTORY
.range(deps.storage, None, None, cosmwasm_std::Order::Descending)
.take(limit)
.collect::<StdResult<Vec<_>>>()
}
/// 检查 Agent 是否健康 (心跳超时检测)
pub fn is_agent_healthy(deps: Deps, env: Env, max_heartbeat_gap: u64) -> StdResult<bool> {
let state = STATE_KEY.load(deps.storage)?;
let current_time = env.block.time.seconds();
let gap = current_time.saturating_sub(state.last_heartbeat);
Ok(gap < max_heartbeat_gap && state.active)
}
2.6 Gas 优化方案
链上状态操作消耗 Gas，需要严格控制：
/// 批量更新状态以减少 Gas 消耗
pub fn batch_update(
deps: DepsMut,
env: Env,
updates: Vec<StateUpdate>,
) -> StdResult<Response> {
STATE_KEY.update(deps.storage, |mut state| -> StdResult<AgentState> {
for update in updates {
match update {
StateUpdate::Heartbeat => {
state.last_heartbeat = env.block.time.seconds();
}
StateUpdate::Reputation(amount) => {
state.reputation = state.reputation.saturating_add(amount);
}
StateUpdate::Metadata(uri) => {
state.metadata_uri = uri;
}
StateUpdate::Capabilities(caps) => {
state.capabilities = caps;
}
StateUpdate::Deactivate => {
state.active = false;
}
StateUpdate::Earnings(amount) => {
state.total_earnings = state.total_earnings.checked_add(amount)
.map_err(|_| StdError::generic_err("earnings overflow"))?;
}
}
}
state.updated_at = env.block.time.seconds();
Ok(state)
})?;
Ok(Response::new()
.add_attribute("action", "batch_update"))
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub enum StateUpdate {
Heartbeat,
Reputation(u64),
Metadata(String),
Capabilities(Vec<String>),
Deactivate,
Earnings(Uint128),
}
/// 状态压缩: 将小整数打包存储以减少存储槽
pub fn pack_flags(active: bool, public_allowed: bool, frozen: bool) -> u8 {
(active as u8) << 2 | (public_allowed as u8) << 1 | (frozen as u8)
}
pub fn unpack_flags(flags: u8) -> (bool, bool, bool) {
let active = (flags >> 2) & 1 == 1;
let public_allowed = (flags >> 1) & 1 == 1;
let frozen = flags & 1 == 1;
(active, public_allowed, frozen)
}
3. 离线数据存储
3.1 多层离线存储架构
离线存储层处理 Agent 的大规模数据，包括对话历史、知识库、日志和完整状态快照。采用多层架构以平衡访问速度和存储成本：
import json
import time
import hashlib
import asyncio
import sqlite3
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from abc import ABC, abstractmethod
logger = logging.getLogger("msg_agent.storage")
@dataclass
class Message:
role: str
content: str
timestamp: float
tool_calls: Optional[List[Dict]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ConversationRecord:
session_id: str
agent_id: str
user_id: str
messages: List[Message]
created_at: float
updated_at: float
message_count: int
total_tokens: int
class StorageBackend(ABC):
@abstractmethod
async def store(self, key: str, value: bytes) -> bool: ...
@abstractmethod
async def load(self, key: str) -> Optional[bytes]: ...
@abstractmethod
async def delete(self, key: str) -> bool: ...
@abstractmethod
async def exists(self, key: str) -> bool: ...
@abstractmethod
async def list_keys(self, prefix: str) -> List[str]: ...
class SQLiteStorage(StorageBackend):
def __init__(self, db_path: str = "/data/agent.db"):
self.db_path = db_path
self.conn: Optional[sqlite3.Connection] = None
self._init_db()
def _init_db(self):
self.conn = sqlite3.connect(self.db_path)
self.conn.execute("PRAGMA journal_mode=WAL")
self.conn.execute("PRAGMA synchronous=NORMAL")
self.conn.row_factory = sqlite3.Row
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS kv_store (
key TEXT PRIMARY KEY,
value BLOB NOT NULL,
created_at REAL NOT NULL,
updated_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS conversations (
session_id TEXT PRIMARY KEY,
agent_id TEXT NOT NULL,
user_id TEXT NOT NULL,
data TEXT NOT NULL,
message_count INTEGER DEFAULT 0,
created_at REAL NOT NULL,
updated_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS checkpoints (
id TEXT PRIMARY KEY,
agent_id TEXT NOT NULL,
cid TEXT NOT NULL,
timestamp REAL NOT NULL,
checksum TEXT NOT NULL,
state_version INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS backup_records (
id TEXT PRIMARY KEY,
agent_id TEXT NOT NULL,
cid TEXT NOT NULL,
backup_type TEXT NOT NULL,
timestamp REAL NOT NULL,
size_bytes INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_conversations_agent
ON conversations(agent_id);
CREATE INDEX IF NOT EXISTS idx_checkpoints_agent
ON checkpoints(agent_id, timestamp);
CREATE INDEX IF NOT EXISTS idx_backup_agent
ON backup_records(agent_id, timestamp);
""")
self.conn.commit()
async def store(self, key: str, value: bytes) -> bool:
now = time.time()
loop = asyncio.get_event_loop()
def _store():
self.conn.execute(
"INSERT OR REPLACE INTO kv_store (key, value, created_at, updated_at) "
"VALUES (?, ?, COALESCE((SELECT created_at FROM kv_store WHERE key=?), ?), ?)",
(key, value, key, now, now)
)
self.conn.commit()
return True
return await loop.run_in_executor(None, _store)
async def load(self, key: str) -> Optional[bytes]:
loop = asyncio.get_event_loop()
def _load():
cursor = self.conn.execute("SELECT value FROM kv_store WHERE key = ?", (key,))
row = cursor.fetchone()
return bytes(row[0]) if row else None
return await loop.run_in_executor(None, _load)
async def delete(self, key: str) -> bool:
loop = asyncio.get_event_loop()
def _delete():
self.conn.execute("DELETE FROM kv_store WHERE key = ?", (key,))
self.conn.commit()
return True
return await loop.run_in_executor(None, _delete)
async def exists(self, key: str) -> bool:
loop = asyncio.get_event_loop()
def _exists():
cursor = self.conn.execute("SELECT 1 FROM kv_store WHERE key = ?", (key,))
return cursor.fetchone() is not None
return await loop.run_in_executor(None, _exists)
async def list_keys(self, prefix: str) -> List[str]:
loop = asyncio.get_event_loop()
def _list():
cursor = self.conn.execute(
"SELECT key FROM kv_store WHERE key LIKE ?", (f"{prefix}%",)
)
return [row[0] for row in cursor.fetchall()]
return await loop.run_in_executor(None, _list)
async def save_conversation(self, record: ConversationRecord):
loop = asyncio.get_event_loop()
data_json = json.dumps(asdict(record))
def _save():
self.conn.execute(
"INSERT OR REPLACE INTO conversations "
"(session_id, agent_id, user_id, data, message_count, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(record.session_id, record.agent_id, record.user_id,
data_json, record.message_count, record.created_at, record.updated_at)
)
self.conn.commit()
return await loop.run_in_executor(None, _save)
async def load_conversation(self, session_id: str) -> Optional[ConversationRecord]:
loop = asyncio.get_event_loop()
def _load():
cursor = self.conn.execute(
"SELECT data FROM conversations WHERE session_id = ?", (session_id,)
)
row = cursor.fetchone()
if row:
return ConversationRecord(**json.loads(row[0]))
return None
return await loop.run_in_executor(None, _load)
async def list_agent_conversations(
self, agent_id: str, limit: int = 100, offset: int = 0
) -> List[ConversationRecord]:
loop = asyncio.get_event_loop()
def _list():
cursor = self.conn.execute(
"SELECT data FROM conversations "
"WHERE agent_id = ? ORDER BY updated_at DESC LIMIT ? OFFSET ?",
(agent_id, limit, offset)
)
return [ConversationRecord(**json.loads(row[0])) for row in cursor.fetchall()]
return await loop.run_in_executor(None, _list)
def close(self):
if self.conn:
self.conn.close()
3.2 IPFS 分布式存储适配器
class IPFSClient:
def __init__(self, api_url: str = "http://localhost:5001", timeout: int = 60):
self.api_url = api_url.rstrip("/")
self.timeout = timeout
async def add(self, data: bytes, pin: bool = True) -> str:
import aiohttp
form = aiohttp.FormData()
form.add_field("file", data, filename="data.bin")
if pin:
form.add_field("pin", "true")
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.api_url}/api/v0/add", data=form, timeout=self.timeout
) as resp:
if resp.status != 200:
raise IOError(f"IPFS add failed: {await resp.text()}")
result = await resp.json()
return result["Hash"]
async def get(self, cid: str) -> bytes:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.api_url}/api/v0/cat?arg={cid}", timeout=self.timeout
) as resp:
if resp.status != 200:
raise IOError(f"IPFS cat failed: {await resp.text()}")
return await resp.read()
async def pin(self, cid: str) -> bool:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_url}/api/v0/pin/add?arg={cid}") as resp:
return resp.status == 200
async def unpin(self, cid: str) -> bool:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_url}/api/v0/pin/rm?arg={cid}") as resp:
return resp.status == 200
async def is_pinned(self, cid: str) -> bool:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_url}/api/v0/pin/ls?arg={cid}") as resp:
if resp.status != 200:
return False
result = await resp.json()
return cid in result.get("Keys", {})
class RedisCache(StorageBackend):
def __init__(self, host: str = "localhost", port: int = 6379, db: int = 0):
self.host = host
self.port = port
self.db = db
self._client: Optional[Any] = None
async def _get_client(self):
if self._client is None:
import redis.asyncio as redis
self._client = redis.Redis(
host=self.host, port=self.port, db=self.db, decode_responses=False
)
return self._client
async def store(self, key: str, value: bytes, ttl: Optional[int] = None) -> bool:
client = await self._get_client()
if ttl:
return await client.setex(key, ttl, value)
return await client.set(key, value)
async def load(self, key: str) -> Optional[bytes]:
client = await self._get_client()
return await client.get(key)
async def delete(self, key: str) -> bool:
client = await self._get_client()
return bool(await client.delete(key))
async def exists(self, key: str) -> bool:
client = await self._get_client()
return bool(await client.exists(key))
async def list_keys(self, prefix: str) -> List[str]:
client = await self._get_client()
cursor = 0
keys = []
while True:
cursor, batch = await client.scan(cursor=cursor, match=f"{prefix}*", count=100)
keys.extend([k.decode() if isinstance(k, bytes) else k for k in batch])
if cursor == 0:
break
return keys
async def expire(self, key: str, ttl: int) -> bool:
client = await self._get_client()
return bool(await client.expire(key, ttl))
async def flush_prefix(self, prefix: str) -> int:
keys = await self.list_keys(prefix)
if not keys:
return 0
client = await self._get_client()
return await client.delete(*keys)
3.3 完整离线存储管理器
class OffChainStorage:
def __init__(
self,
db_path: str = "/data/agent.db",
ipfs_url: str = "http://localhost:5001",
redis_host: str = "localhost",
redis_port: int = 6379,
):
self.local = SQLiteStorage(db_path)
self.ipfs = IPFSClient(ipfs_url)
self.cache = RedisCache(redis_host, redis_port)
self._session_cache_ttl = 3600
async def store_conversation(
self, session_id: str, messages: List[Message],
agent_id: str, user_id: str,
) -> str:
recent = messages[-100:] if len(messages) > 100 else messages
cache_data = json.dumps([asdict(m) for m in recent]).encode()
await self.cache.store(
f"msg:session:{session_id}", cache_data, ttl=self._session_cache_ttl
)
full_data = json.dumps({
"session_id": session_id,
"agent_id": agent_id,
"user_id": user_id,
"messages": [asdict(m) for m in messages],
"message_count": len(messages),
"archived_at": time.time(),
"format_version": 1,
}).encode()
cid = await self.ipfs.add(full_data)
await self.ipfs.pin(cid)
record = ConversationRecord(
session_id=session_id, agent_id=agent_id, user_id=user_id,
messages=messages,
created_at=messages[0].timestamp if messages else time.time(),
updated_at=time.time(),
message_count=len(messages), total_tokens=0,
)
await self.local.save_conversation(record)
logger.info(f"Stored conversation {session_id} -> IPFS {cid}")
return cid
async def load_conversation(
self, session_id: str, prefer_cache: bool = True,
) -> Optional[List[Message]]:
if prefer_cache:
cached = await self.cache.load(f"msg:session:{session_id}")
if cached:
try:
data = json.loads(cached)
return [Message(**m) for m in data]
except (json.JSONDecodeError, TypeError):
pass
record = await self.local.load_conversation(session_id)
if record:
return record.messages
return None
async def archive_and_purge(self, session_id: str, older_than_days: int = 30) -> Optional[str]:
record = await self.local.load_conversation(session_id)
if not record:
return None
age_days = (time.time() - record.updated_at) / 86400
if age_days < older_than_days:
return None
cid = await self.store_conversation(
session_id, record.messages, record.agent_id, record.user_id
)
await self.local.delete(f"conv:{session_id}")
await self.cache.delete(f"msg:session:{session_id}")
logger.info(f"Archived old session {session_id} -> {cid}")
return cid
async def health_check(self) -> Dict[str, bool]:
results = {}
try:
await self.local.exists("health_check_key")
results["sqlite"] = True
except Exception:
results["sqlite"] = False
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.ipfs.api_url}/api/v0/id", timeout=5) as resp:
results["ipfs"] = resp.status == 200
except Exception:
results["ipfs"] = False
try:
client = await self.cache._get_client()
await client.ping()
results["redis"] = True
except Exception:
results["redis"] = False
return results
async def close(self):
self.local.close()
3.4 LevelDB 备选存储
class LevelDBStorage(StorageBackend):
def __init__(self, db_path: str = "/data/agent.ldb"):
self.db_path = db_path
self._db = None
async def _get_db(self):
if self._db is None:
import plyvel
self._db = plyvel.DB(self.db_path, create_if_missing=True)
return self._db
async def store(self, key: str, value: bytes) -> bool:
db = await self._get_db()
db.put(key.encode(), value)
return True
async def load(self, key: str) -> Optional[bytes]:
db = await self._get_db()
return db.get(key.encode())
async def delete(self, key: str) -> bool:
db = await self._get_db()
db.delete(key.encode())
return True
async def exists(self, key: str) -> bool:
db = await self._get_db()
return db.get(key.encode()) is not None
async def list_keys(self, prefix: str) -> List[str]:
db = await self._get_db()
keys = []
prefix_bytes = prefix.encode()
for key_bytes, _ in db:
if key_bytes.startswith(prefix_bytes):
keys.append(key_bytes.decode())
return keys
def close(self):
if self._db:
self._db.close()
3.5 数据序列化与压缩
import zlib
import pickle
class DataSerializer:
COMPRESSION_LEVEL = 6
@classmethod
def serialize(cls, data: Any, compress: bool = True) -> bytes:
pickled = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)
if compress and len(pickled) > 1024:
return zlib.compress(pickled, cls.COMPRESSION_LEVEL)
return pickled
@classmethod
def deserialize(cls, data: bytes) -> Any:
try:
try:
decompressed = zlib.decompress(data)
return pickle.loads(decompressed)
except zlib.error:
return pickle.loads(data)
except Exception as e:
raise ValueError(f"Deserialization failed: {e}") from e
@classmethod
def json_serialize(cls, data: Any, compress: bool = True) -> bytes:
text = json.dumps(data, ensure_ascii=False, default=str)
encoded = text.encode('utf-8')
if compress and len(encoded) > 1024:
return zlib.compress(encoded, cls.COMPRESSION_LEVEL)
return encoded
@classmethod
def json_deserialize(cls, data: bytes) -> Any:
try:
try:
decompressed = zlib.decompress(data)
return json.loads(decompressed.decode('utf-8'))
except zlib.error:
return json.loads(data.decode('utf-8'))
except Exception as e:
raise ValueError(f"JSON deserialization failed: {e}") from e
3.6 存储策略配置
from enum import Enum
class StorageTier(Enum):
HOT = "hot"
WARM = "warm"
COLD = "cold"
CHAIN = "chain"
@dataclass
class StoragePolicy:
conversation_cache_size: int = 100
conversation_cache_ttl: int = 3600
conversation_archive_days: int = 30
checkpoint_interval: int = 600
checkpoint_keep_count: int = 10
checkpoint_cache_ttl: int = 7200
full_backup_interval: int = 86400
incremental_backup_interval: int = 3600
backup_retention_days: int = 90
log_retention_days: int = 7
max_local_db_size_mb: int = 1024
max_cache_memory_mb: int = 512
ipfs_pin_by_default: bool = True
ipfs_timeout: int = 60
class TierRouter:
def __init__(self, policy: StoragePolicy):
self.policy = policy
def determine_tier(
self, data_type: str, data_size: int,
access_frequency: str, persistence: str,
) -> StorageTier:
if access_frequency == "high" and data_type in ("session", "cache"):
return StorageTier.HOT
if data_type in ("config", "metadata", "recent_conversations"):
return StorageTier.WARM
if data_size > 1024 * 1024:
return StorageTier.COLD
if persistence == "permanent":
return StorageTier.COLD
if data_type in ("ownership", "permissions", "reputation") and data_size < 256:
return StorageTier.CHAIN
return StorageTier.WARM
4. 检查点与快照
4.1 检查点管理系统
import hashlib
from dataclasses import dataclass, asdict
@dataclass
class Checkpoint:
id: str
agent_id: str
timestamp: float
version: int
state: Dict[str, Any]
checksum: str
parent_id: Optional[str] = None
is_full: bool = True
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
class CheckpointManager:
def __init__(
self, agent_id: str,
offchain_storage: OffChainStorage,
ipfs_client: IPFSClient,
local_storage: SQLiteStorage,
policy: StoragePolicy = None,
):
self.agent_id = agent_id
self.offchain = offchain_storage
self.ipfs = ipfs_client
self.local = local_storage
self.policy = policy or StoragePolicy()
self._checkpoint_task: Optional[asyncio.Task] = None
self._running = False
async def create_checkpoint(self, is_full: bool = True) -> str:
logger.info(f"Creating {'full' if is_full else 'incremental'} checkpoint for {self.agent_id}")
state = await self.collect_full_state()
parent_id = None
if not is_full:
parent_id = await self.get_latest_checkpoint_id()
checkpoint_state = state
if not is_full and parent_id:
parent_cp = await self.load_checkpoint_by_id(parent_id)
if parent_cp:
checkpoint_state = self.compute_diff(parent_cp.state, state)
state_json = json.dumps(checkpoint_state, sort_keys=True, default=str)
checksum = hashlib.sha256(state_json.encode()).hexdigest()
checkpoint_id = f"cp_{self.agent_id}_{int(time.time())}"
checkpoint = Checkpoint(
id=checkpoint_id, agent_id=self.agent_id,
timestamp=time.time(), version=1, state=checkpoint_state,
checksum=checksum, parent_id=parent_id, is_full=is_full,
metadata={"node_id": self.agent_id, "created_by": "checkpoint_manager"},
)
cp_data = json.dumps(asdict(checkpoint), default=str).encode()
cid = await self.ipfs.add(cp_data)
await self.ipfs.pin(cid)
await self.local.store(
f"checkpoint:{checkpoint_id}",
json.dumps({"id": checkpoint_id, "cid": cid,
"timestamp": checkpoint.timestamp,
"is_full": is_full, "checksum": checksum}).encode()
)
await self.local.store(
f"agent:{self.agent_id}:latest_checkpoint",
json.dumps({"id": checkpoint_id, "cid": cid}).encode()
)
logger.info(f"Checkpoint {checkpoint_id} created -> IPFS {cid}")
await self.cleanup_old_checkpoints()
return cid
async def collect_full_state(self) -> Dict[str, Any]:
state = {}
conversations = await self.local.list_agent_conversations(self.agent_id, limit=1000)
state["conversations"] = [asdict(c) for c in conversations]
kb_data = await self.local.load(f"agent:{self.agent_id}:knowledge_base")
state["knowledge_base"] = json.loads(kb_data) if kb_data else {}
prefs_data = await self.local.load(f"agent:{self.agent_id}:user_preferences")
state["user_preferences"] = json.loads(prefs_data) if prefs_data else {}
tasks_data = await self.local.load(f"agent:{self.agent_id}:task_history")
state["task_history"] = json.loads(tasks_data) if tasks_data else []
metrics_data = await self.local.load(f"agent:{self.agent_id}:metrics")
state["performance_metrics"] = json.loads(metrics_data) if metrics_data else {}
custom_data = await self.local.load(f"agent:{self.agent_id}:custom")
state["custom_data"] = json.loads(custom_data) if custom_data else {}
return state
async def restore_from_checkpoint(self, cid: str) -> Dict[str, Any]:
logger.info(f"Restoring from checkpoint {cid}")
cp_data = await self.ipfs.get(cid)
checkpoint_dict = json.loads(cp_data)
checkpoint = Checkpoint(**checkpoint_dict)
state_json = json.dumps(checkpoint.state, sort_keys=True, default=str)
calculated = hashlib.sha256(state_json.encode()).hexdigest()
if calculated != checkpoint.checksum:
raise ValueError(f"Checksum mismatch: {calculated[:16]} != {checkpoint.checksum[:16]}")
state = checkpoint.state
if not checkpoint.is_full and checkpoint.parent_id:
parent_cp_data = await self.local.load(f"checkpoint:{checkpoint.parent_id}")
if parent_cp_data:
parent_info = json.loads(parent_cp_data)
parent_state = await self.restore_from_checkpoint(parent_info["cid"])
state = self.merge_diffs(parent_state, state)
await self.restore_state(state)
logger.info(f"Restored from checkpoint {cid}")
return state
async def restore_state(self, state: Dict[str, Any]):
if "conversations" in state:
for conv_data in state["conversations"]:
conv = ConversationRecord(**conv_data)
await self.local.save_conversation(conv)
if "knowledge_base" in state:
await self.local.store(
f"agent:{self.agent_id}:knowledge_base",
json.dumps(state["knowledge_base"]).encode()
)
if "user_preferences" in state:
await self.local.store(
f"agent:{self.agent_id}:user_preferences",
json.dumps(state["user_preferences"]).encode()
)
if "task_history" in state:
await self.local.store(
f"agent:{self.agent_id}:task_history",
json.dumps(state["task_history"]).encode()
)
if "performance_metrics" in state:
await self.local.store(
f"agent:{self.agent_id}:metrics",
json.dumps(state["performance_metrics"]).encode()
)
if "custom_data" in state:
await self.local.store(
f"agent:{self.agent_id}:custom",
json.dumps(state["custom_data"]).encode()
)
def compute_diff(self, old_state: Dict, new_state: Dict) -> Dict:
diff = {}
for key in new_state:
if key not in old_state:
diff[key] = new_state[key]
elif old_state[key] != new_state[key]:
diff[key] = new_state[key]
deleted = [k for k in old_state if k not in new_state]
if deleted:
diff["__deleted__"] = deleted
return diff
def merge_diffs(self, base_state: Dict, diffs: Dict) -> Dict:
merged = dict(base_state)
for key, value in diffs.items():
if key == "__deleted__":
for del_key in value:
merged.pop(del_key, None)
else:
merged[key] = value
return merged
async def get_latest_checkpoint_id(self) -> Optional[str]:
data = await self.local.load(f"agent:{self.agent_id}:latest_checkpoint")
if data:
return json.loads(data)["id"]
return None
async def load_checkpoint_by_id(self, cp_id: str) -> Optional[Checkpoint]:
data = await self.local.load(f"checkpoint:{cp_id}")
if not data:
return None
info = json.loads(data)
cp_data = await self.ipfs.get(info["cid"])
return Checkpoint(**json.loads(cp_data))
async def list_checkpoints(self, limit: int = 20, offset: int = 0) -> List[Dict]:
keys = await self.local.list_keys(f"checkpoint:cp_{self.agent_id}")
checkpoints = []
for key in keys:
data = await self.local.load(key)
if data:
checkpoints.append(json.loads(data))
checkpoints.sort(key=lambda x: x["timestamp"], reverse=True)
return checkpoints[offset:offset + limit]
async def verify_checkpoint(self, cid: str) -> bool:
try:
cp_data = await self.ipfs.get(cid)
cp = Checkpoint(**json.loads(cp_data))
state_json = json.dumps(cp.state, sort_keys=True, default=str)
expected = hashlib.sha256(state_json.encode()).hexdigest()
return expected == cp.checksum
except Exception as e:
logger.error(f"Verification failed: {e}")
return False
async def cleanup_old_checkpoints(self, keep_count: int = None):
if keep_count is None:
keep_count = self.policy.checkpoint_keep_count
checkpoints = await self.list_checkpoints(limit=1000)
incremental = [cp for cp in checkpoints if not cp.get("is_full")]
to_delete = incremental[keep_count:]
for cp in to_delete:
try:
await self.ipfs.unpin(cp["cid"])
await self.local.delete(f"checkpoint:{cp['id']}")
except Exception as e:
logger.error(f"Cleanup failed: {e}")
async def start_periodic_checkpointing(self):
if self._running:
return
self._running = True
full_interval = self.policy.checkpoint_interval
inc_interval = full_interval // 2
last_full = 0
while self._running:
await asyncio.sleep(inc_interval)
now = time.time()
is_full = (now - last_full) >= full_interval
try:
cid = await self.create_checkpoint(is_full=is_full)
if is_full:
last_full = now
except Exception as e:
logger.error(f"Periodic checkpoint failed: {e}")
async def stop_periodic_checkpointing(self):
self._running = False
if self._checkpoint_task:
self._checkpoint_task.cancel()
self._checkpoint_task = None
async def restore_to_point_in_time(self, target_timestamp: float) -> bool:
checkpoints = await self.list_checkpoints(limit=1000)
valid = [cp for cp in checkpoints if cp["timestamp"] <= target_timestamp]
if not valid:
return False
nearest = max(valid, key=lambda x: x["timestamp"])
await self.restore_from_checkpoint(nearest["cid"])
return True
4.2 快照验证工具
class SnapshotValidator:
@staticmethod
def validate_checkpoint_integrity(checkpoint: Checkpoint) -> List[str]:
issues = []
state_json = json.dumps(checkpoint.state, sort_keys=True, default=str)
expected = hashlib.sha256(state_json.encode()).hexdigest()
if expected != checkpoint.checksum:
issues.append(f"CHECKSUM_MISMATCH")
now = time.time()
if checkpoint.timestamp > now + 300:
issues.append(f"FUTURE_TIMESTAMP")
if checkpoint.timestamp < 1_500_000_000:
issues.append(f"UNREALISTIC_TIMESTAMP")
required = ["agent_id", "version", "state"]
for field in required:
if field not in asdict(checkpoint):
issues.append(f"MISSING_FIELD: {field}")
return issues
@staticmethod
def validate_chain_consistency(onchain_state: dict, offchain_checkpoint: Checkpoint) -> bool:
onchain_hash = onchain_state.get("state_hash", "")
if onchain_hash:
return onchain_hash == offchain_checkpoint.checksum
return False
4.3 检查点恢复流程
┌─────────────────────────────────────────────────────────┐
│ 故障检测 │
│ heartbeat 超时 / 状态不一致 / 用户触发回滚 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 1. 停止 Agent 当前任务 │
│ 保存当前内存中的会话状态 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 2. 选择恢复点 │
│ - 自动选择最新有效检查点 │
│ - 或用户指定的时间点 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 3. 从 IPFS 加载检查点 │
│ get(cid) -> 反序列化 -> 校验校验和 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 4. 恢复状态到各层 │
│ - 写入本地 SQLite │
│ - 填充 Redis 缓存 │
│ - 如为增量检查点，合并父检查点 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 5. 验证恢复结果 │
│ - 检查关键状态字段 │
│ - 执行健康检查 │
│ - 记录恢复日志 │
└─────────────────────┬───────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 6. 恢复 Agent 运行 │
│ 载入会话上下文，继续处理 │
└─────────────────────────────────────────────────────────┘
5. 会话与上下文管理
5.1 会话生命周期管理
from uuid import uuid4
from enum import Enum
class SessionState(Enum):
ACTIVE = "active"
IDLE = "idle"
SUSPENDED = "suspended"
EXPIRED = "expired"
CLOSED = "closed"
@dataclass
class SessionContext:
user_id: str
conversation_history: List[Message] = field(default_factory=list)
user_preferences: Dict[str, Any] = field(default_factory=dict)
active_tools: List[str] = field(default_factory=list)
current_task: Optional[str] = None
variables: Dict[str, Any] = field(default_factory=dict)
token_count: int = 0
last_activity: Optional[str] = None
@dataclass
class Session:
id: str
agent_id: str
user: str
state: SessionState = SessionState.ACTIVE
context: SessionContext = None
created_at: float = 0.0
updated_at: float = 0.0
last_active_at: float = 0.0
expires_at: float = 0.0
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
if self.context is None:
self.context = SessionContext(user_id=self.user)
if not self.created_at:
self.created_at = time.time()
if not self.updated_at:
self.updated_at = time.time()
if not self.last_active_at:
self.last_active_at = time.time()
@property
def is_expired(self) -> bool:
return time.time() > self.expires_at or self.state == SessionState.EXPIRED
def touch(self):
self.last_active_at = time.time()
self.updated_at = time.time()
if self.state == SessionState.IDLE:
self.state = SessionState.ACTIVE
def close(self):
self.state = SessionState.CLOSED
self.updated_at = time.time()
class SessionStore:
def __init__(self, local_storage: SQLiteStorage, cache: RedisCache):
self.local = local_storage
self.cache = cache
self._cache_ttl = 3600
async def save(self, session: Session):
data = json.dumps(asdict(session), default=str).encode()
await self.cache.store(f"session:{session.id}", data, ttl=self._cache_ttl)
await self.local.store(f"session:{session.id}", data)
async def load(self, session_id: str) -> Optional[Session]:
data = await self.cache.load(f"session:{session_id}")
if not data:
data = await self.local.load(f"session:{session_id}")
if data:
sd = json.loads(data)
sd["state"] = SessionState(sd["state"])
if sd.get("context"):
sd["context"] = SessionContext(**sd["context"])
return Session(**sd)
return None
async def delete(self, session_id: str):
await self.cache.delete(f"session:{session_id}")
await self.local.delete(f"session:{session_id}")
class SessionManager:
def __init__(
self, agent_id: str, store: SessionStore,
session_timeout: int = 1800, max_concurrent_sessions: int = 100,
):
self.agent_id = agent_id
self.store = store
self.session_timeout = session_timeout
self.max_concurrent = max_concurrent_sessions
self.active_sessions: Dict[str, Session] = {}
self._cleanup_task: Optional[asyncio.Task] = None
async def create_session(
self, user: str, preferences: Dict[str, Any] = None,
initial_context: Dict[str, Any] = None,
) -> Session:
if len(self.active_sessions) >= self.max_concurrent:
await self._evict_oldest_session()
context_data = await self._load_user_context(user)
if initial_context:
context_data.update(initial_context)
context = SessionContext(
user_id=user, user_preferences=preferences or {},
variables=context_data,
)
now = time.time()
session = Session(
id=uuid4().hex, agent_id=self.agent_id, user=user,
context=context, created_at=now, updated_at=now,
last_active_at=now, expires_at=now + self.session_timeout,
)
self.active_sessions[session.id] = session
await self.store.save(session)
logger.info(f"Session created: {session.id[:12]} for user {user}")
return session
async def get_session(self, session_id: str) -> Optional[Session]:
session = self.active_sessions.get(session_id)
if session:
if session.is_expired:
await self.expire_session(session_id)
return None
session.touch()
return session
session = await self.store.load(session_id)
if session:
if session.is_expired:
await self.expire_session(session_id)
return None
self.active_sessions[session_id] = session
session.touch()
asyncio.create_task(self._preload_context(session))
return session
return None
async def update_session_context(
self, session_id: str,
new_messages: List[Message] = None, variables: Dict[str, Any] = None,
) -> Optional[Session]:
session = await self.get_session(session_id)
if not session:
return None
if new_messages:
session.context.conversation_history.extend(new_messages)
session.context.token_count += sum(len(m.content) for m in new_messages)
if variables:
session.context.variables.update(variables)
session.touch()
await self.store.save(session)
return session
async def close_session(self, session_id: str):
session = self.active_sessions.pop(session_id, None)
if session:
session.close()
await self.store.save(session)
await self._save_user_context(session)
async def expire_session(self, session_id: str):
session = self.active_sessions.pop(session_id, None)
if not session:
session = await self.store.load(session_id)
if session:
session.state = SessionState.EXPIRED
await self.store.save(session)
async def _load_user_context(self, user: str) -> Dict[str, Any]:
data = await self.store.local.load(f"user_context:{self.agent_id}:{user}")
if data:
return json.loads(data)
return {}
async def _save_user_context(self, session: Session):
persistent = {
k: v for k, v in session.context.variables.items()
if not k.startswith("_temp_")
}
await self.store.local.store(
f"user_context:{self.agent_id}:{session.user}",
json.dumps(persistent).encode()
)
async def _preload_context(self, session: Session):
context_data = await self._load_user_context(session.user)
if context_data:
session.context.variables.update(context_data)
async def _evict_oldest_session(self):
if not self.active_sessions:
return
oldest = min(self.active_sessions.values(), key=lambda s: s.last_active_at)
await self.suspend_session(oldest.id)
async def suspend_session(self, session_id: str):
session = self.active_sessions.pop(session_id, None)
if session:
session.state = SessionState.SUSPENDED
await self.store.save(session)
await self._save_user_context(session)
async def suspend_all_idle(self, max_idle: int = 300):
now = time.time()
to_suspend = [
sid for sid, s in self.active_sessions.items()
if now - s.last_active_at > max_idle and s.state == SessionState.ACTIVE
]
for sid in to_suspend:
await self.suspend_session(sid)
async def start_cleanup_loop(self, interval: int = 60):
while True:
try:
await asyncio.sleep(interval)
now = time.time()
expired = [sid for sid, s in self.active_sessions.items() if s.is_expired]
for sid in expired:
self.active_sessions.pop(sid, None)
await self.suspend_all_idle(max_idle=self.session_timeout)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Cleanup error: {e}")
async def shutdown(self):
for session in self.active_sessions.values():
await self.store.save(session)
await self._save_user_context(session)
self.active_sessions.clear()
5.2 上下文窗口管理
class ContextWindowManager:
def __init__(self, max_tokens: int = 4096, max_messages: int = 50, compression_threshold: float = 0.8):
self.max_tokens = max_tokens
self.max_messages = max_messages
self.compression_threshold = compression_threshold
def should_compress(self, messages: List[Message]) -> bool:
total = sum(len(m.content) for m in messages)
return total > self.max_tokens * self.compression_threshold
def compress(self, messages: List[Message]) -> List[Message]:
if not self.should_compress(messages):
return messages
system_msgs = [m for m in messages if m.role == "system"]
other = [m for m in messages if m.role != "system"]
keep = max(int(len(other) * 0.6), self.max_messages // 2)
recent = other[-keep:]
old = other[:-keep]
if old:
summary = self._summarize(old)
summary_msg = Message(role="system", content=f"[摘要]: {summary}", timestamp=time.time())
return system_msgs + [summary_msg] + recent
return system_msgs + recent
def _summarize(self, messages: List[Message]) -> str:
points = []
for msg in messages[-10:]:
c = msg.content[:100]
if msg.role == "user":
points.append(f"用户: {c}")
elif msg.role == "assistant":
points.append(f"助手: {c}")
return f"共 {len(messages)} 条消息。{'；'.join(points[-5:])}"
5.3 上下文继承
class ContextInheritor:
def __init__(self, store: SessionStore):
self.store = store
async def inherit_context(
self, from_session: Session, to_session: Session, inherit_keys: List[str] = None,
) -> Session:
if inherit_keys:
for key in inherit_keys:
if key in from_session.context.variables:
to_session.context.variables[key] = from_session.context.variables[key]
else:
for k, v in from_session.context.variables.items():
if not k.startswith("_temp_"):
to_session.context.variables[k] = v
to_session.context.user_preferences.update(from_session.context.user_preferences)
recent = from_session.context.conversation_history[-3:]
to_session.context.conversation_history = recent
await self.store.save(to_session)
return to_session
6. 备份与恢复策略
6.1 自动化备份系统
class BackupType(Enum):
FULL = "full"
INCREMENTAL = "incremental"
DIFFERENTIAL = "differential"
class BackupStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
VERIFIED = "verified"
@dataclass
class BackupRecord:
id: str
agent_id: str
cid: str
backup_type: BackupType
status: BackupStatus
timestamp: float
size_bytes: int
checksum: str
parent_id: Optional[str] = None
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
class BackupManager:
def __init__(
self, agent_id: str, offchain: OffChainStorage,
ipfs: IPFSClient, local: SQLiteStorage,
encryption_key: Optional[bytes] = None, policy: StoragePolicy = None,
):
self.agent_id = agent_id
self.offchain = offchain
self.ipfs = ipfs
self.local = local
self.encryption_key = encryption_key
self.policy = policy or StoragePolicy()
self._backup_task: Optional[asyncio.Task] = None
self._running = False
async def full_backup(self) -> str:
logger.info(f"Starting full backup for {self.agent_id}")
backup_id = f"bk_{self.agent_id}_full_{int(time.time())}"
try:
onchain = await self._query_onchain_state()
offchain = await self._collect_offchain_data()
data = {
"backup_id": backup_id, "agent_id": self.agent_id,
"backup_type": "full", "created_at": time.time(),
"format_version": 1, "onchain": onchain, "offchain": offchain,
}
data_json = json.dumps(data, sort_keys=True, default=str)
checksum = hashlib.sha256(data_json.encode()).hexdigest()
data["checksum"] = checksum
encrypted = await self._encrypt(json.dumps(data, default=str).encode())
cid = await self.ipfs.add(encrypted)
await self.ipfs.pin(cid)
record = BackupRecord(
id=backup_id, agent_id=self.agent_id, cid=cid,
backup_type=BackupType.FULL, status=BackupStatus.COMPLETED,
timestamp=time.time(), size_bytes=len(encrypted),
checksum=checksum,
)
await self._save_record(record)
asyncio.create_task(self._anchor_on_chain(cid, backup_id))
logger.info(f"Full backup done: {cid}")
return cid
except Exception as e:
logger.error(f"Full backup failed: {e}")
failed = BackupRecord(
id=backup_id, agent_id=self.agent_id, cid="",
backup_type=BackupType.FULL, status=BackupStatus.FAILED,
timestamp=time.time(), size_bytes=0, checksum="",
metadata={"error": str(e)},
)
await self._save_record(failed)
raise
async def incremental_backup(self) -> str:
backup_id = f"bk_{self.agent_id}_inc_{int(time.time())}"
try:
last = await self._get_last_backup()
if not last:
return await self.full_backup()
current = await self._collect_offchain_data()
last_data = await self._load_backup_data(last.cid)
last_state = last_data.get("offchain", {})
diff = self._compute_diff(last_state, current)
data = {
"backup_id": backup_id, "agent_id": self.agent_id,
"backup_type": "incremental", "created_at": time.time(),
"format_version": 1, "parent_id": last.id, "parent_cid": last.cid,
"diff": diff,
}
data_json = json.dumps(data, sort_keys=True, default=str)
checksum = hashlib.sha256(data_json.encode()).hexdigest()
data["checksum"] = checksum
encrypted = await self._encrypt(json.dumps(data, default=str).encode())
cid = await self.ipfs.add(encrypted)
await self.ipfs.pin(cid)
record = BackupRecord(
id=backup_id, agent_id=self.agent_id, cid=cid,
backup_type=BackupType.INCREMENTAL, status=BackupStatus.COMPLETED,
timestamp=time.time(), size_bytes=len(encrypted),
checksum=checksum, parent_id=last.id,
)
await self._save_record(record)
return cid
except Exception as e:
logger.error(f"Incremental backup failed: {e}")
raise
async def restore_from_backup(self, backup_id: str) -> bool:
logger.info(f"Restoring from backup {backup_id}")
record = await self._load_record(backup_id)
if not record:
raise ValueError(f"Backup not found: {backup_id}")
encrypted = await self.ipfs.get(record.cid)
decrypted = await self._decrypt(encrypted)
data = json.loads(decrypted)
stored_checksum = data.pop("checksum", "")
data_json = json.dumps(data, sort_keys=True, default=str)
calc = hashlib.sha256(data_json.encode()).hexdigest()
if stored_checksum != calc:
raise ValueError(f"Checksum mismatch")
if record.backup_type == BackupType.INCREMENTAL:
parent_cid = data.get("parent_cid")
if parent_cid:
pe = await self.ipfs.get(parent_cid)
pd = await self._decrypt(pe)
parent = json.loads(pd)
full = self._apply_diff(parent.get("offchain", {}), data.get("diff", {}))
data["offchain"] = full
await self._restore_state(data)
record.status = BackupStatus.VERIFIED
await self._save_record(record)
logger.info(f"Restored from {backup_id}")
return True
async def _query_onchain_state(self) -> Dict:
return {"agent_id": self.agent_id, "queried_at": time.time()}
async def _collect_offchain_data(self) -> Dict:
data = {}
session_keys = await self.local.list_keys("session:")
sessions = []
for key in session_keys:
sd = await self.local.load(key)
if sd:
sessions.append(json.loads(sd))
data["sessions"] = sessions
convs = await self.local.list_agent_conversations(self.agent_id, limit=500)
data["conversations"] = [asdict(c) for c in convs]
kb = await self.local.load(f"agent:{self.agent_id}:knowledge_base")
data["knowledge_base"] = json.loads(kb) if kb else {}
tasks = await self.local.load(f"agent:{self.agent_id}:task_history")
data["task_history"] = json.loads(tasks) if tasks else []
prefs = await self.local.load(f"agent:{self.agent_id}:user_preferences")
data["user_preferences"] = json.loads(prefs) if prefs else {}
cp_keys = await self.local.list_keys(f"checkpoint:cp_{self.agent_id}")
cps = []
for k in cp_keys:
cd = await self.local.load(k)
if cd:
cps.append(json.loads(cd))
data["checkpoints"] = cps
return data
async def _encrypt(self, data: bytes) -> bytes:
if not self.encryption_key:
return data
from cryptography.fernet import Fernet
return Fernet(self.encryption_key).encrypt(data)
async def _decrypt(self, data: bytes) -> bytes:
if not self.encryption_key:
return data
from cryptography.fernet import Fernet
return Fernet(self.encryption_key).decrypt(data)
def _compute_diff(self, old: Dict, new: Dict) -> Dict:
diff = {}
all_keys = set(list(old.keys()) + list(new.keys()))
for k in all_keys:
if k not in old:
diff[k] = {"type": "added", "value": new[k]}
elif k not in new:
diff[k] = {"type": "deleted"}
elif old[k] != new[k]:
diff[k] = {"type": "modified", "old": old[k], "new": new[k]}
return diff
def _apply_diff(self, base: Dict, diff: Dict) -> Dict:
result = dict(base)
for k, ch in diff.items():
if ch["type"] in ("added", "modified"):
result[k] = ch.get("value", ch.get("new"))
elif ch["type"] == "deleted":
result.pop(k, None)
return result
async def _save_record(self, record: BackupRecord):
await self.local.store(
f"backup:{record.id}", json.dumps(asdict(record), default=str).encode()
)
await self.local.store(
f"agent:{self.agent_id}:latest_backup",
json.dumps({"id": record.id, "cid": record.cid}).encode()
)
async def _load_record(self, backup_id: str) -> Optional[BackupRecord]:
data = await self.local.load(f"backup:{backup_id}")
if data:
d = json.loads(data)
d["backup_type"] = BackupType(d["backup_type"])
d["status"] = BackupStatus(d["status"])
return BackupRecord(**d)
return None
async def _load_backup_data(self, cid: str) -> Dict:
encrypted = await self.ipfs.get(cid)
decrypted = await self._decrypt(encrypted)
return json.loads(decrypted)
async def _get_last_backup(self) -> Optional[BackupRecord]:
data = await self.local.load(f"agent:{self.agent_id}:latest_backup")
if data:
info = json.loads(data)
return await self._load_record(info["id"])
return None
async def _anchor_on_chain(self, cid: str, backup_id: str):
logger.info(f"Anchoring backup {backup_id} -> {cid} on MSG Chain")
await asyncio.sleep(0.1)
async def _restore_state(self, data: Dict):
offchain = data.get("offchain", {})
for sd in offchain.get("sessions", []):
await self.local.store(f"session:{sd['id']}", json.dumps(sd).encode())
for cp in offchain.get("checkpoints", []):
await self.local.store(f"checkpoint:{cp['id']}", json.dumps(cp).encode())
if "knowledge_base" in offchain:
await self.local.store(
f"agent:{self.agent_id}:knowledge_base",
json.dumps(offchain["knowledge_base"]).encode()
)
if "task_history" in offchain:
await self.local.store(
f"agent:{self.agent_id}:task_history",
json.dumps(offchain["task_history"]).encode()
)
if "user_preferences" in offchain:
await self.local.store(
f"agent:{self.agent_id}:user_preferences",
json.dumps(offchain["user_preferences"]).encode()
)
async def list_backups(self, limit: int = 20, backup_type: Optional[BackupType] = None) -> List[BackupRecord]:
keys = await self.local.list_keys(f"backup:bk_{self.agent_id}")
records = []
for key in keys:
data = await self.local.load(key)
if data:
d = json.loads(data)
d["backup_type"] = BackupType(d["backup_type"])
d["status"] = BackupStatus(d["status"])
records.append(BackupRecord(**d))
if backup_type:
records = [r for r in records if r.backup_type == backup_type]
records.sort(key=lambda r: r.timestamp, reverse=True)
return records[:limit]
async def verify_backup(self, backup_id: str) -> bool:
try:
record = await self._load_record(backup_id)
if not record:
return False
encrypted = await self.ipfs.get(record.cid)
decrypted = await self._decrypt(encrypted)
data = json.loads(decrypted)
stored = data.get("checksum", "")
copy = dict(data)
copy.pop("checksum", None)
calc = hashlib.sha256(json.dumps(copy, sort_keys=True, default=str).encode()).hexdigest()
valid = stored == calc
if valid:
record.status = BackupStatus.VERIFIED
await self._save_record(record)
return valid
except Exception as e:
logger.error(f"Verify failed: {e}")
return False
async def start_scheduled_backups(self):
if self._running:
return
self._running = True
full_int = self.policy.full_backup_interval
inc_int = self.policy.incremental_backup_interval
last_full = 0
while self._running:
await asyncio.sleep(inc_int)
now = time.time()
try:
if now - last_full >= full_int:
await self.full_backup()
last_full = now
else:
await self.incremental_backup()
except Exception as e:
logger.error(f"Scheduled backup failed: {e}")
async def stop_scheduled_backups(self):
self._running = False
if self._backup_task:
self._backup_task.cancel()
self._backup_task = None
async def disaster_recovery_drill(self) -> Dict[str, Any]:
logger.info("=== DISASTER RECOVERY DRILL ===")
results = {"steps": [], "success": True}
try:
last = await self._get_last_backup()
if not last:
results["steps"].append({"step": "find_backup", "success": False, "error": "No backup"})
results["success"] = False
return results
results["steps"].append({"step": "find_backup", "success": True, "id": last.id})
valid = await self.verify_backup(last.id)
results["steps"].append({"step": "verify", "success": valid})
if not valid:
results["success"] = False
return results
await self.restore_from_backup(last.id)
results["steps"].append({"step": "restore", "success": True})
kb = await self.local.load(f"agent:{self.agent_id}:knowledge_base")
results["steps"].append({"step": "verify_restore", "success": kb is not None})
except Exception as e:
results["steps"].append({"step": "error", "error": str(e)})
results["success"] = False
return results
6.2 备份保留策略
@dataclass
class BackupPolicy:
full_backup_cron: str = "0 3 * * *"
incremental_interval_minutes: int = 60
full_keep_days: int = 30
inc_keep_days: int = 7
min_backups_to_keep: int = 5
encrypt_backups: bool = True
verify_after_backup: bool = True
class BackupRotationManager:
def __init__(self, manager: BackupManager, policy: BackupPolicy):
self.manager = manager
self.policy = policy
async def rotate_backups(self):
all_bk = await self.manager.list_backups(limit=1000)
now = time.time()
to_delete = []
for bk in all_bk:
age = (now - bk.timestamp) / 86400
if bk.backup_type == BackupType.FULL and age > self.policy.full_keep_days:
to_delete.append(bk)
elif bk.backup_type != BackupType.FULL and age > self.policy.inc_keep_days:
to_delete.append(bk)
keep = len(all_bk) - len(to_delete)
if keep < self.policy.min_backups_to_keep:
excess = self.policy.min_backups_to_keep - keep
to_delete = sorted(to_delete, key=lambda b: b.timestamp, reverse=True)[excess:] if excess > 0 else to_delete
for bk in to_delete:
try:
await self.manager.ipfs.unpin(bk.cid)
await self.manager.local.delete(f"backup:{bk.id}")
except Exception as e:
logger.error(f"Rotation failed: {e}")
7. 数据清理与归档
7.1 数据保留策略
class DataCategory(Enum):
CONVERSATION = "conversation"
SESSION = "session"
LOG = "log"
CHECKPOINT = "checkpoint"
BACKUP = "backup"
TEMP_CACHE = "temp_cache"
TASK_RESULT = "task_result"
METRICS = "metrics"
@dataclass
class RetentionRule:
category: DataCategory
max_age_days: int
max_count: int
archive_before_delete: bool
priority: int
class DataRetentionPolicy:
DEFAULT_RULES = [
RetentionRule(DataCategory.CONVERSATION, 90, 500, True, 5),
RetentionRule(DataCategory.SESSION, 7, 0, False, 8),
RetentionRule(DataCategory.LOG, 30, 0, False, 3),
RetentionRule(DataCategory.CHECKPOINT, 30, 20, True, 6),
RetentionRule(DataCategory.BACKUP, 90, 10, False, 7),
RetentionRule(DataCategory.TEMP_CACHE, 1, 0, False, 10),
RetentionRule(DataCategory.TASK_RESULT, 30, 200, True, 4),
RetentionRule(DataCategory.METRICS, 365, 0, False, 1),
]
def __init__(self, rules: List[RetentionRule] = None):
self.rules = rules or self.DEFAULT_RULES
def get_rule(self, category: DataCategory) -> Optional[RetentionRule]:
for r in self.rules:
if r.category == category:
return r
return None
7.2 数据清理执行器
class DataCleanupManager:
def __init__(
self, agent_id: str, local_storage: SQLiteStorage,
ipfs: IPFSClient, cache: RedisCache, offchain: OffChainStorage,
policy: DataRetentionPolicy = None,
):
self.agent_id = agent_id
self.local = local_storage
self.ipfs = ipfs
self.cache = cache
self.offchain = offchain
self.policy = policy or DataRetentionPolicy()
self._running = False
async def cleanup_by_category(self, category: DataCategory) -> Dict[str, int]:
rule = self.policy.get_rule(category)
if not rule:
return {"deleted": 0, "archived": 0, "freed_bytes": 0}
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
cutoff = time.time() - (rule.max_age_days * 86400)
if category == DataCategory.CONVERSATION:
stats = await self._cleanup_conversations(cutoff, rule)
elif category == DataCategory.SESSION:
stats = await self._cleanup_sessions(cutoff, rule)
elif category == DataCategory.LOG:
stats = await self._cleanup_logs(cutoff, rule)
elif category == DataCategory.CHECKPOINT:
stats = await self._cleanup_checkpoints(rule)
elif category == DataCategory.BACKUP:
stats = await self._cleanup_backups(rule)
elif category == DataCategory.TEMP_CACHE:
stats = await self._cleanup_temp_cache()
elif category == DataCategory.TASK_RESULT:
stats = await self._cleanup_task_results(cutoff, rule)
elif category == DataCategory.METRICS:
stats = await self._cleanup_metrics(cutoff, rule)
return stats
async def _cleanup_conversations(self, cutoff: float, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys("conv:")
convs = []
for k in keys:
d = await self.local.load(k)
if d:
try:
convs.append((k, json.loads(d)))
except json.JSONDecodeError:
continue
convs.sort(key=lambda x: x[1].get("updated_at", 0))
to_remove = convs[:-rule.max_count] if rule.max_count > 0 and len(convs) > rule.max_count else []
for k, c in convs:
if c.get("updated_at", 0) < cutoff and (k, c) not in to_remove:
to_remove.append((k, c))
for k, c in to_remove:
try:
if rule.archive_before_delete:
cid = await self.offchain.archive_and_purge(c.get("session_id", ""), 0)
if cid:
stats["archived"] += 1
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(json.dumps(c))
except Exception as e:
logger.error(f"Cleanup failed: {e}")
return stats
async def _cleanup_sessions(self, cutoff: float, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys("session:")
for k in keys:
d = await self.local.load(k)
if d:
try:
s = json.loads(d)
if s.get("updated_at", 0) < cutoff:
await self.local.delete(k)
await self.cache.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(d)
except json.JSONDecodeError:
continue
return stats
async def _cleanup_logs(self, cutoff: float, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys("log:")
for k in keys:
d = await self.local.load(k)
if d:
try:
entry = json.loads(d)
if entry.get("timestamp", 0) < cutoff:
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(d)
except json.JSONDecodeError:
continue
return stats
async def _cleanup_checkpoints(self, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys(f"checkpoint:cp_{self.agent_id}")
cps = []
for k in keys:
d = await self.local.load(k)
if d:
cps.append((k, json.loads(d)))
cps.sort(key=lambda x: x[1].get("timestamp", 0))
to_remove = cps[:-rule.max_count] if rule.max_count > 0 and len(cps) > rule.max_count else []
for k, cp in to_remove:
try:
if cp.get("cid"):
await self.ipfs.unpin(cp["cid"])
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(json.dumps(cp))
except Exception as e:
logger.error(f"Cleanup failed: {e}")
return stats
async def _cleanup_backups(self, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys(f"backup:bk_{self.agent_id}")
bks = []
for k in keys:
d = await self.local.load(k)
if d:
bks.append((k, json.loads(d)))
bks.sort(key=lambda x: x[1].get("timestamp", 0))
to_remove = bks[:-rule.max_count] if rule.max_count > 0 and len(bks) > rule.max_count else []
for k, bk in to_remove:
try:
if bk.get("cid"):
await self.ipfs.unpin(bk["cid"])
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(json.dumps(bk))
except Exception as e:
logger.error(f"Cleanup failed: {e}")
return stats
async def _cleanup_temp_cache(self) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.cache.list_keys("temp:")
for k in keys:
await self.cache.delete(k)
stats["deleted"] += 1
local_keys = await self.local.list_keys("temp:")
for k in local_keys:
d = await self.local.load(k)
await self.local.delete(k)
if d:
stats["deleted"] += 1
stats["freed_bytes"] += len(d)
return stats
async def _cleanup_task_results(self, cutoff: float, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys(f"task:{self.agent_id}:")
tasks = []
for k in keys:
d = await self.local.load(k)
if d:
try:
tasks.append((k, json.loads(d)))
except json.JSONDecodeError:
continue
tasks.sort(key=lambda x: x[1].get("completed_at", 0))
to_remove = tasks[:-rule.max_count] if rule.max_count > 0 and len(tasks) > rule.max_count else []
for k, t in tasks:
if t.get("completed_at", 0) < cutoff and (k, t) not in to_remove:
to_remove.append((k, t))
for k, t in to_remove:
try:
if rule.archive_before_delete:
archived = json.dumps(t).encode()
cid = await self.ipfs.add(archived)
await self.ipfs.pin(cid)
stats["archived"] += 1
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(json.dumps(t))
except Exception as e:
logger.error(f"Cleanup failed: {e}")
return stats
async def _cleanup_metrics(self, cutoff: float, rule: RetentionRule) -> Dict[str, int]:
stats = {"deleted": 0, "archived": 0, "freed_bytes": 0}
keys = await self.local.list_keys(f"metrics:{self.agent_id}:")
for k in keys:
d = await self.local.load(k)
if d:
try:
m = json.loads(d)
if m.get("timestamp", 0) < cutoff:
await self.local.delete(k)
stats["deleted"] += 1
stats["freed_bytes"] += len(d)
except json.JSONDecodeError:
continue
return stats
async def run_full_cleanup(self) -> Dict[str, Dict[str, int]]:
results = {}
for cat in DataCategory:
try:
results[cat.value] = await self.cleanup_by_category(cat)
except Exception as e:
results[cat.value] = {"error": str(e)}
total_freed = sum(s.get("freed_bytes", 0) for s in results.values() if isinstance(s, dict))
total_del = sum(s.get("deleted", 0) for s in results.values() if isinstance(s, dict))
logger.info(f"Cleanup: {total_del} items, {total_freed/1024/1024:.2f} MB freed")
return results
async def get_storage_usage(self) -> Dict[str, Any]:
usage = {"by_category": {}}
prefix_map = {
DataCategory.CONVERSATION: "conv:",
DataCategory.SESSION: "session:",
DataCategory.LOG: "log:",
DataCategory.CHECKPOINT: f"checkpoint:cp_{self.agent_id}",
DataCategory.BACKUP: f"backup:bk_{self.agent_id}",
DataCategory.TEMP_CACHE: "temp:",
DataCategory.TASK_RESULT: f"task:{self.agent_id}:",
DataCategory.METRICS: f"metrics:{self.agent_id}:",
}
for cat, prefix in prefix_map.items():
keys = await self.local.list_keys(prefix)
total = 0
for k in keys:
d = await self.local.load(k)
if d:
total += len(d)
usage["by_category"][cat.value] = {"count": len(keys), "size_bytes": total}
return usage
async def start_periodic_cleanup(self, interval: int = 3600):
if self._running:
return
self._running = True
while self._running:
await asyncio.sleep(interval)
try:
await self.run_full_cleanup()
except Exception as e:
logger.error(f"Cleanup failed: {e}")
async def stop_periodic_cleanup(self):
self._running = False
7.3 存储成本管理
class StorageCostManager:
def __init__(self, agent_id: str, ipfs: IPFSClient, local: SQLiteStorage, max_monthly_cost_msg: float = 100.0):
self.agent_id = agent_id
self.ipfs = ipfs
self.local = local
self.max_monthly_cost = max_monthly_cost_msg
async def estimate_store_cost(self, data_size: int, storage_type: str) -> float:
if storage_type == "chain":
return data_size * 0.001
elif storage_type == "ipfs":
return (data_size / (1024**3)) * 0.05
return 0.0
async def should_archive(self, data_size: int, age_days: int) -> bool:
if age_days > 30 and data_size > 1024 * 1024:
return True
local_cost = await self.estimate_store_cost(data_size, "local")
archive_cost = await self.estimate_store_cost(data_size, "ipfs")
return local_cost > archive_cost * 2
async def get_monthly_cost_report(self) -> Dict[str, float]:
return {
"chain_store": await self.estimate_store_cost(1024, "chain") * 30,
"ipfs_pinning": await self.estimate_store_cost(500 * 1024 * 1024, "ipfs"),
"local_storage": 0.0, "total": 0.0,
}
async def optimize_storage(self) -> Dict[str, Any]:
optimizations = []
keys = await self.local.list_keys("conv:")
for k in keys:
d = await self.local.load(k)
if d and len(d) > 1024 * 1024:
try:
conv = json.loads(d)
age = time.time() - conv.get("updated_at", 0)
if age > 7 * 86400:
optimizations.append({
"key": k, "size": len(d), "age_days": age / 86400,
"action": "archive_to_ipfs",
"savings": await self.estimate_store_cost(len(d), "local"),
})
except json.JSONDecodeError:
continue
return {"optimizations": optimizations, "current_cost": await self.get_monthly_cost_report()}
8. 完整 Agent 状态管理示例
8.1 AgentStateManager 综合实现
以下代码整合了前面所有模块，提供一个统一的 Agent 状态管理入口：
class AgentStateManager:
def __init__(
self, agent_id: str,
db_path: str = "/data/agent.db",
ipfs_url: str = "http://localhost:5001",
redis_host: str = "localhost",
redis_port: int = 6379,
chain_client: Optional[Any] = None,
encryption_key: Optional[bytes] = None,
state_version: int = 1,
):
self.agent_id = agent_id
self.chain_client = chain_client
self.state_version = state_version
self._running = False
# 存储层
self.offchain = OffChainStorage(
db_path=db_path, ipfs_url=ipfs_url,
redis_host=redis_host, redis_port=redis_port,
)
# 检查点管理器
self.checkpoints = CheckpointManager(
agent_id=agent_id, offchain_storage=self.offchain,
ipfs_client=self.offchain.ipfs, local_storage=self.offchain.local,
)
# 会话管理器
self._session_store = SessionStore(
local_storage=self.offchain.local, cache=self.offchain.cache,
)
self.sessions = SessionManager(agent_id=agent_id, store=self._session_store)
# 备份管理器
self.backups = BackupManager(
agent_id=agent_id, offchain=self.offchain,
ipfs=self.offchain.ipfs, local=self.offchain.local,
encryption_key=encryption_key,
)
# 数据清理
self.cleanup = DataCleanupManager(
agent_id=agent_id, local_storage=self.offchain.local,
ipfs=self.offchain.ipfs, cache=self.offchain.cache,
offchain=self.offchain,
)
logger.info(f"AgentStateManager initialized for {agent_id}")
async def start(self):
if self._running:
return
self._running = True
tasks = [
self.checkpoints.start_periodic_checkpointing(),
self.sessions.start_cleanup_loop(interval=60),
self.backups.start_scheduled_backups(),
self.cleanup.start_periodic_cleanup(interval=3600),
]
self._background_tasks = [asyncio.create_task(t) for t in tasks]
logger.info(f"Started {len(tasks)} background tasks")
async def shutdown(self, graceful: bool = True):
self._running = False
if graceful:
logger.info("Graceful shutdown...")
await self.sessions.suspend_all_idle(max_idle=0)
try:
cid = await self.checkpoints.create_checkpoint(is_full=True)
logger.info(f"Final checkpoint: {cid[:16]}...")
if self.chain_client:
await self._update_onchain_pointer(cid)
try:
await self.backups.full_backup()
except Exception as e:
logger.error(f"Final backup failed: {e}")
except Exception as e:
logger.error(f"Final checkpoint failed: {e}")
for task in getattr(self, '_background_tasks', []):
task.cancel()
await self.offchain.close()
async def _update_onchain_pointer(self, cid: str):
logger.info(f"Updated on-chain pointer to {cid}")
async def collect_full_state(self) -> Dict[str, Any]:
return {
"agent_id": self.agent_id,
"collected_at": time.time(),
"version": self.state_version,
"session_count": len(self.sessions.active_sessions),
}
async def restore_from_disaster(self, backup_id: Optional[str] = None) -> bool:
logger.info(f"=== DISASTER RECOVERY for {self.agent_id} ===")
try:
if backup_id:
success = await self.backups.restore_from_backup(backup_id)
else:
last = await self.backups._get_last_backup()
if not last:
return False
success = await self.backups.restore_from_backup(last.id)
if success:
await self.checkpoints.create_checkpoint(is_full=True)
return success
except Exception as e:
logger.error(f"Recovery failed: {e}")
return False
async def health_check(self) -> Dict[str, Any]:
storage_health = await self.offchain.health_check()
return {
"agent_id": self.agent_id,
"timestamp": time.time(),
"status": "healthy" if all(storage_health.values()) else "degraded",
"storage": storage_health,
"active_sessions": len(self.sessions.active_sessions),
}
8.2 快速集成示例
async def main():
"""快速集成示例: 创建和运行 Agent 状态管理器"""
manager = AgentStateManager(
agent_id="msg1agent000000000000000000000000000000001",
db_path="/data/my_agent.db",
ipfs_url="http://localhost:5001",
redis_host="localhost",
chain_client=None, # CosmWasm 客户端
)
try:
# 启动后台服务
await manager.start()
# 创建用户会话
session = await manager.sessions.create_session(
user="msg1user000000000000000000000000000000002",
preferences={"language": "zh", "model": "gpt-4"},
)
# 更新会话上下文
await manager.sessions.update_session_context(
session.id,
new_messages=[Message(role="user", content="你好", timestamp=time.time())],
variables={"topic": "区块链"},
)
# 关闭会话
await manager.sessions.close_session(session.id)
# 获取存储使用统计
usage = await manager.cleanup.get_storage_usage()
print(f"Storage usage: {usage}")
# 健康检查
health = await manager.health_check()
print(f"Health: {health}")
finally:
# 优雅关闭
await manager.shutdown(graceful=True)
# asyncio.run(main())
8.3 链上/链下状态一致性检查
/// 链上/链下状态一致性检查 (Rust 合约端)
pub fn verify_state_consistency(
deps: Deps,
env: Env,
claimed_offchain_hash: String,
) -> StdResult<ConsistencyReport> {
let onchain_state = STATE_KEY.load(deps.storage)?;
// 链上关键字段的哈希
let onchain_core = format!(
"{}{}{}{}{}",
onchain_state.reputation,
onchain_state.total_earnings,
onchain_state.last_heartbeat,
onchain_state.active,
onchain_state.metadata_uri,
);
let onchain_hash = format!("{:x}", md5::compute(onchain_core.as_bytes()));
let consistent = onchain_hash == claimed_offchain_hash;
Ok(ConsistencyReport {
agent: onchain_state.owner.to_string(),
onchain_hash,
claimed_offchain_hash,
consistent,
checked_at: env.block.time.seconds(),
})
}
#[derive(Serialize, Deserialize, Clone, Debug, JsonSchema)]
pub struct ConsistencyReport {
pub agent: String,
pub onchain_hash: String,
pub claimed_offchain_hash: String,
pub consistent: bool,
pub checked_at: u64,
}
8.4 部署清单
部署 AI Agent 状态管理系统时的检查列表：
| 检查项 | 说明 | 优先级 |
|---|---|---|
| 确定链上/链下数据边界 | 定义哪些字段上链，哪些存 IPFS | P0 |
| 实现状态 schema 版本化 | 确保未来可迁移 | P0 |
| 配置 SQLite 数据库 | 设置 WAL 模式，定期 VACUUM | P0 |
| 部署 IPFS 节点 | 或使用托管 pinning 服务 | P0 |
| 配置 Redis 缓存 | 设置内存上限和淘汰策略 | P0 |
| 实现检查点系统 | 全量+增量检查点 | P0 |
| 实现备份调度 | 全量每日，增量每小时 | P0 |
| 配置数据清理策略 | 按分类设置保留期限 | P1 |
| 加密敏感数据 | 备份数据加密 + 传输加密 | P1 |
| 灾难恢复演练 | 每月至少一次 | P1 |
| 监控存储成本 | 跟踪 MSG 代币消耗 | P1 |
| 会话上下文继承 | 跨会话的用户上下文传递 | P2 |
| 上下文窗口压缩 | LLM token 管理 | P2 |
附录
A. 常用命令速查
# 查询 Agent 链上状态
msgcli query wasm contract-state smart msg1agent... '{"get_state":{}}'
# 更新 Agent metadata_uri
msgcli tx wasm execute msg1agent... '{"update_metadata":{"uri":"ipfs://Qm..."}}' --from admin
# 手动创建检查点 (通过链上触发)
msgcli tx wasm execute msg1agent... '{"create_checkpoint":{}}' --from admin
# 查询备份记录
msgcli query wasm contract-state smart msg1agent... '{"list_backups":{"limit":10}}'
B. 状态管理架构总览
┌─────────────────────────────────────────────────────────────────┐
│ AgentStateManager │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌──────────────┐ ┌────────────────────────┐ │
│ │ OnChainState │ │ OffChainStore│ │ CheckpointManager │ │
│ │ (Rust) │ │ (Python) │ │ (Python) │ │
│ │ -owner │ │ -SQLite │ │ -full snapshots │ │
│ │ -reputation │ │ -IPFS │ │ -incremental diffs │ │
│ │ -earnings │ │ -Redis │ │ -integrity verify │ │
│ └──────┬───────┘ └──────┬───────┘ └───────────┬────────────┘ │
│ │ │ │ │
│ ┌──────┴───────┐ ┌──────┴───────┐ ┌───────────┴────────────┐ │
│ │ SessionMgr │ │ BackupMgr │ │ DataCleanupMgr │ │
│ │ (Python) │ │ (Python) │ │ (Python) │ │
│ │ -lifecycle │ │ -full/inc │ │ -retention policy │ │
│ │ -context │ │ -encryption │ │ -cost optimization │ │
│ │ -inheritance │ │ -DR drill │ │ -archive │ │
│ └──────────────┘ └─────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
本文档针对 MSG Chain (msg-chain-1) 的 AI Agent 开发，使用 msg 地址前缀。所有代码示例仅供参考，生产环境部署请结合具体业务场景调整。
本文档内容基于 MSGChain 代码库真实状态编写，非 AI 自动生成。
主网状态: No-Go | 白皮书: https://msgchain.org/whitepaper/