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 还是简单的文件系统。它的核心价值在于解耦业务逻辑Agent 的大脑与数据持久化细节分离。今天用 Postgres明天想换为向量数据库只需实现一套新的协议适配器业务代码几乎不用动。统一为系统中所有需要存储的组件用户记忆、会话历史、工具结果、知识文档提供一致的 API降低认知和开发成本。可测试可以轻松实现一个内存态的 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 的独特优势一专多能All in One结构化数据存储用户配置、Agent 状态、工具调用记录等利用其强大的关系模型和 ACID 事务保证数据一致性。半结构化/非结构化数据使用JSONB数据类型可以灵活地存储 Agent 的思维链、LLM 的响应、从网页抓取的结构化内容等。JSONB支持索引和高效的查询性能远超普通文本字段。全文检索内置pg_trgm三元组和zhparser等扩展提供不错的词法检索能力是本文后半部分“词法检索”的基石。向量检索未来可期通过pgvector扩展Postgres 可以直接存储和检索向量实现语义搜索。这为 Agent 融合关键词检索和语义检索提供了统一的数据平台。可靠性与生态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 检索词法检索仍有不可替代的优势精确术语匹配企业文档中包含大量专业术语、产品型号、错误代码如“ERR-5043”、“量子退火算法”。词法检索能确保这些精确术语被高优先级匹配而向量检索可能将其语义泛化。零延迟无需调用 Embedding 模型生成向量检索速度极快尤其适合海量文档的初筛。可解释性强搜索结果可以高亮显示匹配的关键词用户和开发者都容易理解“为什么这篇文档被召回”。成本低廉不需要为存储海量向量付费也不需要为每一次查询支付 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 高级技巧权重、短语与模糊匹配字段权重如果文档有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;短语搜索使用phraseto_tsquery可以确保查询词以特定顺序出现这对于搜索固定短语如产品名称“DeepSeek Chat”非常有用。ts_query phraseto_tsquery(english, $1)模糊匹配与纠错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 性能优化实践连接池管理务必使用像asyncpg或SQLAlchemy提供的连接池避免频繁创建和销毁数据库连接带来的巨大开销。读写分离对于读多写少的 KB 检索场景可以配置 Postgres 的只读副本Replica将搜索请求路由到副本减轻主库压力。索引优化定期使用EXPLAIN ANALYZE分析慢查询。确保查询条件如namespace和排序字段如rank_score都有合适的索引支持。避免在JSONB字段上直接使用-操作符进行无索引的LIKE查询。结果分页一定要实现分页LIMIT/OFFSET或更优的keyset pagination避免一次性返回海量数据。5.2 向混合检索Hybrid Search演进当词法检索无法满足语义搜索需求时例如用户问“系统慢怎么办”而知识库里只有“性能优化指南”就需要引入向量检索。架构可以这样演进数据双写当一份文档存入agent_store时同时触发一个异步任务将其内容通过 Embedding 模型如 OpenAI text-embedding-3-small, BGE, 本地模型转换为向量并存储到专门的向量表或向量数据库如pgvector扩展的表中。混合查询并行查询用户发起搜索后系统同时执行词法检索在 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_weight0.4, vector_weight0.6) return fused_results[:limit]5.3 监控与维护关键指标监控检索延迟 P95/P99确保搜索响应时间在可接受范围内如 200ms 内。召回率与准确率定期用一批标准问题测试评估检索系统是否能找到正确答案。数据库负载监控 Postgres 的 CPU、内存、连接数、慢查询日志。知识库更新策略建立知识文档的增、删、改流程。更新文档后需要同步更新全文检索的tsvector列和向量存储中的嵌入向量。考虑使用消息队列如 RabbitMQ, Kafka来解耦和异步处理这些更新任务。6. 常见问题与排查技巧实录在实际部署和运维这套存储检索系统时你几乎一定会遇到下面这些问题。这里记录了我的踩坑实录和解决方案。6.1 全文检索不生效或结果不符合预期问题现象搜索中文关键词无结果或英文单词的复数形式搜不到单数形式的文档。排查步骤检查文本搜索配置确认to_tsvector和to_tsquery使用了正确的配置如‘english’,‘simple’,‘zhparser’。执行SELECT * FROM pg_ts_config;查看已安装的配置。检查tsvector内容SELECT key, search_text FROM agent_store WHERE key ‘some_doc’;查看生成的词位是否正确。中文需要确保正确安装了分词插件并创建了对应配置。检查tsquery内容SELECT plainto_tsquery(‘english’, ‘your query’);查看你的查询被解析成了什么词位。验证匹配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 升高。排查步骤查看慢查询日志在postgresql.conf中设置log_min_duration_statement 1000记录超过1秒的语句然后分析日志。使用 EXPLAIN ANALYZE在数据库客户端中对慢查询 SQL 前缀EXPLAIN ANALYZE执行查看执行计划。重点关注是否有“Seq Scan”全表扫描。检查索引\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。评估准备一个测试集QA对用不同的权重组合进行搜索计算MRR (Mean Reciprocal Rank)或NDCG (Normalized Discounted Cumulative Gain)等指标看哪个权重组合下正确答案的平均排名最靠前。业务调优如果业务更强调精确匹配如错误码、型号提高词法权重如 0.7。如果业务更强调语义理解如概念性、描述性问题提高向量权重如 0.7。6.4 如何处理文档更新与一致性场景知识库中的一篇文档内容更新了。方案事务更新在同一个数据库事务中更新主表 (agent_store) 的value字段。由于search_text是GENERATED列它会自动更新。异步更新向量在事务提交后发布一个消息到队列如 “kb_updated”包含文档ID。一个独立的向量更新服务消费该消息重新生成该文档的 Embedding 并更新向量存储。保证最终一致性在向量更新完成前混合检索可能暂时返回旧的语义信息。对于大多数场景这是可接受的。如果要求强一致则需要设计更复杂的同步机制但会牺牲性能。构建 Agent 的存储与检索层就像为一座智能大厦铺设水电和网络管线。它不直接产生“智能”但所有的“智能”都依赖于它稳定、高效地输送“养料”数据。从定义清晰的 Store 协议开始选择像 Postgres 这样坚实可靠的基础再针对企业知识检索的核心场景打磨词法检索能力并规划好向混合检索演进的路径你就能为你的 Agent 系统打造一个足以支撑其复杂思考和行动的数据基石。