feat: valves on_startup & on_shutdown

main.py

@@ -255,6 +255,7 @@ async def update_valves(pipeline_id: str, form_data: dict):
 
     pipeline_module = PIPELINE_MODULES[pipeline_id]
 
+    await pipeline_module.on_shutdown()
     try:
         ValvesModel = pipeline_module.valves.__class__
         valves = ValvesModel(**form_data)
@@ -265,6 +266,7 @@ async def update_valves(pipeline_id: str, form_data: dict):
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"{str(e)}",
         )
+    await pipeline_module.on_startup()
 
     return pipeline_module.valves
 

pipelines/examples/applescript_pipeline.py

@@ -18,13 +18,13 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
 
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/azure_openai_pipeline.py

@@ -14,12 +14,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/filter_pipeline.py

@@ -34,12 +34,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/haystack_pipeline.py

@@ -75,7 +75,7 @@ class Pipeline:
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         pass
 
     def pipe(

pipelines/examples/langfuse_filter_pipeline.py

@@ -43,12 +43,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         self.langfuse.flush()
         pass

pipelines/examples/litellm_manifold_pipeline.py

@@ -24,32 +24,38 @@ class Pipeline:
             LITELLM_BASE_URL: str
 
         # Initialize rate limits
-        self.valves = Valves(**{"LITELLM_BASE_URL": "http://localhost:4000"})
-
-        self.pipelines = self.get_litellm_models()
+        self.valves = Valves(**{"LITELLM_BASE_URL": "http://localhost:4001"})
+        self.pipelines = []
         pass
 
     def get_litellm_models(self):
         if self.valves.LITELLM_BASE_URL:
-            r = requests.get(f"{self.valves.LITELLM_BASE_URL}/v1/models")
-            models = r.json()
-            return [
-                {
-                    "id": model["id"],
-                    "name": model["name"] if "name" in model else model["id"],
-                }
-                for model in models["data"]
-            ]
+            try:
+                r = requests.get(f"{self.valves.LITELLM_BASE_URL}/v1/models")
+                models = r.json()
+                return [
+                    {
+                        "id": model["id"],
+                        "name": model["name"] if "name" in model else model["id"],
+                    }
+                    for model in models["data"]
+                ]
+            except Exception as e:
+                print(f"Error: {e}")
+                return [
+                    {"id": "litellm", "name": "LiteLLM: Please configure LiteLLM URL"},
+                ]
         else:
             return []
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
+        self.pipelines = self.get_litellm_models()
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/llama_cpp_pipeline.py

@@ -15,7 +15,7 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         from llama_cpp import Llama
 
@@ -29,7 +29,7 @@ class Pipeline:
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/llamaindex_ollama_github_pipeline.py

@@ -66,7 +66,7 @@ class Pipeline:
         print(self.index)
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         pass
 
     def pipe(

pipelines/examples/llamaindex_ollama_pipeline.py

@@ -18,7 +18,7 @@ class Pipeline:
         )
         Settings.llm = Ollama(model="llama3")
 
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         global documents, index
 
         self.documents = SimpleDirectoryReader("./data").load_data()
@@ -26,7 +26,7 @@ class Pipeline:
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         pass
 
     def pipe(

pipelines/examples/llamaindex_pipeline.py

@@ -17,11 +17,11 @@ class Pipeline:
 
         self.documents = SimpleDirectoryReader("./data").load_data()
         self.index = VectorStoreIndex.from_documents(self.documents)
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         pass
 
     def pipe(

pipelines/examples/manifold_pipeline.py

@@ -30,12 +30,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/ollama_manifold_pipeline.py

@@ -32,12 +32,12 @@ class Pipeline:
         ]
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/ollama_pipeline.py

@@ -14,12 +14,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/openai_pipeline.py

@@ -14,12 +14,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/pipeline_example.py

@@ -14,12 +14,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/python_code_pipeline.py

@@ -11,12 +11,12 @@ class Pipeline:
         pass
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/examples/rate_limit_filter_pipeline.py

@@ -47,12 +47,12 @@ class Pipeline:
         self.user_requests = {}
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/ollama_manifold_pipeline.py

@@ -31,12 +31,12 @@ class Pipeline:
         ]
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass
 

pipelines/rate_limit_filter_pipeline.py

@@ -47,12 +47,12 @@ class Pipeline:
         self.user_requests = {}
 
     async def on_startup(self):
-        # This function is called when the server is started.
+        # This function is called when the server is started or after valves are updated.
         print(f"on_startup:{__name__}")
         pass
 
     async def on_shutdown(self):
-        # This function is called when the server is stopped.
+        # This function is called when the server is stopped or before valves are updated.
         print(f"on_shutdown:{__name__}")
         pass