基于openEuler鲲鹏平台的Agent Memory记忆管理系统实现指南

Agent Memory 记忆管理系统，可以理解为给 AI Agent 增加一套长期记忆仓库：对话结束后，关键信息仍然保留；下次交互，Agent 能直接读取旧记忆并继续工作。在 2026 中国国际大学生创新大赛的 openEuler 方向赛题里，这套系统还要求跑在鲲鹏平台上，这就把问题从“写一个功能 Demo”变成了“在 ARM 架构 + openEuler 环境里做一次完整可验证的工程实现”。这篇文章适合准备参赛、需要从零搭建项目的学生团队，也适合刚接触 openEuler 和 Agent 应用的后端开发者。我会按照“先理解命题，再搭环境，再设计系统，再实现验证”的顺序，把整个项目的关键环节拆开讲。

1. 先想清楚 Agent Memory 在这道赛题里要解决什么问题

1.1 记忆能力为什么是 Agent 从“能用”到“好用”的拐点

大部分 Agent 在没有记忆系统时，本质上只是一个“无状态函数调用器”：每次请求进来，它只能看到当前用户输入和有限的上下文窗口。用户上次说过什么、做过什么选择、有哪些长期偏好，它一概不知道。比如用户在第一轮对话里明确说“我以后希望所有报告都用简洁风格”，下一轮如果 Agent 生成内容时完全忽略这句话，体验就会非常割裂。

Agent Memory 要解决的就是这个状态管理问题。记忆系统通常在 Agent 外部保存结构化或半结构化的记忆数据，在每次交互前把相关记忆取出，拼入 Prompt 或作为上下文输入。这样 Agent 不用把全部历史都塞进有限窗口，也能做到“记住该记住的，忘掉该忘掉的”。

从大赛角度看，这个方向的价值在于：大模型能力已经比较强，但工程落地时常卡在上下文长度、多轮一致性、个性化体验这些问题上。一个设计得当的 Agent Memory 模块，能让整个 Agent 系统的智能感提升一个档次，而且可以独立演示、独立测试，非常适合做项目创新点。

1.2 赛题隐藏的考察点：不只是功能，而是平台适配能力

这道赛题的题目里明确出现了“鲲鹏平台”和“openEuler”，这是很多团队容易低估的地方。许多学生在笔记本上写完 Python 代码，x86 架构运行没问题，就以为项目完成了。但到比赛演示或评测时，评委可能要求现场跑在鲲鹏服务器上，或者要求提供在 ARM 架构下的运行记录。这时候才发现依赖包没有 ARM 版本、Docker 镜像拉不下来、某个 C 扩展需要本地编译，整个项目直接卡住。

所以，Agent Memory 这道题真正考核的，不只是“你会不会用 Redis 存记忆”，而是：

• 能不能在 openEuler 系统上稳定安装依赖；
• 能不能在 ARM 架构下处理兼容性问题；
• 能不能给出清晰的部署、验证和性能测试流程；
• 能不能展示记忆系统在鲲鹏硬件上的实际运行效果。

这个理解会直接影响你的项目计划。我建议团队一拿到题，不要先急着写 AI 逻辑，而是先把 openEuler 环境搭起来，把最基础的“写入一条记忆、读取一条记忆”跑通。平台稳了，后面所有功能才有意义。

2. 基于鲲鹏平台的项目前置准备：openEuler、SSH、yum 源和 Docker

2.1 环境选型：openEuler 版本、架构和机器规格

openEuler 的版本选择，直接影响软件包兼容性。常见做法是选择长期支持（LTS）版本，比如 openEuler 22.03 LTS SP4 或更新的 SP 版本。这类版本维护周期长，软件源比较稳定，适合比赛项目踩坑。如果你们是从 CentOS 7.5 这类旧系统迁移过来，查兼容性时也要优先看 openEuler 版本升级文档，而不是凭印象直接替换。

装系统前，先确认机器是鲲鹏 920 这类 ARM 处理器，还是 x86 架构。可以用这条命令看体系结构：

uname -m

如果在鲲鹏上，正常输出是aarch64。这个信息很重要，因为后面安装任何软件、拉取 Docker 镜像、下载预编译 wheel 包，都要判断是否有对应的 ARM 版本。

硬件规格方面，我建议最低 4 核 CPU、8GB 内存。如果 Agent 记忆系统里要跑向量检索或者同时演示多个服务，内存最好升到 16GB。磁盘至少预留 40GB，因为 Docker 镜像和编译工具链会占用不少空间。要注意：能在低配机器上跑通，和能在批量任务下长期稳定跑，是完全不同的两件事。比赛演示可以先用小规格，但写文档时要说明生产环境建议的规格。

2.2 系统初始化：SSH、用户权限、yum 源

openEuler 安装完成后，第一步是开启 SSH 远程登录。很多团队习惯在自己电脑上写代码，然后把代码传到服务器上跑，没有 SSH 会非常难受。安装系统时如果没有选择安装 SSH Server，可以用下面的命令启动：

systemctl enable --now sshd

然后检查服务状态：

systemctl status sshd

如果用的是云主机或实验平台，还要在安全组或防火墙放行 22 端口。这一步看起来基础，但每年都有团队到了演示前才发现远程连不上。

接下来是配置 yum 源。openEuler 默认会带官方软件源，但如果你的机器在同一时间大量安装依赖，或者网络环境特殊，建议先确认源可用：

yum repolist

如果需要更换为镜像源或内网源，先备份原始配置：

cp /etc/yum.repos.d/openEuler.repo /etc/yum.repos.d/openEuler.repo.bak

然后根据你的 openEuler 版本，选择合适的镜像源地址。这里不要照抄别人的配置，因为不同版本的仓库路径可能不一样。配置完成后执行：

yum clean all yum makecache

看到元数据缓存完成，再继续后续安装。如果这一步失败，不要急着装其他软件，先把源修好，否则后面会连环报错。

2.3 在 openEuler 上安装 Docker 并准备项目运行环境

Agent Memory 系统一般会依赖 Redis、数据库或向量检索服务，用手动安装方式也能跑，但 Docker 能把环境隔离做得更干净，尤其适合比赛现场演示：一个docker-compose up就能拉起整套服务。

openEuler 上安装 Docker，可以使用系统软件源，也可以使用 Docker 官方源。以系统源安装为例：

yum install -y docker

安装完成后启动服务：

systemctl enable --now docker

然后验证：

docker info

如果没有报错，说明 Docker 基本可用。但要注意：鲲鹏平台的 Docker 镜像必须支持linux/arm64架构。拉取 Redis 镜像时，可以指定平台参数：

docker pull --platform linux/arm64 redis:7

如果网络环境下载很慢，可以配置镜像加速器。但比赛节点上不要过度依赖公网拉取，最好提前把需要的镜像docker save保存下来，现场docker load导入。这是很容易被忽略的细节，但也是最容易在演示时救命的操作。

如果你们的 Agent 逻辑用 Python 写，我还会先装好编译工具链和虚拟环境依赖：

yum install -y gcc gcc-c++ make python3-devel python3 -m venv venv source venv/bin/activate pip install --upgrade pip

为什么要先装python3-devel？因为有些 Python 依赖包在 ARM 架构下没有预编译 wheel，pip 会尝试从源码编译。没有编译工具链，安装会直接失败。提前装好，能少踩很多坑。

如果项目里必须使用 conda 管理环境，也要先确认 conda 是否能安装在aarch64架构上，并选择对应版本。同样，如果团队里有人习惯 Java 1.8 环境，也要先验证鲲鹏 + openEuler 下 JDK 的兼容性。这些都属于前置准备，越早确认越好。

3. Agent Memory 系统的核心设计：存储什么、怎么存、怎么读

3.1 记忆类型的划分：短期会话、长期偏好、事实型知识

设计记忆系统之前，首先要区分记忆的类型。不同类型的记忆，留存时间、访问频率和一致性要求都不一样。我见过很多项目把用户所有历史都丢进一个 JSON 字段里，看着简单，但后续检索和清理会非常痛苦。

比较常用的划分方式有三种：

• 短期会话记忆：当前一次会话中的上下文。比如上一轮的 query、回复、临时状态。这种记忆只在会话活跃期有意义，过期时间通常设置为几十分钟到几小时。
• 长期偏好记忆：用户明确的偏好和习惯。比如“使用中文回答”“报告要简洁”“我经常在晚上使用”。这类记忆需要跨会话保留，并且可以由用户主动修改或删除。
• 事实型知识记忆：从对话中提取出来的客观信息。比如“用户所在城市是深圳”“项目截止日期是 6 月 1 日”。这类信息需要准确、可核对，不能随便被覆盖。

在项目代码里，应该用memory_type字段明确标记每一条记忆。这样后续写清理策略时，就能针对不同类别使用不同的 TTL 和优先级，而不是一刀切。

3.2 存储选型：Redis、关系型、向量检索的取舍

存储选型是 Agent Memory 系统里最核心的决策之一。比赛项目不需要追求大而全，但要能讲清楚为什么选某个存储。

如果你的重点是“记忆管理”，也就是存储、更新、过期、检索，那么 Redis 是一个非常合适的选择。Redis 的键值结构天然适合写入和读取高频记忆，TTL 机制可以方便地处理短期记忆过期。项目演示时，用redis-cli直接查看记忆条目的变化，也非常直观。

如果记忆数据需要复杂的条件查询，比如按用户 ID 和时间范围筛选，可以引入关系型数据库，比如 openEuler 上运行 PostgreSQL 或 openGauss。但这会增加部署复杂度。比赛项目里，我更建议以 Redis 为主，关系型作为可选项。

如果 Agent 需要做“语义检索”，也就是根据用户当前的表述，找到历史上含义相近的记忆，而不是只做精确匹配，那么需要引入向量数据库或向量检索能力。常见做法是把记忆内容用 Embedding 模型转成向量，然后用 faiss、Milvus 等工具做相似度检索。但要注意：Embedding 模型参数量不小，在鲲鹏平台 CPU 环境下运行耗时不可忽略。如果比赛时间紧张，可以先不做向量检索，用关键词匹配 + 标签组合检索来演示，效果也足够。

下面是一个简单的存储选型对比：

3.3 记忆生命周期：写入、更新、过期、清理策略

记忆不能只负责写进去，还要考虑什么时候失效、什么时候清理。没有清理机制，系统跑一天内存就会涨满；没有更新机制，用户改了口径，系统还保留旧记忆，就会答非所问。

我建议给每条记忆定义这样几个字段：

{ "memory_id": "uuid", "agent_id": "agent_01", "user_id": "user_42", "memory_type": "preference", "content": "用户喜欢简洁风格", "metadata": { "source": "chat", "importance": 0.8 }, "created_at": "2026-06-01T10:00:00Z", "updated_at": "2026-06-01T10:00:00Z", "expire_at": "2026-06-08T10:00:00Z" }

写入时要注意幂等。同一用户同一类型的同一内容，不应该反复写入多条。更合理的做法是先查重，如果已经存在，就更新updated_at和content即可。

过期策略可以根据memory_type区分：

• 短期会话记忆：TTL 设为 30 分钟或 1 小时。
• 长期偏好记忆：不设置过期，但允许用户主动删除或修改。
• 事实型知识记忆：设置较长 TTL，比如 30 天，并且可以人工回滚。

清理策略建议用定期扫描而非等内存爆了再做。在 Redis 中可以用SCAN命令遍历带有过期时间的键，或者依赖 Redis 自身的 TTL 机制惰性清理。如果整体偏工程化，可以写一个调度任务，每小时清理一次过期记忆，并记录清理数量到日志。

还要考虑隐私边界。比赛项目在宣传时不要承诺“永久保存用户所有数据”，而是应该在设计文档里说明：用户有权删除记忆，系统不会主动记录明文密码、银行卡号等敏感信息。这一点评委很看重，也符合行业规范。

4. 从零实现一个可演示的 Agent Memory 模块

4.1 工程目录和最小能跑通的结构

一个适合比赛的 Agent Memory 项目，不需要一开始就设计成微服务架构。我建议用单服务 + Redis 的方式，先把主流程跑通，再考虑扩展。下面是一个参考目录：

agent-memory/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI / Flask 入口 │ ├── memory_store.py # 记忆存储接口 │ ├── agent_service.py # Agent 主逻辑 │ ├── config.py # 配置项 │ └── prompts.py # Prompt 拼接 ├── scripts/ │ ├── reset_demo.sh # 清理数据并重启服务 │ └── test_demo.py # 演示测试脚本 ├── requirements.txt ├── docker-compose.yml └── README.md

这个结构的好处是：入口、存储、Agent 逻辑分离，即使团队分工，也不会互相阻塞。而且演示时可以直接说“这是存储层，这是业务层”，让评委看到你的设计思路。

4.2 记忆写入与检索接口示例

我用 Python + Redis 写一个简化版存储接口。代码不复杂，但足够演示核心逻辑。

import json import uuid from datetime import datetime, timedelta, timezone import redis r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) def memory_key(user_id: str, memory_id: str) -> str: return f"agent:main:user:{user_id}:memory:{memory_id}" def save_memory( user_id: str, memory_type: str, content: str, metadata: dict | None = None, ttl_seconds: int | None = None, ) -> str: memory_id = str(uuid.uuid4()) now = datetime.now(timezone.utc).isoformat() expire_at = None if ttl_seconds: expire_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat() memory_data = { "memory_id": memory_id, "agent_id": "main", "user_id": user_id, "memory_type": memory_type, "content": content, "metadata": metadata or {}, "created_at": now, "updated_at": now, "expire_at": expire_at, } key = memory_key(user_id, memory_id) r.set(key, json.dumps(memory_data, ensure_ascii=False)) # 维护一个按用户区分的记忆ID列表，便于批量获取 list_key = f"agent:main:user:{user_id}:memory_ids" r.sadd(list_key, memory_id) # 如果设置了 TTL，同时给 key 设置 Redis 过期时间 if ttl_seconds: r.expire(key, ttl_seconds) return memory_id

检索时，可以按用户拿到全部记忆 ID，再读取内容。如果要做简单的关键词过滤，可以在 Python 层做，也可以使用 Redis 的集合和哈希组合。下面是一个按用户和记忆类型读取的示例：

def get_memories(user_id: str, memory_type: str | None = None) -> list[dict]: list_key = f"agent:main:user:{user_id}:memory_ids" memory_ids = r.smembers(list_key) result = [] for memory_id in memory_ids: key = memory_key(user_id, memory_id) raw = r.get(key) if not raw: r.srem(list_key, memory_id) continue data = json.loads(raw) if memory_type and data.get("memory_type") != memory_type: continue result.append(data) # 按创建时间倒序 result.sort(key=lambda x: x.get("created_at", ""), reverse=True) return result

这段代码里有两个细节值得注意：一是存储时用ensure_ascii=False，保证中文内容在 Redis 里可读；二是读取时如果 key 已经不存在，就从集合里移除。这个“惰性清理”能在演示时避免脏数据残留。

4.3 把记忆接入 Agent 对话流程

有了存储接口，下一步就是接进 Agent 对话流程。最简单的接法是：用户输入前，先取出历史记忆，拼入系统 Prompt。

def build_agent_prompt(user_id: str, user_input: str) -> str: memories = get_memories(user_id) memory_context = "\n".join( f"[{m['memory_type']}] {m['content']}" for m in memories ) system_prompt = f""" 你是基于鲲鹏平台运行的 Agent 助手。 以下是用户的长期记忆，回答时请合理参考： {memory_context} 当前用户输入：{user_input} """ return system_prompt

这里没有强制要求 Agent 一定要用某一个大模型。比赛项目里可以用云端 API，也可以本地部署一个较小的开源模型。但要注意：如果现场网络不稳定，模型调用会失败。更稳妥的方式是提供一个“演示模式”，在模型不可用时，用规则匹配返回预设答案，保证记忆模块本身可以被独立验证。

对话主流程可以这样设计：

1. 用户输入。
2. 调用get_memories获取相关记忆。
3. 拼到 Prompt 中。
4. 调用大模型获得回复。
5. 回复结束后，从对话中提取新的记忆，调用save_memory保存。
6. 返回回复给用户。

记忆提取这一步，最简单的方法是规则提取，比如识别“我喜欢 XXX”“我不喜欢 XXX”“我在 XXX”这样的句式。如果要做得更智能，可以额外调用一次大模型做信息抽取，但会增加延迟。比赛演示时，规则提取已经足够。

5. 在鲲鹏环境里做验证：性能、并发和稳定性怎么判断

5.1 功能验证：一条消息记住上下文

功能验证的目标是回答一个问题：Agent 到底有没有真的记住。我建议写一个自动化演示脚本，流程如下：

第1轮：用户输入 "我喜欢极简风格" 第2轮：用户输入 "以后生成的报表都用什么风格？" 预期：Agent 回答包含 "极简风格"

这个测试看起来很基础，但非常关键。它能同时验证记忆写入、记忆读取、Prompt 拼接和 Agent 回复四个环节是否正常。如果第 2 轮没有命中记忆，说明问题可能出在写入、读取或排序任一环节，需要用日志和 Redis 里的实际数据进一步定位。

除了正确性，还要验证“记忆更新”是否生效。比如第 1 轮说“我喜欢极简风格”，第 2 轮说“我现在更喜欢商务风格”，第 3 轮问“报表风格”，应该返回“商务风格”。如果系统返回旧记忆，说明更新逻辑有问题。

5.2 性能验证：吞吐、延迟、资源占用

比赛答辩时，评委很可能会问：“你们的记忆系统性能怎么样？”这时候如果没有数据，回答就会很空。我建议至少测三个指标：

• 延迟：从发起请求到返回结果，统计 P50、P95、P99 延迟。
• 吞吐：单位时间内能处理多少次记忆写入或读取。
• 资源占用：Redis 内存、Python 进程 CPU、系统内存等。

可以写一个简单的并发压测脚本，用协程模拟多个用户同时写入记忆。注意不要一上来就开 1000 并发，先跑 10、50、100 个用户，观察延迟变化。一个常见的结果是：并发数较低时，延迟稳定；并发数升高后，P99 延迟快速上升，这时候要检查 Redis 连接池、Python GIL 和机器核数。

在鲲鹏平台上的一个优势是核心数通常较多，适合并发任务。你可以尝试把并发数设置为 CPU 核数的 2 到 4 倍，观察资源利用情况。如果 CPU 没有跑满但延迟已经很高，问题往往不在硬件，而在代码里的串行等待。

5.3 稳定性验证：批量任务和长时间运行

性能好的系统不一定稳定。比赛项目至少要跑一次长时间稳定性测试，比如连续运行 2 小时，不断写入临时记忆，然后观察 Redis 内存是否持续增长、是否出现连接超时、短期记忆是否按 TTL 过期。

批量任务还要看失败重试和输出一致性。写入 10000 条记忆，中间有网络抖动，Redis 连接断开，系统能不能恢复？建议在代码里给 Redis 操作加简单的重试机制：

def save_with_retry(user_id, memory_type, content, retries=3): for i in range(retries): try: return save_memory(user_id, memory_type, content) except redis.exceptions.ConnectionError: print(f"Redis connection error, retry {i + 1}") time.sleep(0.5) raise RuntimeError("save memory failed after retries")

如果是在鲲鹏平台演示 Docker 服务，稳定性测试还可以包含“重启 Docker 容器后数据是否还在”。这取决于 Redis 是否开启了持久化。如果使用默认配置，容器重启可能丢失内存数据。比赛演示前，最好确认 Redis 持久化策略，或者演示脚本里明确“这是一个可重置的示例环境”，避免评委误以为数据永久丢失。

6. 参赛落地最容易踩的坑和对应排查顺序

6.1 环境坑：版本、架构、依赖编译

很多团队刚开始在本地 x86 上一切正常，一到鲲鹏服务器就报错。最常见的问题有：

• uname -m显示 aarch64，但 pip 尝试安装 x86 的 wheel。
• openEuler 默认 Python 版本和本地不同，某些依赖没有 ARM 编译产物。
• Docker 镜像没有 arm64 版本，拉下来后启动失败。
• 编译 C 扩展时报缺少头文件，本质是没装python3-devel和 gcc。

遇到这类问题，不要急着改代码。先按这个顺序排查：

1. 确认系统架构：uname -m 2. 确认 openEuler 版本：cat /etc/openEuler-release 3. 确认 Python 版本：python3 --version 4. 确认是否安装了编译工具链：gcc --version 5. 确认 Redis 是否正常运行：redis-cli ping 6. 确认 Docker 镜像平台：docker inspect <image> | grep Architecture

只要环境没问题，很多“代码报错”其实不会发生。

6.2 数据坑：序列化、过期、脏数据

记忆系统最容易出的数据问题有三个：

第一个是序列化混乱。Python 里存的是 dict，取出来如果是字符串，没有做json.loads，后续访问字段就会报错。排查时直接在 Redis 里查看 key 对应的值，确认保存的格式。

第二个是过期时间设错。有些短期记忆本来应该几十秒后过期，结果忘了设置ttl_seconds，导致 Redis 内存一直上涨。排查时可以执行：

redis-cli info memory

看到used_memory持续上涨，优先检查有没有大量没有 TTL 的 key。

第三个是脏数据残留。用户删除了某条记忆，但记忆 ID 列表里还保留着 ID。所以读取时要像 4.2 节那样，做一次“key 不存在就从集合里移除”的清理。否则列表越来越大，最终影响性能。

6.3 演示坑：日志、可观测性、恢复演示

比赛演示不像平时开发，现场压力下大概率会出现意外。最怕的不是报错，而是报错后不知道发生了什么。所以我建议从第一天就重视日志。

日志至少要包含这些信息：

• 每次记忆写入的关键字段：用户 ID、记忆类型、TTL。
• 每次记忆读取命中的条数和内容摘要。
• Redis 连接异常时的堆栈。
• 模型调用失败时的降级方案。

可以在入口处打印简化日志：

print(f"[MEMORY] save user={user_id} type={memory_type} content={content}") print(f"[MEMORY] get user={user_id} hit={len(memories)}")

演示脚本要准备一个reset_demo.sh，一键清理数据、重启 Redis、启动服务。这样如果现场数据被改乱，可以快速恢复：

#!/bin/bash echo "Stopping services..." docker-compose down echo "Cleaning data..." rm -rf ./data echo "Starting services..." docker-compose up -d echo "Waiting for Redis..." sleep 3 echo "Demo environment reset done."

最后一点，大模型调用必须有降级方案。比赛现场如果公网模型 API 超时，不要卡在那里。可以用“离线模式”返回基于记忆的规则回答，让评委看到记忆系统本身是工作的，只是模型服务临时不可用。这个细节做得好，会给评委留下工程准备充分的印象。

踩过几次之后我发现，这类赛题真正拉开差距的不是谁用了更炫的模型，而是谁能把环境、数据、验证这三件事处理得更干净。用鲲鹏平台的时候，很多问题其实不是代码逻辑，而是系统的版本和依赖没有对齐。如果你们团队准备报名，我建议第一天就把 openEuler 环境装好，把最小的“写入-读取-清理”流程跑通，再往里面加 Agent 对话和大模型能力。