Pregel采用基于Checkpoint的持久化记录下一个以thread_id标识的对话历史这种基于会话的持久化属于短期存储。复杂的流程在运行的时候还需要跨越多个对话的长期甚至永久存储这类存储被抽象成一个BaseStore类型采用内存存储的InMemoryStore它的常用实现。1. BaseStorePregel的长期存储不再仅仅是存储简单的键值对它具有一个具有层次结构命名空间并支持基于向量检索的自然语言查询以及生命周期管理TTL。classBaseStore(ABC):supports_ttl:boolFalsettl_config:TTLConfig|NoneNone…classTTLConfig(TypedDict,totalFalse):refresh_on_read:booldefault_ttl:float|Nonesweep_interval_minutes:int|None我们先来是说基于生命周期的TTL如何使用。存储是否提供针对TTL的支持可以利用其supports_ttl字段来判断而ttl_config字段返回的TTLConfig提供了TTL的相关配置它包括如下的配置项refresh_on_read决定了读取动作是否会延长数据的寿命类似于滑动窗口缓存机制。如果不显式配置通常默认为True。当执行get或search方法时系统会自动重置该数据条目的过期计时器。开启此选项适用于活跃用户会话只要用户还在持续提问或交互其长期记忆就一直保留。只有当用户长时间不使用时数据才会被清理。否则数据寿命自创建起固定无论读取多少次到期即删default_ttl定义数据项默认过期时间单位为分钟。默认为None意味着数据永久有效。当新的数据项被存入时它会被赋予一个初始倒计时。为存储的数据设置一个合理的生命周期有助于防止数据库随着时间推移被无用的僵尸数据填满sweep_interval_minutes决定了后端引擎清理过期数据的频率。默认为None表示不执行定期扫描。系统会维护一个异步数据清除任务以设定的频率扫描数据库删除那些倒计时归零的条目。如果没有配置此项某些后端可能只在读取时发现数据过期惰性删除而不会主动回收磁盘空间。对于InMemoryStore这非常关键定期清理能防止内存会持续膨胀。对于支持原生 TTL 的数据库如Redis此配置可能被忽略因为数据库引擎具有自己的清理策略。数据项的存入利用如下所示的put/aput方法完成数据以键值对命名空间的形式被存储其中值以字典的形式提供字段和值的映射关系具有层级的命名空间以字符串元组的形式提供。ttl参数可以为存入的数据指定一个针对性的过期时间。classBaseStore(ABC):defput(self,namespace:tuple[str,...],key:str,value:dict[str,Any],index:Literal[False]|list[str]|NoneNone,*,ttl:float|None|NotProvidedNOT_PROVIDED,)-None:asyncdefaput(self,namespace:tuple[str,...],key:str,value:dict[str,Any],index:Literal[False]|list[str]|NoneNone,*,ttl:float|None|NotProvidedNOT_PROVIDED,)-Noneput/aput方法的参数index决定该数据项是否进行向量索引。如果我们只是简单地存入数据而不指定index这些数据就像锁在保险箱里只能通过精确的命名空间和键取回。如果我们采用list[str]的形式之定义多个索引字段系统会提取这些字段的内容利用LLM生成嵌入向量进行存储那么后续就能针对它们进行语义搜索。get/aget方法根据提供的命名空间键的组合将对应的数据项提取出来refresh_ttl参数决定是否重置该数据项的过期时间。方法返回的是一个Item对象我们可以从中提取数据项的键、值、命名空间以及创建和最近修改的时间戳。Item类型的dict方法将自身转换成字典两个时间类型会转换成ISO格式的字符串相当于提供序列化的能力datetime对象能直接被序列化。classBaseStore(ABC):defget(self,namespace:tuple[str,...],key:str,*,refresh_ttl:bool|NoneNone,)-Item|Noneasyncdefaget(self,namespace:tuple[str,...],key:str,*,refresh_ttl:bool|NoneNone,)-Item|None…classItem:__slots__(value,key,namespace,created_at,updated_at)defdict(self)-dict…BaseStore提供了用于查询的search/asearch方法我们调用这两个方法是时必须指定命名空间前缀。我们可以利用query参数实施基于自然语言的查询它会触发底层存储的向量相似度搜索。filter参数以字典的形式提供基于多个字段的组合查询条件。limit和offset参数作为典型的分页参数前者指定本次搜索返回结果的最大条目数后者指定在返回结果集之前要跳过的条目数量。至于refresh_ttl参数同样是用于决定是否重置过期时间。classBaseStore(ABC):defsearch(self,namespace_prefix:tuple[str,...],/,*,query:str|NoneNone,filter:dict[str,Any]|NoneNone,limit:int10,offset:int0,refresh_ttl:bool|NoneNone,)-list[SearchItem]:asyncdefasearch(self,namespace_prefix:tuple[str,...],/,*,query:str|NoneNone,filter:dict[str,Any]|NoneNone,limit:int10,offset:int0,refresh_ttl:bool|NoneNone,)-list[SearchItem]…classSearchItem(Item):__slots__(score,)defdict(self)-dict:resultsuper().dict()result[score]self.scorereturnresultsearch/asearch方法返回的不是迭代器而是一个固化的SearchItem列表。SearchItem继承自Item字段score是该类型存在的唯一理由它代表搜索结果与查询之间的相关程度。具体数值取决于底层的向量存储通常会采用余弦相似度或欧氏距离。它的dict方法会在基类返回的字典上将score代表的相似度得分添加进去。除了上述的这些常规方法BaseStore还提供用于删除指定数据项的delete方法和get/aget方法一样我们调用此方法时需要指定准确的命名空间和键。list_namespaces/alist_namespaces方法会根据指定的前缀/后缀查询命名空间的。classBaseStore(ABC):defdelete(self,namespace:tuple[str,...],key:str)-Nonedeflist_namespaces(self,*,prefix:NamespacePath|NoneNone,suffix:NamespacePath|NoneNone,max_depth:int|NoneNone,limit:int100,offset:int0,)-list[tuple[str,...]]asyncdefalist_namespaces(self,*,prefix:NamespacePath|NoneNone,suffix:NamespacePath|NoneNone,max_depth:int|NoneNone,limit:int100,offset:int0,)-list[tuple[str,...]]...NamespacePathtuple[str|Literal[*],...]BaseStore的实现类型必须实现它的batch/abatch方法。这两个抽象方法也是BaseStore最核心的方法因为其他的快捷方法如get、put、search在底层通常都是封装成一个单操作的batch来执行的。在分布式图计算中工作节点可能位于不同的服务器合并多个操作对于减少时延尤为重要。classBaseStore(ABC):abstractmethoddefbatch(self,ops:Iterable[Op])-list[Result]abstractmethodasyncdefabatch(self,ops:Iterable[Op])-list[Result]...OpGetOp|SearchOp|PutOp|ListNamespacesOp ResultItem|list[Item]|list[SearchItem]|list[tuple[str,...]]|None被batch/abatch方法打包的四个操作get、search、put和list_namespaces分别定义成如下的类型各自的字段基本上对应这个快捷方法的参数。batch/abatch返回的是元素类型为Result的列表。classGetOp(NamedTuple):namespace:tuple[str,...]key:strrefresh_ttl:boolTrueclassSearchOp(NamedTuple):namespace_prefix:tuple[str,...]filter:dict[str,Any]|NoneNonelimit:int10offset:int0query:str|NoneNonerefresh_ttl:boolTrueclassMatchCondition(NamedTuple):match_type:NamespaceMatchType path:NamespacePathclassListNamespacesOp(NamedTuple):match_conditions:tuple[MatchCondition,...]|NoneNonemax_depth:int|NoneNonelimit:int100offset:int0classPutOp(NamedTuple):namespace:tuple[str,...]key:strvalue:dict[str,Any]|Noneindex:Literal[False]|list[str]|Nonettl:float|NoneNone2.InMemoryStoreInMemoryStore在内部利用一个字典来存储数据项即众多具有命名空间的键值对并针对这个结构提供单记录的读取和多记录的查询。由于需要提供自然语言查询功能InMemoryStore需要利用语言模型将文本转换成多维嵌入向量的能力。我们知道嵌入是语言模型的一项核心概念它来源于拓扑学和几何学它表示的是将一个低维的对象嵌入到一个高维的连续空间中并保持其结构不变。基本的原理是定义一个N维的语义空间将文本根据其表达的语义转换成该空间的一个点并通过N维的向量来表示。我们一般将这个语义空间称为嵌入空间对应点的向量称为嵌入向量。嵌入的原则很明确那就是让具有相似语义的文本对应的点在嵌入空间尽量靠近否则使它们尽量远离。如果我们将数据库记录相关的文本字段转换成对应的嵌入向量那么实现基于自然语言的查询就很容易实现将查询文本采用相同的嵌入规则转换成向量以两个点在嵌入空间的距离比如欧氏距离或者向量之间的夹角余弦相似度作为相似语义的度量对数据项实施查询。我们将这种查询方式称为向量检索。InMemoryStore采用的是基于余弦相似度的检索方式。初始化InMemoryStore对象的时候需要提供一个IndexConfig对象来对采用的向量索引进行配置。IndexConfig的dims字段表示嵌入空间的维度而fields字段提供参与嵌入转换的字段。嵌入功能可以由一个Embeddings对象提供它利用两组方法embed_documents/aembed_documents和embed_query/aembed_query提供了针对待检索的文档针对数据项和查询文本实施嵌入。EmbeddingsFunc和AEmbeddingsFunc以可执行对象的形式提供了相同的功能表达。这三种形式对embed字段的设置都是支持的。classInMemoryStore(BaseStore):def__init__(self,*,index:IndexConfig|NoneNone)-None:classIndexConfig(TypedDict,totalFalse):dims:intembed:Embeddings|EmbeddingsFunc|AEmbeddingsFunc|strfields:list[str]|NoneclassEmbeddings(ABC):abstractmethoddefembed_documents(self,texts:list[str])-list[list[float]]abstractmethoddefembed_query(self,text:str)-list[float]asyncdefaembed_documents(self,texts:list[str])-list[list[float]]asyncdefaembed_query(self,text:str)-list[float]EmbeddingsFuncCallable[[Sequence[str]],list[list[float]]]AEmbeddingsFuncCallable[[Sequence[str]],Awaitable[list[list[float]]]]对于2.0之前的版本还支持将其设置为一个预定义的Embeddings名称内部会调用langchain.embeddings.init_embeddings函数提取内部的映射得到对应的Embeddings对象但是这个langchain.embeddings包已经与最新版本的LangChain不兼容了。指定的Embeddings应该与嵌入空间维度相匹配而嵌入是语言模型提供的能力我们可以根据选择的语言模型和版本指定对应的维度。比如我们可以通过如下的方式得到text-embedding-3-small这个模型采用的嵌入空间维度是1536。fromlangchain_openaiimportOpenAIEmbeddings embeddingsOpenAIEmbeddings(modeltext-embedding-3-small,api_keyapi_key,base_urlbase_url,)vectorembeddings.embed_query(Hello world)print(1536len(vector))在如下的演示程序中我们针对模型text-embedding-3-small创建了一个OpenAIEmbeddings并据此创建了一个InMemoryStore对象。然后我们以命名空间profile.instrests存入了两个分别针对饮食和运动喜好的数据项。fromlangchain_openaiimportOpenAIEmbeddingsfromlanggraph.store.baseimportIndexConfigfromlanggraph.store.memoryimportInMemoryStoreimportjson base_urlbase URL of you deployed modelmodel_nametext-embedding-3-smallapi_keyyour api keyembeddingsOpenAIEmbeddings(modeltext-embedding-3-small,api_keyapi_key,base_urlbase_url,)storeInMemoryStore(indexIndexConfig(embedembeddings,dims1536))namespace(profile,interests)store.put(namespace,foods,{text:I love eating spicy Sichuan food.})store.put(namespace,sports,{text:My favorite sport is swimming.})resultsstore.search(namespace,queryWhat are my food preferences?,limit1)foriteminresults:print(json.dumps(item.dict(),indent2))我们指定命名空间调用InMemoryStore对象的search方法以自然语言的形式What are my food preferences?实施查询。由于我们只有两条数据所以我们将limit参数设置为1返回匹配度最高的那一条我们会得到希望的第一条数据。如下所示的是查询结果。{namespace:[profile,interests],key:foods,value:{text:I love eating spicy Sichuan food.},created_at:2026-01-28T02:53:58.64168000:00,updated_at:2026-01-28T02:53:58.64168200:00,score:0.3216021020504674}3. 在Pregel中的使用如果需要在Pregel中使用长期存储需要在初始化Pregel对象的时候提供对应的BaseStore对象。由于节点处理函数不支持针对BaseStore的直接注入我们还得像使用静态上下文和StreamWriter一样从注入的RunnableConfig配置中得到当前的Runtime对象它的store字段返回的就是我们指定的BaseStore对象。我们使用上面创建用于存储用于个人喜好的InMemoryStore编写了如下这个演示程序这次我们将用户ID放进命名空间{user_id}.{interests}。我们使用静态上下文类型User来传递用户ID所以在初始化Pregel对象的时候除了需要指定store之外还需要将User类型设置为context_schema。fromlangchain_openaiimportOpenAIEmbeddingsfromlanggraph.store.baseimportBaseStore,IndexConfigfromlanggraph.store.memoryimportInMemoryStorefromlanggraph.typesimportRunnableConfigfromlanggraph.runtimeimportRuntimefromlanggraph.pregelimportPregel,NodeBuilderfromlanggraph.channelsimportLastValueclassUser:user_id:strdef__init__(self,user_id:str):self.user_iduser_iddefhandle(query:str,config:RunnableConfig)-str:runtime:Runtimeconfig[configurable].get(__pregel_runtime)store:BaseStoreruntime.store user_idruntime.context.user_id namespace(user_id,interests)resultsstore.search(namespace,queryquery,limit1)iflen(results)0:returnresults[0].value[text]returnNo interests found.node(NodeBuilder().subscribe_only(query).do(handle).write_to(answer))defbuild_store()-BaseStore:embeddingsOpenAIEmbeddings(modeltext-embedding-3-small,api_keyyour api key,base_url base URL of you deployed model)storeInMemoryStore(indexIndexConfig(embedembeddings,dims1536))namespace(jyden,interests)store.put(namespace,foods,{text:I love eating spicy Sichuan food.})store.put(namespace,sports,{text:My favorite sport is swimming.})returnstore appPregel(nodes{body:node},channels{query:LastValue(str),answer:LastValue(str)},input_channels[query],output_channels[answer],storebuild_store(),context_schemaUser)resultapp.invoke(input{query:What is my favorite sport?},contextUser(user_idjyden))assertresult{answer:My favorite sport is swimming.}resultapp.invoke(input{query:What are my food preferences?},contextUser(user_idjyden))assertresult{answer:I love eating spicy Sichuan food.}调用Pregel对象时利用指定的静态上下文指定用户ID并将针对用户喜好的查询文本写入输入通道query并触发唯一的用于响应查询的节点。在节点的处理函数中我们直接将查询文本中作为参数并从另一个RunnableConfig类型的参数中提取静态上下文和BaseStore对象。在从静态上下文得到用户ID后我们用它生成对应的命名空间针对得到BaseStore对象实施查询并返回最终的查询结果。