Source code for langchain_community.storage.mongodb
from typing import Iterator, List, Optional, Sequence, Tuple
from langchain_core.documents import Document
from langchain_core.stores import BaseStore
[docs]class MongoDBStore(BaseStore[str, Document]):
"""使用MongoDB作为底层存储的BaseStore实现。
示例:
创建一个MongoDBStore实例并对其执行操作:
.. code-block:: python
# 使用MongoDB连接实例化MongoDBStore
from langchain.storage import MongoDBStore
mongo_conn_str = "mongodb://localhost:27017/"
mongodb_store = MongoDBStore(mongo_conn_str, db_name="test-db",
collection_name="test-collection")
# 为键设置值
doc1 = Document(...)
doc2 = Document(...)
mongodb_store.mset([("key1", doc1), ("key2", doc2)])
# 获取键的值
values = mongodb_store.mget(["key1", "key2"])
# [doc1, doc2]
# 遍历键
for key in mongodb_store.yield_keys():
print(key)
# 删除键
mongodb_store.mdelete(["key1", "key2"])"""
[docs] def __init__(
self,
connection_string: str,
db_name: str,
collection_name: str,
*,
client_kwargs: Optional[dict] = None,
) -> None:
"""使用MongoDB连接字符串初始化MongoDBStore。
参数:
connection_string(str):MongoDB连接字符串
db_name(str):要使用的名称
collection_name(str):要使用的集合名称
client_kwargs(dict):要传递给Mongo客户端的关键字参数
"""
try:
from pymongo import MongoClient
except ImportError as e:
raise ImportError(
"The MongoDBStore requires the pymongo library to be "
"installed. "
"pip install pymongo"
) from e
if not connection_string:
raise ValueError("connection_string must be provided.")
if not db_name:
raise ValueError("db_name must be provided.")
if not collection_name:
raise ValueError("collection_name must be provided.")
self.client = MongoClient(connection_string, **(client_kwargs or {}))
self.collection = self.client[db_name][collection_name]
[docs] def mget(self, keys: Sequence[str]) -> List[Optional[Document]]:
"""获取与给定键相关联的文档列表。
参数:
keys(list[str]):表示文档ID的键列表。
返回:
list[Document]:与提供的键对应的文档列表,其中每个文档如果成功检索则为实际文档,如果未找到则表示为None。
"""
result = self.collection.find({"_id": {"$in": keys}})
result_dict = {doc["_id"]: Document(**doc["value"]) for doc in result}
return [result_dict.get(key) for key in keys]
[docs] def mset(self, key_value_pairs: Sequence[Tuple[str, Document]]) -> None:
"""设置给定的键值对。
参数:
key_value_pairs (list[tuple[str, Document]]): 一个id-document对的列表。
返回:
None
"""
from pymongo import UpdateOne
updates = [{"_id": k, "value": v.__dict__} for k, v in key_value_pairs]
self.collection.bulk_write(
[UpdateOne({"_id": u["_id"]}, {"$set": u}, upsert=True) for u in updates]
)
[docs] def mdelete(self, keys: Sequence[str]) -> None:
"""删除给定的ids。
参数:
keys(list[str]):表示文档ID的键列表。
"""
self.collection.delete_many({"_id": {"$in": keys}})
[docs] def yield_keys(self, prefix: Optional[str] = None) -> Iterator[str]:
"""在存储中生成键。
参数:
prefix (str): 要检索的键的前缀。
"""
if prefix is None:
for doc in self.collection.find(projection=["_id"]):
yield doc["_id"]
else:
for doc in self.collection.find(
{"_id": {"$regex": f"^{prefix}"}}, projection=["_id"]
):
yield doc["_id"]