在 Mastra 中使用 @mastra/elasticsearch：向量存储、语义检索与状态存储完整实战指南

在 Mastra 中使用 mastra/elasticsearch向量存储、语义检索与状态存储完整实战指南【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra本文以 Mastra 开源仓库中的stores/elasticsearch包为核心系统讲解如何基于 Elasticsearch 8.x 搭建向量存储Vector Store、完成索引创建、向量写入、相似度检索、按 ID/过滤条件更新与删除等完整数据生命周期操作并延伸介绍包内提供的ElasticSearchStore存储适配器Memory / Workflows / Scores 领域与本地测试环境搭建方式。读完本文你将掌握在 Mastra 应用中接入 Elasticsearch 的全部配置项、核心 API 参数与底层实现原理可直接复制代码用于生产级 RAG 与 Agent 记忆场景。一、包概览mastra/elasticsearch提供了什么mastra/elasticsearch是 Mastra 官方维护的 Elasticsearch 集成包位于仓库的 stores/elasticsearch 目录。从包的入口文件 src/index.ts 可以看到它对外暴露了两大类能力向量能力ElasticSearchVector向量存储类与ElasticSearchVectorFilter过滤条件类型用于向量相似度检索和索引管理存储能力ElasticSearchStore通用存储适配器以及MemoryElasticSearch、WorkflowsElasticSearch、ScoresElasticSearch三个领域存储分别对应 Agent 记忆threads/messages、工作流状态与评分数据的持久化。包的依赖声明见 package.json显示它基于官方elastic/elasticsearch客户端^8.19.2并将mastra/core1.61.0-0作为 peerDependency因此它要求项目本身已安装对应版本的 Mastra 核心包。二、安装与本地环境准备1. 安装依赖npm install mastra/elasticsearch2. 本地启动 Elasticsearch测试/开发仓库在 docker-compose.yaml 中提供了开箱即用的单节点 Elasticsearch 9.4.3 配置覆盖了开发与测试所需的全部要点services: elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:9.4.3 platform: linux/amd64 ports: - 9200:9200 volumes: - elasticsearch_data:/usr/share/elasticsearch/data environment: - discovery.typesingle-node - xpack.security.enabledfalse - xpack.security.http.ssl.enabledfalse - ES_JAVA_OPTS-Xms512m -Xmx512m restart: always healthcheck: test: [CMD-SHELL, curl -s http://localhost:9200/_cluster/health || exit 1] interval: 10s timeout: 5s retries: 30 volumes: elasticsearch_data: driver: local其中discovery.typesingle-node用于单节点部署xpack.security.enabledfalse关闭安全认证便于本地调试healthcheck通过_cluster/health接口做健康检查。包内的测试脚本见 package.json正是先docker compose up -d再轮询健康接口直至集群状态变为green或yellow后才运行 vitest运行结束执行docker compose down -v清理你可以照此流程复现完整的本地测试环境。三、向量存储核心用法继承 README 并深入1. 创建客户端两种连接方式ElasticSearchVector的构造函数见 src/vector/index.ts支持两种互斥的配置形态由ElasticSearchVectorConfig类型约束方式一传入连接参数由包内部基于elastic/elasticsearch创建客户端方式二传入已配置好的Client实例便于与ElasticSearchStore或其他组件共享同一个连接。import { ElasticSearchVector } from mastra/elasticsearch; // 方式一url 可选 auth const vectorDB new ElasticSearchVector({ url: http://localhost:9200, id: my-vector-store, auth: { apiKey: insert-api-key }, }); // 方式二复用外部 client import { Client } from elastic/elasticsearch; const client new Client({ node: http://localhost:9200 }); const vectorDB2 new ElasticSearchVector({ id: my-vector-store, client });auth字段支持三种形式见源码中导出的ElasticSearchAuth类型认证方式配置形态适用场景API Key{ apiKey: string }使用 Elasticsearch 生成的 API Key推荐生产使用用户名密码{ username: string; password: string }基础认证需服务端开启安全特性Bearer Token{ bearer: string }基于 Token 的认证如云服务需要注意的是源码在构造时做了严格校验两者都未提供时会抛出ELASTIC_SEARCH_CONSTRUCTOR_ERROR错误提示 Invalid config: provide either { client } or { url }.该行为有对应的测试用例覆盖见 src/vector/index.test.ts。此外包内创建的客户端会设置name: mastra-elasticsearch与携带包版本号的user-agent便于在服务端日志中识别请求来源。2. 创建索引createIndexawait vectorDB.createIndex({ indexName: my_vectors, dimension: 3, metric: cosine, // 可选 euclidean | dotproduct默认 cosine });createIndex的参数说明参数类型说明indexNamestring要创建的索引名对应 Elasticsearch index 名称dimensionnumber向量维度必须为正整数否则抛出CREATE_INDEX/INVALID_ARGS用户错误metriccosine \| euclidean \| dotproduct向量相似度度量默认cosine从源码看createIndex底层调用client.indices.create生成的 mapping 结构为metadata字段使用type: objectembedding字段使用type: dense_vector并设置dims、index: true与similarity。值得注意的细节是度量与 Elasticsearch 原生类型的映射关系源码中定义了正反两张映射表cosine → cosine euclidean → l2_norm dotproduct → dot_producteuclidean对应 Elasticsearch 的l2_norm、dotproduct对应dot_productdescribeIndex查询索引信息时再通过反向映射还原为 Mastra 的度量名称。幂等语义重点如果索引已存在createIndex不会直接失败而是进入validateExistingIndex校验逻辑源码中通过捕获 already exists 错误触发维度一致跳过创建并记录 info 日志维度一致但度量不同跳过创建并记录 warn 日志提示需要删除重建索引才能切换度量维度不一致抛出VALIDATE_INDEX/DIMENSION_MISMATCH错误。对应的测试用例见 src/vector/index.test.ts明确验证了重复创建同维度索引不抛错、维度变更抛错的幂等行为因此你可以放心地在应用启动阶段重复调用createIndex。3. 写入向量upsertconst ids await vectorDB.upsert({ indexName: my_vectors, vectors: [ [0.1, 0.2, 0.3], [0.3, 0.4, 0.5], ], metadata: [{ text: doc1 }, { text: doc2 }], });upsert的完整参数为{ indexName, vectors, metadata?, ids? }参数说明vectors二维数组每个元素是一个向量写入前会与索引维度做一致性校验validateVectorDimensions维度不匹配直接报错metadata与向量一一对应的元数据对象数组缺省为空对象{}ids可选的自定义 ID 数组不传时内部使用crypto.randomUUID()自动生成底层实现使用 ElasticsearchBulk API批量写入每条操作为index 文档交替排列并设置refresh: true保证写入后立即可检索。源码对批量响应做了逐条错误解析若存在部分失败会抛出UPSERT/BULK_PARTIAL_FAILURE错误并在错误详情中携带失败项 ID 与原因便于你定位脏数据全部成功时返回写入的 ID 数组。4. 相似度检索queryconst results await vectorDB.query({ indexName: my_vectors, queryVector: [0.1, 0.2, 0.3], topK: 10, filter: { text: doc1 }, includeVector: false, });query的参数与行为参数默认值说明indexName-查询的索引名queryVector-查询向量必填。源码明确抛出QUERY/MISSING_VECTOR错误说明仅按元数据查询不带向量不被该向量存储支持topK10返回结果条数会经过validateTopK校验filter无可选过滤条件会被翻译为 Elasticsearch Query DSL详见下一节includeVectorfalse是否在结果中回传原始向量vector字段底层使用 Elasticsearch 的kNN 查询knn关键字指定field: embedding、query_vector、k: topK与num_candidates: topK * 2当存在过滤条件时将其并入 kNN 的filter子句。返回结果的score字段即 Elasticsearch 的_score每条结果包含id、score、metadata以及当includeVector为 true 时vector字段。5. 更新向量updateVectorawait vectorDB.updateVector({ indexName: my_vectors, id: vector-id, update: { vector: [0.5, 0.6, 0.7], metadata: { text: updated }, }, });updateVector支持按 ID 或按过滤条件两种定位方式且二者互斥同时提供会抛UPDATE_VECTOR/MUTUALLY_EXCLUSIVE错误update中至少要提供vector或metadata之一否则抛NO_UPDATES过滤条件不允许为空对象抛EMPTY_FILTER。底层行为分两条路径按 ID 更新updateVectorById先通过client.get读取现有文档不存在时报 Document with ID xxx not found再合并向量与元数据后调用client.index整体覆写并设置refresh: true按过滤条件批量更新updateVectorsByFilter将过滤条件翻译为 DSL 后调用update_by_query使用 Painless 脚本ctx._source.embedding params.embedding等批量更新命中文档。6. 删除向量deleteVector / deleteVectors// 按 ID 删除单条 await vectorDB.deleteVector({ indexName: my_vectors, id: vector-id, }); // 按过滤条件批量删除 await vectorDB.deleteVectors({ indexName: my_vectors, filter: { source: old-document.pdf }, });deleteVector按 ID 删除底层调用client.delete并refresh: true若文档不存在404则静默返回不会抛错保证删除操作的幂等性deleteVectors支持两种互斥方式传入ids时用 Bulk API 批量删除并逐条解析失败项传入filter时将过滤条件翻译为 DSL 后调用delete_by_query。二者均不能同时为空NO_TARGET错误空数组/空对象同样会被拒绝。四、过滤条件MongoDB 风格 DSL 到 Elasticsearch Query DSLElasticSearchVectorFilter继承了 Mastra 核心包mastra/core/vector/filter统一的过滤语法由ElasticSearchFilterTranslator见 src/vector/filter.ts翻译为 Elasticsearch Query DSL。该翻译器支持的操作符如下类别支持的操作符逻辑$and、$or、$not、$nor数组$in、$nin、$all正则$regex基础比较$eq、$ne由BaseFilterTranslator.DEFAULT_OPERATORS提供数值范围$gt、$gte、$lt、$lte等翻译规则中有几个值得一提的工程细节字段一律加metadata.前缀因为向量文档的结构是{ embedding, metadata }过滤条件作用于元数据字符串字段自动追加.keyword后缀addKeywordIfNeeded确保使用 keyword 字段做精确匹配避免被分词多个数值比较操作符会被优化合并为单个range查询canOptimizeToRangeQuery/createRangeQuery减少查询开销$eq: null翻译为must_not: [{ exists }]字段不存在$ne: null翻译为exists字段存在正确处理空值语义逻辑操作符映射为bool查询$and→bool.must$or→bool.shouldminimum_should_match: 1$not/$nor→bool.must_not空$and匹配所有文档空$or不匹配任何文档带锚点的正则^.../...$会转换为wildcard查询并转义*、?通配元字符其余正则透传为regexp查询。典型示例// 逻辑组合来源是 doc1 或 doc2且评分大于 0.8 const results await vectorDB.query({ indexName: my_vectors, queryVector: [0.1, 0.2, 0.3], topK: 10, filter: { $or: [{ source: doc1 }, { source: doc2 }], score: { $gt: 0.8 }, }, includeVector: false, });五、存储适配器ElasticSearchStore 与三个领域除向量检索外mastra/elasticsearch还提供通用存储适配器ElasticSearchStore见 src/storage/store.ts其配置面ElasticSearchConfig见 src/storage/types.ts与ElasticSearchVector完全对称——同样支持{ id, url, auth? }与{ id, client }两种形态因此两者可以共享同一个连接配置或同一个Client实例import { ElasticSearchStore, ElasticSearchVector } from mastra/elasticsearch; // 共享同一个 client import { Client } from elastic/elasticsearch; const client new Client({ node: http://localhost:9200 }); const storage new ElasticSearchStore({ id: my-store, client }); const vector new ElasticSearchVector({ id: my-vector, client }); // 访问 memory 领域保存会话线程 const memory await storage.getStore(memory); await memory?.saveThread({ thread: { id: thread-1, resourceId: user-1, title: demo } });ElasticSearchStore内部维护三个领域存储见 src/storage/index.ts 的导出领域类职责MemoryMemoryElasticSearch实现MemoryStorage接口持久化 threads / messages / resources支持分页、日期过滤、克隆会话等见 src/storage/domains/memory/index.tsWorkflowsWorkflowsElasticSearch工作流快照与状态持久化ScoresScoresElasticSearch评分数据持久化如 eval 结果构造时还支持disableInit选项为 true 时关闭自动初始化索引创建需手动调用storage.init()。当存储由包内部创建客户端时shouldManageConnection为 true调用close()会释放连接若传入外部 client 则不会代为关闭。六、与 Mastra 生态的协作方式ElasticSearchVector继承自mastra/core的MastraVector基类ElasticSearchStore继承自MastraStorage这意味着它们遵守 Mastra 的统一抽象契约CreateIndexParams、UpsertVectorParams、QueryResult、StorageDomains等接口均来自 core 包可以无缝接入 Mastra 的Memory、Agent 检索等上层能力。包内测试通过共享的向量测试套件internal/storage-test-utils的createVectorTestSuite验证了这一点见 src/vector/index.test.ts该套件会对所有向量存储实现运行统一的行为断言确保不同后端Elasticsearch、PG、LibSQL 等在 API 层面的行为一致性。七、错误处理与可观测性整个包的实现统一使用 Mastra 的错误体系MastraError、ErrorDomain.STORAGE错误 ID 遵循ELASTICSEARCH_操作_错误类型的命名规范如ELASTICSEARCH_CREATE_INDEX_INVALID_ARGS并通过createVectorErrorId/createStorageErrorId生成。错误按ErrorCategory区分归属参数类错误归为USER服务端故障归为THIRD_PARTY内部状态异常归为SYSTEM便于上层按类别做告警或重试策略。此外代码中通过this.logger?.error(...)与this.logger?.trackException(...)记录错误支持接入 Mastra 的可观测性体系进行链路追踪。八、小结mastra/elasticsearch为 Mastra 应用提供了一套覆盖向量写入—索引管理—相似度检索—过滤更新—批量删除完整生命周期的 Elasticsearch 集成同时以统一的存储抽象补齐了 Agent 记忆、工作流与评分数据的持久化能力。其底层直接利用 Elasticsearch 8.x 的dense_vector与 kNN 查询能力配合幂等的索引创建、MongoDB 风格过滤 DSL 翻译与细粒度的批量错误解析兼顾了开发体验与生产可靠性。你可以基于 docker-compose.yaml 快速搭建本地环境参考 src/vector/index.test.ts 中的测试用例验证各 API 行为随后按本文示例逐步接入自己的 RAG 或 Agent 记忆场景。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考