多智能体记忆系统

为 AI Agent 提供跨会话持久化记忆、团队知识沉淀与语义检索能力 | Memory System API v3.0.2

1. 项目概述

1.1 背景与定位

在多智能体协作场景中，AI Agent 面临以下核心痛点：

• 无状态困境：Agent 每次对话都是从零开始，之前的交互经验无法积累
• 会话间记忆丢失：对话结束后，上下文和决策依据全部消失
• 团队知识孤岛：多个 Agent 各自独立工作，无法共享已学到的知识和经验
• 检索效率低下：当记忆积累到一定规模，关键词匹配无法满足语义级别的回忆需求
• 记忆无限膨胀：缺少自动化的清理和压缩机制，存储成本持续增长
• 知识结构化不足：原始记忆是自由文本，缺少事实提取、场景聚合和画像生成等智能化处理能力

多智能体记忆系统正是为解决这些问题而设计的。它为 AI Agent 提供了一套双层次（工作记忆 + 长期记忆）、LLM 驱动的智能生命周期管理、多团队隔离、语义检索的记忆基础设施，让 Agent 能够像人类一样"记住"、"回忆"、"总结"和"遗忘"。

1.2 核心能力

1.3 记忆层次说明

系统采用双层次记忆模型，模拟人类的短期记忆与长期记忆，同时支持基于 LLM 的场景聚合和用户画像层：

扩展数据类型

1.4 架构流程图

graph LR
    subgraph Client["客户端"]
        A[AI Agent
OpenClaw / Hermes / Claude Code / Codex]
    end

    subgraph Gateway["接入层"]
        B[Nginx
HTTPS + SSL]
        C[Auth Middleware
HMAC-SHA256 签名验证]
    end

    subgraph Core["核心服务 (Flask)"]
        D[API Routes]
        E[Memory System
业务逻辑]
        F[Lifecycle Manager]
        F1[Fact Extractor
原子事实提取]
        F2[Scenario Aggregator
场景聚合]
        F3[Persona Generator
画像生成]
        F4[Pipeline Engine
自动化引擎]
    end

    subgraph Storage["存储层"]
        G[(MySQL
长期记忆 + 向量)]
        H[(Redis
工作记忆缓存)]
        I[BGE Embedding
512 维向量化]
    end

    A -->|"HTTPS 请求
+ HMAC 签名"| B
    B --> C
    C -->|"认证通过"| D
    C -.->|"认证失败 → 401"| A
    D --> E
    E -->|"写入记忆"| G
    E -->|"缓存工作记忆"| H
    E -->|"文本→向量"| I
    E -->|"语义搜索"| G
    F -->|"压缩: 工作→长期"| E
    F -->|"清理: 过期记忆"| G
    F1 -->|"提取事实→长期记忆"| E
    F2 -->|"聚合场景"| G
    F3 -->|"生成画像"| G
    F4 -->|"自动触发"| F
    F4 -->|"自动触发"| F1

    style Client fill:#1a1f2e,stroke:#58a6ff,color:#c9d1d9
    style Gateway fill:#1a1f2e,stroke:#d29922,color:#c9d1d9
    style Core fill:#1a1f2e,stroke:#3fb950,color:#c9d1d9
    style Storage fill:#1a1f2e,stroke:#f85149,color:#c9d1d9

1.5 适用场景

2. 快速开始

2.1 前置条件

2.2 获取 API Key

2.3 第一个请求

获取 API Key 后，先调用 /health 验证连通性（无需认证）：

curl -s https://memory.lsz.name/health
# {"ok": true, "data": {"status": "ok"}}

然后用认证接口验证 Key 是否正常：

import hashlib, hmac, time, json, requests

API_KEY = "your_api_key_here"
SECRET  = "your_secret_here"
BASE    = "https://memory.lsz.name"

ts  = int(time.time())
sig = hmac.new(SECRET.encode(), str(ts).encode(), hashlib.sha256).hexdigest()

r = requests.get(f"{BASE}/stats", headers={
    "X-API-Key": API_KEY,
    "X-Timestamp": str(ts),
    "X-Signature": sig
})
print(r.json())
# {"ok": true, "data": {"personal_memories": 0, "team_memories": 0, "agents": 0, ...}}

2.4 Python SDK（MemoryClient）

推荐使用封装好的客户端类，所有签名细节自动处理：

import hashlib, hmac, time, json, requests

class MemoryClient:
    """Memory System API 客户端 — 自动处理 HMAC 签名"""

    def __init__(self, api_key, secret, base_url="https://memory.lsz.name"):
        self.api_key = api_key
        self.secret = secret
        self.base_url = base_url

    def _sign(self, method, ts, body=None):
        if method == "GET" or body is None:
            msg = str(ts).encode()            # GET: body = b""
        else:
            msg = str(ts).encode() + json.dumps(body, ensure_ascii=True).encode()
        return hmac.new(self.secret.encode(), msg, hashlib.sha256).hexdigest()

    def _headers(self, method, ts, body=None):
        return {
            "X-API-Key": self.api_key,
            "X-Timestamp": str(ts),
            "X-Signature": self._sign(method, ts, body),
        }

    def get(self, path, params=None):
        ts = int(time.time())
        return requests.get(f"{self.base_url}{path}",
                            headers=self._headers("GET", ts), params=params)

    def post(self, path, body):
        ts = int(time.time())
        return requests.post(f"{self.base_url}{path}",
                             json=body, headers=self._headers("POST", ts, body))

    def put(self, path, body):
        ts = int(time.time())
        return requests.put(f"{self.base_url}{path}",
                            json=body, headers=self._headers("PUT", ts, body))

    def delete(self, path):
        ts = int(time.time())
        return requests.delete(f"{self.base_url}{path}",
                               headers=self._headers("DELETE", ts))

    # ---- 便捷方法 ----
    def write_memory(self, agent_id, content, importance=0.5):
        return self.post("/memories/personal",
                         {"agent_id": agent_id, "content": content, "importance": importance})

    def search(self, agent_id, query, limit=10, min_score=0.3, max_chars_per_memory=0, max_total_chars=0):
        body = {"agent_id": agent_id, "query": query, "limit": limit, "min_score": min_score}
        if max_chars_per_memory:
            body["max_chars_per_memory"] = max_chars_per_memory
        if max_total_chars:
            body["max_total_chars"] = max_total_chars
        return self.post("/memories/personal/search", body)

    def get_recent(self, agent_id, limit=10):
        return self.get(f"/memories/personal/recent/{agent_id}", params={"limit": limit})

    def get_stats(self):
        return self.get("/stats")

    # ---- 生命周期方法 ----
    def extract_facts(self, agent_id, max_memories=20, delete_after=False):
        从工作记忆提取原子事实
        return self.post("/lifecycle/extract-facts",
                         {"agent_id": agent_id, "max_memories": max_memories, "delete_after": delete_after})

    def aggregate_scenarios(self, agent_id, max_memories=50):
        将记忆按主题聚合为场景块
        return self.post("/lifecycle/aggregate-scenes",
                         {"agent_id": agent_id, "max_memories": max_memories})

    def get_scenarios(self, agent_id):
        获取已存储的场景块
        return self.get(f"/lifecycle/scenarios/{agent_id}")

    def generate_persona(self, agent_id, max_items=30):
        从记忆和场景中生成用户画像
        return self.post("/lifecycle/generate-persona",
                         {"agent_id": agent_id, "max_items": max_items})

    def get_persona(self, agent_id):
        获取最新用户画像
        return self.get(f"/lifecycle/persona/{agent_id}")

    def set_pipeline_config(self, **kwargs):
        设置 Pipeline 自动化配置
        return self.post("/lifecycle/auto-config", kwargs)

    def get_pipeline_config(self):
        获取 Pipeline 自动化配置
        return self.get("/lifecycle/auto-config")

    def compress(self, agent_id, target_count=5):
        压缩工作记忆为长期记忆
        return self.post("/lifecycle/compress",
                         {"agent_id": agent_id, "target_count": target_count})


# 使用示例
client = MemoryClient("your_api_key_here", "your_secret_here")
r = client.write_memory("assistant", "用户偏好简洁回复", importance=0.8)
print(r.json())

3. 认证机制

3.1 API Key 与签名

除 /health 外，所有接口均需认证。认证方式为 API Key + HMAC-SHA256 签名。

认证模型

• 每个团队分配唯一的 API Key + Secret，Key 绑定到对应 team_id
• 同团队下多个 Agent 共用同一个 Key
• 跨团队数据隔离由系统强制保证

请求头

3.2 签名算法详解

签名公式

GET 请求:  HMAC-SHA256(secret, str(timestamp))             # body 为空字节 b""
POST 请求: HMAC-SHA256(secret, str(timestamp) + raw_body)  # body 为原始 JSON 字节

签名要点

• GET 请求使用空 body b"" 进行签名（不是 b"{}"）
• POST / PUT 请求使用实际发送的原始 JSON 字节。Python requests.post(url, json=body) 默认 ensure_ascii=True，中文字符会被转义（如 \u7b56\u7565），签名必须使用相同的序列化方式
• 时间戳过期后不可直接重试：必须用新的当前时间戳重新计算签名
• 时钟同步：客户端与服务器时钟偏差需在 ±5 分钟以内

3.3 错误码说明

认证失败时返回 {"ok": false, "error": "...", "code": "..."} 格式：

认证错误码

通用错误码

3.4 跨语言签名示例

Python

import hashlib, hmac, time

def sign(secret, timestamp, body=None):
    """body=None 表示 GET 请求，使用空字节"""
    if body is None:
        msg = str(timestamp).encode()
    else:
        msg = str(timestamp).encode() + body  # body 已经是 bytes
    return hmac.new(secret.encode(), msg, hashlib.sha256).hexdigest()

ts = int(time.time())
sig = sign("your_secret_here", ts)  # GET 请求

Node.js

const crypto = require('crypto');

function sign(secret, timestamp, body = null) {
    const msg = body
        ? Buffer.concat([Buffer.from(String(timestamp)), body])
        : Buffer.from(String(timestamp));
    return crypto.createHmac('sha256', secret).update(msg).digest('hex');
}

const ts = Math.floor(Date.now() / 1000);
const sig = sign('your_secret_here', ts);  // GET 请求

Go

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "strconv"
)

func sign(secret string, timestamp int64, body []byte) string {
    msg := strconv.FormatInt(timestamp, 10)
    if body != nil {
        msg += string(body)
    }
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write([]byte(msg))
    return hex.EncodeToString(mac.Sum(nil))
}

ts := time.Now().Unix()
sig := sign("your_secret_here", ts, nil)  // GET 请求

4. API 端点

4.1 系统

curl -s https://memory.lsz.name/health
# {"ok": true, "data": {"status": "ok"}}

curl -X GET "https://memory.lsz.name/stats" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: $(date +%s)" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": {
#   "personal_memories": 11,
#   "team_memories": 4,
#   "agents": 4,
#   "redis_ok": true,
#   "embedding_ok": true
# }}

# 响应字段:
#   personal_memories  int   个人记忆条数
#   team_memories      int   团队共享记忆条数
#   agents             int   Agent 数量
#   redis_ok           bool  Redis 连接状态
#   embedding_ok       bool  Embedding 服务状态

4.2 团队管理

curl -X POST https://memory.lsz.name/teams \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"team_id": "my_team", "name": "My Team", "description": "团队描述"}'
# {"ok": true, "data": {"id": "my_team", "name": "My Team", ...}}

# 参数:
#   team_id     string  必填  团队唯一标识
#   name        string  必填  团队名称
#   description string  可选  描述
#   config      object  可选  配置（JSON）

curl -X GET "https://memory.lsz.name/teams/my_team" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": {"id": "my_team", "name": "My Team", ...}}

curl -X DELETE "https://memory.lsz.name/teams/my_team" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": null}

4.3 Agent 管理

curl -X POST https://memory.lsz.name/agents \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "name": "研究员", "role": "research"}'
# {"ok": true, "data": {"id": "researcher", "team_id": "my_team", "name": "研究员", ...}}

# 参数:
#   agent_id string  必填  Agent 唯一标识（团队内唯一）
#   name     string  必填  显示名称
#   role     string  可选  角色描述
#   config   object  可选  配置（JSON）

curl -X GET "https://memory.lsz.name/agents/my_team/researcher" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": {"id": "researcher", "team_id": "my_team", ...}}

curl -X GET "https://memory.lsz.name/agents/my_team" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": [{"id": "researcher", ...}, {"id": "coder", ...}]}

curl -X DELETE "https://memory.lsz.name/agents/my_team/researcher" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": null}

4.4 个人长期记忆

curl -X POST https://memory.lsz.name/memories/personal \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "content": "用户偏好简洁回复，不喜欢冗长解释", "importance": 0.8}'
# {"ok": true, "data": {"id": 42, "content": "用户偏好简洁回复...", "importance": 0.8, ...}}

# 参数:
#   agent_id        string  必填  Agent 标识
#   content         string  必填  记忆内容
#   importance      float   可选  重要性评分 0.0~1.0（默认 0.5）
#   metadata        object  可选  元数据（JSON）
#   enable_dedup    bool    可选  是否启用去重（默认 true）
#   dedup_threshold float   可选  去重相似度阈值 0.0~1.0（默认 0.85）
#
# 去重行为: 当 enable_dedup=true 时，系统会将新记忆与已有记忆做向量相似度
# 比较。若存在相似度 >= dedup_threshold 的已有记忆，则跳过写入并返回该已有
# 记忆（响应中包含 dedup_skipped=true 和 dedup_score 字段）。

curl -X POST https://memory.lsz.name/memories/personal/search \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "query": "用户的沟通偏好", "limit": 5, "min_score": 0.3}'
# {"ok": true, "data": [
#   {"id": 42, "content": "用户偏好简洁回复...", "importance": 0.8, "score": 0.75,
#    "metadata": {"source": "compression", "source_items": [...]}, ...},
#   ...
# ]}

# 参数:
#   agent_id           string  必填  Agent 标识
#   query              string  可选  查询文本（自然语言），为空时返回最近记忆
#   limit              int     可选  返回条数（默认 10，最大 50）
#   min_score          float   可选  最低相似度阈值 0.0~1.0（默认 0.3）
#   max_chars_per_memory int   可选  单条记忆最大字符数（0=不限制）
#   max_total_chars    int     可选  所有返回记忆的总字符预算（0=不限制）
#
# 字符限制: max_chars_per_memory 会截断超长单条记忆（末尾追加 "...")。
# max_total_chars 在累计达到预算后停止返回更多结果。两者可组合使用。

curl -X GET "https://memory.lsz.name/memories/personal/recent/researcher?limit=10" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": [{"id": 42, "content": "...", "importance": 0.8, ...}, ...]}

# 查询参数:
#   limit  int  可选  返回条数（默认 20，最大 100）

curl -X PUT https://memory.lsz.name/memories/personal/42 \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"content": "用户偏好简洁回复，重要结论需要给出论据", "importance": 0.9}'
# {"ok": true, "data": {"id": 42, "content": "...", "importance": 0.9, ...}}

curl -X DELETE "https://memory.lsz.name/memories/personal/42" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": null}

4.5 团队共享记忆

curl -X POST https://memory.lsz.name/memories/team \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"content": "代码规范：所有接口返回 ResultBean 统一格式", "importance": 0.9}'
# {"ok": true, "data": {"id": 5, "content": "代码规范：...", "importance": 0.9, ...}}

# 参数:
#   content         string  必填  记忆内容
#   importance      float   可选  重要性评分 0.0~1.0（默认 0.5）
#   category        string  可选  分类标签（默认 "general"）
#   metadata        object  可选  元数据（JSON）
#   enable_dedup    bool    可选  是否启用去重（默认 true）
#   dedup_threshold float   可选  去重相似度阈值 0.0~1.0（默认 0.85）

curl -X POST https://memory.lsz.name/memories/team/search \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"query": "接口规范", "limit": 5, "min_score": 0.3}'
# {"ok": true, "data": [{"id": 5, "content": "代码规范：...", "score": 0.82, ...}, ...]}

# 参数:
#   query              string  可选  查询文本，为空时返回最近记忆
#   limit              int     可选  返回条数（默认 10，最大 50）
#   min_score          float   可选  最低相似度阈值（默认 0.3）
#   max_chars_per_memory int   可选  单条记忆最大字符数（0=不限制）
#   max_total_chars    int     可选  所有返回记忆的总字符预算（0=不限制）

curl -X GET "https://memory.lsz.name/memories/team/recent?limit=10" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": [{"id": 5, "content": "...", ...}, ...]}

curl -X PUT https://memory.lsz.name/memories/team/5 \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"content": "代码规范：所有接口返回 ResultBean，含 ok/data/error/code 字段"}'
# {"ok": true, "data": {"id": 5, "content": "...", ...}}

curl -X DELETE "https://memory.lsz.name/memories/team/5" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": null}

4.6 工作记忆

curl -X POST https://memory.lsz.name/memories/working \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "content": "当前正在处理用户关于量化策略的查询"}'
# {"ok": true, "data": {"agent_id": "researcher", "content": "..."}}

# 参数:
#   agent_id  string  必填  Agent 标识
#   content   string  必填  工作记忆内容
#   ttl       int     可选  TTL 秒数（默认 259200，即 72 小时）

curl -X GET "https://memory.lsz.name/memories/working/researcher?limit=20" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": [{"content": "...", "created_at": "..."}, ...]}

# 查询参数:
#   limit  int  可选  返回条数（默认 50）

curl -X DELETE "https://memory.lsz.name/memories/working/researcher" \
  -H "X-API-Key: your_api_key_here" \
  -H "X-Timestamp: <unix-ts>" \
  -H "X-Signature: <signature>"
# {"ok": true, "data": {"cleared": 5}}

4.7 生命周期管理

curl -X POST https://memory.lsz.name/lifecycle/compress \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "target_count": 5}'
# {"ok": true, "data": {
#   "compressed": 12,
#   "remaining": 3,
#   "total_before": 15,
#   "target_count": 5,
#   "summary": "用户主要关注量化策略优化..."
# }}

# 参数:
#   agent_id     string  必填  Agent 标识
#   target_count int     可选  压缩后保留的目标条数（默认 5）

# 响应字段:
#   compressed    int     已压缩的条目数
#   remaining     int     压缩后剩余的工作记忆条目数
#   total_before  int     压缩前的工作记忆总条目数
#   target_count  int     请求的目标压缩条数
#   summary       string  生成的摘要文本（可为 null）

curl -X POST https://memory.lsz.name/lifecycle/cleanup \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"max_age_days": 90, "min_importance": 0.2}'
# {"ok": true, "data": {"cleaned": 3, "remaining": 8}}

# 参数:
#   max_age_days   int     可选  最大未访问天数（默认 90）
#   min_importance float   可选  重要性阈值 0.0~1.0（默认 0.2）
#                                仅清理 importance < min_importance
#                                且 last_accessed > max_age_days 的记忆

curl -X POST https://memory.lsz.name/lifecycle/extract-facts \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "max_memories": 20, "delete_after": false}'
# {"ok": true, "data": {
#   "facts": [
#     {"id": "abc123", "content": "...", "importance": 0.8, "category": "..."},
#     ...
#   ],
#   "count": 4
# }}

# 参数:
#   agent_id      string  必填  Agent 标识
#   max_memories  int     可选  分析的最大工作记忆条数（默认 20）
#   delete_after  bool    可选  提取后是否清空工作记忆（默认 false）

# 响应字段:
#   facts   array   提取的事实列表，每条含 id/content/importance/category
#   count   int     提取的事实数量

curl -X POST https://memory.lsz.name/lifecycle/aggregate-scenes \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "max_memories": 50}'
# {"ok": true, "data": {
#   "scenarios": [
#     {"id": "sc001", "name": "...", "summary": "...", "memory_ids": ["m1","m2"]},
#     ...
#   ],
#   "count": 3
# }}

# 参数:
#   agent_id      string  必填  Agent 标识
#   max_memories  int     可选  分析的最大记忆条数（默认 50）

# 响应字段:
#   scenarios  array  场景列表，每条含 id/name/summary/memory_ids
#   count      int    场景数量

curl https://memory.lsz.name/lifecycle/scenarios/researcher \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>'
# {"ok": true, "data": [
#   {"id": "sc001", "name": "...", "summary": "...", "memory_ids": [...], "created_at": "..."},
#   ...
# ]}

curl -X DELETE https://memory.lsz.name/lifecycle/scenarios/sc001 \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>'
# {"ok": true, "data": {"deleted": "sc001"}}

curl -X POST https://memory.lsz.name/lifecycle/generate-persona \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{"agent_id": "researcher", "max_items": 30}'
# {"ok": true, "data": {
#   "persona": {
#     "id": "p001",
#     "preferences": ["...", "..."],
#     "habits": ["...", "..."],
#     "expertise": ["...", "..."],
#     "communication_style": "...",
#     "summary": "..."
#   }
# }}

# 参数:
#   agent_id   string  必填  Agent 标识
#   max_items  int     可选  分析的最大记忆+场景条数（默认 30）

# 响应字段:
#   persona.preferences        array   用户偏好列表
#   persona.habits             array   用户习惯列表
#   persona.expertise          array   擅长领域列表
#   persona.communication_style string  沟通风格描述
#   persona.summary            string  画像摘要

curl https://memory.lsz.name/lifecycle/persona/researcher \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>'
# {"ok": true, "data": {
#   "id": "p001", "agent_id": "researcher", "preferences": [...], ...
# }}

curl -X DELETE https://memory.lsz.name/lifecycle/persona/p001 \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>'
# {"ok": true, "data": {"deleted": "p001"}}

curl https://memory.lsz.name/lifecycle/auto-config \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>'
# {"ok": true, "data": {
#   "team_id": "hermes1",
#   "compress_every_n": 10,
#   "compress_target_count": 5,
#   "cleanup_idle_days": 30,
#   "cleanup_min_importance": 0.2,
#   "enabled": true,
#   "warmup_max_memories": 5,
#   "warmup_compress_every_n": 2
# }}

curl -X POST https://memory.lsz.name/lifecycle/auto-config \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{
    "compress_every_n": 10,
    "compress_target_count": 5,
    "cleanup_idle_days": 30,
    "cleanup_min_importance": 0.2,
    "enabled": true,
    "warmup_max_memories": 5,
    "warmup_compress_every_n": 2
  }'
# {"ok": true, "data": {...}}

# 参数:
#   compress_every_n         int     可选  工作记忆达到 N 条时自动压缩（0=禁用）
#   compress_target_count    int     可选  压缩后保留的目标条数（默认 5）
#   cleanup_idle_days        int     可选  自动清理未访问天数（0=禁用）
#   cleanup_min_importance   float   可选  清理的重要性阈值（默认 0.2）
#   enabled                  bool    可选  是否启用自动规则（默认 true）
#   warmup_max_memories      int     可选  暖机阈值：个人记忆少于 N 条时进入暖机模式（0=禁用）
#   warmup_compress_every_n  int     可选  暖机模式下的压缩阈值（默认 1）

curl -X POST https://memory.lsz.name/lifecycle/auto-run \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: your_api_key_here' \
  -H 'X-Timestamp: <unix-ts>' \
  -H 'X-Signature: <signature>' \
  -d '{}'
# {"ok": true, "data": {"results": [...]}}

5. 对接示例

以下示例均使用前文的 MemoryClient 类。请将 API_KEY 和 SECRET 替换为实际值。

5.1 OpenClaw 对接

场景：AI 助手 OpenClaw 记住用户偏好和团队知识。

from memory_client import MemoryClient  # 假设已封装为模块

client = MemoryClient("your_api_key_here", "your_secret_here")

# --- 1. 创建团队和 Agent ---
client.post("/teams", {"team_id": "openclaw_team", "name": "OpenClaw 团队"})
client.post("/agents", {"agent_id": "assistant", "name": "AI 助手", "role": "assistant"})

# --- 2. 写入个人记忆（用户偏好）---
client.post("/memories/personal", {
    "agent_id": "assistant",
    "content": "用户偏好简洁直接的回复，不喜欢冗长的解释",
    "importance": 0.9
})

client.post("/memories/personal", {
    "agent_id": "assistant",
    "content": "用户的技术栈：Java 后端 + WSL2 + 腾讯云新加坡",
    "importance": 0.7
})

# --- 3. 写入团队记忆（编码规范）---
client.post("/memories/team", {
    "content": "Java 接口规范：ResultBean 统一返回格式，@ControllerAdvice 全局异常处理",
    "importance": 0.8
})

client.post("/memories/team", {
    "content": "部署规范：代码变更后必须三同步——代码逻辑、内部文档、公网文档",
    "importance": 0.95
})

# --- 4. 语义检索（个人记忆）---
r = client.post("/memories/personal/search", {
    "agent_id": "assistant",
    "query": "用户的沟通风格偏好",
    "limit": 3,
    "min_score": 0.3
})
results = r.json()["data"]
for m in results:
    print(f"[{m['score']:.2f}] {m['content']}")
# 输出:
# [0.75] 用户偏好简洁直接的回复，不喜欢冗长的解释
# [0.52] 用户的技术栈：Java 后端 + WSL2 + 腾讯云新加坡

# --- 5. 语义检索（团队记忆）---
r = client.post("/memories/team/search", {
    "query": "接口返回格式规范",
    "limit": 3,
    "min_score": 0.3
})
for m in r.json()["data"]:
    print(f"[{m['score']:.2f}] {m['content']}")
# [0.82] Java 接口规范：ResultBean 统一返回格式...

5.2 Hermes 对接

场景：量化策略系统 Hermes 的多个 Agent 积累策略研究知识。

from memory_client import MemoryClient

client = MemoryClient("your_api_key_here", "your_secret_here")

# --- 1. 创建团队和 Agent ---
client.post("/teams", {"team_id": "hermes", "name": "Hermes 策略研究部"})
client.post("/agents", {"agent_id": "cloud_strategist", "name": "云策", "role": "决策调度"})
client.post("/agents", {"agent_id": "backtest_engineer", "name": "弈回", "role": "回测工程师"})
client.post("/agents", {"agent_id": "review_analyst", "name": "弈析", "role": "复盘分析"})

# --- 2. 多 Agent 写入记忆 ---
client.post("/memories/personal", {
    "agent_id": "cloud_strategist",
    "content": "趋势策略在震荡行情中表现不佳，需要增加行情识别层",
    "importance": 0.9
})

client.post("/memories/personal", {
    "agent_id": "backtest_engineer",
    "content": "回测数据源：Binance USDT 永续合约，1h K线，2024-01 至今",
    "importance": 0.7
})

client.post("/memories/personal", {
    "agent_id": "review_analyst",
    "content": "2025-06 策略复盘：最大回撤 12%，主要损失来自 6/3 闪崩",
    "importance": 0.85
})

client.post("/memories/team", {
    "content": "风控规则：单策略最大仓位 30%，总杠杆不超过 3x",
    "importance": 0.95
})

# --- 3. 查看统计 ---
r = client.get("/stats")
stats = r.json()["data"]
print(f"个人记忆: {stats['personal_memories']} 条")
print(f"团队记忆: {stats['team_memories']} 条")
print(f"Agent 数: {stats['agents']}")
print(f"Redis: {'OK' if stats['redis_ok'] else 'DOWN'}")
print(f"Embedding: {'OK' if stats['embedding_ok'] else 'DOWN'}")
# 个人记忆: 3 条
# 团队记忆: 1 条
# Agent 数: 3
# Redis: OK
# Embedding: OK

# --- 4. 获取最近记忆 ---
r = client.get("/memories/personal/recent/cloud_strategist", params={"limit": 5})
for m in r.json()["data"]:
    print(f"[{m['importance']:.1f}] {m['content'][:50]}...")
# [0.9] 趋势策略在震荡行情中表现不佳，需要增加行情识别层...

5.3 Claude Code 对接

场景：Claude Code 使用工作记忆记录当前编码上下文，任务完成后压缩为长期记忆。

from memory_client import MemoryClient

client = MemoryClient("your_api_key_here", "your_secret_here")

# --- 1. 创建团队和 Agent ---
client.post("/teams", {"team_id": "claude_code", "name": "Claude Code 团队"})
client.post("/agents", {"agent_id": "coder", "name": "编码助手", "role": "developer"})

# --- 2. 写入工作记忆（当前任务上下文）---
working_items = [
    "正在重构 memory_system.py 中的 compress 方法",
    "目标：将返回值从 {summary: string} 改为结构化 dict",
    "需要同步更新：routes.py, docs/index.html, README_INTERNAL.html",
    "测试方法：E2E curl 验证响应格式",
]
for item in working_items:
    client.post("/memories/working", {
        "agent_id": "coder",
        "content": item
    })

# --- 3. 查看工作记忆 ---
r = client.get("/memories/working/coder")
print(f"工作记忆: {len(r.json()['data'])} 条")
# 工作记忆: 4 条

# --- 4. 任务完成后，压缩为长期记忆 ---
r = client.post("/lifecycle/compress", {
    "agent_id": "coder",
    "target_count": 1
})
result = r.json()["data"]
print(f"压缩: {result['compressed']} 条 → {result['remaining']} 条")
print(f"摘要: {result['summary']}")
# 压缩: 4 条 → 1 条
# 摘要: 重构 memory_system.py 的 compress 方法，将返回值改为结构化 dict...

# --- 5. 验证：语义检索 ---
r = client.post("/memories/personal/search", {
    "agent_id": "coder",
    "query": "compress 方法重构",
    "limit": 3,
    "min_score": 0.3
})
for m in r.json()["data"]:
    print(f"[{m['score']:.2f}] {m['content'][:60]}...")
# [0.71] 重构 memory_system.py 的 compress 方法...

5.4 Codex 对接

场景：Codex Agent 记忆管理 + 验证跨团队隔离。

from memory_client import MemoryClient

client = MemoryClient("your_api_key_here", "your_secret_here")

# --- 1. 创建团队和 Agent ---
client.post("/teams", {"team_id": "codex_team", "name": "Codex 团队"})
client.post("/agents", {"agent_id": "codex_agent", "name": "Codex 助手", "role": "coder"})

# --- 2. 写入记忆 ---
client.post("/memories/personal", {
    "agent_id": "codex_agent",
    "content": "Codex 默认使用 gpt-4 模型，CLI 支持 --model 参数切换",
    "importance": 0.7
})

client.post("/memories/personal", {
    "agent_id": "codex_agent",
    "content": "Codex 配置文件位于 ~/.codex/config.toml",
    "importance": 0.6
})

# --- 3. 验证隔离：用当前 Key 查询其他团队 ---
# 尝试访问不存在的团队（预期：403）
r = client.get("/teams/other_team")
print(f"跨团队查询: {r.status_code} - {r.json().get('code')}")
# 跨团队查询: 403 - FORBIDDEN_CROSS_TEAM

# 尝试查询其他团队的 Agent（预期：403）
r = client.get("/agents/other_team/some_agent")
print(f"跨团队Agent: {r.status_code} - {r.json().get('code')}")
# 跨团队Agent: 403 - FORBIDDEN_CROSS_TEAM

5.5 跨团队隔离验证

验证不同团队之间的数据完全隔离：

from memory_client import MemoryClient

# 两个不同团队的客户端
team_a = MemoryClient("team_a_api_key", "team_a_secret")
team_b = MemoryClient("team_b_api_key", "team_b_secret")

# --- Team A 写入记忆 ---
team_a.post("/teams", {"team_id": "team_a", "name": "Team Alpha"})
team_a.post("/agents", {"agent_id": "agent_1", "name": "Agent 1"})
team_a.post("/memories/personal", {
    "agent_id": "agent_1",
    "content": "这是 Team A 的私密记忆",
    "importance": 0.9
})

# --- Team B 写入记忆 ---
team_b.post("/teams", {"team_id": "team_b", "name": "Team Beta"})
team_b.post("/agents", {"agent_id": "agent_1", "name": "Agent B1"})
team_b.post("/memories/personal", {
    "agent_id": "agent_1",
    "content": "这是 Team B 的私密记忆",
    "importance": 0.8
})

# --- 验证隔离 ---
# Team A 搜索自己的记忆
r = team_a.post("/memories/personal/search", {
    "agent_id": "agent_1", "query": "私密记忆", "limit": 10
})
a_results = [m["content"] for m in r.json()["data"]]
print(f"Team A sees: {a_results}")
# Team A sees: ["这是 Team A 的私密记忆"]  ← 看不到 Team B 的

# Team A 搜索团队记忆（也看不到 Team B 的）
r = team_a.post("/memories/team/search", {
    "query": "私密", "limit": 10
})
print(f"Team A team search: {len(r.json()['data'])} 条")
# Team A team search: 0 条  ← Team B 的团队记忆不可见

# Team A 的 Stats 只包含自己的数据
r = team_a.get("/stats")
stats = r.json()["data"]
print(f"Team A stats: {stats['personal_memories']} personal, {stats['team_memories']} team")
# Team A stats: 1 personal, 0 team  ← 只看到自己的

print("\n✅ 跨团队隔离验证通过")

5.6 记忆生命周期完整示例

场景：完整的记忆生命周期 —— 从工作记忆写入，到原子事实提取、场景聚合、画像生成，最后配置自动化 Pipeline。

from memory_client import MemoryClient

client = MemoryClient("your_api_key_here", "your_secret_here")

# --- 1. 创建团队和 Agent ---
client.post("/teams", {"team_id": "research_lab", "name": "研究实验室"})
client.post("/agents", {"agent_id": "researcher", "name": "研究员", "role": "研究分析"})

# --- 2. 写入工作记忆（模拟日常交互）---
wm_items = [
    "今天决定把量化策略的回测周期从 3 个月改为 6 个月，因为短期数据噪声太大",
    "发现 DeepSeek V4 在代码生成任务上比 MiMo 快约 40%，但中文理解稍弱",
    "用户偏好简洁直接的回复风格，不喜欢冗长的解释",
    "Redis 缓存命中率从 75% 提升到 92% 后，API 响应时间从 120ms 降到 45ms",
    "下周一需要和团队讨论 Q3 的风控模型升级方案",
]
for item in wm_items:
    client.post("/memories/working", {"agent_id": "researcher", "content": item})

print(f"工作记忆: {len(wm_items)} 条")

# --- 3. 原子事实提取 ---
# 从工作记忆中提取结构化事实，每条事实存为独立的长期记忆
r = client.extract_facts("researcher", max_memories=20, delete_after=False)
facts = r.json()["data"]["facts"]
print(f"\n提取了 {len(facts)} 条原子事实:")
for f in facts:
    print(f"  [{f['importance']:.1f}] [{f['category']}] {f['content'][:60]}...")

# --- 4. 场景聚合 ---
# 将长期记忆按主题聚合为场景块
r = client.post("/memories/personal", {
    "agent_id": "researcher",
    "content": "2025-06 策略复盘：最大回撤 12%，主要损失来自 6/3 闪崩",
    "importance": 0.85
})
r = client.post("/memories/personal", {
    "agent_id": "researcher",
    "content": "回测数据源：Binance USDT 永续合约，1h K线，2024-01 至今",
    "importance": 0.7
})
r = client.post("/memories/team", {
    "content": "风控规则：单策略最大仓位 30%，总杠杆不超过 3x",
    "importance": 0.95
})

r = client.aggregate_scenarios("researcher", max_memories=50)
scenarios = r.json()["data"]["scenarios"]
print(f"\n聚合了 {len(scenarios)} 个场景:")
for s in scenarios:
    print(f"  [{s['name']}] {s['summary'][:60]}... (关联 {len(s['memory_ids'])} 条记忆)")

# --- 5. 查看已存储场景 ---
r = client.get_scenarios("researcher")
print(f"\n已存储场景: {len(r.json()['data'])} 个")

# --- 6. 用户画像生成 ---
# 从记忆和场景中提取用户画像
r = client.generate_persona("researcher", max_items=30)
persona = r.json()["data"]["persona"]
print(f"\n用户画像:")
print(f"  偏好: {persona.get('preferences', [])}")
print(f"  习惯: {persona.get('habits', [])}")
print(f"  擅长: {persona.get('expertise', [])}")
print(f"  沟通风格: {persona.get('communication_style', '')}")
print(f"  摘要: {persona.get('summary', '')[:80]}...")

# --- 7. 查看用户画像 ---
r = client.get_persona("researcher")
print(f"\n画像已存储, ID: {r.json()['data'].get('id')}")

# --- 8. 配置 Pipeline 自动化 ---
# 设置自动压缩：每 10 条工作记忆自动压缩
# 设置自动清理：60 天未访问且重要性 < 0.2 的记忆自动清理
# 设置暖机：新 Agent 前 5 条记忆使用更频繁的压缩
r = client.set_pipeline_config(
    compress_every_n=10,
    compress_target_count=5,
    cleanup_idle_days=60,
    cleanup_min_importance=0.2,
    enabled=True,
    warmup_max_memories=5,
    warmup_compress_every_n=2
)
print(f"\nPipeline 配置: {r.json()['data']}")

# --- 9. 查看 Pipeline 配置 ---
r = client.get_pipeline_config()
config = r.json()["data"]
print(f"自动压缩: 每 {config.get('compress_every_n')} 条触发")
print(f"自动清理: {config.get('cleanup_idle_days')} 天未访问, 重要性 < {config.get('cleanup_min_importance')}")
print(f"暖机模式: 前 {config.get('warmup_max_memories')} 条记忆启用")

# --- 10. 语义检索（验证所有数据可检索）---
r = client.post("/memories/personal/search", {
    "agent_id": "researcher",
    "query": "策略回测",
    "limit": 5,
    "min_score": 0.3
})
print(f"\n搜索 '策略回测': {len(r.json()['data'])} 条结果")
for m in r.json()["data"]:
    src = m.get("metadata", {}).get("source", "normal")
    print(f"  [{m['score']:.2f}] [{src}] {m['content'][:60]}...")

print("\n✅ 记忆生命周期完整示例结束")

6. 数据管理

6.1 数据持久化说明

6.2 记忆清理与压缩策略

工作记忆压缩（/lifecycle/compress）

将 Redis 中的工作记忆压缩为 MySQL 长期记忆：

• 从 Redis 读取指定 Agent 的全部工作记忆
• 调用 LLM 生成摘要
• 摘要写入 MySQL 长期记忆
• 原始工作记忆从 Redis 删除

长期记忆清理（/lifecycle/cleanup）

基于双维度清理过期的长期记忆：

• 时间维度：max_age_days — 清理超过指定天数未访问的记忆
• 重要性维度：min_importance — 仅清理重要性评分低于阈值的记忆

两个条件同时满足才会被清理：即 低重要性 + 长期未访问 的记忆才会被移除。高重要性的记忆即使长期未访问也不会被清理。

6.3 智能生命周期策略

原子事实提取（/lifecycle/extract-facts）

替代简单压缩，从工作记忆中提取独立的结构化事实：

• LLM 分析工作记忆内容，识别独立的、长期有价值的观察、决策和知识
• 每条事实作为独立的长期记忆存储，便于精确检索
• 适合需要精确回忆的场景（如技术决策、用户偏好、关键数据）
• 相比压缩，事实提取保留了信息的独立性和可检索性
• 系统对 LLM 返回结果和工作记忆条目均有类型安全校验，非标准格式数据会被自动跳过并记录警告日志

场景聚合（/lifecycle/aggregate-scenes）

将相关的长期记忆按主题聚合为场景块：

• LLM 分析多条记忆，识别共同主题并生成场景摘要
• 场景块保留原始记忆 ID 列表，支持溯源
• 适合将碎片化的记忆整合为完整的上下文（如项目进展、研究发现）

用户画像生成（/lifecycle/generate-persona）

从记忆和场景中提取结构化用户画像：

• 输出：偏好列表、习惯列表、擅长领域、沟通风格、综合摘要
• 适合为 Agent 提供长期的用户/上下文理解
• 建议定期更新（如每周或每月），画像会随记忆积累而丰富

Pipeline 自动化（/lifecycle/auto-config）

配置自动化的记忆管理规则，无需手动触发：

Pipeline 在每次写入工作记忆时自动检查是否达到压缩阈值，无需手动调用。

6.4 last_accessed 字段维护

系统自动维护每条长期记忆的 last_accessed 时间戳，确保清理策略基于"最近使用时间"而非"创建时间"：

7. 常见问题