Python openaiパッケージのlangfuseによるトレース手順

概要

検証時にopenaiパッケージとトークン取得のためにlangfuseを使用しました。
トレースの取得方法や、独自のメタデータの付与のし方が分からず少し悩んだため、要点のみを記録します。
記事執筆に使用したlangfuseはバージョン3.2.1です。

トレースに必要な要素

LLMへのリクエスト状況をトレースするためには以下の点を最低限押さえておけば良さそうです。

1. langfuseの起動
   プログラムから投げられたトレースを受け取るlangfuseサーバーを実行しておく必要があります。
   dockerを使えば簡単に立ち上げられます。

   1. langfuseをCloneし、コンテナを起動します。
      初回起動時はVolumeの準備があるので少し時間がかかるようです。

      git clone https://github.com/langfuse/langfuse
      cd langfuse
      docker compose up -d
      

   2. ブラウザでhttp://localhost:3000 にアクセス。
   3. 初回アクセス時はユーザーを作成する必要があるので「Sign up」をクリックします。
      （ローカル実行しているサーバーですので、ユーザーはコンテナが管理するDBに保存されるものです。）
      
   4. 最初のユーザーの情報を入力し、「Sign up」をクリックします。
      
   5. 組織を作成します。
      自分のみが使用するローカル実行の場合は、適当で良いと思います。
      
      
   6. テスト用に自分だけが使用する場合、メンバー追加は不要ですので、「Next」をクリックします。
      
   7. プロジェクトを追加します。
      
   8. APIキーを発行します。
      
   9. APIキーが表示されます。（ローカルで実行しているサーバーのキーなのですが一応伏せました）
      このキーとHostの情報をこの先の手順で使います。
      

2. 環境変数の設定
   .envなどに環境変数として先ほどのキーを設定します。

   LANGFUSE_HOST=http://localhost:3000
   LANGFUSE_SECRET_KEY=sk-から始まるSecret Keyの値
   LANGFUSE_PUBLIC_KEY=pk-から始まるPublic Keyの値
   

3. importの変更

   • 元々.envを使用していなければ、pip install python-dotenvなどでdotenvを追加し、スクリプトの先頭で.envの内容を環境変数に読み込みます。

     from dotenv import load_dotenv
     load_dotenv()
     

   • pip install langfuse等でlangfuseパッケージも追加し、openaiのimportをしている所を変更します。
     こちらの環境はAzureOpenAIを使用しています。
     • 変更前

       from openai import AzureOpenAI
       

     • 変更後

       from langfuse.openai import AzureOpenAI
       

これで、LLM呼び出し時のトレースがlocalhost:3000で起動しているlangfuseに送信され、確認できるようになります。

独自メタデータの付与

• 設定の上書き処理
  複数の検証を行う時に、どの処理で呼び出したトレースであるかを識別したい時があります。
  モデルにプロンプトを投げるところで情報を付与できます。
  以下の例では、トレース名に「History Summary Wave-1」、タグに「Wave-1」と付けています。

  # Azure OpenAIクライアント初期化
  client = AzureOpenAI(
      azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
      api_version="2024-02-15-preview",
      api_key=os.getenv("AZURE_OPENAI_API_KEY")
  )
  deployment = os.getenv("AZURE_OPENAI_DEPLOYMENT")
  response = client.chat.completions.create(
      model=deployment,
      messages=[{"role": "user", "content": prompt}],
      name = "History Summary Wave-1",
      metadata={"langfuse_tags": ["Wave-1"]}
  )
  return response.choices[0].message.content
  

• langfuse上でTraceに識別子が付くことで比較しやすくなりました。