AI Agent存储架构设计：基于PostgreSQL的Store协议与混合检索实践

1. 项目概述为什么我们需要一个健壮的 Agent 存储层如果你正在搭建一个 AI Agent 系统无论是个人项目还是企业级应用迟早会撞上一个核心问题Agent 的记忆和知识放在哪里这听起来像是个简单的存储问题但背后牵扯的复杂性远超想象。一个简单的对话 Agent 可能只需要记住上文的几句话但一个面向复杂任务、需要长期记忆、并能从海量文档中检索知识的 Agent其存储和检索需求就完全是另一个量级了。我最近在重构一个企业内部的智能客服 Agent 项目就深刻体会到了这一点。最初的版本Agent 的状态、对话历史、乃至从知识库KB里检索到的片段都一股脑地塞在内存里或者用简单的键值对文件存储。当并发用户数上来或者知识文档超过几百篇时系统就开始变得迟缓、不稳定甚至出现“记忆错乱”——Agent 把不同会话的内容混在了一起。这迫使我停下来思考一个生产级的 Agent 系统其“基础设施”到底应该是什么样子这正是“Store 协议”要解决的问题。它不是一个具体的数据库而是一个抽象层一个契约。它定义了 Agent 系统如何与底层存储介质进行交互无论是保存临时的会话状态还是查询庞大的企业知识库。把 Postgres 作为存储路径则是这个抽象协议下一个非常经典且强大的具体实现。而“企业 KB 词法检索”则是这个存储层之上面向特定业务场景知识问答的核心能力增强。今天我就结合自己的踩坑和重构经验来拆解这三者如何协同工作构建出 Agent 系统坚实的数据地基。2. 核心需求解析Agent 系统对存储的独特要求在深入技术细节之前我们必须先搞清楚一个 AI Agent 系统到底对存储提出了哪些不同于传统应用的需求。理解这些才能明白为什么不能随便找个数据库就往上套。2.1 状态管理的复杂性与实时性Agent 的核心是拥有“状态”。这个状态可能包括会话上下文当前多轮对话的历史消息。工作记忆为解决当前任务而临时记住的中间信息、工具调用结果。长期记忆用户偏好、历史交互的关键结论等需要持久化的信息。执行状态一个复杂任务被分解成多个步骤后当前执行到哪一步了。这些状态需要被频繁、低延迟地读写。想象一下Agent 每说一句话都可能需要更新它的工作记忆每执行一个工具都需要记录结果。这就要求存储层必须有极高的写入性能和毫秒级的读取延迟。同时状态之间可能存在复杂的关联比如一个会话状态关联多个工具调用记录这又要求存储能支持一定程度的关系型查询。注意把 Agent 的所有状态都塞进同一个大 JSON 对象存到数据库的一个字段里是最初级的做法。虽然简单但在高并发下对这个字段的频繁更新会成为性能瓶颈和锁冲突的重灾区。2.2 知识检索的精准度与效率矛盾企业知识库KB检索是 Agent 能力的放大器。但这里的挑战在于“语义”与“词法”的平衡。语义检索如向量检索优点是能理解意图。用户问“怎么报销差旅费”即使知识库里只有一篇名为《员工费用报销流程》的文档也能被找出来。但它依赖嵌入模型计算开销大且对于非常具体的术语、代码、型号如“ERROR-404A”、“Spring Boot 2.7.x”可能不够精准。词法检索如全文搜索优点是快、准。对于上述具体的术语传统的倒排索引能瞬间找到精确匹配的文档。但它无法处理表述差异用户说“电脑开不了机”知识库里是“主机无法启动”就可能检索不到。一个健壮的 Agent 系统尤其是企业级应用必须能融合两者。词法检索在这里不是落后的代名词而是对语义检索的必要补充和兜底确保关键信息不被遗漏。2.3 数据模式的灵活性与演化能力Agent 系统在快速迭代。今天你可能只需要存储对话明天可能就需要记录每个决策的置信度后天又需要关联外部业务系统的 ID。存储层的数据模式Schema必须具备灵活性。传统关系型数据库严格的表结构在项目早期可能会成为阻碍而 NoSQL 的完全无模式又可能在后期数据一致性上埋坑。我们需要一种折中有基本的结构约束以保证核心数据的可靠性同时又允许部分字段能动态扩展。这正是像 Postgres 的JSONB数据类型这类技术大放异彩的地方。3. Store 协议设计定义数据访问的通用语言“Store 协议”听起来高大上其实核心思想就是面向接口编程。它为 Agent 系统内部各种需要存储的组件如记忆、知识库、工具缓存定义了一套统一的、标准化的读写接口。3.1 协议的核心接口抽象一个最小化的 Store 协议通常包含以下几个核心接口# 这是一个概念示例并非特定框架代码 from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, Generic, TypeVar T TypeVar(T) # 实体类型 class Store(ABC, Generic[T]): 存储抽象基类 abstractmethod async def put(self, key: str, value: T, **kwargs) - bool: 插入或更新一个键值对。 pass abstractmethod async def get(self, key: str, **kwargs) - Optional[T]: 根据键获取值。 pass async def delete(self, key: str, **kwargs) - bool: 根据键删除值。 pass abstractmethod async def search(self, query: str, filter_dict: Optional[Dict] None, limit: int 10, **kwargs) - List[T]: 搜索接口。对于不同存储query的含义不同可能是文本也可能是向量。 pass class StateStore(Store[Dict]): 专门用于存储Agent状态的Store值通常是字典。 pass class KnowledgeStore(Store[Document]): 专门用于存储知识文档的Store。 abstractmethod async def lexical_search(self, query: str, field: str content, **kwargs) - List[Document]: 词法检索接口。 pass abstractmethod async def semantic_search(self, query_embedding: List[float], **kwargs) - List[Document]: 语义向量检索接口。 pass为什么这么设计解耦Agent 的业务逻辑如推理、决策不再关心数据是存在 Redis、Postgres 还是云存储里。它只调用store.get()或store.search()。可替换性今天你用 SQLite 做原型开发明天要上线了只需要实现一个基于 Postgres 的PostgresKnowledgeStore类替换掉原来的实现业务代码一行都不用改。测试友好你可以轻松实现一个MockStore用于单元测试而不需要搭建真实的数据库环境。3.2 协议中的关键设计决策在设计协议时有几个细节决定了它的好用程度异步优先现代 AI 应用框架如 FastAPI、LangChain普遍基于异步 I/O。存储操作尤其是网络 I/O是主要的阻塞源因此 Store 协议的方法应该设计为async以充分利用异步生态。泛型支持使用Generic[T]可以让协议更类型安全。一个StateStore返回Dict一个KnowledgeStore返回Document对象IDE 和类型检查器能提供更好的支持。扩展性通过**kwargs参数为不同的后端存储实现提供传递特殊参数的通道。例如向PostgresKnowledgeStore.search()传递use_lexical_firstTrue参数来控制检索策略。实操心得在定义协议时不要试图一开始就设计一个“万能”的接口。从最核心的get、put、search开始在实际开发中遇到新的需求比如按范围查询、批量操作时再谨慎地添加到协议中。过度设计的前期协议往往会变得臃肿且难以实现。4. Postgres 作为存储路径的深度实践为什么是 Postgres在众多数据库中Postgres 因其惊人的“全能性”成为了实现 Store 协议的绝佳选择。它不仅仅是一个关系型数据库。4.1 利用 JSONB 实现灵活的模式对于 Agent 的状态存储我们可以在 Postgres 中设计这样一张表CREATE TABLE agent_sessions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), session_id VARCHAR(255) NOT NULL UNIQUE, -- 业务会话ID agent_id VARCHAR(100) NOT NULL, -- 哪个Agent state_data JSONB NOT NULL DEFAULT {}::jsonb, -- 核心状态JSON格式 metadata JSONB DEFAULT {}::jsonb, -- 扩展元数据 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 为常用查询字段和JSONB中的关键路径创建索引 CREATE INDEX idx_sessions_agent ON agent_sessions(agent_id); CREATE INDEX idx_sessions_created ON agent_sessions(created_at); CREATE INDEX idx_sessions_state ON agent_sessions USING GIN (state_data); -- GIN索引加速JSONB查询关键点state_data字段使用JSONB类型可以存储任意结构的 Agent 状态对话历史、工作记忆等。你可以直接在里面嵌套数组、对象。metadata字段同样使用JSONB用于存放那些未来可能增加、但当前模式不明确的扩展信息。在JSONB字段上创建GIN 索引可以极大地加速对其中特定键值的查询。例如如果你想快速找到所有state_data-user_preference-theme为dark的会话GIN 索引能派上大用场。updated_at字段的自动更新可通过触发器实现对于清理过期会话和监控非常有用。4.2 实现具体的 Store 类基于上述表结构我们可以实现一个PostgresStateStoreimport asyncpg from your_store_protocol import StateStore class PostgresStateStore(StateStore): def __init__(self, connection_pool: asyncpg.Pool): self.pool connection_pool async def put(self, key: str, value: dict, **kwargs) - bool: 插入或更新会话状态。这里key对应session_id。 query INSERT INTO agent_sessions (session_id, agent_id, state_data, metadata) VALUES ($1, $2, $3, $4) ON CONFLICT (session_id) DO UPDATE SET state_data EXCLUDED.state_data, metadata EXCLUDED.metadata, updated_at NOW() RETURNING id; agent_id kwargs.get(agent_id, default) metadata kwargs.get(metadata, {}) async with self.pool.acquire() as conn: try: await conn.execute(query, key, agent_id, value, metadata) return True except Exception as e: # 实际项目中应有更细致的异常处理和日志 return False async def get(self, key: str, **kwargs) - Optional[dict]: 获取会话状态。 query SELECT state_data FROM agent_sessions WHERE session_id $1; async with self.pool.acquire() as conn: row await conn.fetchrow(query, key) return dict(row[state_data]) if row else None async def search(self, query: str, filter_dict: Optional[Dict] None, limit: int 10, **kwargs) - List[dict]: 示例根据JSONB字段内的内容进行查询。 # 这里构建一个基于JSONB路径的简单查询 base_sql SELECT state_data FROM agent_sessions WHERE 11 params [] param_counter 1 if filter_dict: for field, value in filter_dict.items(): # 假设filter_dict的key是JSONB路径如 user_id base_sql f AND state_data-${{{param_counter}}} ${param_counter 1} params.extend([field, value]) param_counter 2 base_sql f LIMIT ${param_counter}; params.append(limit) async with self.pool.acquire() as conn: rows await conn.fetch(base_sql, *params) return [dict(row[state_data]) for row in rows]这个实现展示了如何将抽象的协议映射到具体的 SQL 操作。使用asyncpg这样的异步驱动能保证整个数据访问层是非阻塞的。5. 企业 KB 词法检索的工程化实现词法检索的核心是全文搜索引擎。Postgres 内置了强大的全文搜索功能足以应对大多数企业知识库的场景。5.1 知识库表结构与全文搜索索引首先设计存储知识文档的表CREATE TABLE knowledge_documents ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), doc_id VARCHAR(255) NOT NULL UNIQUE, -- 外部文档ID title TEXT NOT NULL, content TEXT NOT NULL, -- 文档全文内容 content_tsvector TSVECTOR, -- 用于全文搜索的向量列 category VARCHAR(100), tags TEXT[], -- 使用数组类型存储标签 metadata JSONB DEFAULT {}::jsonb, embedding vector(1536), -- 假设使用1536维的向量例如OpenAI text-embedding-3 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 创建GIN索引加速全文搜索 CREATE INDEX idx_knowledge_fts ON knowledge_documents USING GIN(content_tsvector); -- 创建向量索引例如使用pgvector的ivfflat或hnsw CREATE INDEX idx_knowledge_embedding ON knowledge_documents USING ivfflat (embedding vector_cosine_ops) WITH (lists 100); -- 为常用过滤条件创建索引 CREATE INDEX idx_knowledge_category ON knowledge_documents(category);关键的一步是自动生成content_tsvector。我们可以创建一个触发器CREATE OR REPLACE FUNCTION knowledge_documents_tsvector_update() RETURNS TRIGGER AS $$ BEGIN NEW.content_tsvector setweight(to_tsvector(english, coalesce(NEW.title, )), A) || setweight(to_tsvector(english, coalesce(NEW.content, )), B); RETURN NEW; END; $$ LANGUAGE plpgsql; CREATE TRIGGER tsvector_update BEFORE INSERT OR UPDATE ON knowledge_documents FOR EACH ROW EXECUTE FUNCTION knowledge_documents_tsvector_update();这个触发器做了两件事to_tsvector(english, ...)将文本解析为词位lexeme并进行词干提取、移除停用词等。english是文本搜索配置针对英文优化中文需要其他配置如zhparser。setweight(... , A)给标题A和内容B赋予不同的权重这样在搜索结果中标题匹配的文档排名会更靠前。5.2 实现混合检索策略现在我们可以在PostgresKnowledgeStore中实现混合检索方法class PostgresKnowledgeStore(KnowledgeStore): def __init__(self, connection_pool: asyncpg.Pool): self.pool connection_pool async def lexical_search(self, query: str, field: str content, limit: int 5, **kwargs) - List[Document]: 基于Postgres全文搜索的词法检索。 # 构建全文搜索查询plainto_tsquery 将查询字符串转换为tsquery sql SELECT id, doc_id, title, content, category, tags, ts_rank_cd(content_tsvector, plainto_tsquery(english, $1)) as rank FROM knowledge_documents WHERE content_tsvector plainto_tsquery(english, $1) ORDER BY rank DESC LIMIT $2; async with self.pool.acquire() as conn: rows await conn.fetch(sql, query, limit) return [self._row_to_document(row) for row in rows] async def semantic_search(self, query_embedding: List[float], limit: int 5, **kwargs) - List[Document]: 基于pgvector的向量语义检索。 # 假设已安装pgvector扩展并且embedding列类型为vector sql SELECT id, doc_id, title, content, category, tags, 1 - (embedding $1) as cosine_similarity -- pgvector的运算符计算余弦距离 FROM knowledge_documents WHERE embedding IS NOT NULL ORDER BY embedding $1 LIMIT $2; async with self.pool.acquire() as conn: rows await conn.fetch(sql, query_embedding, limit) return [self._row_to_document(row) for row in rows] async def hybrid_search(self, query_text: str, query_embedding: List[float], lexical_weight: float 0.3, semantic_weight: float 0.7, limit: int 10, **kwargs) - List[Document]: 混合检索结合词法检索和语义检索的分数。 这是一种简单的线性加权融合方式。 lexical_results await self.lexical_search(query_text, limitlimit*2) # 多取一些 semantic_results await self.semantic_search(query_embedding, limitlimit*2) # 将结果合并到字典key为doc_id scored_docs {} # 给词法检索结果赋分 (归一化排名分数) for i, doc in enumerate(lexical_results): # 排名越靠前分数越高例如第1名得1分第2名得0.9分... lexical_score 1.0 / (i 1) base_score scored_docs.get(doc.doc_id, {doc: doc, lexical: 0.0, semantic: 0.0}) base_score[lexical] lexical_score scored_docs[doc.doc_id] base_score # 给语义检索结果赋分 (使用余弦相似度) for i, doc in enumerate(semantic_results): # 假设doc对象有一个similarity属性来自SQL查询 semantic_score getattr(doc, cosine_similarity, 1.0 / (i 1)) base_score scored_docs.get(doc.doc_id, {doc: doc, lexical: 0.0, semantic: 0.0}) base_score[semantic] semantic_score scored_docs[doc.doc_id] base_score # 计算加权总分并排序 def calc_final_score(item): scores item[1] return (scores[lexical] * lexical_weight scores[semantic] * semantic_weight) sorted_items sorted(scored_docs.items(), keycalc_final_score, reverseTrue) final_docs [item[1][doc] for item in sorted_items[:limit]] return final_docs def _row_to_document(self, row) - Document: 将数据库行转换为业务层的Document对象。 # 这里是一个简单示例实际项目中的Document类可能更复杂 from your_models import Document return Document( idrow[id], doc_idrow[doc_id], titlerow[title], contentrow[content], metadata{ category: row[category], tags: row[tags], rank: getattr(row, rank, None), cosine_similarity: getattr(row, cosine_similarity, None) } )这个hybrid_search方法是混合检索的核心。它分别进行词法和语义检索然后通过加权分数进行融合。lexical_weight和semantic_weight参数需要根据你的具体数据和查询类型进行调优。例如对于术语性很强的技术文档查询可以调高词法权重对于开放性的、重语义的理解类查询则调高语义权重。6. 系统集成与性能调优实战设计好各个组件后如何将它们优雅地集成到 Agent 系统中并保证高性能、高可用是下一个挑战。6.1 依赖注入与配置化管理不要在业务代码里硬编码new PostgresStateStore(...)。应该使用依赖注入容器来管理这些存储实例的生命周期和配置。# 示例使用FastAPI的依赖注入系统 from fastapi import Depends import asyncpg async def get_db_pool(): 创建数据库连接池单例。 pool await asyncpg.create_pool( hostsettings.db_host, portsettings.db_port, usersettings.db_user, passwordsettings.db_password, databasesettings.db_name, min_size5, max_size20 ) yield pool await pool.close() async def get_state_store(pool: asyncpg.Pool Depends(get_db_pool)) - StateStore: 获取状态存储实例。 return PostgresStateStore(pool) async def get_knowledge_store(pool: asyncpg.Pool Depends(get_db_pool)) - KnowledgeStore: 获取知识存储实例。 return PostgresKnowledgeStore(pool) # 在Agent服务中使用 app.post(/chat) async def chat_endpoint( message: str, session_id: str, state_store: StateStore Depends(get_state_store), knowledge_store: KnowledgeStore Depends(get_knowledge_store) ): # 1. 获取当前会话状态 current_state await state_store.get(session_id) or {} # 2. 如果需要从知识库检索 if need_knowledge(message): # 先获取查询的向量嵌入这里简化实际可能调用嵌入模型API query_embedding await get_embedding(message) # 使用混合检索 relevant_docs await knowledge_store.hybrid_search( query_textmessage, query_embeddingquery_embedding, lexical_weight0.4, semantic_weight0.6 ) current_state[retrieved_context] [doc.content for doc in relevant_docs[:3]] # 3. 调用LLM生成回复... # agent_logic YourAgentLogic(statecurrent_state, context...) # response await agent_logic.generate(message) # 4. 更新状态 # current_state[conversation_history].append({role:user, content:message}) # current_state[conversation_history].append({role:assistant, content:response}) # await state_store.put(session_id, current_state) # return response这种方式使得存储后端的更换比如从开发环境的 SQLite 切换到生产环境的 Postgres只需要修改配置和依赖提供函数业务逻辑完全不受影响。6.2 性能调优关键点当数据量增长后以下几个方面的调优至关重要连接池配置asyncpg.create_pool中的min_size和max_size需要根据你的应用负载调整。设置过小会导致频繁创建连接过大则会浪费资源。监控数据库连接数是一个好习惯。索引优化JSONB GIN 索引确保对state_data和metadata中需要频繁查询的路径创建索引例如CREATE INDEX idx_state_user ON agent_sessions USING GIN ((state_data-user))。全文搜索 GIN 索引content_tsvector上的索引是全文搜索快的根本。向量索引pgvector提供的ivfflat或hnsw索引对于加速向量相似度搜索是必须的。创建索引时lists参数的选择需要权衡查询速度和索引构建速度/精度。查询优化避免 N1 查询在 Agent 处理中如果需要根据多个 ID 获取知识文档应使用WHERE id IN (...)一次性查询而不是循环查询。合理使用事务对于需要原子性更新的多个状态操作使用数据库事务。限制返回字段SELECT *是性能杀手。在查询中明确指定需要的字段尤其是当表中有TEXT或JSONB这类大字段时。缓存策略状态缓存Agent 的会话状态在短时间内可能被频繁读取。可以考虑在StateStore之上增加一层内存缓存如 Redis缓存最近活跃的会话。StateStore.get方法先查缓存未命中再查数据库并回填缓存。知识缓存对于热点知识文档或者混合检索的结果也可以进行缓存。但要注意知识库更新的缓存失效问题。7. 常见问题与排查技巧实录在实际部署和运行中我遇到了不少典型问题这里分享一些排查思路和解决方案。7.1 问题排查速查表问题现象可能原因排查步骤与解决方案Agent 响应突然变慢1. 数据库连接池耗尽。2. 关键查询缺少索引。3. 知识库表体积暴涨向量检索变慢。1. 查看数据库监控检查活跃连接数。适当调大连接池max_size并检查是否有连接泄漏未正确关闭。2. 使用EXPLAIN ANALYZE分析慢查询 SQL检查是否进行了全表扫描。为WHERE和ORDER BY子句中的字段加索引。3. 检查knowledge_documents表行数。为embedding列重建更优化的向量索引如调整hnsw的m和ef_construction参数。词法检索查不到明明存在的关键词1. 全文搜索配置语言不匹配。2. 停用词Stop Words导致关键词被过滤。3. 词干提取Stemming导致词形变化。1. 确认to_tsvector和plainto_tsquery使用的配置如english与文档语言一致。对于中文需使用zhparser等插件。2. 查询SELECT * FROM ts_debug(english, your-word);查看该词是否被识别为停用词。必要时创建自定义词典。3. 理解这是全文搜索的特性。如需精确匹配可考虑结合LIKE或ILIKE操作符或使用tsquery的短语搜索操作符-。混合检索结果不理想1. 词法和语义检索的权重比例不合适。2. 两种检索返回的结果集重叠度低简单加权融合效果差。1.进行 A/B 测试。准备一批标准查询和预期文档手动调整lexical_weight和semantic_weight计算召回率Recall和平均精度MAP。2. 尝试更复杂的融合策略如倒数融合排名Reciprocal Rank Fusion, RRF。这种方法不依赖绝对分数只依赖排名通常对异构检索系统的融合更鲁棒。更新 Agent 状态时出现并发冲突多个请求同时读写同一个session_id的状态。1.乐观锁在state_data中增加一个版本号字段。更新时WHERE session_id$1 AND version$2如果版本不对则更新失败业务层重试或合并。2.悲观锁对于冲突概率极高的场景在事务开始时SELECT ... FOR UPDATE锁定该行。但这会降低并发度需谨慎使用。3.最终一致性如果业务允许可以将状态更新放入消息队列由单个消费者串行处理。向量索引占用磁盘空间过大hnsw索引为了追求速度会占用比原始数据大得多的空间。1. 这是空间换时间的典型权衡。确保服务器磁盘空间充足。2. 评估是否可以接受稍低的召回率通过调整hnsw索引的ef_search参数来降低搜索时的内存和CPU开销但索引体积不会减小。3. 定期清理或归档旧的、不常用的知识文档。7.2 一个真实的调试案例模糊匹配的陷阱有一次用户反馈 Agent 在回答关于“Python 装饰器”的问题时总是引用一篇讲“Java 注解”的文档。词法检索显示“装饰器”和“注解”的英文都是“decorator”所以排名很高。但这两者其实是不同的概念。解决方案我们改进了词法检索的查询构造。不再仅仅使用plainto_tsquery而是结合了短语搜索和领域词权重提升。-- 改进前的简单查询 SELECT ... WHERE content_tsvector plainto_tsquery(english, python decorator); -- 改进后的查询要求“python”和“decorator”以一定距离接近并提升标题中匹配的权重。 SELECT ..., ts_rank_cd( content_tsvector, to_tsquery(english, python - decorator), -- - 表示相邻 32 -- 排名标准化参数 ) as rank_phrase, ts_rank_cd( setweight(to_tsvector(english, coalesce(title, )), A), -- 标题权重高 to_tsquery(english, decorator) ) as rank_title FROM knowledge_documents WHERE content_tsvector to_tsquery(english, python decorator) -- 必须同时包含 OR title_tsvector to_tsquery(english, decorator) -- 或者在标题里 ORDER BY (rank_phrase * 0.7 rank_title * 0.3) DESC; -- 综合排名这个案例说明词法检索的精度高度依赖于查询的构建技巧。简单的关键词匹配往往不够需要结合业务知识设计更精细的查询逻辑。构建 Agent 系统的存储层远不止是选个数据库那么简单。它要求我们在抽象与具体、灵活与规范、语义与词法、性能与精度之间做出持续的权衡和设计。通过定义清晰的 Store 协议我们为系统奠定了可扩展的基础通过深度利用 Postgres 这样的多模数据库我们获得了关系模型、JSON 文档、全文搜索和向量检索于一体的强大能力而通过精心设计混合检索策略我们让 Agent 的知识查找能力既智能又可靠。这套架构不是一蹴而就的。我的建议是从最简单的内存存储开始快速验证 Agent 的核心逻辑。当遇到状态持久化、知识检索的需求时再引入 Store 协议和 Postgres。先实现基础版本然后在真实流量的考验下逐步迭代出适合你业务场景的混合检索权重、缓存策略和索引优化方案。记住基础设施的价值最终体现在它让上层的 Agent 智能变得多么稳定、强大和易用。