【LangChain中級】CallbackHandlerでトークン消費量と実行ログを完全掌握する

はじめに

LLMアプリを本番運用する際、最も気になるのは「一体今、AIは何をしているのか？」「いくら課金されたのか？」という点です。

LangChainには、処理の開始・終了・エラー発生などのイベントごとに追加処理をフックできる「Callbacks」という仕組みがあります。これを使えば、ログ出力や課金計算をスマートに実装できます。

実行中の様子が見えない恐怖

Monitoring

Chainが複雑になると、どこで時間がかかっているのか、本当にプロンプトが正しく送られているのかがブラックボックス化します。

Printデバッグを仕込むのも手ですが、コードが汚くなります。CallbackHandlerを使えば、ビジネスロジックとログ処理を綺麗に分離できます。

StdOutCallbackHandler（標準出力）

開発中に一番お世話になるのがこれです。コンソールに詳細な実行ログを表示してくれます。

from langchain_core.callbacks import StdOutCallbackHandler

handler = StdOutCallbackHandler()
llm.invoke("こんにちは", config={"callbacks": [handler]})

これを渡すだけで、「プロンプトとして何が送られ、いつ生成が始まり、最終的に何が返ってきたか」が色付きで表示されます。

get_openai_callback（コスト計算）

OpenAIのAPI利用料を計算したい場合は、コンテキストマネージャーを使います。

from langchain_community.callbacks import get_openai_callback

with get_openai_callback() as cb:
    llm.invoke("長い文章を作って")
    print(f"消費トークン: {cb.total_tokens}")
    print(f"コスト($): {cb.total_cost}")

複数のChainをまとめて実行しても、合計コストを一発で算出してくれます。

カスタムCallbackの実装

「エラーが起きたらSlackに通知する」といった独自の処理も簡単に作れます。

from langchain_core.callbacks import BaseCallbackHandler

class SlackAlertHandler(BaseCallbackHandler):
    def on_llm_error(self, error, **kwargs):
        # ここでSlack APIを叩く
        send_to_slack(f"エラー発生: {error}")

まとめ

Callbacksは地味ですが、運用フェーズでは生命線となる機能です。早いうちから使い方に慣れておきましょう。

次回は、複数のプロンプトやチェーンを条件に応じて切り替える「RouterChain」について解説します。

なんでも技術ブログ

このブログを検索