Langchain統合のためのクックブック: Langfuse編

こんにちは！今日は、LangchainとLangfuseを組み合わせて使う方法について、初心者にもわかりやすく解説していきたいと思います。
Langchainは自然言語処理に特化したPythonライブラリで、LangfuseはLangchainの機能を拡張するプラグインのようなものです。
二つを組み合わせることで、より強力な自然言語処理システムを構築することができるのです。

セットアップ
順次実行チェーンの例
会話チェーンの例
@observe()デコレータでLangfuseのトレースを取得
まとめ
リポジトリ
参考サイト
1. 関連

セットアップ

まずは必要なライブラリをインストールしましょう。以下のコマンドを実行してください。

%pip install langfuse langchain langchain_openai --upgrade

次に、LangfuseのAPIキーを環境変数に設定します。APIキーは、Langfuseのプロジェクト設定ページから取得できます。

import os

# Langfuseのプロジェクト設定からAPIキーを取得
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-***"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-***"
# データリージョンに応じてホストを設定
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # EUの場合
# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # USの場合

# OpenAIのAPIキー
os.environ["OPENAI_API_KEY"] = "***"

これでセットアップは完了です。接続をテストしてみましょう。

from langfuse.callback import CallbackHandler

langfuse_handler = CallbackHandler()

# SDKとサーバーの接続をテスト
langfuse_handler.auth_check()

テストがうまくいけば、Langchain+Langfuseを使う準備は整ったことになります。
では、具体的な使用例を見ていきましょう。

順次実行チェーンの例

Langfuseでは、Langchainの順次実行チェーンをトレースすることができます。以下はその例です。

from langchain_openai import OpenAI
from langchain.chains import LLMChain, SimpleSequentialChain  
from langchain.prompts import PromptTemplate

llm = OpenAI()
template = """あなたは脚本家です。与えられた戯曲のタイトルに基づいて、あらすじを書くのがあなたの仕事です。
    タイトル: {title}
    脚本家: 上記の戯曲のあらすじはこちらです:"""
prompt_template = PromptTemplate(input_variables=["title"], template=template)
synopsis_chain = LLMChain(llm=llm, prompt=prompt_template)  

template = """あなたはニューヨークタイムズの演劇評論家です。与えられた戯曲のあらすじに基づいて、批評を書くのがあなたの仕事です。
    戯曲のあらすじ:
    {synopsis} 
    ニューヨークタイムズの演劇評論家による上記の戯曲の批評:"""
prompt_template = PromptTemplate(input_variables=["synopsis"], template=template)
review_chain = LLMChain(llm=llm, prompt=prompt_template)

overall_chain = SimpleSequentialChain(
    chains=[synopsis_chain, review_chain],  
)

# 呼び出し
review = overall_chain.invoke("Tragedy at sunset on the beach", {"callbacks":[langfuse_handler]}) 

# 実行
review = overall_chain.run("Tragedy at sunset on the beach", callbacks=[langfuse_handler])

注目すべきは、invokeメソッドとrunメソッドの両方でコールバックハンドラーを渡している点です。
これにより、チェーンの実行がLangfuseにトレースされるようになります。

会話チェーンの例

次は会話チェーンの例です。Langfuseのセッション機能を使って、一連の会話をトラッキングしてみましょう。
各呼び出しが1つのトレースとなり、セッションビューで会話全体を見ることができます。

from langchain.chains import ConversationChain  
from langchain.memory import ConversationBufferMemory
from langchain_openai import OpenAI

llm = OpenAI(temperature=0)

conversation = ConversationChain(
    llm=llm, memory=ConversationBufferMemory()  
)

# セッション付きのコールバックハンドラーを作成
langfuse_handler = CallbackHandler(session_id="conversation_chain")

conversation.predict(input="Hi there!", callbacks=[langfuse_handler])

conversation.predict(input="How to build great developer tools?", callbacks=[langfuse_handler]) 

conversation.predict(input="Summarize your last response", callbacks=[langfuse_handler])

会話の流れがセッションとしてトレースされ、Langfuseで一覧できるようになります。
自然な会話の流れを分析するのに役立つでしょう。

@observe()デコレータでLangfuseのトレースを取得

Langfuseの@observe()デコレータを使うことで、LangChain以外のコードもトレースできます。
また、複数のLangChainの呼び出しを1つのトレースにまとめることもできます。

以下は、@observe()デコレータを使った例です。

from langfuse.decorators import langfuse_context, observe

# Langfuseのデコレータでトレースを作成し、そのトレースのLangChainコールバックハンドラーを取得
@observe() # 関数をLangfuseのトレースとして自動的にログ
def main():
    # トレース属性を更新（名前、セッションID、ユーザーIDなど）
    langfuse_context.update_current_trace(
        name="custom-trace",
        session_id="user-1234", 
        user_id="session-1234",
    )
    # 現在のトレースのLangChainハンドラーを取得
    langfuse_context.get_current_langchain_handler()

    # ハンドラーを使ってLangChainの実行をトレース...

main()

@observe()デコレータを付けた関数は、自動的にLangfuseのトレースとしてログされます。
langfuse_context.update_current_trace()で、トレースの属性（名前、セッションIDなど）を更新できます。
そして、langfuse_context.get_current_langchain_handler()で、現在のトレースのLangChainハンドラーを取得できます。

これらを駆使することで、LangChainの実行を柔軟にトレースできるようになります。