✨

OpenLLMetryへの入門

2025/07/06に公開

今回はLLMアプリケーションのオブザーバビリティを実現するためのOpenLLMetryについて紹介しようと思います。

OpenLLMetryとは？

OpenLLMetryとは、OpenTelemetryをベースに構築されており、LLMアプリケーションに関してオブザーバビリティを提供してくれるツールになります。OpenTelemetryの拡張であることから、既存のアプリケーション監視基盤の形式を踏襲しつつLLMに適したオブザーバビリティを導入することができます。

早速使ってみましょう

OpenLLMetryはtraceloopによって展開されており、公式サイトおよびGitHubレポジトリが公開されています。

https://traceloop.com/docs/introduction

今回はPythonのSDKを用いてテストをしてみます！なお、可視化にはtraceloopのプラットフォームを利用します。以下をお試しいただくときはご登録お願いします。

環境構築

uvを利用してtraceloop SDKをインストールします。

uv init openllmetry_test -p 3.12
cd openllmetry_test
uv add traceloop-sdk openai dotenv

タスクの実装

まずはシンプルに入力した日本語を英語に翻訳するコードを作ってみます。

translate.py

from dotenv import load_dotenv
from openai import OpenAI
from traceloop.sdk import Traceloop
from traceloop.sdk.decorators import task

Traceloop.init(
  disable_batch=True, 
  api_key="<traceloop API key>"
)

client = OpenAI()


@task(name="translate")
def translate_japanese(japanese_text: str) -> str:
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "assistant", "content": "You are a translate. Translate Japanese into English."},
            {"role": "user", "content": japanese_text},
        ]
    )
    return completion.choices[0].message.content


def main():
    japanese_text = "今日の天気はなんですか？"
    english_text = translate_japanese(japanese_text)
    print(f"{japanese_text} -> {english_text}")


if __name__ == "__main__":
    main()

まずtraceloopにログを送るための設定を以下のようにして実装します。APIキーはtraceloopに登録すると発行することができます。

from traceloop.sdk import Traceloop

Traceloop.init(
  disable_batch=True, 
  api_key="<traceloop API key>"
)

次に、翻訳をさせるためのタスクを作成します。タスクは1つのLLMを呼び出す最小単位になります。タスクを作成するには、関数にtaskデコレータを設定することで対応できます。タスクの中では一般的なOpenAIのチャット機能を利用して実装しています。アシスタントに対して日本語から英語に翻訳させるように指示し、ユーザプロンプトに指定された日本語文を設定するようにしました。

from traceloop.sdk.decorators import task


@task(name="translate")
def translate_japanese(japanese_text: str) -> str:
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "assistant", "content": "You are a translate. Translate Japanese into English."},
            {"role": "user", "content": japanese_text},
        ]
    )
    return completion.choices[0].message.content

それではこのコードを実行してみましょう。実行した結果、実行結果がトレース結果をtraceloopに連携しているメッセージと共に、翻訳された結果が表示されたことがわかります。

uv run translate.py

それではtraceloopのWeb UIにアクセスしてみます。アクセスするとまず以下のように記録されたスパンの一覧が表示されます。

次にスパンをクリックすると、スパンが含まれているトレースの詳細を取得することができます。どのようなプロンプトが利用されたか、トークン利用料や課金見積もり、実行時間など諸々が記録されています。

ワークフローの実装

ワークフローとはLLM処理に始まりその他エージェントやプラグインなどを利用する一連の処理を定義します。今回は先ほどの翻訳のコードにて、英語訳されたテキストを再び日本語に戻すというタスクを追加で定義したものをワークフローとして定義してみます。

translate_reverse.py

from dotenv import load_dotenv
from openai import OpenAI
from traceloop.sdk import Traceloop
from traceloop.sdk.decorators import workflow, task

Traceloop.init(
  disable_batch=True, 
  api_key="<traceloop API key>"
)

client = OpenAI()


@task(name="translate_ja2en")
def translate_japanese(japanese_text: str) -> str:
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "assistant", "content": "You are a translate. Translate Japanese into English."},
            {"role": "user", "content": japanese_text},
        ]
    )
    return completion.choices[0].message.content


@task(name="translate_en2ja")
def translate_english(english_text: str) -> str:
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "assistant", "content": "You are a translate. Translate English into Japanese."},
            {"role": "user", "content": english_text},
        ]
    )
    return completion.choices[0].message.content


@workflow(name="translate")
def translate(japanese_text: str) -> tuple[str, str]:
    english_text = translate_japanese(japanese_text)
    regenerated_japanese_text = translate_english(english_text)
    return english_text, regenerated_japanese_text


def main():
    japanese_text = "今日の天気はなんですか？"
    english_text, regenerated_japanese_text = translate(japanese_text)
    print(f"{japanese_text} -> {english_text} -> {regenerated_japanese_text}")


if __name__ == "__main__":
    main()

先ほど説明したように、二つのタスク（日本語から英語、英語から日本語）を定義し、以下のようにしてそれらを呼び出すワークフローを定義しました。

from traceloop.sdk.decorators import workflow

@workflow(name="translate")
def translate(japanese_text: str) -> tuple[str, str]:
    english_text = translate_japanese(japanese_text)
    regenerated_japanese_text = translate_english(english_text)
    return english_text, regenerated_japanese_text

それではこのコードも実行してみましょう。結果を見ると、最終的に生成された日本語は元の文章とは違うものの、ワークフローは実行できているようです。

uv run translate_reverse.py

それではtraceloopのWeb UIでも確認してみましょう。トレースを見ると以下のようになっていました。Durationの項目にて、スパンが実行されている範囲が青色で示されており、ワークフローの実行順や経過時間などが可視化されていることがわかります。また、今回は二つのタスクにてそれぞれLLMを呼び出していますが、Outputにて両方の出力を表示してくれているようです。

まとめ

今回はOpenLLMetryに入門してみました。LLMのオブザーバビリティはLLMアプリケーションを運用するには必須の要素であり、どのような呼び出しがされているか、不正アクセスはないかなど知る上で重要な項目です。ぜひ皆さんも利用してみてください。