個人的ベスト LLMのObservabilityツール

はじめに

LLMのObservabilityツールでよく聞くものとしてLangSmith、Langfuse、MLflowなどがあります。

しかしこれらはSelf-hostが環境によっては面倒です。（私はメモリかつかつなのでDockerはできるだけ使いたくない状況です。。。）

また、業務で使用する際などは外部に保存はしたくありませんし、ほんとにちょっとだけ確認したいことがある際にいちいちDocker立てるのも面倒でした。

何かないかなぁと思い探しているとAlizeのphoenixというObservabilityツールがあり、非常に使いやすく私の状況にピッタリだったので備忘録としてまとめました。

また、本記事はご紹介としてさらっとだけ記述します。
より詳しい内容は別の方がまとめていらっしゃったのでそちらをご覧ください！（人任せ...)

試してみる

今回使用したのはphoenixのNotebook-selfHostです。

Notebook内でコードを実行すると、localhostにダッシュボードが作成されます。

Runtimeを削除すると当然記録は消えますが簡単にLLMの動きをトレースしたい際などとても便利です。
（私は試したことありませんが、記録を残しておきたい場合はDockerコンテナとして立ち上げ、別途DBなどに保存すればいいそうです。）

インストール

pip で簡単にインストールできます。

pip install -qU arize-phoenix

ダッシュボードの立ち上げ

下記コードでダッシュボードを立ち上げます。

ほんの数行で立ち上げることができるので非常に簡単です。

import phoenix as px
px.launch_app()

デフォルトではlocalhost:6006に立ち上がります。

トレースの設定をする

ダッシュボードが立ち上がったのでLLMの出力をトレースする準備をしていきます。

トレースの方法はいくつかあるのですが、まずは最も簡単なregisterを使用していきます。

registerを使用するとphoenixが対応している一部のサードパーティーライブラリ（OpenAI、langchain、llamaindexなど）を自動的にトレースしてくれます。

後ほど実際にトレースをしますが、registerを使用すると非常に簡単にトレースをしてくれます。

下記コードでregisterを登録します。

from phoenix.otel import register

tracer_provider = register(
  project_name="test", # Default is 'default'
  auto_instrument=True # Auto-instrument your app based on installed OI dependencies
)

langchainをトレースしてみる

registerを作成したので実際にトレースしてみます。

この部分も非常に簡単です。というかほぼ何もしません。
まずはopeninference-instrumentation-langchainをインストールして、先ほど作成したregisterと紐づけます。

!pip install -qU openinference-instrumentation-langchain


from openinference.instrumentation.langchain import LangChainInstrumentor

LangChainInstrumentor().instrument(tracer_provider=tracer_provider)

あとは、同じRuntime上でいつもどおりlangchainにinvokeするだけです！

試しに下記コードを実行してみます。（APIキーはご自身でご用意ください。）

!pip install -qU langchain-openai


from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
llm.invoke("こんにちは")

すると先ほどのダッシュボード上にtestというプロジェクトが追加され、更にlangchainの出力がトレースされているのが確認できるはずです！

中身はこんな感じ

Latencyや使用したToken料などをトレースできます！
今回はシンプルにChatModelにinvoke()しただけなのであっさりしていますが、langgraphなどで複雑なワークフローを組んでいるときなどは各ノードで何が行われたのかなどを細かく確認することができます。

そして何より簡単です！

もし既存のアプリがlangchainなどで書かれてあり、一時的にテストとしてトレースしたいときなどはコードの上部に数行付け足すだけでトレースが出来てしまいます！

より柔軟なトレース（補足）

openinferenceのTracerProviderを使用することにより、トレースをより細かく（スパン単位で）制御可能です。

例えば先ほど使用したregisterが対応していない独自の関数などを使用した際などに有効です。

こちらの手法を私はあまり使ったことが無いため触りの部分だけまとめます。

まずは下記コード例でtracerを作成します。

from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor

from openinference.instrumentation import TracerProvider
from openinference.semconv.resource import ResourceAttributes

endpoint = "http://localhost:6006/v1/traces"
resource = Resource(attributes={ResourceAttributes.PROJECT_NAME: "tracer-test"})
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint)))
tracer = tracer_provider.get_tracer(__name__)

このtracerを使用してトレースを行っていきますが、主にデコレーターとwith句の2つの使い方があります。

デコレーターの場合の基本的使い方はこちら

@tracer.chain
def my_func(input: str) -> str:
    return "output"

# トレース結果見たいのでとりあえず実行
my_func("あいうえお")

with句の場合の基本的使い方はこちら

from opentelemetry.trace import Status, StatusCode

with tracer.start_as_current_span(
    "my-span-name",
    openinference_span_kind="chain",
) as span:
    span.set_input("input")
    span.set_output("output")
    span.set_status(Status(StatusCode.OK))

結果はこんな感じです。

ちなみにスパンとはこのレコード1行ずつのことを指していると私は理解しています。

またスパンの種類にもいくつかあり、今回はchainのみを使用していますがほかにもllm、agent、toolなどがあります。

まとめ

Arize phoenix について簡単にまとめました！

phoenixは非常に簡単に使い始めることのできるObservabilityツールとして今後も使っていこうかと思います。

また、今回はトレースのみ行いましたが他にもRAGなどの評価を行うことができる様なので勉強したいと思います!