咨询:LangGraph当前是否提供MongoDB检查点类?
LangGraph MongoDB检查点类现状与自定义实现
目前LangGraph官方并未提供基于MongoDB的原生检查点类,你提到的langchain_postgres.checkpoint.PostgresCheckpoint(PostgreSQL检查点)和langgraph.checkpoint.sqlite.SqliteSaver(SQLite检查点)是官方仅有的SQL类检查点实现。
如果需要将MongoDB作为LangGraph的检查点存储,可通过以下方式实现:
- 自定义检查点类:基于LangGraph的
BaseCheckpointSaver抽象类,实现核心方法对接MongoDB的CRUD操作 - 适配LangChain组件:利用
langchain_mongodb中的存储工具,封装为符合LangGraph检查点接口的类
自定义MongoDB检查点类示例
from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint, CheckpointMetadata from pymongo import MongoClient from typing import Optional, dict, list, tuple class MongoDBSaver(BaseCheckpointSaver): def __init__(self, connection_string: str, db_name: str, collection_name: str = "langgraph_checkpoints"): self.client = MongoClient(connection_string) self.collection = self.client[db_name][collection_name] def get(self, config: dict) -> Optional[Checkpoint]: doc = self.collection.find_one({"config": config}) return Checkpoint(**doc["checkpoint"]) if doc else None def put(self, config: dict, checkpoint: Checkpoint, metadata: Optional[CheckpointMetadata] = None) -> None: update_data = { "checkpoint": checkpoint.dict(), "metadata": metadata.dict() if metadata else {} } self.collection.update_one( {"config": config}, {"$set": update_data}, upsert=True ) def list(self, config: Optional[dict] = None, limit: Optional[int] = None) -> list[tuple[Checkpoint, CheckpointMetadata]]: query = {} if config is None else {"config": config} cursor = self.collection.find(query) if limit: cursor = cursor.limit(limit) return [ (Checkpoint(**doc["checkpoint"]), CheckpointMetadata(**doc["metadata"])) for doc in cursor ]
内容的提问来源于stack exchange,提问作者JCm
相关产品推荐
相关产品推荐

