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_first=True参数，来控制检索策略。
实操心得：在定义协议时，不要试图一开始就设计一个“万能”的接口。从最核心的
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 类
基于上述表结构，我们可以实现一个PostgresStateStore：
import 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 1=1" 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, limit=limit*2) # 多取一些 semantic_results = await self.semantic_search(query_embedding, limit=limit*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(), key=calc_final_score, reverse=True) 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( id=row['id'], doc_id=row['doc_id'], title=row['title'], content=row['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( host=settings.db_host, port=settings.db_port, user=settings.db_user, password=settings.db_password, database=settings.db_name, min_size=5, max_size=20 ) 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_text=message, query_embedding=query_embedding, lexical_weight=0.4, semantic_weight=0.6 ) current_state['retrieved_context'] = [doc.content for doc in relevant_docs[:3]] # 3. 调用LLM生成回复... # agent_logic = YourAgentLogic(state=current_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参数的选择需要权衡查询速度和索引构建速度/精度。
- JSONB GIN 索引：确保对
- 查询优化：
- 避免 N+1 查询：在 Agent 处理中，如果需要根据多个 ID 获取知识文档，应使用
WHERE id IN (...)一次性查询，而不是循环查询。 - 合理使用事务：对于需要原子性更新的多个状态操作，使用数据库事务。
- 限制返回字段：
SELECT *是性能杀手。在查询中明确指定需要的字段，尤其是当表中有TEXT或JSONB这类大字段时。
- 避免 N+1 查询：在 Agent 处理中，如果需要根据多个 ID 获取知识文档，应使用
- 缓存策略：
- 状态缓存：Agent 的会话状态在短时间内可能被频繁读取。可以考虑在
StateStore之上增加一层内存缓存（如 Redis），缓存最近活跃的会话。StateStore.get方法先查缓存，未命中再查数据库并回填缓存。 - 知识缓存：对于热点知识文档，或者混合检索的结果，也可以进行缓存。但要注意知识库更新的缓存失效问题。
- 状态缓存：Agent 的会话状态在短时间内可能被频繁读取。可以考虑在
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 智能变得多么稳定、强大和易用。
