news 2026/8/9 11:18:21

构建Agent系统存储层：从Store协议到Postgres词法检索的工程实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
构建Agent系统存储层：从Store协议到Postgres词法检索的工程实践

1. 项目概述：为什么我们需要一个“聪明”的存储层？

在构建一个复杂的 Agent 系统时，我们常常会把注意力集中在那些“聪明”的部分：大语言模型（LLM）的调用、复杂的推理链、多智能体协作的编排。然而，一个经常被忽视、却又至关重要的部分是：数据如何被存储、组织和检索。你可以把 Agent 想象成一个经验丰富的侦探，它的“大脑”（LLL）负责推理和决策，但如果它的“档案室”（存储系统）一团糟，所有案件卷宗都堆在一起，找不到关键线索，那么再聪明的大脑也无用武之地。

这就是我们这一期要深入探讨的核心：Agent 系统的存储基础设施。具体来说，我们将聚焦于三个紧密关联的层面：Store 协议、Postgres 的实现路径，以及企业级知识库（KB）的词法检索。这不仅仅是选择一个数据库那么简单，而是为你的 Agent 设计一套从数据接入、标准化存储到高效检索的完整“消化系统”。

想象一下这样的场景：你的客服 Agent 需要从海量的产品手册、历史工单和内部 Wiki 中，快速找到用户问题的准确答案；你的研发 Agent 需要在代码库、设计文档和会议纪要中，关联出某个 Bug 的所有相关上下文。如果没有一个设计良好的存储与检索层，Agent 要么会“胡言乱语”（检索到无关信息），要么会“反应迟钝”（检索速度慢）。因此，构建一个可靠、高效且易于扩展的 Store，是 Agent 系统从玩具走向生产级应用的关键一步。

2. 核心设计：Store 协议——定义数据交互的“通用语言”

在分布式和多模块的 Agent 系统中，不同的组件（如记忆模块、工具调用模块、知识库模块）都可能需要存取数据。如果每个模块都直接操作数据库，会导致代码高度耦合、难以维护，并且更换底层存储引擎会是一场灾难。因此，我们需要一个抽象层——Store 协议。

2.1 Store 协议的核心价值与设计原则

Store 协议本质上是一组接口（Interface）或抽象基类（ABC），它定义了 Agent 系统与存储后端交互的“标准动作”，而不关心后端具体是 PostgreSQL、Redis、Chromadb 还是简单的文件系统。

它的核心价值在于：

  1. 解耦：业务逻辑（Agent 的大脑）与数据持久化细节分离。今天用 Postgres，明天想换为向量数据库，只需实现一套新的协议适配器，业务代码几乎不用动。
  2. 统一：为系统中所有需要存储的组件（用户记忆、会话历史、工具结果、知识文档）提供一致的 API，降低认知和开发成本。
  3. 可测试：可以轻松实现一个内存态的 Mock Store 用于单元测试，而不必依赖外部数据库。

一个典型的 Agent Store 协议会包含以下核心方法：

# 这是一个概念性示例，并非完整实现 from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional class AgentStoreProtocol(ABC): """Agent 存储协议抽象基类""" @abstractmethod async def put(self, key: str, value: Dict[str, Any], namespace: str = "default") -> bool: """存储一个键值对。""" pass @abstractmethod async def get(self, key: str, namespace: str = "default") -> Optional[Dict[str, Any]]: """根据键获取值。""" pass @abstractmethod async def search( self, query: str, namespace: str = "default", filters: Optional[Dict[str, Any]] = None, limit: int = 10 ) -> List[Dict[str, Any]]: """根据查询文本进行检索。这是最核心的方法，不同后端实现差异最大。""" pass @abstractmethod async def delete(self, key: str, namespace: str = "default") -> bool: """删除一个键值对。""" pass

注意：这里的search方法是灵魂所在。对于简单的键值存储，它可能退化为前缀扫描；对于 SQL 数据库，它可能转换为LIKE或全文检索查询；对于向量数据库，则是向量相似度搜索。协议的设计要能包容这些差异。

2.2 命名空间（Namespace）的设计巧思

你可能注意到了上面代码中的namespace参数。这不是一个可有可无的设计，而是管理多租户、多 Agent 或多数据类型的关键。

  • 按 Agent 实例隔离：namespace=”agent_001_session”和namespace=”agent_002_session”，确保不同 Agent 的会话记忆不会互相污染。
  • 按数据类型隔离：namespace=”knowledge_base”和namespace=”user_profiles”，将知识文档和用户数据分开存储，便于管理和实施不同的安全策略。
  • 按组织/租户隔离：在 SaaS 化的 Agent 平台中，namespace=”company_A”和namespace=”company_B”是实现数据隔离的简洁方案。

通过协议层统一处理命名空间，底层存储实现可以灵活应对。例如，在 Postgres 中，namespace可以映射为一张表名或一个 schema；在 Redis 中，可以作为键的前缀。

3. 实现路径：为什么选择 Postgres 作为主力存储？

当协议定义好后，我们需要为其选择一个可靠的“实体”。在众多数据库中，PostgreSQL（简称 Postgres）因其强大的功能、极高的可靠性和活跃的生态，成为实现 Agent Store 的绝佳选择，尤其适合从零开始构建、对数据一致性和复杂查询有要求的项目。

3.1 Postgres 的独特优势

  1. 一专多能，All in One：

    • 结构化数据：存储用户配置、Agent 状态、工具调用记录等，利用其强大的关系模型和 ACID 事务保证数据一致性。
    • 半结构化/非结构化数据：使用JSONB数据类型，可以灵活地存储 Agent 的思维链、LLM 的响应、从网页抓取的结构化内容等。JSONB支持索引和高效的查询，性能远超普通文本字段。
    • 全文检索：内置pg_trgm（三元组）和zhparser等扩展，提供不错的词法检索能力，是本文后半部分“词法检索”的基石。
    • 向量检索（未来可期）：通过pgvector扩展，Postgres 可以直接存储和检索向量，实现语义搜索。这为 Agent 融合关键词检索和语义检索提供了统一的数据平台。

  2. 可靠性与生态：Postgres 是经过数十年验证的工业级数据库，拥有完善的备份、复制和监控方案。其庞大的生态意味着你几乎能找到任何问题的解决方案。

3.2 基于 Postgres 的 Store 协议实现

让我们实现一个最基础的 Postgres 后端。假设我们有一张表来存储数据：

-- 创建存储表 CREATE TABLE agent_store ( id BIGSERIAL PRIMARY KEY, namespace VARCHAR(255) NOT NULL, key VARCHAR(1024) NOT NULL, value JSONB NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW(), -- 复合索引，针对 namespace 和 key 的查询进行优化 UNIQUE(namespace, key) ); CREATE INDEX idx_namespace ON agent_store(namespace); CREATE INDEX idx_value_gin ON agent_store USING GIN(value); -- 为 JSONB 内容创建 GIN 索引以加速内部查询

对应的 Python 实现可能如下：

import asyncpg from typing import List, Dict, Any, Optional class PostgresStore: """基于 asyncpg 的 Postgres 存储实现""" def __init__(self, connection_pool: asyncpg.Pool): self.pool = connection_pool async def put(self, key: str, value: Dict[str, Any], namespace: str = "default") -> bool: """插入或更新数据""" query = """ INSERT INTO agent_store (namespace, key, value) VALUES ($1, $2, $3) ON CONFLICT (namespace, key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW(); """ async with self.pool.acquire() as conn: await conn.execute(query, namespace, key, value) return True async def get(self, key: str, namespace: str = "default") -> Optional[Dict[str, Any]]: """获取数据""" query = "SELECT value FROM agent_store WHERE namespace = $1 AND key = $2;" async with self.pool.acquire() as conn: row = await conn.fetchrow(query, namespace, key) return row['value'] if row else None async def search(self, query: str, namespace: str = "default", filters: Optional[Dict] = None, limit: int = 10) -> List[Dict]: """ 基础关键词搜索。 这里先实现一个简单的 JSONB 字段内容扫描，后续会升级为真正的全文检索。 """ # 这是一个非常初级、低效的实现，仅用于演示逻辑 sql = "SELECT key, value FROM agent_store WHERE namespace = $1" params = [namespace] # 简单地在 JSONB 的文本内容中模糊匹配（生产环境不推荐！） if query: sql += " AND value::text ILIKE $2" params.append(f'%{query}%') sql += f" LIMIT {limit};" async with self.pool.acquire() as conn: rows = await conn.fetch(sql, *params) return [dict(row) for row in rows]

实操心得：在生产环境中，上述search方法的简单ILIKE查询是性能杀手，尤其当数据量变大时。它无法利用索引，会进行全表扫描。这正引出了我们接下来要解决的核心问题：如何实现高效检索？答案就是为企业知识库构建专门的词法检索能力。

4. 核心环节：为企业知识库（KB）构建高效的词法检索

Agent 需要从企业知识库（如产品文档、技术手册、客服问答对）中精准查找信息。传统的“模糊匹配”远远不够。我们需要的是类似搜索引擎的体验：输入一个问题，能快速返回最相关的文档片段。这就是词法检索（Lexical Search）的用武之地，它主要基于关键词的匹配、频率和位置等信息计算相关性。

4.1 为什么不是一开始就用向量检索？

向量检索（语义搜索）很火，它通过 Embedding 模型将文本转换为向量，计算余弦相似度来找到“意思相近”的内容。但对于企业 KB 检索，词法检索仍有不可替代的优势：

  1. 精确术语匹配：企业文档中包含大量专业术语、产品型号、错误代码（如“ERR-5043”、“量子退火算法”）。词法检索能确保这些精确术语被高优先级匹配，而向量检索可能将其语义泛化。
  2. 零延迟：无需调用 Embedding 模型生成向量，检索速度极快，尤其适合海量文档的初筛。
  3. 可解释性强：搜索结果可以高亮显示匹配的关键词，用户和开发者都容易理解“为什么这篇文档被召回”。
  4. 成本低廉：不需要为存储海量向量付费，也不需要为每一次查询支付 Embedding API 调用成本。

一个成熟的方案往往是“词法检索”先行粗筛，再结合“向量检索”进行精排和语义兜底的混合检索（Hybrid Search）策略。

4.2 基于 Postgres 全文检索实现词法检索

Postgres 提供了强大的全文检索功能。我们不需要引入 Elasticsearch 这样的外部系统，就能构建一个相当不错的词法检索服务。关键步骤是创建全文索引。

首先，我们需要调整表结构，添加一个专门用于全文检索的字段和索引：

-- 为 agent_store 表添加一个用于全文检索的生成列和索引 ALTER TABLE agent_store ADD COLUMN IF NOT EXISTS search_text tsvector GENERATED ALWAYS AS ( -- 这里假设我们的 value JSONB 中有一个 'content' 字段存储文本内容 -- 你可以根据实际数据结构调整，例如合并多个字段：to_tsvector('english', value->>'title' || ' ' || value->>'content') to_tsvector('english', coalesce(value->>'content', '')) ) STORED; -- 创建 GIN 索引，极大加速 @ 操作符的查询 CREATE INDEX idx_search_text_gin ON agent_store USING GIN(search_text);

关键点解析：

  • tsvector是 Postgres 的一种数据类型，它将文本预处理成一系列词位（lexemes），并记录其位置信息，是全文检索的基础。
  • to_tsvector(‘english’, text)函数将文本解析并转换成tsvector。’english’是文本搜索配置，指定了语言相关的停用词和词干提取规则。对于中文，你需要安装zhparser等扩展，并使用to_tsvector(‘zhparser’, text)。
  • GENERATED ALWAYS AS ... STORED表示这是一个生成的列，其值会自动从value->>’content’计算并物理存储，便于索引。
  • GIN(Generalized Inverted Index) 索引是全文检索的标准索引类型，对于@（包含）操作符的查询效率极高。

接下来，升级我们PostgresStore的search方法：

class PostgresStore: # ... 之前的 __init__, put, get 方法保持不变 ... async def search_lexical( self, query: str, namespace: str = "knowledge_base", # 通常为知识库指定独立的命名空间 limit: int = 10, offset: int = 0 ) -> List[Dict[str, Any]]: """ 使用 Postgres 全文检索进行高效的词法搜索。 使用 tsquery 和 ts_rank 进行相关性排序。 """ # 将用户查询字符串转换为 tsquery # plainto_tsquery 会将查询词转换为词位，并用 & (AND) 连接，适合简单搜索 # 如果需要更复杂的逻辑（OR, NOT），可以使用 websearch_to_tsquery (PG 11+) 或 phraseto_tsquery ts_query = "plainto_tsquery('english', $1)" search_sql = f""" SELECT key, value, -- 计算相关性得分，用于排序 ts_rank(search_text, {ts_query}) AS rank_score FROM agent_store WHERE namespace = $2 AND search_text @@ {ts_query} -- @@ 操作符表示“匹配” ORDER BY rank_score DESC LIMIT $3 OFFSET $4; """ async with self.pool.acquire() as conn: rows = await conn.fetch(search_sql, query, namespace, limit, offset) results = [] for row in rows: result = dict(row['value']) # 原始数据 result['_score'] = float(row['rank_score']) # 加入相关性得分 result['_key'] = row['key'] results.append(result) return results

现在，当用户搜索“如何重置产品密码”时，Postgres 会利用idx_search_text_gin索引快速找到所有包含“重置”、“产品”、“密码”这些词位（经过词干提取，如“重置”可能不变，“产品”和“密码”是原词）的文档，并按照ts_rank计算出的相关性分数进行排序返回。

4.3 高级技巧：权重、短语与模糊匹配

  1. 字段权重：如果文档有title和content字段，通常title的匹配权重应该更高。可以在生成tsvector时设置权重标签（A, B, C, D），并在ts_rank中为不同权重设置不同的系数。

    -- 生成列示例，为 title 和 content 设置不同权重 search_text tsvector GENERATED ALWAYS AS ( setweight(to_tsvector('english', coalesce(value->>'title', '')), 'A') || setweight(to_tsvector('english', coalesce(value->>'content', '')), 'B') ) STORED;

  2. 短语搜索：使用phraseto_tsquery可以确保查询词以特定顺序出现，这对于搜索固定短语（如产品名称“DeepSeek Chat”）非常有用。

    ts_query = "phraseto_tsquery('english', $1)"

  3. 模糊匹配与纠错：Postgres 的pg_trgm扩展提供了%操作符和similarity函数，支持基于三元组的模糊匹配。这对于处理拼写错误很有帮助。可以将其作为词法检索的补充。

    -- 启用 pg_trgm 扩展 CREATE EXTENSION IF NOT EXISTS pg_trgm; -- 创建 GIN 索引支持模糊匹配 CREATE INDEX idx_content_trgm ON agent_store USING GIN ((value->>'content') gin_trgm_ops); -- 查询示例：查找与‘configuraton’相似度超过0.3的内容 SELECT * FROM agent_store WHERE (value->>'content') % 'configuraton' AND similarity(value->>'content', 'configuraton') > 0.3;

5. 企业级考量：性能、扩展与混合检索架构

将上述组件组合起来，我们就得到了一个面向生产环境的企业级 Agent 存储与检索架构的雏形。

5.1 性能优化实践

  1. 连接池管理：务必使用像asyncpg或SQLAlchemy提供的连接池，避免频繁创建和销毁数据库连接带来的巨大开销。
  2. 读写分离：对于读多写少的 KB 检索场景，可以配置 Postgres 的只读副本（Replica），将搜索请求路由到副本，减轻主库压力。
  3. 索引优化：定期使用EXPLAIN ANALYZE分析慢查询。确保查询条件（如namespace）和排序字段（如rank_score）都有合适的索引支持。避免在JSONB字段上直接使用->>操作符进行无索引的LIKE查询。
  4. 结果分页：一定要实现分页（LIMIT/OFFSET或更优的keyset pagination），避免一次性返回海量数据。

5.2 向混合检索（Hybrid Search）演进

当词法检索无法满足语义搜索需求时（例如，用户问“系统慢怎么办”，而知识库里只有“性能优化指南”），就需要引入向量检索。架构可以这样演进：

  1. 数据双写：当一份文档存入agent_store时，同时触发一个异步任务，将其内容通过 Embedding 模型（如 OpenAI text-embedding-3-small, BGE, 本地模型）转换为向量，并存储到专门的向量表或向量数据库（如pgvector扩展的表）中。
  2. 混合查询：
    • 并行查询：用户发起搜索后，系统同时执行词法检索（在 Postgres）和语义检索（在向量库）。
    • 结果融合：收到两组结果后，使用RRF (Reciprocal Rank Fusion)或加权分数融合等算法，将两者的排序列表合并成一个最终的、既考虑关键词匹配又考虑语义相似度的结果列表。
# 简化的混合检索控制器示例 class HybridSearchService: def __init__(self, lexical_store: PostgresStore, vector_store: VectorStore): self.lexical = lexical_store self.vector = vector_store async def hybrid_search(self, query: str, namespace: str, limit: int = 10): # 并行发起两种检索 lexical_future = self.lexical.search_lexical(query, namespace, limit*2) # 多取一些用于融合 vector_future = self.vector.semantic_search(query, namespace, limit*2) lexical_results, vector_results = await asyncio.gather(lexical_future, vector_future) # 结果融合 (这里使用简单的加权分数，生产环境可用 RRF) fused_results = self._fuse_results(lexical_results, vector_results, lexical_weight=0.4, vector_weight=0.6) return fused_results[:limit]

5.3 监控与维护

  1. 关键指标监控：
    • 检索延迟 P95/P99：确保搜索响应时间在可接受范围内（如 200ms 内）。
    • 召回率与准确率：定期用一批标准问题测试，评估检索系统是否能找到正确答案。
    • 数据库负载：监控 Postgres 的 CPU、内存、连接数、慢查询日志。
  2. 知识库更新策略：建立知识文档的增、删、改流程。更新文档后，需要同步更新全文检索的tsvector列和向量存储中的嵌入向量。考虑使用消息队列（如 RabbitMQ, Kafka）来解耦和异步处理这些更新任务。

6. 常见问题与排查技巧实录

在实际部署和运维这套存储检索系统时，你几乎一定会遇到下面这些问题。这里记录了我的踩坑实录和解决方案。

6.1 全文检索不生效或结果不符合预期

  • 问题现象：搜索中文关键词无结果，或英文单词的复数形式搜不到单数形式的文档。
  • 排查步骤：
    1. 检查文本搜索配置：确认to_tsvector和to_tsquery使用了正确的配置（如‘english’,‘simple’,‘zhparser’）。执行SELECT * FROM pg_ts_config;查看已安装的配置。
    2. 检查tsvector内容：SELECT key, search_text FROM agent_store WHERE key = ‘some_doc’;查看生成的词位是否正确。中文需要确保正确安装了分词插件并创建了对应配置。
    3. 检查tsquery内容：SELECT plainto_tsquery(‘english’, ‘your query’);查看你的查询被解析成了什么词位。
    4. 验证匹配：SELECT search_text @@ plainto_tsquery(‘english’, ‘query’) AS matches FROM agent_store WHERE key=‘some_doc’;直接测试匹配逻辑。
  • 解决方案：
    • 对于中文，必须安装并配置zhparser或pg_jieba等中文分词插件。
    • 对于英文，确保使用‘english’配置以获得词干提取（stemming）能力。如果希望精确匹配，可以使用‘simple’配置。

6.2 检索性能突然下降

  • 问题现象：随着数据量增长，搜索响应时间变长，数据库 CPU 升高。
  • 排查步骤：
    1. 查看慢查询日志：在postgresql.conf中设置log_min_duration_statement = 1000（记录超过1秒的语句），然后分析日志。
    2. 使用 EXPLAIN ANALYZE：在数据库客户端中，对慢查询 SQL 前缀EXPLAIN ANALYZE执行，查看执行计划。重点关注是否有“Seq Scan”（全表扫描）。
    3. 检查索引：\d+ agent_store查看表结构和索引。确认search_text列上有GIN索引，并且namespace等常用过滤条件也有索引。
  • 解决方案：
    • 确保查询条件能命中索引。避免在WHERE子句中对索引列进行函数操作（如WHERE lower(namespace)=…）。
    • 定期对表执行ANALYZE agent_store;更新统计信息，帮助查询优化器选择最佳计划。
    • 如果tsvector列非常大，考虑将其与主表分离，使用外键关联。

6.3 向量检索与词法检索结果如何平衡？

  • 问题：在混合检索中，给词法检索和向量检索的权重（lexical_weight, vector_weight）设置为多少合适？
  • 经验：没有银弹，需要A/B 测试。
    • 初期：可以设置为 0.5:0.5。
    • 评估：准备一个测试集（Q&A对），用不同的权重组合进行搜索，计算MRR (Mean Reciprocal Rank)或NDCG (Normalized Discounted Cumulative Gain)等指标，看哪个权重组合下，正确答案的平均排名最靠前。
    • 业务调优：如果业务更强调精确匹配（如错误码、型号），提高词法权重（如 0.7）。如果业务更强调语义理解（如概念性、描述性问题），提高向量权重（如 0.7）。

6.4 如何处理文档更新与一致性？

  • 场景：知识库中的一篇文档内容更新了。
  • 方案：
    1. 事务更新：在同一个数据库事务中，更新主表 (agent_store) 的value字段。由于search_text是GENERATED列，它会自动更新。
    2. 异步更新向量：在事务提交后，发布一个消息到队列（如 “kb_updated”，包含文档ID）。一个独立的向量更新服务消费该消息，重新生成该文档的 Embedding 并更新向量存储。
    3. 保证最终一致性：在向量更新完成前，混合检索可能暂时返回旧的语义信息。对于大多数场景，这是可接受的。如果要求强一致，则需要设计更复杂的同步机制，但会牺牲性能。

构建 Agent 的存储与检索层，就像为一座智能大厦铺设水电和网络管线。它不直接产生“智能”，但所有的“智能”都依赖于它稳定、高效地输送“养料”（数据）。从定义清晰的 Store 协议开始，选择像 Postgres 这样坚实可靠的基础，再针对企业知识检索的核心场景打磨词法检索能力，并规划好向混合检索演进的路径，你就能为你的 Agent 系统打造一个足以支撑其复杂思考和行动的数据基石。

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/8/9 11:18:14

CFD云仿真中的许可证管理技术演进与实践

1. CFD云仿真与许可证管理的现状与挑战计算流体力学（CFD）作为工程仿真领域的重要工具，正经历着从本地部署向云端迁移的深刻变革。过去五年间，我们看到越来越多的企业开始将CFD工作负载迁移到云端，这种转变带来了显著的…

作者头像 李华
网站建设 2026/8/9 11:11:36

2024年深度解析：为什么您的清远企业网站建设需要告别模板化选择定制化开发策略

在如今的数字时代，互联网早已不是那些遥不可及的技术名词，而是每一个实体企业必须掌握的“第二生命线”。特别是对于身处广东北部、山水甲天下的清远而言，这里的企业生态正在发生着深刻的变化。传统的酒家、特色的农产品基地、新兴的文旅度假村，甚至是深耕工业制造的加工厂…

作者头像 李华
网站建设 2026/8/9 11:10:28

3步解锁Steam游戏清单管理：Onekey工具完全实战指南

3步解锁Steam游戏清单管理：Onekey工具完全实战指南 【免费下载链接】Onekey Onekey Steam Depot Manifest Downloader 项目地址: https://gitcode.com/gh_mirrors/one/Onekey 当你面对Steam游戏库中数百个游戏，需要跨设备同步游戏文件&#xff0c…

作者头像 李华
网站建设 2026/8/9 11:09:50

开源信息简报系统BriefingAutoFlow：从信息焦虑到工程化解决方案

你有没有过这样的经历：每天早上打开电脑，面对十几个需要关注的平台、几十个甚至上百条行业动态、技术资讯、竞品消息，感觉信息像潮水一样涌来，根本看不过来？手动收集、整理、筛选、摘要，一套流程下来&#…

作者头像 李华