お客様と対話するインテリジェントなボットをビジネスに導入することを想像してみてください。チャットボットは一般的に、顧客と対話し、彼らに助けや情報を提供するために使用されます。しかし、通常のチャットボットは複雑な質問に答えるのに苦労することがあります。
Retrieval Augmented Generation (RAG)は、チャットボットが難しい質問を理解し、応答する能力を向上させる方法です。この生成AIデザインパターンは、大規模言語モデル（LLM）と外部知識の取得を組み合わせています。 これにより、リアルタイムデータを生成プロセス（推論時間）中にAIアプリケーションに統合することが可能になります。この文脈情報をLLMに提供することで、RAGは生成された出力の精度と品質を大幅に向上さ せます。
RAGを使用する利点の一部は次のとおりです：
Pineconeのベクトルデータベースは、複雑なデータ検索をピンポイントで管理するのに優れている一方、Databricksのデータインテリジェンスプラットフォームは 、大量のデータセットの取り扱いと分析を効率化します。
Pineconeとの統合はシームレスで、Databricksが大規模にベクトル埋め込みを効率的に保存および取得できます。 この統合により、PineconeとDatabricksを活用した高性能ベクトル検索アプリケーションの開発が簡素化されます。
DatabricksとPineconeを一緒に使用することで、従来のチャットボットよりも正確で効率的なチャットボットを作成することができます。
このブログでは、Databricksのドキュメントやホワイトペーパーを活用して、Databricksに関するあらゆる質問に答えられるチャットボットの構築方法をご紹介します。
チャットボットの構築には、4つの重要なステージがあります。最初のステージは、データのインジェストと準備です。次に、効率的な情報検索のために、データをPineconeのようなベクターデータベースに保存します。3つ目のステージでは、Pineconeを使った検索と、Llama 3.1のようなLLM（大規模言語モデル）を使用して応答を生成するRAGリトリーバーとチェーンを設定します。最後に、チャットボットをDatabricks Unity Catalogに登録し、Databricks Mosaic AI Model Servingを通じてデプロイします。このプロセスの詳細な手順については、続きをご覧ください。
# Pineconeの初期化
import pinecone
from pinecone import Pinecone
pinecone_api_key = dbutils.secrets.get("your_secrets_scope", "PINECONE_API_KEY")
project_name = "Starter" # your-pinecone-project-name
index_name = "dbdemo-index" # your-pinecone-index-name
# connect to pinecone index
pc = Pinecone(api_key=api_key)
index = pc.Index(index_name)pc.create_index(
name= index_name, # pineconeのインデックス名
dimension=1536,
metric="cosine",
spec=ServerlessSpec(
cloud="aws",
region="us-east-1"
)
)from pyspark.sql.functions import col, lit, struct, to_json
from pyspark.sql.functions import encode
df = spark.table('databricks_pdf_documentation')\
.withColumn("metadata", to_json(struct(col("content"), col("url"), col("id"))))\
.withColumn("namespace", lit("dbdemo-namespace")) \
.withColumn("values", col("embedding")) \
.withColumn("sparse_values", lit(None)) \
.select("id", "values", "namespace", "metadata", "sparse_values")#write to pinecone
(
df.write
.option("pinecone.apiKey", api_key)
.option("pinecone.indexName", index_name)
.format("io.pinecone.spark.pinecone.Pinecone")
.mode("append")
.save()
)次に、クエリAPIを使用してPineconeのベクトルインデックスをクエリすることができます。このAPIは、質問のエンベッディングを入力として受け取ります。
# UDF for embedding
from pyspark.sql.types import *
def get_embedding_for_string(text):
response = deploy_client.predict(endpoint="databricks-bge-large-en", inputs={"input": text})
e = response.data
return e[0]['embedding']
#udfとして登録
get_embedding_for_string_udf = udf(get_embedding_for_string, ArrayType(FloatType()))
# Pineconeベクトルデータベースのクエリ
question = "How can I track billing usage on my workspaces?"
# クエリ埋め込みを作成
xq = get_embedding_for_string(question)
# Pineconeに最も類似した上位5つの結果をクエリ
query_response = index.query(
namespace='dbdemo-namespace',
top_k=5,
include_values=True,
include_metadata=True,
vector=xq
)
#print(query_response)
query_response_docs = []
for match in query_response['matches']:
query_response_docs.append([match['metadata']['url'],match['metadata']['content'],match['score']])
print(query_response_docs)APIを通じてPineconeを直接クエリすることで、PineconeとDatabricksを任意のコードに統合できます。
次のセクションでは、人気のあるLangChainフレームワークを使用してこのワークフローを簡素化する方法を示します。
Langchainは、LLM（大規模言語モデル）によるアプリケーション構築を簡素化するフレームワークです。Databricksの埋め込みは、埋め込みモデルとの対話を簡素化し、Pineconeとの統合は簡素化されたクエリインターフェースを提供します。
Langchainのラッパーは、すべての基礎的なロジックとAPI呼び出しを処理することで、簡単にします。下記のLangChainコードは、クエリテキストをベクトルに明示的に変換する必要性を抽象化します。
import pinecone
from langchain_community.embeddings import DatabricksEmbeddings
from langchain.chains import RetrievalQA
from pinecone import Pinecone
from langchain_pinecone import PineconeVectorStore
import os
#Creating the input question embeddings (with Databricks `bge-large-en`)
embedding_model = DatabricksEmbeddings(endpoint="databricks-bge-large-en")
# connect to pinecone index
pc = Pinecone(api_key=pinecone_api_key)
index_pc = pc.Index(pinecone_index_name)
vectorstore = PineconeVectorStore(
index=index_pc,
namespace=pinecone_namespace,
embedding=embedding_model,
text_key="content"
)
#Calling the Pinecone vector database to find similar documents
query = "What is Apache Spark?"
docs = vectorstore.similarity_search(
クエリ、 # 検索クエリ
k=3 # 最も関連性の高いドキュメントを3つ返す
)
pprint(docs[0])上記では、Pineconeベクトルインデックスでの類似性検索の方法を示しました。RAGチャットボットを作成するために、LangChain Retrieverインターフェースを使用してインデックスをラップします。
まず、APIキーと環境を設定するためにPineconeを初期化します。次に、正しいネームスペースとキーを持つ既存のPineconeインデックスからVectorStoreインスタンスを作成します。
from langchain_community.embeddings import DatabricksEmbeddings
from langchain.chains import RetrievalQA
from pinecone import Pinecone
from langchain_pinecone import PineconeVectorStore
embedding_model = DatabricksEmbeddings(endpoint="databricks-bge-large-en")
def get_retriever(persist_dir: str = None):
# initialize pinecone and connect to pinecone index
pc = Pinecone(api_key=pinecone_api_key)
index_pc = pc.Index(pinecone_index_name)
vectorstore = PineconeVectorStore(
index=index_pc,
namespace=pinecone_namespace,
embedding=embedding_model,
text_key="content"
)
return vectorstore.as_retriever()
retriever = get_retriever()これで、チャットボットを定義するチェーンにリトリーバーを入れることができます！
retrieve_document_chain = (
itemgetter("messages")
| RunnableLambda(extract_question)
| retriever
)
#test the retriever chain
print(retrieve_document_chain.invoke({"messages": [{"role": "user", "content": "Apache Sparkとは何ですか？"}]}))チャットボットがチャットメッセージから質問を正しく抽出し、Pineconeから関連するコンテキストを取得できるかどうかを見てみましょう。
チャットボットを反復処理するにつれて、モデルオブジェクト、モデルバージョン、メタデータを追跡し、アクセス制御を管理することが求められます。そのためには、Unity Catalogと統合されたMLflowのModel Registryを使用します。
Unityカタログを使用して、チャットボットチェーンをmlflow.langchain.log_modelを使用してモデルとして登録できます。モデルのシグネチャは、mlflowのinfer_signatureを使用して推測することができます。 pinecone-clientを依存関係に入れることを忘れないでください。チェーンを定義したノートブックで"mlflow.models.set_model(model=full_chain)"を設定します。新しいドライバーノートブックで、チャットボットを登録し、モデルサービングにチャットボットをデプロイします。
from mlflow.models import infer_signature
import mlflow
import langchain
import pandas as pd
mlflow.set_registry_uri("databricks-uc")
model_name = f"{catalog_name}.{schema_name}.rag_with_pinecone_model" #catalog_name, schema_name and model name
# Specify the full path to the chain notebook
chain_notebook_file = "2.1 - Advanced-Chatbot-Chain - Using Pinecone" # the name of the notebook that has the chain definition
chain_notebook_path = os.path.join(os.getcwd(), chain_notebook_file)
mlflow.start_run()で：
signature = infer_signature(input_example, output_example)
logged_chain_info = mlflow.langchain.log_model(
lc_model=chain_notebook_path,
artifact_path="chain",
registered_model_name=model_name,
input_example=input_example,
signature=signature,
example_no_conversion=True, # スキーマを動作させるために必要
extra_pip_requirements=[
"mlflow==" + mlflow.__version__,
"langchain==" + langchain.__version__,
"pinecone-client==3.2.2",
"langchain-pinecone==0.1.1",
"langchain-community"、
]
)モデルはDatabricks のUnity Catalogに登録されており、すべてのデータとAI資産のアクセス制御、監査、系統、発見を一元化します。
さあ、チャットボットチェーンモードをモデルサービングエンドポイントとしてデプロイしましょう。以下では、PINECONE_API_KEYとDATABRICKS_TOKENを環境変数に入れます。これは、サービングエンドポイントがPineconeとDatabricks Foundation Modelsと通信するために使用します。これにより、コードやユーザーにこれらの機密情報を明かすことなく、サーブされたモデルへのアクセスを許可することができます。
# サービングエンドポイントの作成または更新
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedModelInput, ServedModelInputWorkloadSize
import requests
# 最新のモデルバージョンをチェック
def get_latest_model_version(model_name):
from mlflow import MlflowClient
mlflow_client = MlflowClient()
latest_version = 1
for mv in mlflow_client.search_model_versions(f"name='{model_name}'"):
version_int = int(mv.version)
if version_int > latest_version:
latest_version = version_int
return latest_version
# これでサービングエンドポイントを作成または更新します
serving_endpoint_name = "pinecone_rag_chain"
latest_model_version = get_latest_model_version(model_name)
databricks_api_token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()
w = WorkspaceClient()
endpoint_config = EndpointCoreConfigInput(
name=serving_endpoint_name,
served_models=[
ServedModelInput(
model_name=model_name,
model_version=latest_model_version,
workload_size=ServedModelInputWorkloadSize.SMALL,
scale_to_zero_enabled=True,
environment_vars={
"PINECONE_API_KEY": "{{secrets/prasad_kona/PINECONE_API_KEY}}",
"DATABRICKS_TOKEN": "{{secrets/dbdemos/rag_sp_token}}",
}
)
]
)
existing_endpoint = next(
(e for e in w.serving_endpoints.list() if e.name == serving_endpoint_name), None
)
serving_endpoint_url = f"{host}/ml/endpoints/{serving_endpoint_name}"
if existing_endpoint == None:
print(f"エンドポイント{serving_endpoint_url}を作成しています、これにはパッケージ化とデプロイに数分かかります...")
w.serving_endpoints.create_and_wait(name=serving_endpoint_name, config=endpoint_config)
else:
print(f"エンドポイント{serving_endpoint_url}をバージョン{latest_model_version}に更新します。これには、エンドポイントのパッケージ化とデプロイに数分かかります...")
w.serving_endpoints.update_config_and_wait(served_models=endpoint_config.served_models, name=serving_endpoint_name)
displayHTML(f'あなたのモデルエンドポイントの提供が利用可能になりました。詳細は<a href="/ml/endpoints/{serving_endpoint_name}">モデルサービングエンドポイントページ</a>をご覧ください。')モデルサービングUIは、提供されているモデルのヘルスに関するリアルタイム情報を提供します。
チャットボットをデプロイした後、REST APIまたはDatabricks SDKでテストすることができます。
from databricks.sdk.service.serving import DataframeSplitInput
test_dialog = DataframeSplitInput(
columns=["messages"],
data=[
{
"messages": [
{"role": "user", "content": "Apache Sparkとは何ですか？"},
{
"role": "assistant",
"content": "Apache Sparkは、ビッグデータ分析で広く使用されているオープンソースのデータ処理エンジンです。",
},
{"role": "user", "content": "ストリーミングはサポートしていますか？"},
]
}
],
)
answer = w.serving_endpoints.query(serving_endpoint_name, dataframe_split=test_dialog)
print(answer.predictions[0])モデル提供の一部として利用可能なQuery UIを使用してテストすることもできます。
DatabricksとPineconeを活用して、最先端のRAGチャットボットを導入し、顧客サービスを向上させましょう。従来のボットとは異なり、これらの高度なチャットボットはDatabricksのデータインテリジェンスプラットフォームとPineconeのベクターデータベースを活用して、正確かつタイムリーな回答を提供します。膨大なデータを素早く検索し、必要な情報を即座に見つけ、数秒でお客様に正確な答えを提供します。これにより、顧客体験が向上するだけでなく、デジタルエンゲージメントの新しい基準が確立されます。
ビジネスリーダーにとって、このテクノロジーの導入は単なるアップグレードではなく、顧客サービスの革新をリードするための戦略的な一手です。データ駆動型のインテリジェントなソリューションを採用することで、ビジネスを顧客対応の最前線に置き、優れたサービスへのコミットメントを示し、顧客の共感を得ることができます。
Databricksでの生成AIについ てもっと学べる無料トレーニングをぜひチェックしてください。また、PineconeおよびDatabricksの追加ドキュメントもこちらからご覧いただけます。このブログのサンプルノートブックもこちらからアクセスできます。
