community: Retry retriable errors in Neo4j (#26211)

Co-authored-by: Erick Friis <erick@langchain.dev>
This commit is contained in:
Tomaz Bratanic 2024-09-19 13:01:07 +09:00 committed by GitHub
parent acbb4e4701
commit 03b9aca55d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 69 additions and 38 deletions

View File

@ -411,7 +411,9 @@ class Neo4jGraph(GraphStore):
return self.structured_schema return self.structured_schema
def query( def query(
self, query: str, params: dict = {}, retry_on_session_expired: bool = True self,
query: str,
params: dict = {},
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
"""Query Neo4j database. """Query Neo4j database.
@ -423,26 +425,44 @@ class Neo4jGraph(GraphStore):
List[Dict[str, Any]]: The list of dictionaries containing the query results. List[Dict[str, Any]]: The list of dictionaries containing the query results.
""" """
from neo4j import Query from neo4j import Query
from neo4j.exceptions import CypherSyntaxError, SessionExpired from neo4j.exceptions import Neo4jError
with self._driver.session(database=self._database) as session: try:
try: data, _, _ = self._driver.execute_query(
data = session.run(Query(text=query, timeout=self.timeout), params) Query(text=query, timeout=self.timeout),
json_data = [r.data() for r in data] database=self._database,
if self.sanitize: parameters_=params,
json_data = [value_sanitize(el) for el in json_data] )
return json_data json_data = [r.data() for r in data]
except CypherSyntaxError as e: if self.sanitize:
raise ValueError(f"Generated Cypher Statement is not valid\n{e}") json_data = [value_sanitize(el) for el in json_data]
except ( return json_data
SessionExpired except Neo4jError as e:
) as e: # Session expired is a transient error that can be retried if not (
if retry_on_session_expired: (
return self.query( ( # isCallInTransactionError
query, params=params, retry_on_session_expired=False e.code == "Neo.DatabaseError.Statement.ExecutionFailed"
or e.code
== "Neo.DatabaseError.Transaction.TransactionStartFailed"
) )
else: and "in an implicit transaction" in e.message
raise e )
or ( # isPeriodicCommitError
e.code == "Neo.ClientError.Statement.SemanticError"
and (
"in an open transaction is not possible" in e.message
or "tried to execute in an explicit transaction" in e.message
)
)
):
raise
# fallback to allow implicit transactions
with self._driver.session() as session:
data = session.run(Query(text=query, timeout=self.timeout), params)
json_data = [r.data() for r in data]
if self.sanitize:
json_data = [value_sanitize(el) for el in json_data]
return json_data
def refresh_schema(self) -> None: def refresh_schema(self) -> None:
""" """

View File

@ -595,11 +595,8 @@ class Neo4jVector(VectorStore):
query: str, query: str,
*, *,
params: Optional[dict] = None, params: Optional[dict] = None,
retry_on_session_expired: bool = True,
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
""" """Query Neo4j database with retries and exponential backoff.
This method sends a Cypher query to the connected Neo4j database
and returns the results as a list of dictionaries.
Args: Args:
query (str): The Cypher query to execute. query (str): The Cypher query to execute.
@ -608,24 +605,38 @@ class Neo4jVector(VectorStore):
Returns: Returns:
List[Dict[str, Any]]: List of dictionaries containing the query results. List[Dict[str, Any]]: List of dictionaries containing the query results.
""" """
from neo4j.exceptions import CypherSyntaxError, SessionExpired from neo4j import Query
from neo4j.exceptions import Neo4jError
params = params or {} params = params or {}
with self._driver.session(database=self._database) as session: try:
try: data, _, _ = self._driver.execute_query(
data = session.run(query, params) query, database=self._database, parameters_=params
return [r.data() for r in data] )
except CypherSyntaxError as e: return [r.data() for r in data]
raise ValueError(f"Cypher Statement is not valid\n{e}") except Neo4jError as e:
except ( if not (
SessionExpired (
) as e: # Session expired is a transient error that can be retried ( # isCallInTransactionError
if retry_on_session_expired: e.code == "Neo.DatabaseError.Statement.ExecutionFailed"
return self.query( or e.code
query, params=params, retry_on_session_expired=False == "Neo.DatabaseError.Transaction.TransactionStartFailed"
) )
else: and "in an implicit transaction" in e.message
raise e )
or ( # isPeriodicCommitError
e.code == "Neo.ClientError.Statement.SemanticError"
and (
"in an open transaction is not possible" in e.message
or "tried to execute in an explicit transaction" in e.message
)
)
):
raise
# Fallback to allow implicit transactions
with self._driver.session() as session:
data = session.run(Query(text=query), params)
return [r.data() for r in data]
def verify_version(self) -> None: def verify_version(self) -> None:
""" """