Data Driven Knowledgebase

functions-framework==3.*

google-cloud-discoveryengine==0.11.6

oauth2client==4.1.3

from typing import Any, Mapping

from google.cloud.discoveryengine_v1alpha import SearchServiceClient, SearchRequest

from google.protobuf.json_format import MessageToDict

from oauth2client import client



import functions_framework, flask, os



PROJECT_ID = os.environ.get("PROJECT_ID")

PROJECT_NUMBER = os.environ.get("PROJECT_NUMBER")

DATA_STORE = os.environ.get("DATA_STORE")



#Google Chatから送られてくるBearer Tokenの整合に使用 CHAT_ISSUER = os.environ.get("CHAT_ISSUER") PUBLIC_CERT_URL_PREFIX = os.environ.get("PUBLIC_CERT_URL_PREFIX") def search_document(text: str) -> Mapping[str, Any]: 　 discov_client = SearchServiceClient() 　 #Vertex AI Searchのアプリ等基本的な内容を設定 　 serving_config = discov_client.serving_config_path( 　 project=PROJECT_ID, 　 location="global", 　 data_store=DATA_STORE, serving_config="default_config" 　) #Vertex AI Searchの出力内容に関する設定 　 content_search_spec = SearchRequest.ContentSearchSpec( 　 #スニペットを出力させない snippet_spec=SearchRequest.ContentSearchSpec().SnippetSpec( 　 　return_snippet=False ), #要約文を出力させる summary_spec=SearchRequest.ContentSearchSpec().SummarySpec( 　 summary_result_count=3, 　 include_citations=False, 　 #Gemini Proを用いるように指定 　 model_spec=SearchRequest.ContentSearchSpec().SummarySpec().ModelSpec( 　 version="preview" 　 ) 　 ) ) 　 #Vertex AI Searchにクエリを投げる 　 response = discov_client.search( 　 SearchRequest( 　 serving_config=serving_config, 　 query=text, 　 page_size=3, 　 content_search_spec=content_search_spec 　 ) 　 ) 　 #要約文取得 　 summary = response.summary.summary_text.replace("", "").replace("", "") 　 #関連ファイル取得 　 references = [] 　 for r in response.results: 　 r_dct = MessageToDict(r._pb) 　 link = r_dct["document"]["derivedStructData"]["link"] 　 references.append(link.split("/")[-1]) 　 result = { 　 "summary":summary, 　 "references":references 　 } 　 　 return result #チャットボットが実際に出力するメッセージを作成する def create_message(text: str) -> Mapping[str, Any]: result = search_document(text=text) reference_sentence = " ・".join(result["references"]) #返信文作成 cards = { "cardsV2": [ 　 { 　 "cardId": "searchResults", 　 "card": { 　 "sections": [ 　 { 　 "collapsible": False, 　 "widgets": [ { "textParagraph": { "text": "回答文" + result["summary"] } }, { "divider": {} }, { "textParagraph": { "text": "関連ファイル・ " + reference_sentence } } ] } ] } } ] } return cards #チャットボットからメッセージを受信 #create_message() で作成した JSON を返す @functions_framework.http def chatbot(req: flask.Request): """ 正しいGoogle Chatから送られてきたリクエストかを確認 """ if req.method == "GET": return flask.make_response(flask.jsonify({"message": "Bad Request"}), 400) auth_header = req.headers.get("Authorization") #「Authorization」ヘッダーが存在するか確認 if not auth_header: print("Missing Authorization header") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) #「Authorization」ヘッダーが「Bearer」から始まるか確認 if auth_header.split()[0] != "Bearer": print("Authorization header format is incorrect") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) bearer_token = auth_header.split()[1] try: token = client.verify_id_token( bearer_token, PROJECT_NUMBER, cert_uri=PUBLIC_CERT_URL_PREFIX + CHAT_ISSUER) if token['iss'] != CHAT_ISSUER: print("Invalid issuee") return flask.make_response(flask.jsonify({"message": "Unauthorized"}),401) except: print("Invalid token") return flask.make_response(flask.jsonify({"message": "Unauthorized"}), 401) """ 受け取ったメッセージでVertex AI Searchに検索をかける 検索結果を整形し送信する """ request_json = req.get_json(silent=True) text = request_json["message"]["text"] response = create_message(text=text) return response