LangfuseのDatabricks連携を試す

先日のイベントでLangfuseというサービスのことを伺いました。

https://langfuse.com/jp

LLMアプリケーションをデバッグ、改善するためのトレース、評価、プロンプト管理、メトリクスを提供しているオープンソースのLLMエンジニアリングプラットフォームとのこと。[MLflow](https://www.mlflow.org/docs/latest/index.html)や[LangSmith](https://www.langchain.com/langsmith)と同じようなモチベーションを持つプラットフォームですね。

調べてみたら、[Databricksとも連携](https://langfuse.com/docs/integrations/databricks/example-python)しているので試してみます。こちらのノートブックを動かします。

https://github.com/langfuse/langfuse-docs/blob/main/cookbook/integration_databricks.ipynb

翻訳版はこちらです。

https://github.com/taka-yayoi/langfuse_integration

# Langfuseを使用したDatabricksモデルの可観測性

Databricksは、大規模な言語モデルをホスティングおよび提供するための強力なプラットフォームを提供します。Databricksの提供エンドポイントと**Langfuse**を組み合わせることで、開発および本番環境でAIワークロードをトレース、監視、および分析できます。

このノートブックでは、Langfuseを使用してDatabricksモデルを使用する**3つ**の異なる方法を示します：
1. **OpenAI SDK:** OpenAI SDKを介してDatabricksモデルエンドポイントを使用します。
2. **LangChain:** LangChainパイプラインでDatabricks LLMインターフェースと統合します。
3. **LlamaIndex:** LlamaIndex内でDatabricksエンドポイントを使用します。

> **Databricks Model Servingとは？**  
Databricks Model Servingは、大規模なモデルを本番環境で提供するための自動スケーリングと堅牢なインフラストラクチャを備えています。また、プライベートデータでLLMを微調整することも可能で、データプライバシーを維持しながら独自の情報を活用できます。

> **Langfuseとは？**  
[Langfuse](https://langfuse.com)は、LLMの可観測性と監視のためのオープンソースプラットフォームです。メタデータ、プロンプトの詳細、トークン使用量、レイテンシーなどをキャプチャすることで、AIアプリケーションをトレースおよび監視するのに役立ちます。

## 1. 依存関係のインストール

始める前に、Python環境に必要なパッケージをインストールします：

- **openai**: OpenAI SDKを介してDatabricksエンドポイントを呼び出すために必要です。
- **databricks-langchain**: "OpenAIのような"インターフェースを介してDatabricksエンドポイントを呼び出すために必要です。
- **llama-index** および **llama-index-llms-databricks**: LlamaIndex内でDatabricksエンドポイントを使用するために必要です。
- **langfuse**: Langfuseプラットフォームにトレースデータを送信するために必要です。

```py
%pip install openai langfuse databricks-langchain llama-index llama-index-llms-databricks
%restart_python
```

## 2. 環境変数の設定

**Langfuse**のクレデンシャルと**Databricks**のクレデンシャルを環境変数として設定します。以下のダミーキーをそれぞれのアカウントから取得した実際のキーに置き換えてください。

 - `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY`: Langfuseプロジェクト設定から取得します。
 - `LANGFUSE_HOST`: `https://cloud.langfuse.com`（EUリージョン）または `https://us.cloud.langfuse.com`（USリージョン）。
 - `DATABRICKS_TOKEN`: Databricksのパーソナルアクセストークン。
 - `DATABRICKS_HOST`: DatabricksワークスペースのURL（例：`https://dbc-xxxxxxx.cloud.databricks.com`）。

Langfuseにサインアップして、Organization、Projectを作成することでAPIキーを取得することができます。Secret KeyとPublic Keyをコピーしておきます。

![Screenshot 2025-03-07 at 9.37.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/82db4ad8-c8df-4823-a621-6ba6074d67cb.png)


```py
import os

# 環境変数の例（実際のキー/トークンに置き換えてください）
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."  # あなたのパブリックキー
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."  # あなたのシークレットキー
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # または https://us.cloud.langfuse.com

os.environ["DATABRICKS_TOKEN"] = "dapi..."  # Databricks パーソナルアクセストークン
os.environ["DATABRICKS_HOST"] = "https://.....cloud.databricks.com/"
```

## アプローチ1: OpenAI SDKを介したDatabricksモデルの使用

Databricksエンドポイントは、OpenAI APIの代替として機能できます。これにより、`openai`ライブラリに依存する既存のコードと簡単に統合できます。内部では、`langfuse.openai.OpenAI`がリクエストを自動的にLangfuseにトレースします。

### 手順
1. `langfuse.openai`から`OpenAI`クライアントをインポートします。
2. クライアントを作成し、`api_key`にDatabricksトークンを、`base_url`にDatabricksワークスペースエンドポイントを設定します。
3. クライアントの`chat.completions.create()`メソッドを使用してプロンプトを送信します。
4. Langfuseダッシュボードでトレースを確認します。

**注:** Langfuseを使用したOpenAIのトレースに関する詳細な例については、[OpenAI統合ドキュメント](https://langfuse.com/docs/integrations/openai/python/get-started)を参照してください。

```py
# Langfuse OpenAI クライアント
from langfuse.openai import OpenAI

# 環境変数を取得
databricks_token = os.environ.get("DATABRICKS_TOKEN")
databricks_host = os.environ.get("DATABRICKS_HOST")

# Databricks を指す OpenAI ライクなクライアントを作成
client = OpenAI(
    api_key=databricks_token,  # Databricks パーソナルアクセストークン
    base_url=f"{databricks_host}/serving-endpoints",  # あなたの Databricks ワークスペース
)
```

```py
response = client.chat.completions.create(
    messages=[
        {"role": "system", "content": "あなたはAIアシスタントです。"},
        {"role": "user", "content": "Databricksとは何ですか？"}
    ],
    model="databricks-meta-llama-3-3-70b-instruct",  # Databricksモデルサービングエンドポイント名に基づいて調整
    max_tokens=256
)

# モデルからの応答を表示
display(response.choices[0].message.content)
```
```
'Databricks は、Apache Spark をベースにした、データエンジニアリング、データサイエンス、データ分析のためのクラウドベースのプラットフォームです。Databricks は、データの統合、処理、分析を容易にし、データ駆動型の意思決定を促進することを目的としています。\n\nDatabricks の主な機能には、以下のものがあります。\n\n1. Apache Spark: Databricks は Apache Spark を基盤としており、Spark のすべての機能を利用できます。\n2. クラウドネイティブ: Databricks はクラウドベースのプラットフォームであり、AWS、Azure、Google Cloud Platform などの主要なクラウドプロバイダーで利用できます。\n3. データレイク: Databricks では、データレイクを使用して構造化されたデータと非構造化されたデータを統合できます。\n4. ノートブック: Databricks では、Python、R、SQL などの言語を使用したノートブックを使用してコードを書き、データを分析できます。\n5. コラボレーション: Databricks'
```

Langfuseのダッシュボードを確認するとトレースを確認できます。プロンプト、応答、レイテンシ、トークン使用量などの詳細が表示されます。

![Screenshot 2025-03-07 at 9.41.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c7de2c92-40c0-435f-bc4d-d7599d806d7f.png)

## アプローチ2: LangChainの使用

DatabricksモデルはLangChainを介しても使用できます。[`ChatDatabricks`](https://python.langchain.com/docs/integrations/chat/databricks/)クラスは、Databricksのモデルサービングエンドポイントをラップします。

### 手順
1. 環境変数として`DATABRICKS_HOST`を設定します。
2. トレースデータを自動的に収集するLangfuseの`CallbackHandler`を初期化します。
3. エンドポイント名、温度、その他のパラメータを使用して`ChatDatabricks`を使用します。
4. メッセージでモデルを呼び出し、Langfuseのコールバックハンドラーを渡します。
5. Langfuseダッシュボードでトレースを確認します。

**注:** Langfuseを使用したLangChainのトレースに関する詳細な例については、[LangChain統合ドキュメント](https://langfuse.com/docs/integrations/langchain/tracing)を参照してください。

```py
from langfuse.callback import CallbackHandler

# Langfuseコールバックハンドラーを初期化
langfuse_handler = CallbackHandler(
    secret_key=os.environ.get("LANGFUSE_SECRET_KEY"),
    public_key=os.environ.get("LANGFUSE_PUBLIC_KEY"),
    host=os.environ.get("LANGFUSE_HOST")
)
```

```py
from databricks_langchain import ChatDatabricks

chat_model = ChatDatabricks(
    endpoint="databricks-meta-llama-3-3-70b-instruct",   # Databricksモデルサービングエンドポイント名
    temperature=0.1,
    max_tokens=256,
    # 他のパラメータをここに追加できます
)

# プロンプトをシステム/ユーザーメッセージのリストとして構築
messages = [
    ("system", "あなたはDatabricksに関する質問に答えることができるチャットボットです。"),
    ("user", "Databricks Model Servingとは何ですか？")
]

# LangChainの.invoke()メソッドを使用してモデルを呼び出す
chat_model.invoke(messages, config={"callbacks": [langfuse_handler]})
```

```
AIMessage(content='Databricks Model Servingは、機械学習モデルを簡単にデプロイして管理できるようにする、Databricksのクラウドベースのサービスです。データ サイエンティストとエンジニアは、Databricks Model Serving を使用して、トレーニング済みのモデルを簡単にデプロイして、リアルタイムの予測とバッチ予測のために REST エンドポイントとして公開できます。モデルは、Databricks のマネージド サービスとして自動的にホストされ、スケーリングされ、セキュリティが保護されます。Databricks Model Serving を使用すると、モデルを迅速にデプロイして、ビジネス上の意思決定を支援する予測を生成できます。', additional_kwargs={}, response_metadata={'prompt_tokens': 44, 'completion_tokens': 170, 'total_tokens': 214}, id='run-f62a4015-0dd5-44ae-a627-be83abd34baf-0')
```

コードを実行した後、Langfuseダッシュボードを開いて記録された会話を確認します。

![Screenshot 2025-03-07 at 9.43.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c7f6aeea-bd62-480b-b474-3fc3ea862950.png)

## アプローチ3: LlamaIndexの使用

データの取り込み、インデックス作成、またはRAGに[LlamaIndex](https://github.com/jerryjliu/llama_index)を使用する場合、デフォルトのLLMをDatabricksエンドポイントに置き換えることができます。

### 手順
1. `llama_index.llms.databricks`から`Databricks`をインポートします。
2. エンドポイント名とDatabricksの認証情報を使用して`Databricks` LLMを初期化します。
3. `langfuse.llama_index`から`LlamaIndexInstrumentor`を使用して自動トレースを有効にします。
4. チャットリクエストでLLMを呼び出します。
5. Langfuseダッシュボードでトレースを確認します。

**注:** Langfuseを使用したLlamaIndexのトレースに関する詳細な例については、[LlamaIndex統合ドキュメント](https://langfuse.com/docs/integrations/llama-index/get-started)を参照してください。

```py
from llama_index.llms.databricks import Databricks

# Databricks LLMインスタンスを作成
llm = Databricks(
    model="databricks-meta-llama-3-3-70b-instruct",  # Databricksモデルサービングエンドポイント名
    api_key=os.environ.get("DATABRICKS_TOKEN"),
    api_base=f"{os.environ.get('DATABRICKS_HOST')}/serving-endpoints/"
)
```

```py
from langfuse.llama_index import LlamaIndexInstrumentor
from llama_index.core.llms import ChatMessage

# LlamaIndexの操作をトレースするためにLlamaIndexInstrumentorを初期化
instrumentor = LlamaIndexInstrumentor(
    secret_key=os.environ.get("LANGFUSE_SECRET_KEY"),
    public_key=os.environ.get("LANGFUSE_PUBLIC_KEY"),
    host=os.environ.get("LANGFUSE_HOST")
)

# 自動トレースを開始
instrumentor.start()
```

```py
messages = [
    ChatMessage(role="system", content="あなたは役に立つアシスタントです。"),
    ChatMessage(role="user", content="Databricksとは何ですか？")
]

response = llm.chat(messages)
display(response)

# Flush any pending events to Langfuse
instrumentor.flush()
```
```
ChatResponse(message=ChatMessage(role=<MessageRole.ASSISTANT: 'assistant'>, additional_kwargs={}, blocks=[TextBlock(block_type='text', text='Databricksは、Apache Sparkをベースにしたデータエンジニアリング、データサイエンス、データ分析のためのクラウドベースのプラットフォームです。Databricksは、データレイクハウス、データウェアハウス、リアルタイムデータ処理、機械学習などの機能を提供し、ビッグデータの処理と分析を効率化することを目的としています。')]), raw=ChatCompletion(id='chatcmpl_c5974af8-6da9-4787-83d8-31a2b0219a07', choices=[Choice(finish_reason='stop', index=0, logprobs=None, message=ChatCompletionMessage(content='Databricksは、Apache Sparkをベースにしたデータエンジニアリング、データサイエンス、データ分析のためのクラウドベースのプラットフォームです。Databricksは、データレイクハウス、データウェアハウス、リアルタイムデータ処理、機械学習などの機能を提供し、ビッグデータの処理と分析を効率化することを目的としています。', refusal=None, role='assistant', audio=None, function_call=None, tool_calls=None))], created=1741304951, model='meta-llama-3.3-70b-instruct-121024', object='chat.completion', service_tier=None, system_fingerprint=None, usage=CompletionUsage(completion_tokens=93, prompt_tokens=35, total_tokens=128, completion_tokens_details=None, prompt_tokens_details=None)), delta=None, logprobs=None, additional_kwargs={'prompt_tokens': 35, 'completion_tokens': 93, 'total_tokens': 128})
```

LlamaIndexの呼び出しを確認するためにLangfuseにログインし、プロンプト、トークン使用量、完了データなどの詳細を確認できます。

![Screenshot 2025-03-07 at 9.45.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e540dfbf-4ad6-42de-9558-ccfed2a07fdb.png)

そして、Langfuseのダッシュボードでは様々なメトリクスを概観することができます。これは便利。

![Screenshot 2025-03-07 at 10.11.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cdf79d7f-2e9e-4e90-91ea-1754562e032b.png)
![Screenshot 2025-03-07 at 10.11.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/875a0387-b202-4eec-80ee-3344fe452235.png)


## トレースの強化 (オプション)

Langfuseは、より豊富なトレースデータのための追加機能をサポートしています:

- トレースに[メタデータ](https://langfuse.com/docs/tracing-features/metadata)、[タグ](https://langfuse.com/docs/tracing-features/tags)、[ログレベル](https://langfuse.com/docs/tracing-features/log-levels)、および[ユーザーID](https://langfuse.com/docs/tracing-features/users)を追加
- トレースを[セッション](https://langfuse.com/docs/tracing-features/sessions)でグループ化
- 追加のアプリケーションロジックをトレースするための[`@observe()`デコレータ](https://langfuse.com/docs/sdk/python/decorators)
- [Langfuseプロンプト管理](https://langfuse.com/docs/prompts/get-started)を使用し、プロンプトをトレースにリンク
- トレースに[スコア](https://langfuse.com/docs/scores/custom)を追加

詳細については、[Langfuseのドキュメント](https://langfuse.com/docs)を参照してください。

## 次のステップ
- Langfuse PlaygroundでDatabricksモデルを使用する方法やLLM-as-a-Judge評価については[こちら](https://langfuse.com/docs/integrations/databricks/use-with-playground-and-evals.mdx)をご覧ください。
- 高度なモデルサービング構成については[Databricksのドキュメント](https://docs.databricks.com/aws/ja/machine-learning/model-serving/manage-serving-endpoints)を参照してください。
- アプリケーション全体のフローを追跡するための[Langfuseトレース機能](https://langfuse.com/docs)について学びましょう。
- Langfuseの[プロンプト管理](https://langfuse.com/docs/prompts/get-started)を試すか、[LLM-as-a-Judge評価](https://langfuse.com/docs/scores/model-based-evals)を設定してみましょう。

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)