Dialogflow CXで生成AIを使う方法をまとめてみた

import functions_framework
import requests
import vertexai
from vertexai.generative_models import GenerationConfig, GenerativeModel, Part

def interview(project_id: str,location: str,model: str,text: str) -> str:
    # Initialize Vertex AI
    vertexai.init(project=project_id, location=location)

    # Load the model
    model = GenerativeModel(model_name=model)

    # Generation Config
    config = GenerationConfig(
        max_output_tokens=2048, temperature=0.4, top_p=1, top_k=32
    )

    # Generate text
    response = model.generate_content(
        text, generation_config=config
    )

    return response.text

@functions_framework.http
def webhook(request):
    response = request.get_json()
    text = response["sessionInfo"]["parameters"]["question"]
    tag = response["fulfillmentInfo"]["tag"]
    model = "gemini-1.5-flash"
    output_text = interview("プロジェクト名","リージョン",model,text)

    response["fulfillmentResponse"] = {
            "messages": [
                {
                    "text": {
                        "text": [
                            output_text
                        ],
                        "allowPlaybackInterruption": False
                    }
                }
            ]
        }
    return response

webhookをAgentに設定し、テストした結果は以下の通りです。

import functions_framework
import vertexai
from google.cloud import discoveryengine
from vertexai.preview.generative_models import GenerativeModel, ChatSession
from typing import List
from google.cloud import storage
from google.oauth2 import service_account
import json
import requests

#Search用のparametersを定義
project_id = "プロジェクト名"
location = "global"
search_engine_id = "データストアのID"
#configIDはAgent Builderのコンソール→アプリ→統合で確認することができます。
serving_config_id = "configID"
location_v = "ロケーション"
model_name = "gemini-pro"

vertexai.init(project=project_id, location=location_v)
model = GenerativeModel(model_name)
chat = model.start_chat()

def extract_path_from_link(link: str):
    return "/".join(link[5:].split("/")[1:])

def search(
    project_id: str,
    location: str,
    search_engine_id: str,
    serving_config_id: str,
    search_query: str
    ) -> List[discoveryengine.SearchResponse.SearchResult]:

    client = discoveryengine.SearchServiceClient()

    serving_config = client.serving_config_path(
        project=project_id,
        location=location,
        data_store=search_engine_id,
        serving_config=serving_config_id,
    )

    request = discoveryengine.SearchRequest(
        serving_config=serving_config,
        query=search_query,
        page_size=2,
        content_search_spec=discoveryengine.SearchRequest.ContentSearchSpec(
            extractive_content_spec=discoveryengine.SearchRequest.ContentSearchSpec.ExtractiveContentSpec(
                max_extractive_segment_count=1,
                max_extractive_answer_count=1,
            ),
            snippet_spec=discoveryengine.SearchRequest.ContentSearchSpec.SnippetSpec(
                return_snippet=True,
            ),
            summary_spec=discoveryengine.SearchRequest.ContentSearchSpec.SummarySpec(
                summary_result_count=5,
                include_citations=True,
            ),
        ),
    )

    response = client.search(request)

    search_result = {}
    search_result = {
        "answer": response.summary.summary_text,
        "retrieved_texts": [
            {
                "document_path": extract_path_from_link(result.document.derived_struct_data["link"]),
                "link": result.document.derived_struct_data["link"],
                "snippets": result.document.derived_struct_data["snippets"][0]["snippet"],
                "text": result.document.derived_struct_data["extractive_segments"][0]["content"],
            }
            for result in response.results
        ],
    }

    search_result_text = search_result["answer"]

    return search_result_text

@functions_framework.http
def search_document(request):
    response = request.get_json()
    sentence = response["sessionInfo"]["parameters"]["question"]
    tag = response["fulfillmentInfo"]["tag"]

    output_text= search(project_id, location, search_engine_id, serving_config_id, sentence)

    response["fulfillmentResponse"] = {
            "messages": [
                {
                    "text": {
                        "text": [
                            output_text
                        ],
                        "allowPlaybackInterruption": False
                    }
                }
            ]
        }

    return response