x

langgraph_tutorials/customer_support/db.py

@@ -113,10 +113,13 @@ class DatabaseManager:
         # potential users. This is not meant to be production code, but
         # rather a tutorial example.
         if not os.path.exists(self.dirty_file):
-            raise FileNotFoundError(
+            msg = (
                 f"The dirty database file '{self.dirty_file}' does not exist.\n"
                 f"Please run 'manager.initialize()' first to set up the database."
             )
+            raise FileNotFoundError(
+                msg
+            )
         conn: sqlite3.Connection = sqlite3.connect(self.dirty_file)
         cursor: sqlite3.Cursor = conn.cursor()
         try:

langgraph_tutorials/customer_support/excursions.py

@@ -1,5 +1,4 @@
-"""Information about excursions"""
-
+"""Information about excursions."""
 
 from langgraph_tutorials.customer_support.db import DB
 

langgraph_tutorials/customer_support/flights.py

@@ -13,20 +13,21 @@ def fetch_user_flight_information(config: RunnableConfig) -> list[dict]:
     configuration = config.get("configurable", {})
     passenger_id = configuration.get("passenger_id")
     if not passenger_id:
-        raise ValueError("No passenger ID configured.")
+        msg = "No passenger ID configured."
+        raise ValueError(msg)
 
     with DB.get_cursor() as cursor:
         query = """
-        SELECT 
+        SELECT
             t.ticket_no, t.book_ref,
             f.flight_id, f.flight_no, f.departure_airport, f.arrival_airport, f.scheduled_departure, f.scheduled_arrival,
             bp.seat_no, tf.fare_conditions
-        FROM 
+        FROM
             tickets t
             JOIN ticket_flights tf ON t.ticket_no = tf.ticket_no
             JOIN flights f ON tf.flight_id = f.flight_id
             JOIN boarding_passes bp ON bp.ticket_no = t.ticket_no AND bp.flight_id = f.flight_id
-        WHERE 
+        WHERE
             t.passenger_id = ?
         """
         cursor.execute(query, (passenger_id,))
@@ -79,7 +80,8 @@ def update_ticket_to_new_flight(
     configuration = config.get("configurable", {})
     passenger_id = configuration.get("passenger_id")
     if not passenger_id:
-        raise ValueError("No passenger ID configured.")
+        msg = "No passenger ID configured."
+        raise ValueError(msg)
 
     with DB.get_cursor() as cursor:
         cursor.execute(
@@ -132,7 +134,8 @@ def cancel_ticket(ticket_no: str, *, config: RunnableConfig) -> str:
     configuration = config.get("configurable", {})
     passenger_id = configuration.get("passenger_id")
     if not passenger_id:
-        raise ValueError("No passenger ID configured.")
+        msg = "No passenger ID configured."
+        raise ValueError(msg)
 
     with DB.get_cursor() as cursor:
         cursor.execute(

langgraph_tutorials/customer_support/policy.py

@@ -10,7 +10,7 @@ import numpy as np
 import requests
 from langchain.embeddings import init_embeddings
 from langchain_core.embeddings import Embeddings
-from langchain_core.tools import tool
+from langchain_core.tools import BaseTool, tool
 
 
 class PolicyRetriever:
@@ -32,8 +32,9 @@ class PolicyRetriever:
         elif isinstance(embedding_model, Embeddings):
             self._model = embedding_model
         else:
+            msg = "embedding must be an Embeddings instance or a model name string"
             raise ValueError(
-                "embedding must be an Embeddings instance or a model name string"
+                msg
             )
 
         self._docs: list[dict] | None = None
@@ -76,10 +77,13 @@ class PolicyRetriever:
             # run an empty retriever by mistake.
             # So while this is not a good practice in production (as an empty index
             # is a valid state), we raise an error here to prevent misuse.
-            raise RuntimeError(
+            msg = (
                 "Retriever is not initialized. "
                 "Please call retriever.initialize() first."
             )
+            raise RuntimeError(
+                msg
+            )
 
         query_vector = np.array(self._model.embed_query(query))
         scores = query_vector @ self._arr.T
@@ -90,7 +94,7 @@ class PolicyRetriever:
             {**self._docs[idx], "similarity": scores[idx]} for idx in top_k_idx_sorted
         ]
 
-    def as_tool(self, k: int = 2):
+    def as_tool(self, k: int = 2) -> BaseTool:
         """Return a LangChain tool for looking up policy information.
 
         Args: