Build a RAG chatbot with GKE and Cloud Storage

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 flask 
  
 import 
 Flask 
 , 
 jsonify 
 from 
  
 flask 
  
 import 
 request 
 import 
  
 logging 
 import 
  
 sys 
 , 
 os 
 , 
  
 time 
 from 
  
 kubernetes 
  
 import 
 client 
 , 
 config 
 , 
 utils 
 import 
  
 kubernetes.client 
 from 
  
 kubernetes.client.rest 
  
 import 
 ApiException 
 app 
 = 
 Flask 
 ( 
 __name__ 
 ) 
 @app 
 . 
 route 
 ( 
 '/check' 
 ) 
 def 
  
 message 
 (): 
 return 
 jsonify 
 ({ 
 "Message" 
 : 
 "Hi there" 
 }) 
 @app 
 . 
 route 
 ( 
 '/' 
 , 
 methods 
 = 
 [ 
 'POST' 
 ]) 
 def 
  
 bucket 
 (): 
 request_data 
 = 
 request 
 . 
 get_json 
 () 
 print 
 ( 
 request_data 
 ) 
 bckt 
 = 
 request_data 
 [ 
 'bucket' 
 ] 
 f_name 
 = 
 request_data 
 [ 
 'name' 
 ] 
 id 
 = 
 request_data 
 [ 
 'generation' 
 ] 
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ) 
 return 
 "ok" 
 # Set logging 
 logging 
 . 
 basicConfig 
 ( 
 stream 
 = 
 sys 
 . 
 stdout 
 , 
 level 
 = 
 logging 
 . 
 INFO 
 ) 
 # Setup K8 configs 
 config 
 . 
 load_incluster_config 
 () 
 def 
  
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bucket_name 
 , 
 f_name 
 , 
 namespace 
 = 
 "qdrant" 
 , 
 container_name 
 = 
 "jobcontainer" 
 , 
 env_vars 
 = 
 {}): 
 body 
 = 
 client 
 . 
 V1Job 
 ( 
 api_version 
 = 
 "batch/v1" 
 , 
 kind 
 = 
 "Job" 
 ) 
 body 
 . 
 metadata 
 = 
 client 
 . 
 V1ObjectMeta 
 ( 
 namespace 
 = 
 namespace 
 , 
 name 
 = 
 name 
 ) 
 body 
 . 
 status 
 = 
 client 
 . 
 V1JobStatus 
 () 
 template 
 = 
 client 
 . 
 V1PodTemplate 
 () 
 template 
 . 
 template 
 = 
 client 
 . 
 V1PodTemplateSpec 
 () 
 env_list 
 = 
 [ 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "QDRANT_URL" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "QDRANT_URL" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "COLLECTION_NAME" 
 , 
 value 
 = 
 "training-docs" 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "FILE_NAME" 
 , 
 value 
 = 
 f_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "BUCKET_NAME" 
 , 
 value 
 = 
 bucket_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "APIKEY" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "api-key" 
 , 
 name 
 = 
 "qdrant-database-apikey" 
 ))), 
 ] 
 container 
 = 
 client 
 . 
 V1Container 
 ( 
 name 
 = 
 container_name 
 , 
 image 
 = 
 container_image 
 , 
 env 
 = 
 env_list 
 ) 
 template 
 . 
 template 
 . 
 spec 
 = 
 client 
 . 
 V1PodSpec 
 ( 
 containers 
 = 
 [ 
 container 
 ], 
 restart_policy 
 = 
 'Never' 
 , 
 service_account 
 = 
 'embed-docs-sa' 
 ) 
 body 
 . 
 spec 
 = 
 client 
 . 
 V1JobSpec 
 ( 
 backoff_limit 
 = 
 3 
 , 
 ttl_seconds_after_finished 
 = 
 60 
 , 
 template 
 = 
 template 
 . 
 template 
 ) 
 return 
 body 
 def 
  
 kube_test_credentials 
 (): 
 try 
 : 
 api_response 
 = 
 api_instance 
 . 
 get_api_resources 
 () 
 logging 
 . 
 info 
 ( 
 api_response 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling API: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 def 
  
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ): 
 container_image 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_IMAGE" 
 ) 
 namespace 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_NAMESPACE" 
 ) 
 name 
 = 
 "docs-embedder" 
 + 
 id 
 body 
 = 
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bckt 
 , 
 f_name 
 ) 
 v1 
 = 
 client 
 . 
 BatchV1Api 
 () 
 try 
 : 
 v1 
 . 
 create_namespaced_job 
 ( 
 namespace 
 , 
 body 
 , 
 pretty 
 = 
 True 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling BatchV1Api->create_namespaced_job: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 return 
 if 
 __name__ 
 == 
 '__main__' 
 : 
 app 
 . 
 run 
 ( 
 '0.0.0.0' 
 , 
 port 
 = 
 5001 
 , 
 debug 
 = 
 True 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 flask 
  
 import 
 Flask 
 , 
 jsonify 
 from 
  
 flask 
  
 import 
 request 
 import 
  
 logging 
 import 
  
 sys 
 , 
 os 
 , 
  
 time 
 from 
  
 kubernetes 
  
 import 
 client 
 , 
 config 
 , 
 utils 
 import 
  
 kubernetes.client 
 from 
  
 kubernetes.client.rest 
  
 import 
 ApiException 
 app 
 = 
 Flask 
 ( 
 __name__ 
 ) 
 @app 
 . 
 route 
 ( 
 '/check' 
 ) 
 def 
  
 message 
 (): 
 return 
 jsonify 
 ({ 
 "Message" 
 : 
 "Hi there" 
 }) 
 @app 
 . 
 route 
 ( 
 '/' 
 , 
 methods 
 = 
 [ 
 'POST' 
 ]) 
 def 
  
 bucket 
 (): 
 request_data 
 = 
 request 
 . 
 get_json 
 () 
 print 
 ( 
 request_data 
 ) 
 bckt 
 = 
 request_data 
 [ 
 'bucket' 
 ] 
 f_name 
 = 
 request_data 
 [ 
 'name' 
 ] 
 id 
 = 
 request_data 
 [ 
 'generation' 
 ] 
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ) 
 return 
 "ok" 
 # Set logging 
 logging 
 . 
 basicConfig 
 ( 
 stream 
 = 
 sys 
 . 
 stdout 
 , 
 level 
 = 
 logging 
 . 
 INFO 
 ) 
 # Setup K8 configs 
 config 
 . 
 load_incluster_config 
 () 
 def 
  
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bucket_name 
 , 
 f_name 
 , 
 namespace 
 = 
 "elastic" 
 , 
 container_name 
 = 
 "jobcontainer" 
 , 
 env_vars 
 = 
 {}): 
 body 
 = 
 client 
 . 
 V1Job 
 ( 
 api_version 
 = 
 "batch/v1" 
 , 
 kind 
 = 
 "Job" 
 ) 
 body 
 . 
 metadata 
 = 
 client 
 . 
 V1ObjectMeta 
 ( 
 namespace 
 = 
 namespace 
 , 
 name 
 = 
 name 
 ) 
 body 
 . 
 status 
 = 
 client 
 . 
 V1JobStatus 
 () 
 template 
 = 
 client 
 . 
 V1PodTemplate 
 () 
 template 
 . 
 template 
 = 
 client 
 . 
 V1PodTemplateSpec 
 () 
 env_list 
 = 
 [ 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "ES_URL" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "ES_URL" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "INDEX_NAME" 
 , 
 value 
 = 
 "training-docs" 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "FILE_NAME" 
 , 
 value 
 = 
 f_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "BUCKET_NAME" 
 , 
 value 
 = 
 bucket_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "PASSWORD" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "elastic" 
 , 
 name 
 = 
 "elasticsearch-ha-es-elastic-user" 
 ))), 
 ] 
 container 
 = 
 client 
 . 
 V1Container 
 ( 
 name 
 = 
 container_name 
 , 
 image 
 = 
 container_image 
 , 
 image_pull_policy 
 = 
 'Always' 
 , 
 env 
 = 
 env_list 
 ) 
 template 
 . 
 template 
 . 
 spec 
 = 
 client 
 . 
 V1PodSpec 
 ( 
 containers 
 = 
 [ 
 container 
 ], 
 restart_policy 
 = 
 'Never' 
 , 
 service_account 
 = 
 'embed-docs-sa' 
 ) 
 body 
 . 
 spec 
 = 
 client 
 . 
 V1JobSpec 
 ( 
 backoff_limit 
 = 
 3 
 , 
 ttl_seconds_after_finished 
 = 
 60 
 , 
 template 
 = 
 template 
 . 
 template 
 ) 
 return 
 body 
 def 
  
 kube_test_credentials 
 (): 
 try 
 : 
 api_response 
 = 
 api_instance 
 . 
 get_api_resources 
 () 
 logging 
 . 
 info 
 ( 
 api_response 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling API: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 def 
  
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ): 
 container_image 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_IMAGE" 
 ) 
 namespace 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_NAMESPACE" 
 ) 
 name 
 = 
 "docs-embedder" 
 + 
 id 
 body 
 = 
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bckt 
 , 
 f_name 
 ) 
 v1 
 = 
 client 
 . 
 BatchV1Api 
 () 
 try 
 : 
 v1 
 . 
 create_namespaced_job 
 ( 
 namespace 
 , 
 body 
 , 
 pretty 
 = 
 True 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling BatchV1Api->create_namespaced_job: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 return 
 if 
 __name__ 
 == 
 '__main__' 
 : 
 app 
 . 
 run 
 ( 
 '0.0.0.0' 
 , 
 port 
 = 
 5001 
 , 
 debug 
 = 
 True 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 flask 
  
 import 
 Flask 
 , 
 jsonify 
 from 
  
 flask 
  
 import 
 request 
 import 
  
 logging 
 import 
  
 sys 
 , 
 os 
 , 
  
 time 
 from 
  
 kubernetes 
  
 import 
 client 
 , 
 config 
 , 
 utils 
 import 
  
 kubernetes.client 
 from 
  
 kubernetes.client.rest 
  
 import 
 ApiException 
 app 
 = 
 Flask 
 ( 
 __name__ 
 ) 
 @app 
 . 
 route 
 ( 
 '/check' 
 ) 
 def 
  
 message 
 (): 
 return 
 jsonify 
 ({ 
 "Message" 
 : 
 "Hi there" 
 }) 
 @app 
 . 
 route 
 ( 
 '/' 
 , 
 methods 
 = 
 [ 
 'POST' 
 ]) 
 def 
  
 bucket 
 (): 
 request_data 
 = 
 request 
 . 
 get_json 
 () 
 print 
 ( 
 request_data 
 ) 
 bckt 
 = 
 request_data 
 [ 
 'bucket' 
 ] 
 f_name 
 = 
 request_data 
 [ 
 'name' 
 ] 
 id 
 = 
 request_data 
 [ 
 'generation' 
 ] 
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ) 
 return 
 "ok" 
 # Set logging 
 logging 
 . 
 basicConfig 
 ( 
 stream 
 = 
 sys 
 . 
 stdout 
 , 
 level 
 = 
 logging 
 . 
 INFO 
 ) 
 # Setup K8 configs 
 config 
 . 
 load_incluster_config 
 () 
 def 
  
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bucket_name 
 , 
 f_name 
 , 
 namespace 
 = 
 "pg-ns" 
 , 
 container_name 
 = 
 "jobcontainer" 
 , 
 env_vars 
 = 
 {}): 
 body 
 = 
 client 
 . 
 V1Job 
 ( 
 api_version 
 = 
 "batch/v1" 
 , 
 kind 
 = 
 "Job" 
 ) 
 body 
 . 
 metadata 
 = 
 client 
 . 
 V1ObjectMeta 
 ( 
 namespace 
 = 
 namespace 
 , 
 name 
 = 
 name 
 ) 
 body 
 . 
 status 
 = 
 client 
 . 
 V1JobStatus 
 () 
 template 
 = 
 client 
 . 
 V1PodTemplate 
 () 
 template 
 . 
 template 
 = 
 client 
 . 
 V1PodTemplateSpec 
 () 
 env_list 
 = 
 [ 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "POSTGRES_HOST" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "POSTGRES_HOST" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "DATABASE_NAME" 
 , 
 value 
 = 
 "app" 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "COLLECTION_NAME" 
 , 
 value 
 = 
 "training-docs" 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "FILE_NAME" 
 , 
 value 
 = 
 f_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "BUCKET_NAME" 
 , 
 value 
 = 
 bucket_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "PASSWORD" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "password" 
 , 
 name 
 = 
 "gke-pg-cluster-app" 
 ))), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "USERNAME" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "username" 
 , 
 name 
 = 
 "gke-pg-cluster-app" 
 ))), 
 ] 
 container 
 = 
 client 
 . 
 V1Container 
 ( 
 name 
 = 
 container_name 
 , 
 image 
 = 
 container_image 
 , 
 image_pull_policy 
 = 
 'Always' 
 , 
 env 
 = 
 env_list 
 ) 
 template 
 . 
 template 
 . 
 spec 
 = 
 client 
 . 
 V1PodSpec 
 ( 
 containers 
 = 
 [ 
 container 
 ], 
 restart_policy 
 = 
 'Never' 
 , 
 service_account 
 = 
 'embed-docs-sa' 
 ) 
 body 
 . 
 spec 
 = 
 client 
 . 
 V1JobSpec 
 ( 
 backoff_limit 
 = 
 3 
 , 
 ttl_seconds_after_finished 
 = 
 60 
 , 
 template 
 = 
 template 
 . 
 template 
 ) 
 return 
 body 
 def 
  
 kube_test_credentials 
 (): 
 try 
 : 
 api_response 
 = 
 api_instance 
 . 
 get_api_resources 
 () 
 logging 
 . 
 info 
 ( 
 api_response 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling API: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 def 
  
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ): 
 container_image 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_IMAGE" 
 ) 
 namespace 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_NAMESPACE" 
 ) 
 name 
 = 
 "docs-embedder" 
 + 
 id 
 body 
 = 
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bckt 
 , 
 f_name 
 ) 
 v1 
 = 
 client 
 . 
 BatchV1Api 
 () 
 try 
 : 
 v1 
 . 
 create_namespaced_job 
 ( 
 namespace 
 , 
 body 
 , 
 pretty 
 = 
 True 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling BatchV1Api->create_namespaced_job: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 return 
 if 
 __name__ 
 == 
 '__main__' 
 : 
 app 
 . 
 run 
 ( 
 '0.0.0.0' 
 , 
 port 
 = 
 5001 
 , 
 debug 
 = 
 True 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 flask 
  
 import 
 Flask 
 , 
 jsonify 
 from 
  
 flask 
  
 import 
 request 
 import 
  
 logging 
 import 
  
 sys 
 , 
 os 
 , 
  
 time 
 from 
  
 kubernetes 
  
 import 
 client 
 , 
 config 
 , 
 utils 
 import 
  
 kubernetes.client 
 from 
  
 kubernetes.client.rest 
  
 import 
 ApiException 
 app 
 = 
 Flask 
 ( 
 __name__ 
 ) 
 @app 
 . 
 route 
 ( 
 '/check' 
 ) 
 def 
  
 message 
 (): 
 return 
 jsonify 
 ({ 
 "Message" 
 : 
 "Hi there" 
 }) 
 @app 
 . 
 route 
 ( 
 '/' 
 , 
 methods 
 = 
 [ 
 'POST' 
 ]) 
 def 
  
 bucket 
 (): 
 request_data 
 = 
 request 
 . 
 get_json 
 () 
 print 
 ( 
 request_data 
 ) 
 bckt 
 = 
 request_data 
 [ 
 'bucket' 
 ] 
 f_name 
 = 
 request_data 
 [ 
 'name' 
 ] 
 id 
 = 
 request_data 
 [ 
 'generation' 
 ] 
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ) 
 return 
 "ok" 
 # Set logging 
 logging 
 . 
 basicConfig 
 ( 
 stream 
 = 
 sys 
 . 
 stdout 
 , 
 level 
 = 
 logging 
 . 
 INFO 
 ) 
 # Setup K8 configs 
 config 
 . 
 load_incluster_config 
 () 
 def 
  
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bucket_name 
 , 
 f_name 
 , 
 namespace 
 , 
 container_name 
 = 
 "jobcontainer" 
 , 
 env_vars 
 = 
 {}): 
 body 
 = 
 client 
 . 
 V1Job 
 ( 
 api_version 
 = 
 "batch/v1" 
 , 
 kind 
 = 
 "Job" 
 ) 
 body 
 . 
 metadata 
 = 
 client 
 . 
 V1ObjectMeta 
 ( 
 namespace 
 = 
 namespace 
 , 
 name 
 = 
 name 
 ) 
 body 
 . 
 status 
 = 
 client 
 . 
 V1JobStatus 
 () 
 template 
 = 
 client 
 . 
 V1PodTemplate 
 () 
 template 
 . 
 template 
 = 
 client 
 . 
 V1PodTemplateSpec 
 () 
 env_list 
 = 
 [ 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "WEAVIATE_ENDPOINT" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_ENDPOINT" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "WEAVIATE_GRPC_ENDPOINT" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_GRPC_ENDPOINT" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "FILE_NAME" 
 , 
 value 
 = 
 f_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "BUCKET_NAME" 
 , 
 value 
 = 
 bucket_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "APIKEY" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "AUTHENTICATION_APIKEY_ALLOWED_KEYS" 
 , 
 name 
 = 
 "apikeys" 
 ))), 
 ] 
 container 
 = 
 client 
 . 
 V1Container 
 ( 
 name 
 = 
 container_name 
 , 
 image 
 = 
 container_image 
 , 
 image_pull_policy 
 = 
 'Always' 
 , 
 env 
 = 
 env_list 
 ) 
 template 
 . 
 template 
 . 
 spec 
 = 
 client 
 . 
 V1PodSpec 
 ( 
 containers 
 = 
 [ 
 container 
 ], 
 restart_policy 
 = 
 'Never' 
 , 
 service_account 
 = 
 'embed-docs-sa' 
 ) 
 body 
 . 
 spec 
 = 
 client 
 . 
 V1JobSpec 
 ( 
 backoff_limit 
 = 
 3 
 , 
 ttl_seconds_after_finished 
 = 
 60 
 , 
 template 
 = 
 template 
 . 
 template 
 ) 
 return 
 body 
 def 
  
 kube_test_credentials 
 (): 
 try 
 : 
 api_response 
 = 
 api_instance 
 . 
 get_api_resources 
 () 
 logging 
 . 
 info 
 ( 
 api_response 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling API: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 def 
  
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ): 
 container_image 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_IMAGE" 
 ) 
 namespace 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_NAMESPACE" 
 ) 
 name 
 = 
 "docs-embedder" 
 + 
 id 
 body 
 = 
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bckt 
 , 
 f_name 
 , 
 namespace 
 ) 
 v1 
 = 
 client 
 . 
 BatchV1Api 
 () 
 try 
 : 
 v1 
 . 
 create_namespaced_job 
 ( 
 namespace 
 , 
 body 
 , 
 pretty 
 = 
 True 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling BatchV1Api->create_namespaced_job: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 return 
 if 
 __name__ 
 == 
 '__main__' 
 : 
 app 
 . 
 run 
 ( 
 '0.0.0.0' 
 , 
 port 
 = 
 5001 
 , 
 debug 
 = 
 True 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 ChatVertexAI 
 from 
  
 langchain.prompts 
  
 import 
 ChatPromptTemplate 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain.memory 
  
 import 
 ConversationBufferWindowMemory 
 from 
  
 langchain_community.vectorstores 
  
 import 
 Qdrant 
 from 
  
 qdrant_client 
  
 import 
 QdrantClient 
 import 
  
 streamlit 
  
 as 
  
 st 
 import 
  
 os 
 vertexAI 
 = 
 ChatVertexAI 
 ( 
 model_name 
 = 
 os 
 . 
 getenv 
 ( 
 "VERTEX_AI_MODEL_NAME" 
 , 
 "gemini-2.5-flash-preview-04-17" 
 ), 
 streaming 
 = 
 True 
 , 
 convert_system_message_to_human 
 = 
 True 
 ) 
 prompt_template 
 = 
 ChatPromptTemplate 
 . 
 from_messages 
 ( 
 [ 
 ( 
 "system" 
 , 
 "You are a helpful assistant who helps in finding answers to questions using the provided context." 
 ), 
 ( 
 "human" 
 , 
 """ 
 The answer should be based on the text context given in "text_context" and the conversation history given in "conversation_history" along with its Caption: 
 \n 
 Base your response on the provided text context and the current conversation history to answer the query. 
 Select the most relevant information from the context. 
 Generate a draft response using the selected information. Remove duplicate content from the draft response. 
 Generate your final response after adjusting it to increase accuracy and relevance. 
 Now only show your final response! 
 If you do not know the answer or context is not relevant, response with "I don't know". 
 text_context: 
  
 {context} 
 conversation_history: 
  
 {history} 
 query: 
  
 {query} 
 """ 
 ), 
 ] 
 ) 
 embedding_model 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 client 
 = 
 QdrantClient 
 ( 
 url 
 = 
 os 
 . 
 getenv 
 ( 
 "QDRANT_URL" 
 ), 
 api_key 
 = 
 os 
 . 
 getenv 
 ( 
 "APIKEY" 
 ), 
 ) 
 collection_name 
 = 
 os 
 . 
 getenv 
 ( 
 "COLLECTION_NAME" 
 ) 
 vector_search 
 = 
 Qdrant 
 ( 
 client 
 , 
 collection_name 
 , 
 embeddings 
 = 
 embedding_model 
 ) 
 def 
  
 format_docs 
 ( 
 docs 
 ): 
 return 
 " 
 \n\n 
 " 
 . 
 join 
 ([ 
 d 
 . 
 page_content 
 for 
 d 
 in 
 docs 
 ]) 
 st 
 . 
 title 
 ( 
 "🤖 Chatbot" 
 ) 
 if 
 "messages" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "messages" 
 ] 
 = 
 [{ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 "How can I help you?" 
 }] 
 if 
 "memory" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "memory" 
 ] 
 = 
 ConversationBufferWindowMemory 
 ( 
 memory_key 
 = 
 "history" 
 , 
 ai_prefix 
 = 
 "Bob" 
 , 
 human_prefix 
 = 
 "User" 
 , 
 k 
 = 
 3 
 , 
 ) 
 for 
 message 
 in 
 st 
 . 
 session_state 
 . 
 messages 
 : 
 with 
 st 
 . 
 chat_message 
 ( 
 message 
 [ 
 "role" 
 ]): 
 st 
 . 
 write 
 ( 
 message 
 [ 
 "content" 
 ]) 
 if 
 chat_input 
 := 
 st 
 . 
 chat_input 
 (): 
 with 
 st 
 . 
 chat_message 
 ( 
 "human" 
 ): 
 st 
 . 
 write 
 ( 
 chat_input 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "human" 
 , 
 "content" 
 : 
 chat_input 
 }) 
 found_docs 
 = 
 vector_search 
 . 
 similarity_search 
 ( 
 chat_input 
 ) 
 context 
 = 
 format_docs 
 ( 
 found_docs 
 ) 
 prompt_value 
 = 
 prompt_template 
 . 
 format_messages 
 ( 
 name 
 = 
 "Bob" 
 , 
 query 
 = 
 chat_input 
 , 
 context 
 = 
 context 
 , 
 history 
 = 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 load_memory_variables 
 ({})) 
 with 
 st 
 . 
 chat_message 
 ( 
 "ai" 
 ): 
 with 
 st 
 . 
 spinner 
 ( 
 "Typing..." 
 ): 
 content 
 = 
 "" 
 with 
 st 
 . 
 empty 
 (): 
 for 
 chunk 
 in 
 vertexAI 
 . 
 stream 
 ( 
 prompt_value 
 ): 
 content 
 += 
 chunk 
 . 
 content 
 st 
 . 
 write 
 ( 
 content 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 content 
 }) 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 save_context 
 ({ 
 "input" 
 : 
 chat_input 
 }, 
 { 
 "output" 
 : 
 content 
 }) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain_community.document_loaders 
  
 import 
 PyPDFLoader 
 from 
  
 langchain.text_splitter 
  
 import 
 RecursiveCharacterTextSplitter 
 from 
  
 elasticsearch 
  
 import 
 Elasticsearch 
 from 
  
 langchain_community.vectorstores.elasticsearch 
  
 import 
 ElasticsearchStore 
 from 
  
 google.cloud 
  
 import 
  storage 
 
 import 
  
 os 
 bucketname 
 = 
 os 
 . 
 getenv 
 ( 
 "BUCKET_NAME" 
 ) 
 filename 
 = 
 os 
 . 
 getenv 
 ( 
 "FILE_NAME" 
 ) 
 storage_client 
 = 
  storage 
 
 . 
  Client 
 
 () 
 bucket 
 = 
 storage_client 
 . 
  bucket 
 
 ( 
 bucketname 
 ) 
 blob 
 = 
 bucket 
 . 
 blob 
 ( 
 filename 
 ) 
 blob 
 . 
  download_to_filename 
 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 loader 
 = 
 PyPDFLoader 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 text_splitter 
 = 
 RecursiveCharacterTextSplitter 
 ( 
 chunk_size 
 = 
 1000 
 , 
 chunk_overlap 
 = 
 0 
 ) 
 documents 
 = 
 loader 
 . 
 load_and_split 
 ( 
 text_splitter 
 ) 
 embeddings 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 client 
 = 
 Elasticsearch 
 ( 
 [ 
 os 
 . 
 getenv 
 ( 
 "ES_URL" 
 )], 
 verify_certs 
 = 
 False 
 , 
 ssl_show_warn 
 = 
 False 
 , 
 basic_auth 
 = 
 ( 
 "elastic" 
 , 
 os 
 . 
 getenv 
 ( 
 "PASSWORD" 
 )) 
 ) 
 db 
 = 
 ElasticsearchStore 
 . 
 from_documents 
 ( 
 documents 
 , 
 embeddings 
 , 
 es_connection 
 = 
 client 
 , 
 index_name 
 = 
 os 
 . 
 getenv 
 ( 
 "INDEX_NAME" 
 ) 
 ) 
 db 
 . 
 client 
 . 
 indices 
 . 
 refresh 
 ( 
 index 
 = 
 os 
 . 
 getenv 
 ( 
 "INDEX_NAME" 
 )) 
 print 
 ( 
 filename 
 + 
 " was successfully embedded" 
 ) 
 print 
 ( 
 f 
 "# of vectors = 
 { 
 len 
 ( 
 documents 
 ) 
 } 
 " 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain_community.document_loaders 
  
 import 
 PyPDFLoader 
 from 
  
 langchain.text_splitter 
  
 import 
 RecursiveCharacterTextSplitter 
 from 
  
 langchain_community.vectorstores.pgvector 
  
 import 
 PGVector 
 from 
  
 google.cloud 
  
 import 
  storage 
 
 import 
  
 os 
 bucketname 
 = 
 os 
 . 
 getenv 
 ( 
 "BUCKET_NAME" 
 ) 
 filename 
 = 
 os 
 . 
 getenv 
 ( 
 "FILE_NAME" 
 ) 
 storage_client 
 = 
  storage 
 
 . 
  Client 
 
 () 
 bucket 
 = 
 storage_client 
 . 
  bucket 
 
 ( 
 bucketname 
 ) 
 blob 
 = 
 bucket 
 . 
 blob 
 ( 
 filename 
 ) 
 blob 
 . 
  download_to_filename 
 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 loader 
 = 
 PyPDFLoader 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 text_splitter 
 = 
 RecursiveCharacterTextSplitter 
 ( 
 chunk_size 
 = 
 1000 
 , 
 chunk_overlap 
 = 
 0 
 ) 
 documents 
 = 
 loader 
 . 
 load_and_split 
 ( 
 text_splitter 
 ) 
 for 
 document 
 in 
 documents 
 : 
 document 
 . 
 page_content 
 = 
 document 
 . 
 page_content 
 . 
 replace 
 ( 
 ' 
 \x00 
 ' 
 , 
 '' 
 ) 
 embeddings 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 CONNECTION_STRING 
 = 
 PGVector 
 . 
 connection_string_from_db_params 
 ( 
 driver 
 = 
 "psycopg2" 
 , 
 host 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "POSTGRES_HOST" 
 ), 
 port 
 = 
 5432 
 , 
 database 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "DATABASE_NAME" 
 ), 
 user 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "USERNAME" 
 ), 
 password 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "PASSWORD" 
 ), 
 ) 
 COLLECTION_NAME 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "COLLECTION_NAME" 
 ) 
 db 
 = 
 PGVector 
 . 
 from_documents 
 ( 
 embedding 
 = 
 embeddings 
 , 
 documents 
 = 
 documents 
 , 
 collection_name 
 = 
 COLLECTION_NAME 
 , 
 connection_string 
 = 
 CONNECTION_STRING 
 , 
 use_jsonb 
 = 
 True 
 ) 
 print 
 ( 
 filename 
 + 
 " was successfully embedded" 
 ) 
 print 
 ( 
 f 
 "# of vectors = 
 { 
 len 
 ( 
 documents 
 ) 
 } 
 " 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain_community.document_loaders 
  
 import 
 PyPDFLoader 
 from 
  
 langchain.text_splitter 
  
 import 
 RecursiveCharacterTextSplitter 
 import 
  
 weaviate 
 from 
  
 weaviate.connect 
  
 import 
 ConnectionParams 
 from 
  
 langchain_weaviate.vectorstores 
  
 import 
 WeaviateVectorStore 
 from 
  
 google.cloud 
  
 import 
  storage 
 
 import 
  
 os 
 bucketname 
 = 
 os 
 . 
 getenv 
 ( 
 "BUCKET_NAME" 
 ) 
 filename 
 = 
 os 
 . 
 getenv 
 ( 
 "FILE_NAME" 
 ) 
 storage_client 
 = 
  storage 
 
 . 
  Client 
 
 () 
 bucket 
 = 
 storage_client 
 . 
  bucket 
 
 ( 
 bucketname 
 ) 
 blob 
 = 
 bucket 
 . 
 blob 
 ( 
 filename 
 ) 
 blob 
 . 
  download_to_filename 
 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 loader 
 = 
 PyPDFLoader 
 ( 
 "/documents/" 
 + 
 filename 
 ) 
 text_splitter 
 = 
 RecursiveCharacterTextSplitter 
 ( 
 chunk_size 
 = 
 1000 
 , 
 chunk_overlap 
 = 
 0 
 ) 
 documents 
 = 
 loader 
 . 
 load_and_split 
 ( 
 text_splitter 
 ) 
 embeddings 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 auth_config 
 = 
 weaviate 
 . 
 auth 
 . 
 AuthApiKey 
 ( 
 api_key 
 = 
 os 
 . 
 getenv 
 ( 
 "APIKEY" 
 )) 
 client 
 = 
 weaviate 
 . 
 WeaviateClient 
 ( 
 connection_params 
 = 
 ConnectionParams 
 . 
 from_params 
 ( 
 http_host 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_ENDPOINT" 
 ), 
 http_port 
 = 
 "80" 
 , 
 http_secure 
 = 
 False 
 , 
 grpc_host 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_GRPC_ENDPOINT" 
 ), 
 grpc_port 
 = 
 "50051" 
 , 
 grpc_secure 
 = 
 False 
 , 
 ), 
 auth_client_secret 
 = 
 auth_config 
 ) 
 client 
 . 
 connect 
 () 
 if 
 not 
 client 
 . 
 collections 
 . 
 exists 
 ( 
 "trainingdocs" 
 ): 
 collection 
 = 
 client 
 . 
 collections 
 . 
 create 
 ( 
 name 
 = 
 "trainingdocs" 
 ) 
 db 
 = 
 WeaviateVectorStore 
 . 
 from_documents 
 ( 
 documents 
 , 
 embeddings 
 , 
 client 
 = 
 client 
 , 
 index_name 
 = 
 "trainingdocs" 
 ) 
 print 
 ( 
 filename 
 + 
 " was successfully embedded" 
 ) 
 print 
 ( 
 f 
 "# of vectors = 
 { 
 len 
 ( 
 documents 
 ) 
 } 
 " 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 flask 
  
 import 
 Flask 
 , 
 jsonify 
 from 
  
 flask 
  
 import 
 request 
 import 
  
 logging 
 import 
  
 sys 
 , 
 os 
 , 
  
 time 
 from 
  
 kubernetes 
  
 import 
 client 
 , 
 config 
 , 
 utils 
 import 
  
 kubernetes.client 
 from 
  
 kubernetes.client.rest 
  
 import 
 ApiException 
 app 
 = 
 Flask 
 ( 
 __name__ 
 ) 
 @app 
 . 
 route 
 ( 
 '/check' 
 ) 
 def 
  
 message 
 (): 
 return 
 jsonify 
 ({ 
 "Message" 
 : 
 "Hi there" 
 }) 
 @app 
 . 
 route 
 ( 
 '/' 
 , 
 methods 
 = 
 [ 
 'POST' 
 ]) 
 def 
  
 bucket 
 (): 
 request_data 
 = 
 request 
 . 
 get_json 
 () 
 print 
 ( 
 request_data 
 ) 
 bckt 
 = 
 request_data 
 [ 
 'bucket' 
 ] 
 f_name 
 = 
 request_data 
 [ 
 'name' 
 ] 
 id 
 = 
 request_data 
 [ 
 'generation' 
 ] 
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ) 
 return 
 "ok" 
 # Set logging 
 logging 
 . 
 basicConfig 
 ( 
 stream 
 = 
 sys 
 . 
 stdout 
 , 
 level 
 = 
 logging 
 . 
 INFO 
 ) 
 # Setup K8 configs 
 config 
 . 
 load_incluster_config 
 () 
 def 
  
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bucket_name 
 , 
 f_name 
 , 
 namespace 
 = 
 "qdrant" 
 , 
 container_name 
 = 
 "jobcontainer" 
 , 
 env_vars 
 = 
 {}): 
 body 
 = 
 client 
 . 
 V1Job 
 ( 
 api_version 
 = 
 "batch/v1" 
 , 
 kind 
 = 
 "Job" 
 ) 
 body 
 . 
 metadata 
 = 
 client 
 . 
 V1ObjectMeta 
 ( 
 namespace 
 = 
 namespace 
 , 
 name 
 = 
 name 
 ) 
 body 
 . 
 status 
 = 
 client 
 . 
 V1JobStatus 
 () 
 template 
 = 
 client 
 . 
 V1PodTemplate 
 () 
 template 
 . 
 template 
 = 
 client 
 . 
 V1PodTemplateSpec 
 () 
 env_list 
 = 
 [ 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "QDRANT_URL" 
 , 
 value 
 = 
 os 
 . 
 getenv 
 ( 
 "QDRANT_URL" 
 )), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "COLLECTION_NAME" 
 , 
 value 
 = 
 "training-docs" 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "FILE_NAME" 
 , 
 value 
 = 
 f_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "BUCKET_NAME" 
 , 
 value 
 = 
 bucket_name 
 ), 
 client 
 . 
 V1EnvVar 
 ( 
 name 
 = 
 "APIKEY" 
 , 
 value_from 
 = 
 client 
 . 
 V1EnvVarSource 
 ( 
 secret_key_ref 
 = 
 client 
 . 
 V1SecretKeySelector 
 ( 
 key 
 = 
 "api-key" 
 , 
 name 
 = 
 "qdrant-database-apikey" 
 ))), 
 ] 
 container 
 = 
 client 
 . 
 V1Container 
 ( 
 name 
 = 
 container_name 
 , 
 image 
 = 
 container_image 
 , 
 env 
 = 
 env_list 
 ) 
 template 
 . 
 template 
 . 
 spec 
 = 
 client 
 . 
 V1PodSpec 
 ( 
 containers 
 = 
 [ 
 container 
 ], 
 restart_policy 
 = 
 'Never' 
 , 
 service_account 
 = 
 'embed-docs-sa' 
 ) 
 body 
 . 
 spec 
 = 
 client 
 . 
 V1JobSpec 
 ( 
 backoff_limit 
 = 
 3 
 , 
 ttl_seconds_after_finished 
 = 
 60 
 , 
 template 
 = 
 template 
 . 
 template 
 ) 
 return 
 body 
 def 
  
 kube_test_credentials 
 (): 
 try 
 : 
 api_response 
 = 
 api_instance 
 . 
 get_api_resources 
 () 
 logging 
 . 
 info 
 ( 
 api_response 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling API: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 def 
  
 kube_create_job 
 ( 
 bckt 
 , 
 f_name 
 , 
 id 
 ): 
 container_image 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_IMAGE" 
 ) 
 namespace 
 = 
 os 
 . 
 getenv 
 ( 
 "JOB_NAMESPACE" 
 ) 
 name 
 = 
 "docs-embedder" 
 + 
 id 
 body 
 = 
 kube_create_job_object 
 ( 
 name 
 , 
 container_image 
 , 
 bckt 
 , 
 f_name 
 ) 
 v1 
 = 
 client 
 . 
 BatchV1Api 
 () 
 try 
 : 
 v1 
 . 
 create_namespaced_job 
 ( 
 namespace 
 , 
 body 
 , 
 pretty 
 = 
 True 
 ) 
 except 
 ApiException 
 as 
 e 
 : 
 print 
 ( 
 "Exception when calling BatchV1Api->create_namespaced_job: 
 %s 
 \n 
 " 
 % 
 e 
 ) 
 return 
 if 
 __name__ 
 == 
 '__main__' 
 : 
 app 
 . 
 run 
 ( 
 '0.0.0.0' 
 , 
 port 
 = 
 5001 
 , 
 debug 
 = 
 True 
 ) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 ChatVertexAI 
 from 
  
 langchain.prompts 
  
 import 
 ChatPromptTemplate 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain.memory 
  
 import 
 ConversationBufferWindowMemory 
 from 
  
 elasticsearch 
  
 import 
 Elasticsearch 
 from 
  
 langchain_community.vectorstores.elasticsearch 
  
 import 
 ElasticsearchStore 
 import 
  
 streamlit 
  
 as 
  
 st 
 import 
  
 os 
 vertexAI 
 = 
 ChatVertexAI 
 ( 
 model_name 
 = 
 os 
 . 
 getenv 
 ( 
 "VERTEX_AI_MODEL_NAME" 
 , 
 "gemini-2.5-flash-preview-04-17" 
 ), 
 streaming 
 = 
 True 
 , 
 convert_system_message_to_human 
 = 
 True 
 ) 
 prompt_template 
 = 
 ChatPromptTemplate 
 . 
 from_messages 
 ( 
 [ 
 ( 
 "system" 
 , 
 "You are a helpful assistant who helps in finding answers to questions using the provided context." 
 ), 
 ( 
 "human" 
 , 
 """ 
 The answer should be based on the text context given in "text_context" and the conversation history given in "conversation_history" along with its Caption: 
 \n 
 Base your response on the provided text context and the current conversation history to answer the query. 
 Select the most relevant information from the context. 
 Generate a draft response using the selected information. Remove duplicate content from the draft response. 
 Generate your final response after adjusting it to increase accuracy and relevance. 
 Now only show your final response! 
 If you do not know the answer or context is not relevant, response with "I don't know". 
 text_context: 
  
 {context} 
 conversation_history: 
  
 {history} 
 query: 
  
 {query} 
 """ 
 ), 
 ] 
 ) 
 embedding_model 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 client 
 = 
 Elasticsearch 
 ( 
 [ 
 os 
 . 
 getenv 
 ( 
 "ES_URL" 
 )], 
 verify_certs 
 = 
 False 
 , 
 ssl_show_warn 
 = 
 False 
 , 
 basic_auth 
 = 
 ( 
 "elastic" 
 , 
 os 
 . 
 getenv 
 ( 
 "PASSWORD" 
 )) 
 ) 
 vector_search 
 = 
 ElasticsearchStore 
 ( 
 index_name 
 = 
 os 
 . 
 getenv 
 ( 
 "INDEX_NAME" 
 ), 
 es_connection 
 = 
 client 
 , 
 embedding 
 = 
 embedding_model 
 ) 
 def 
  
 format_docs 
 ( 
 docs 
 ): 
 return 
 " 
 \n\n 
 " 
 . 
 join 
 ([ 
 d 
 . 
 page_content 
 for 
 d 
 in 
 docs 
 ]) 
 st 
 . 
 title 
 ( 
 "🤖 Chatbot" 
 ) 
 if 
 "messages" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "messages" 
 ] 
 = 
 [{ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 "How can I help you?" 
 }] 
 if 
 "memory" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "memory" 
 ] 
 = 
 ConversationBufferWindowMemory 
 ( 
 memory_key 
 = 
 "history" 
 , 
 ai_prefix 
 = 
 "Bot" 
 , 
 human_prefix 
 = 
 "User" 
 , 
 k 
 = 
 3 
 , 
 ) 
 for 
 message 
 in 
 st 
 . 
 session_state 
 . 
 messages 
 : 
 with 
 st 
 . 
 chat_message 
 ( 
 message 
 [ 
 "role" 
 ]): 
 st 
 . 
 write 
 ( 
 message 
 [ 
 "content" 
 ]) 
 if 
 chat_input 
 := 
 st 
 . 
 chat_input 
 (): 
 with 
 st 
 . 
 chat_message 
 ( 
 "human" 
 ): 
 st 
 . 
 write 
 ( 
 chat_input 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "human" 
 , 
 "content" 
 : 
 chat_input 
 }) 
 found_docs 
 = 
 vector_search 
 . 
 similarity_search 
 ( 
 chat_input 
 ) 
 context 
 = 
 format_docs 
 ( 
 found_docs 
 ) 
 prompt_value 
 = 
 prompt_template 
 . 
 format_messages 
 ( 
 name 
 = 
 "Bot" 
 , 
 query 
 = 
 chat_input 
 , 
 context 
 = 
 context 
 , 
 history 
 = 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 load_memory_variables 
 ({})) 
 with 
 st 
 . 
 chat_message 
 ( 
 "ai" 
 ): 
 with 
 st 
 . 
 spinner 
 ( 
 "Typing..." 
 ): 
 content 
 = 
 "" 
 with 
 st 
 . 
 empty 
 (): 
 for 
 chunk 
 in 
 vertexAI 
 . 
 stream 
 ( 
 prompt_value 
 ): 
 content 
 += 
 chunk 
 . 
 content 
 st 
 . 
 write 
 ( 
 content 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 content 
 }) 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 save_context 
 ({ 
 "input" 
 : 
 chat_input 
 }, 
 { 
 "output" 
 : 
 content 
 }) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 ChatVertexAI 
 from 
  
 langchain.prompts 
  
 import 
 ChatPromptTemplate 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain.memory 
  
 import 
 ConversationBufferWindowMemory 
 from 
  
 langchain_community.vectorstores.pgvector 
  
 import 
 PGVector 
 import 
  
 streamlit 
  
 as 
  
 st 
 import 
  
 os 
 vertexAI 
 = 
 ChatVertexAI 
 ( 
 model_name 
 = 
 os 
 . 
 getenv 
 ( 
 "VERTEX_AI_MODEL_NAME" 
 , 
 "gemini-2.5-flash-preview-04-17" 
 ), 
 streaming 
 = 
 True 
 , 
 convert_system_message_to_human 
 = 
 True 
 ) 
 prompt_template 
 = 
 ChatPromptTemplate 
 . 
 from_messages 
 ( 
 [ 
 ( 
 "system" 
 , 
 "You are a helpful assistant who helps in finding answers to questions using the provided context." 
 ), 
 ( 
 "human" 
 , 
 """ 
 The answer should be based on the text context given in "text_context" and the conversation history given in "conversation_history" along with its Caption: 
 \n 
 Base your response on the provided text context and the current conversation history to answer the query. 
 Select the most relevant information from the context. 
 Generate a draft response using the selected information. Remove duplicate content from the draft response. 
 Generate your final response after adjusting it to increase accuracy and relevance. 
 Now only show your final response! 
 If you do not know the answer or context is not relevant, response with "I don't know". 
 text_context: 
  
 {context} 
 conversation_history: 
  
 {history} 
 query: 
  
 {query} 
 """ 
 ), 
 ] 
 ) 
 embedding_model 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 CONNECTION_STRING 
 = 
 PGVector 
 . 
 connection_string_from_db_params 
 ( 
 driver 
 = 
 "psycopg2" 
 , 
 host 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "POSTGRES_HOST" 
 ), 
 port 
 = 
 5432 
 , 
 database 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "DATABASE_NAME" 
 ), 
 user 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "USERNAME" 
 ), 
 password 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "PASSWORD" 
 ), 
 ) 
 COLLECTION_NAME 
 = 
 os 
 . 
 environ 
 . 
 get 
 ( 
 "COLLECTION_NAME" 
 ), 
 vector_search 
 = 
 PGVector 
 ( 
 collection_name 
 = 
 COLLECTION_NAME 
 , 
 connection_string 
 = 
 CONNECTION_STRING 
 , 
 embedding_function 
 = 
 embedding_model 
 , 
 ) 
 def 
  
 format_docs 
 ( 
 docs 
 ): 
 return 
 " 
 \n\n 
 " 
 . 
 join 
 ([ 
 d 
 . 
 page_content 
 for 
 d 
 in 
 docs 
 ]) 
 st 
 . 
 title 
 ( 
 "🤖 Chatbot" 
 ) 
 if 
 "messages" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "messages" 
 ] 
 = 
 [{ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 "How can I help you?" 
 }] 
 if 
 "memory" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "memory" 
 ] 
 = 
 ConversationBufferWindowMemory 
 ( 
 memory_key 
 = 
 "history" 
 , 
 ai_prefix 
 = 
 "Bot" 
 , 
 human_prefix 
 = 
 "User" 
 , 
 k 
 = 
 3 
 , 
 ) 
 for 
 message 
 in 
 st 
 . 
 session_state 
 . 
 messages 
 : 
 with 
 st 
 . 
 chat_message 
 ( 
 message 
 [ 
 "role" 
 ]): 
 st 
 . 
 write 
 ( 
 message 
 [ 
 "content" 
 ]) 
 if 
 chat_input 
 := 
 st 
 . 
 chat_input 
 (): 
 with 
 st 
 . 
 chat_message 
 ( 
 "human" 
 ): 
 st 
 . 
 write 
 ( 
 chat_input 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "human" 
 , 
 "content" 
 : 
 chat_input 
 }) 
 found_docs 
 = 
 vector_search 
 . 
 similarity_search 
 ( 
 chat_input 
 ) 
 context 
 = 
 format_docs 
 ( 
 found_docs 
 ) 
 prompt_value 
 = 
 prompt_template 
 . 
 format_messages 
 ( 
 name 
 = 
 "Bot" 
 , 
 query 
 = 
 chat_input 
 , 
 context 
 = 
 context 
 , 
 history 
 = 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 load_memory_variables 
 ({})) 
 with 
 st 
 . 
 chat_message 
 ( 
 "ai" 
 ): 
 with 
 st 
 . 
 spinner 
 ( 
 "Typing..." 
 ): 
 content 
 = 
 "" 
 with 
 st 
 . 
 empty 
 (): 
 for 
 chunk 
 in 
 vertexAI 
 . 
 stream 
 ( 
 prompt_value 
 ): 
 content 
 += 
 chunk 
 . 
 content 
 st 
 . 
 write 
 ( 
 content 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 content 
 }) 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 save_context 
 ({ 
 "input" 
 : 
 chat_input 
 }, 
 { 
 "output" 
 : 
 content 
 }) 
 

  # Copyright 2024 Google LLC 
 # 
 # Licensed under the Apache License, Version 2.0 (the "License"); 
 # you may not use this file except in compliance with the License. 
 # You may obtain a copy of the License at 
 # 
 #     https://www.apache.org/licenses/LICENSE-2.0 
 # 
 # Unless required by applicable law or agreed to in writing, software 
 # distributed under the License is distributed on an "AS IS" BASIS, 
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 # See the License for the specific language governing permissions and 
 # limitations under the License. 
 from 
  
 langchain_google_vertexai 
  
 import 
 ChatVertexAI 
 from 
  
 langchain.prompts 
  
 import 
 ChatPromptTemplate 
 from 
  
 langchain_google_vertexai 
  
 import 
 VertexAIEmbeddings 
 from 
  
 langchain.memory 
  
 import 
 ConversationBufferWindowMemory 
 import 
  
 weaviate 
 from 
  
 weaviate.connect 
  
 import 
 ConnectionParams 
 from 
  
 langchain_weaviate.vectorstores 
  
 import 
 WeaviateVectorStore 
 import 
  
 streamlit 
  
 as 
  
 st 
 import 
  
 os 
 vertexAI 
 = 
 ChatVertexAI 
 ( 
 model_name 
 = 
 os 
 . 
 getenv 
 ( 
 "VERTEX_AI_MODEL_NAME" 
 , 
 "gemini-2.5-flash-preview-04-17" 
 ), 
 streaming 
 = 
 True 
 , 
 convert_system_message_to_human 
 = 
 True 
 ) 
 prompt_template 
 = 
 ChatPromptTemplate 
 . 
 from_messages 
 ( 
 [ 
 ( 
 "system" 
 , 
 "You are a helpful assistant who helps in finding answers to questions using the provided context." 
 ), 
 ( 
 "human" 
 , 
 """ 
 The answer should be based on the text context given in "text_context" and the conversation history given in "conversation_history" along with its Caption: 
 \n 
 Base your response on the provided text context and the current conversation history to answer the query. 
 Select the most relevant information from the context. 
 Generate a draft response using the selected information. Remove duplicate content from the draft response. 
 Generate your final response after adjusting it to increase accuracy and relevance. 
 Now only show your final response! 
 If you do not know the answer or context is not relevant, response with "I don't know". 
 text_context: 
  
 {context} 
 conversation_history: 
  
 {history} 
 query: 
  
 {query} 
 """ 
 ), 
 ] 
 ) 
 embedding_model 
 = 
 VertexAIEmbeddings 
 ( 
 "text-embedding-005" 
 ) 
 auth_config 
 = 
 weaviate 
 . 
 auth 
 . 
 AuthApiKey 
 ( 
 api_key 
 = 
 os 
 . 
 getenv 
 ( 
 "APIKEY" 
 )) 
 client 
 = 
 weaviate 
 . 
 WeaviateClient 
 ( 
 connection_params 
 = 
 ConnectionParams 
 . 
 from_params 
 ( 
 http_host 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_ENDPOINT" 
 ), 
 http_port 
 = 
 "80" 
 , 
 http_secure 
 = 
 False 
 , 
 grpc_host 
 = 
 os 
 . 
 getenv 
 ( 
 "WEAVIATE_GRPC_ENDPOINT" 
 ), 
 grpc_port 
 = 
 "50051" 
 , 
 grpc_secure 
 = 
 False 
 , 
 ), 
 auth_client_secret 
 = 
 auth_config 
 ) 
 client 
 . 
 connect 
 () 
 vector_search 
 = 
 WeaviateVectorStore 
 . 
 from_documents 
 ([], 
 embedding_model 
 , 
 client 
 = 
 client 
 , 
 index_name 
 = 
 "trainingdocs" 
 ) 
 def 
  
 format_docs 
 ( 
 docs 
 ): 
 return 
 " 
 \n\n 
 " 
 . 
 join 
 ([ 
 d 
 . 
 page_content 
 for 
 d 
 in 
 docs 
 ]) 
 st 
 . 
 title 
 ( 
 "🤖 Chatbot" 
 ) 
 if 
 "messages" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "messages" 
 ] 
 = 
 [{ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 "How can I help you?" 
 }] 
 if 
 "memory" 
 not 
 in 
 st 
 . 
 session_state 
 : 
 st 
 . 
 session_state 
 [ 
 "memory" 
 ] 
 = 
 ConversationBufferWindowMemory 
 ( 
 memory_key 
 = 
 "history" 
 , 
 ai_prefix 
 = 
 "Bot" 
 , 
 human_prefix 
 = 
 "User" 
 , 
 k 
 = 
 3 
 , 
 ) 
 for 
 message 
 in 
 st 
 . 
 session_state 
 . 
 messages 
 : 
 with 
 st 
 . 
 chat_message 
 ( 
 message 
 [ 
 "role" 
 ]): 
 st 
 . 
 write 
 ( 
 message 
 [ 
 "content" 
 ]) 
 if 
 chat_input 
 := 
 st 
 . 
 chat_input 
 (): 
 with 
 st 
 . 
 chat_message 
 ( 
 "human" 
 ): 
 st 
 . 
 write 
 ( 
 chat_input 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "human" 
 , 
 "content" 
 : 
 chat_input 
 }) 
 found_docs 
 = 
 vector_search 
 . 
 similarity_search 
 ( 
 chat_input 
 ) 
 context 
 = 
 format_docs 
 ( 
 found_docs 
 ) 
 prompt_value 
 = 
 prompt_template 
 . 
 format_messages 
 ( 
 name 
 = 
 "Bot" 
 , 
 query 
 = 
 chat_input 
 , 
 context 
 = 
 context 
 , 
 history 
 = 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 load_memory_variables 
 ({})) 
 with 
 st 
 . 
 chat_message 
 ( 
 "ai" 
 ): 
 with 
 st 
 . 
 spinner 
 ( 
 "Typing..." 
 ): 
 content 
 = 
 "" 
 with 
 st 
 . 
 empty 
 (): 
 for 
 chunk 
 in 
 vertexAI 
 . 
 stream 
 ( 
 prompt_value 
 ): 
 content 
 += 
 chunk 
 . 
 content 
 st 
 . 
 write 
 ( 
 content 
 ) 
 st 
 . 
 session_state 
 . 
 messages 
 . 
 append 
 ({ 
 "role" 
 : 
 "ai" 
 , 
 "content" 
 : 
 content 
 }) 
 st 
 . 
 session_state 
 . 
 memory 
 . 
 save_context 
 ({ 
 "input" 
 : 
 chat_input 
 }, 
 { 
 "output" 
 : 
 content 
 })