gcloud firestore indexes composite create --project=<project id> --collection-group=knowledge_demo --query-scope=COLLECTION_GROUP --field-config=vector-config='{"dimension":"1536","flat": "{}"}',field-path=position_vector

main.py

import json
import logging
import os
import re
import functions_framework
import google.cloud.logging
from openai import OpenAI
from box import Box
from flask import Request
from slack_bolt import App
from slack_bolt.adapter.google_cloud_functions import SlackRequestHandler
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from google.cloud import firestore
from google.cloud.firestore_v1.base_vector_query import DistanceMeasure
from google.cloud.firestore_v1.vector import Vector

# Google Cloud Logging クライアント ライブラリを設定
logging_client = google.cloud.logging.Client()
logging_client.setup_logging(log_level=logging.DEBUG)

# 環境変数からシークレットを取得
slack_token = os.environ.get("SLACK_BOT_TOKEN")
slack_signing_secret = os.environ.get("SLACK_SIGNING_SECRET")
openai_api_key = os.environ.get("OPENAI_API_KEY")

# Slackの設定
slack_client = WebClient(token=slack_token)

# OpenAIクライアントの初期化
client = OpenAI(api_key=openai_api_key)

# Firestoreクライアントの初期化
db = firestore.Client()

# FaaS で実行する場合、応答速度が遅いため process_before_response は True でなければならない
app = App(
    token=slack_token,
    signing_secret=slack_signing_secret,
    process_before_response=True
)
handler = SlackRequestHandler(app)

# Bot アプリにダイレクトメッセージしたイベントに対する応答
@app.event("message")
def handle_app_message_events(body: dict,say):
    try:
        logging.debug(type(body))
        logging.debug(body)
        event = body["event"]

        # メッセージが変更（リンクのプレビュー含む）された場合はむしする。
        subtype = event.get("subtype")
        if subtype in ["message_changed", "message_deleted", "channel_join"]:
            logging.debug(f"Message change or deletion event ignored: {subtype}")
            return

        #DMメッセージからのみ受付し、メッセージ処理を実行。
        if body["event"]["channel_type"] == "im":
            process_message_event(body)
        
    except Exception as e:
        error_message = f"Error processing message event: {str(e)}"
        logging.error(error_message)
        say(error_message)  # エラーをDMに通知
        raise

def process_message_event(body):
    try:
        box = Box(body)
        user_id = box.event.user
        text = box.event.text

        # slack投稿からテキスト抽出
        only_text = re.sub("<@[a-zA-Z0-9]{11}>", "", text)
        logging.debug(f"query: {only_text}")
        
        dm_response = slack_client.conversations_open(users=[user_id])
        dm_channel_id = dm_response["channel"]["id"]

        initial_message = slack_client.chat_postMessage(
            channel=dm_channel_id,
            text=f"お疲れさまです<@{user_id}> さん:exclamation:\n"
                ":book: ご質問をありがとうございます。\n"
                ":hourglass_flowing_sand: 回答生成が終わるまでしばらくお待ちください。"
        )
        ts = initial_message["ts"] 
        
        embedded_query = embed_text(only_text)
        search_results = search_firestore(embedded_query, neighbor_count=10)
        final_response = create_chat_completion(only_text, search_results)
        post_message_with_markdown(dm_channel_id, final_response, ts)

    except Exception as e:
        logging.error(f"Unexpected error: {str(e)}")
        slack_client.chat_postMessage(
            channel=dm_channel_id,
            text="処理中にエラーが発生しました。システム管理者にお問い合わせください。"
        )

def search_firestore(embedded_query, neighbor_count):
    try:
        # コレクションを参照
        collection_ref = db.collection_group('knowledge_demo')
        vector = Vector(embedded_query)

        #近傍検索処理を実行
        resp = collection_ref.find_nearest(
            vector_field="position_vector",
            query_vector=vector,
            distance_measure=DistanceMeasure.EUCLIDEAN,
            limit=neighbor_count
        ).stream()
    except Exception as e:
        logging.error(f"Unexpected error during Firestore search: {e}")
        raise e

    # レスポンスからネイバーを取得
    neighbors_result = []
    if resp:
        for doc in resp:
            if doc.exists:
                doc_data = doc.to_dict()
                logging.debug(f"Document ID: {doc.id}, Data: {doc_data}")
                neighbors_result.append({
                    'id':doc.id,
                    'original_text': doc_data.get('original_text', ''),
                    'created_at': doc_data.get('created_at', '')
                })
            else:
                logging.warning(f"Document {doc.id} does not exist.")

        logging.debug(f"neighbors resp text:{neighbors_result}")
    else:
        logging.debug("クエリの取得に失敗しました。")

    return neighbors_result

def embed_text(text):
    try:
        response = client.embeddings.create(
            model="text-embedding-3-small",  # 埋め込みモデルを指定
            input=text
        )
        embedding = response.data[0].embedding
        return embedding
    except Exception as e:
        logging.error(f"OpenAI API呼び出し中にエラーが発生しました: {e}")
        raise e

def get_data_from_firestore(neighbor_id):
    try:
        doc_ref = db.collection("knowledge_demo").document(neighbor_id)
        doc = doc_ref.get()
        if doc.exists:
            data = doc.to_dict()
            return data.get('original_text', '')
        else:
            logging.error(f"Firestore ドキュメント {neighbor_id} が見つかりませんでした。")
            return '', '', ''
    except Exception as e:
        logging.error(f"Firestore からデータ取得中にエラーが発生しました: {e}")
        raise e

def call_openai_api(model, prompt, temperature=0.7):
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "system", "content": prompt}],
            stop=None,
            temperature=temperature
        )
        return response
    except Exception as e:
        logging.error(f"OpenAI API呼び出し中にエラーが発生しました: {e}")
        raise e

def create_chat_completion(only_text, search_results):
    related_texts = []
    for i,neighbor in enumerate(search_results):
        text, url, created_at = get_data_from_firestore(neighbor["id"])
        logging.debug(f"neighbor no.{i}: {text} \n {url} \n {created_at}")
        if text:
            related_texts.append(f"ナレッジ: {text}")

    combined_related_texts = "\n\n".join(related_texts)
    logging.debug(f"combined_related_texts: {combined_related_texts}")

    prompt = f"""
    - 以下は株式会社テックディレクションの社内文書と質問者からの質問内容です。
    - 適宜、インデント、太字、箇条書きをSlack向けのマークダウン方式を活用してください。
    - 推測に基づいた回答を避け、不足している情報がある場合や次のアクションを質問にて提案してください。
    - 少しウィットに富んだニュアンスを加えることで親しみやすさを持たせてください。
    - プログラムコードや特記事項に関してはslackのコードブロックの形式にしてください。

    {combined_related_texts}
    質問: {only_text}
    """

    logging.debug(f"summary_prompt:{prompt}")

    try:
        summary_response = call_openai_api("gpt-4o", prompt, 0.3)
        final_response = summary_response.choices[0].message.content.replace("**","*")
        return final_response

    except Exception as e:
        logging.error(f"OpenAI API呼び出し中にエラーが発生しました: {e}")
        raise e

def post_message_with_markdown(channel_id, text, thread_ts):
    try:
        max_length = 3000
        text_chunks = [text[i:i+max_length] for i in range(0, len(text), max_length)]
        for i, chunk in enumerate(text_chunks):
            blocks = [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": chunk
                    }
                }
            ]

            if i == 0 and thread_ts:
                # 最初のメッセージを更新
                result = slack_client.chat_update(
                    channel=channel_id,
                    ts=thread_ts,
                    blocks=blocks
                )
            else:
                # 2つ目以降のメッセージはスレッドに投稿
                result = slack_client.chat_postMessage(
                    channel=channel_id,
                    thread_ts=thread_ts,
                    blocks=blocks
                )
            logging.debug(f"Message sent: {result}")
    except SlackApiError as e:
        logging.error(f"Error posting message: {e}")
        raise e

# Cloud Functions で呼び出されるエントリポイント
@functions_framework.http
def hello_http(request: Request):
    """slack のイベントリクエストを受信して各処理を実行する関数

    Args:
        request: Slack のイベントリクエスト

    Returns:
        SlackRequestHandler への接続
    """
    header = request.headers
    logging.debug(f"header: {header}")
    body = request.get_json()
    logging.debug(f"body: {body}")

    # URL確認を通すとき
    if body.get("type") == "url_verification":
        logging.info("url verification started")
        headers = {"Content-Type": "application/json"}
        res = json.dumps({"challenge": body["challenge"]})
        logging.debug(f"res: {res}")
        return (res, 200, headers)
    # 応答が遅いと Slack からリトライを何度も受信してしまうため、リトライ時は処理しない
    elif header.get("x-slack-retry-num"):
        logging.info("slack retry received")
        return {"statusCode": 200, "body": json.dumps({"message": "No need to resend"})}

    # handler への接続 class: flask.wrappers.Response
    return handler.handle(request)

requirements.txt

functions-framework
OpenAI
slack-bolt
slack-sdk
google-cloud-logging
flask == 3.*
python-box == 7.0.0
google-cloud-aiplatform>=1.29.0
google-cloud-firestore