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.
fromflaskimport Flask, jsonify
fromflaskimport request
importlogging
importsys,os,time
fromkubernetesimport client, config, utils
importkubernetes.client
fromkubernetes.client.restimport ApiException
app = Flask(__name__)
@app.route('/check')
defmessage():
 return jsonify({"Message": "Hi there"})
@app.route('/', methods=['POST'])
defbucket():
 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()
defkube_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
defkube_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)
defkube_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.
fromflaskimport Flask, jsonify
fromflaskimport request
importlogging
importsys,os,time
fromkubernetesimport client, config, utils
importkubernetes.client
fromkubernetes.client.restimport ApiException
app = Flask(__name__)
@app.route('/check')
defmessage():
 return jsonify({"Message": "Hi there"})
@app.route('/', methods=['POST'])
defbucket():
 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()
defkube_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
defkube_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)
defkube_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.
fromflaskimport Flask, jsonify
fromflaskimport request
importlogging
importsys,os,time
fromkubernetesimport client, config, utils
importkubernetes.client
fromkubernetes.client.restimport ApiException
app = Flask(__name__)
@app.route('/check')
defmessage():
 return jsonify({"Message": "Hi there"})
@app.route('/', methods=['POST'])
defbucket():
 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()
defkube_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
defkube_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)
defkube_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.
fromflaskimport Flask, jsonify
fromflaskimport request
importlogging
importsys,os,time
fromkubernetesimport client, config, utils
importkubernetes.client
fromkubernetes.client.restimport ApiException
app = Flask(__name__)
@app.route('/check')
defmessage():
 return jsonify({"Message": "Hi there"})
@app.route('/', methods=['POST'])
defbucket():
 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()
defkube_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
defkube_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)
defkube_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.
fromlangchain_google_vertexaiimport ChatVertexAI
fromlangchain.promptsimport ChatPromptTemplate
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain.memoryimport ConversationBufferWindowMemory
fromlangchain_community.vectorstoresimport Qdrant
fromqdrant_clientimport QdrantClient
importstreamlitasst
importos
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)
defformat_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.
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain_community.document_loadersimport PyPDFLoader
fromlangchain.text_splitterimport RecursiveCharacterTextSplitter
fromelasticsearchimport Elasticsearch
fromlangchain_community.vectorstores.elasticsearchimport ElasticsearchStore
fromgoogle.cloudimport storage
importos
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.
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain_community.document_loadersimport PyPDFLoader
fromlangchain.text_splitterimport RecursiveCharacterTextSplitter
fromlangchain_community.vectorstores.pgvectorimport PGVector
fromgoogle.cloudimport storage
importos
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.
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain_community.document_loadersimport PyPDFLoader
fromlangchain.text_splitterimport RecursiveCharacterTextSplitter
importweaviate
fromweaviate.connectimport ConnectionParams
fromlangchain_weaviate.vectorstoresimport WeaviateVectorStore
fromgoogle.cloudimport storage
importos
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.
fromflaskimport Flask, jsonify
fromflaskimport request
importlogging
importsys,os,time
fromkubernetesimport client, config, utils
importkubernetes.client
fromkubernetes.client.restimport ApiException
app = Flask(__name__)
@app.route('/check')
defmessage():
 return jsonify({"Message": "Hi there"})
@app.route('/', methods=['POST'])
defbucket():
 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()
defkube_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
defkube_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)
defkube_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.
fromlangchain_google_vertexaiimport ChatVertexAI
fromlangchain.promptsimport ChatPromptTemplate
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain.memoryimport ConversationBufferWindowMemory
fromelasticsearchimport Elasticsearch
fromlangchain_community.vectorstores.elasticsearchimport ElasticsearchStore
importstreamlitasst
importos
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
)
defformat_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.
fromlangchain_google_vertexaiimport ChatVertexAI
fromlangchain.promptsimport ChatPromptTemplate
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain.memoryimport ConversationBufferWindowMemory
fromlangchain_community.vectorstores.pgvectorimport PGVector
importstreamlitasst
importos
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,
)
defformat_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.
fromlangchain_google_vertexaiimport ChatVertexAI
fromlangchain.promptsimport ChatPromptTemplate
fromlangchain_google_vertexaiimport VertexAIEmbeddings
fromlangchain.memoryimport ConversationBufferWindowMemory
importweaviate
fromweaviate.connectimport ConnectionParams
fromlangchain_weaviate.vectorstoresimport WeaviateVectorStore
importstreamlitasst
importos
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")
defformat_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})