[利用LangGraph SDK调用部署的Agent-06]利用StoreClient访问和管理数据存储

LangGraph的BaseCheckpointSaver采用基于Checkpoint的持久化记录下一个基于Thread的对话历史这种基于会话的持久化属于短期存储。复杂的流程在运行的时候还需要跨越多个对话的长期甚至永久存储这类存储被抽象成一个BaseStore类型。LangGraph客户端SDK提供了一个StoreClient它的作用就调用Agent Server提供的RESTful API来远程操作注册的BaseStore。1. 利用BaseStore实现跨Thread的长期存储BaseStore不仅仅存储简单的键值对它具有一个支持层次化命名空间还支持基于向量检索的自然语言查询以及生命周期管理TTL。BaseStore定义如下它提供以系列方法实现存储数据的增删改查和搜索功能在我的文章“支持自然语言查询的长期存储”中对LangGraph基于BaseStore的长期存储有过系统介绍。classBaseStore(ABC):abstractmethoddefbatch(self,ops:Iterable[Op])-list[Result]abstractmethodasyncdefabatch(self,ops:Iterable[Op])-list[Result]defget(self,namespace:tuple[str,...],key:str,*,refresh_ttl:bool|NoneNone,)-Item|Nonedefsearch(self,namespace_prefix:tuple[str,...],/,*,query:str|NoneNone,filter:dict[str,Any]|NoneNone,limit:int10,offset:int0,refresh_ttl:bool|NoneNone,)-list[SearchItem]defput(self,namespace:tuple[str,...],key:str,value:dict[str,Any],index:Literal[False]|list[str]|NoneNone,*,ttl:float|None|NotProvidedNOT_PROVIDED,)-Nonedeflist_namespaces(self,*,prefix:NamespacePath|NoneNone,suffix:NamespacePath|NoneNone,max_depth:int|NoneNone,limit:int100,offset:int0,)-list[tuple[str,...]]asyncdefaget(self,namespace:tuple[str,...],key:str,*,refresh_ttl:bool|NoneNone,)-Item|Noneasyncdefasearch(self,namespace_prefix:tuple[str,...],/,*,query:str|NoneNone,filter:dict[str,Any]|NoneNone,limit:int10,offset:int0,refresh_ttl:bool|NoneNone,)-list[SearchItem]asyncdefaput(self,namespace:tuple[str,...],key:str,value:dict[str,Any],index:Literal[False]|list[str]|NoneNone,*,ttl:float|None|NotProvidedNOT_PROVIDED,)-Noneasyncdefadelete(self,namespace:tuple[str,...],key:str)-Noneasyncdefalist_namespaces(self,*,prefix:NamespacePath|NoneNone,suffix:NamespacePath|NoneNone,max_depth:int|NoneNone,limit:int100,offset:int0,)-list[tuple[str,...]]BaseStore可以用在RAG的多向量检索MultiVector Retriever。在RAG架构中一个大文档可能被切分成多个摘要或小块Chunk并转化为多个向量存入向量数据库。此时BaseStore被用来存储原始的完整文档数据而向量库仅存储向量和对应的 Key。BaseStore也可以用来做嵌入向量缓存Embedding Caching。为避免对相同的文本重复调用大模型 Embedding API通过BaseStore将文本和其计算出的向量关联并持久化从而极大节约API开销和时间。通过实现BaseStoreLangChain及其社区提供了多种开箱即用的底层存储器InMemoryStore基于内存的本地K-V存储适合开发、测试或轻量级临时缓存RedisStore基于Redis的分布式缓存适用于生产环境的高并发高速读写需求LocalFileStore基于本地文件系统的持久化存储数据库存储基于各种云端或托管数据库的存储适配2. 利用StoreClient远程操作BaseStoreLangGraph客户端SDK提供了一个StoreClient它的作用就调用Agent Server提供的RESTful API来远程操作注册的BaseStore所以你会发现StoreClient和BaseStore定义了基本一致的方法。classStoreClient:asyncdefput_item(self,namespace:Sequence[str],/,key:str,value:Mapping[str,Any],index:Literal[False]|list[str]|NoneNone,ttl:int|NoneNone,headers:Mapping[str,str]|NoneNone,params:QueryParamTypes|NoneNone,)-Noneasyncdefget_item(self,namespace:Sequence[str],/,key:str,*,refresh_ttl:bool|NoneNone,headers:Mapping[str,str]|NoneNone,params:QueryParamTypes|NoneNone,)-Itemasyncdefdelete_item(self,namespace:Sequence[str],/,key:str,headers:Mapping[str,str]|NoneNone,params:QueryParamTypes|NoneNone,)-Noneasyncdefsearch_items(self,namespace_prefix:Sequence[str],/,filter:Mapping[str,Any]|NoneNone,limit:int10,offset:int0,query:str|NoneNone,refresh_ttl:bool|NoneNone,headers:Mapping[str,str]|NoneNone,params:QueryParamTypes|NoneNone,)-SearchItemsResponseasyncdeflist_namespaces(self,prefix:list[str]|NoneNone,suffix:list[str]|NoneNone,max_depth:int|NoneNone,limit:int100,offset:int0,headers:Mapping[str,str]|NoneNone,params:QueryParamTypes|NoneNone,)-ListNamespaceResponseclassItem(TypedDict):namespace:list[str]key:strvalue:dict[str,Any]created_at:datetime updated_at:datetimeclassSearchItemsResponse(TypedDict):items:list[SearchItem]classSearchItem(Item,totalFalse):score:float|None方法说明如下put_item方法在指定的命名空间下存储一个键值对可以选择为该键值对设置TTLTime To Live来指定它的生命周期get_item方法获取指定命名空间下的某个键值对可以选择是否刷新它的TTLdelete_item方法删除指定命名空间下的某个键值对search_items方法在指定的命名空间前缀下搜索满足条件的键值对可以基于内容进行自然语言查询也可以基于索引进行过滤还可以指定分页参数list_namespaces方法列出满足条件的命名空间可以指定前缀、后缀和最大深度等参数来过滤命名空间3. 基于自然语言的查询支持基于自然语言的查询是BaseStore区别于传统的关系型数据库和NoSQL数据库的一个重要特性。用户可以通过自然语言来描述他们想要查询的数据而不需要了解底层的存储结构和查询语法。BaseStore会将用户的自然语言查询转化为相应的搜索条件并返回满足条件的键值对。这种基于自然语言的查询极大地降低了用户的使用门槛使得非技术用户也能够轻松地访问和管理数据存储。下面的程序演示了基于StoreClient的search_items方法来进行基于自然语言的查询以及基本的增删改查操作。fromtypingimportLiteral,TypedDictfromlanggraph_sdkimportget_clientimportasyncio,jsonclassContact(TypedDict):id:strname:strgender:Literal[male,female]email:strphone:straddress:strcontacts[Contact(id1,nameJohn Doe,gendermale,emailjohn.doeexample.com,phone123-456-7890,address123 Main St),Contact(id2,nameJane Doe,genderfemale,emailjane.doeexample.com,phone987-654-3210,address456 Elm St),Contact(id3,nameAlice Smith,genderfemale,emailalice.smithexample.com,phone555-555-5555,address789 Oak St)]namespaces[crm,contacts]asyncdefmain():asyncwithget_client(urlhttp://localhost:2024)asclient:storeclient.store resultawaitstore.search_items(namespaces)foriteminresult[items]:awaitstore.delete_item(item[namespace],keyitem[key])forcontactincontacts:awaitstore.put_item(namespaces,keycontact[id],valuecontact)print(available namespaces:,awaitstore.list_namespaces())contactawaitstore.get_item(namespaces,key3)print(\ncontact (id 3):\n,json.dumps(contact,indent2))print(\nsearch contacts (last name is Doe):)resultawaitstore.search_items(namespaces,queryContacts whose last name is Doe,limit2)print(json.dumps([item[value]foriteminresult[items]],indent2))asyncio.run(main())输出available namespaces: {namespaces: [[crm, contacts]]} contact (id 3): { namespace: [ crm, contacts ], key: 3, value: { id: 3, name: Alice Smith, gender: female, email: alice.smithexample.com, phone: 555-555-5555, address: 789 Oak St }, created_at: 2026-04-29T11:41:56.81273800:00, updated_at: 2026-04-29T11:41:56.81274000:00 } search contacts (last name is Doe): [ { id: 1, name: John Doe, gender: male, email: john.doeexample.com, phone: 123-456-7890, address: 123 Main St }, { id: 2, name: Jane Doe, gender: female, email: jane.doeexample.com, phone: 987-654-3210, address: 456 Elm St } ]