OpenLLMetry + Traceloop試してみた

早速表題のOpenLLMetryについて。

OpenLLMetryは、Traceloop社が公開しているOpenTelemetryを拡張したOSSです。このツールは、LLMアプリケーションのオブザーバビリティ（可観測性）を向上させることを目的としています。

ここ数年でLLMがアプリケーションに組み込まれることが増えていますが、その重要な課題の1つが継続的な監視だと思います。使用されているモデルやAPIの実行時間、プロンプトの内容、トークン消費量、生成内容の一貫性など、通常のシステム運用と同様に効率的な監視が必要です。

OpenLLMetryによる計装をおこなうことで、モデル名、バージョン、プロンプト、トークン、Temperature等の重要なメトリクスを一貫して収集できます。これにより、LLMアプリケーションの動作を詳細に把握し、AIエージェントのツール実行、ワークフロー、LLMモデルの呼び出し、RAGの埋め込みや検索などのプロセスをスパンとして追跡することで、可観測性を強化できます。

OpenLLMetryは、OpenAI、Azure OpenAI、Anthropic、Google Gemini、Amazon Bedrockといった主要なLLMモデルと、LangChain、LlamaIndexなどのAIフレームワークをサポートしています。さらに、Traceloop以外にも、Datadog、Splunk、New Relicなど、多数のモニタリングツールとの統合が可能です。

Traceloop

Traceloopは、LLM（大規模言語モデル）アプリケーションの品質を監視し、実行トレースの記録と可視化を行うツールです。プロンプト管理やLLMの入出力評価などの機能を備えています。

といったところで、なにはともあれ、これらを使ってみようと思います。

試してみる

今回、クイックスタート通り、トレースの送信先はTraceloop、言語はPythonでやってみます。早速、Traceloopにサインインして、APIキーを取得しておきます。

Generate API Key からAPIキーを取得し、コピーしておきます。

pip install traceloop-sdkTraceloop でSDKをインストールし、トレーサーを初期化します。ローカルで実行している場合は、ライブで確認できるようにバッチ送信を無効にしておきます。

python

import os
from traceloop.sdk import Traceloop
from traceloop.sdk.decorators import workflow
from dotenv import load_dotenv

load_dotenv()

Traceloop.init(
  disable_batch=True,
  api_key=os.getenv("TRACELOOP_API_KEY")
)

複数の処理を連鎖させて1つのワークフローとして実行する場合、各処理に注釈を付けることで可視性が向上します。Pythonでは、デコレーターが用意されており、各処理にデコレーターを記述するだけで対応できます。※詳細はこちらのドキュメントをご参照ください。
まずは以下のような短いコードでどのように記録されるか実験です。

python

from openai import OpenAI

# デコレーター (ここではworkflow全体にgpt4oという名前の注釈を付けている)
@workflow(name="gpt4o")
def create_completion(message: str):
  client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
  completion = client.chat.completions.create(
      model="gpt-4o",
      messages=[
        {
          "role": "user",
          "content": message
        }
      ],
  )
  
  print(completion.choices[0].message.content)
  return completion.choices[0].message.content


print("Please input message ↓")
message = input()
create_comletion(message)

bash

$ python llm.py
# Traceloopへの送信開始
Traceloop exporting traces to https://api.traceloop.com authenticating with bearer token

# 質問
Please input message ↓
お元気ですか?

# 回答
はい、元気です！ありがとうございます。あなたはいかがですか？何かお手伝いできることがあれば教えてください。

Traceloop Cloudでトレースを確認すると、トレース全体や個々のLLM呼び出しにおける実行時間、モデルの種類、プロンプト内容、トークン数、推定コストなどを確認できます。

PromptやToken、推定コストなど

もう一例として簡単なRAG (Retrieval-Augmented Generation) を実行した場合のトレースを見てみます。すると、以下のようにトレース全体から個々のスパンを段階的に確認できます。

※Pineconeを使ったRAGのサンプルコード

python

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

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# tracer init
Traceloop.init(app_name="pinecone_app",
            disable_batch=True,
            api_key=os.getenv("TRACELOOP_API_KEY"))

pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index_name = "gen-qa-openai-test"
index = pc.Index(index_name)

open_ai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# decorator: @task(name="my_task")
@task("retrieve")
def retrieve(query):
    context_limit = 3750
    res = open_ai_client.embeddings.create(
        input=[query], model="text-embedding-ada-002"
    )

    # retrieve from Pinecone
    xq = res.data[0].embedding
    xq = res.data[0].embedding

    # get relevant contexts
    res = index.query(vector=xq, top_k=3, include_metadata=True, include_values=True)
    contexts = [x["metadata"]["text"] for x in res.matches]

    # build our prompt with the retrieved contexts included
    prompt_start = "以下の文脈に基づいて質問に答えなさい。\n\n" + "Context:\n"
    prompt_end = f"\n\nQuestion: {query}\nAnswer:"
    # append contexts until hitting limit
    for i in range(1, len(contexts)):
        if len("\n\n---\n\n".join(contexts[:i])) >= context_limit:
            prompt = prompt_start + "\n\n---\n\n".join(contexts[: i - 1]) + prompt_end
            break
        elif i == len(contexts) - 1:
            prompt = prompt_start + "\n\n---\n\n".join(contexts) + prompt_end
    return prompt

# decorator: @task(name="my_task")
@task("chat")
def chat(prompt):
    res = open_ai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=400,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0,
        stop=None,
    )
    return res.choices[0].message.content

# decorator: @workflow(name="my_workflow")
@workflow(name="query_with_retrieve")
def run_query(query: str):
    query_with_contexts = retrieve(query)
    print(query_with_contexts)
    print(chat(query_with_contexts))


query = (
    "関連する文のペアしか持っていない場合、機械学習の手法としてどの方法を使用すべきですか？"
)
run_query(query)

また、これはTraceloopの機能ですが、LLMの入出力されるプロンプトを監視する機能があります。

LLMアプリの運用では、プロンプトで制限している事項やポリシーに沿った運用が継続的にできているか監視する必要があるかと思いますが、Traceloopでは、LLMへの入出力に個人情報や機密情報が含まれていないかの確認や、出力結果に有害なコンテンツが含まれていないかなど、様々な指標での評価が可能です。

例として、以下のような個人情報を含むプロンプトをLLMに与えたとします。

# 入力
Please input message ↓
私は島崎健一です。昭和15年、1940年2月17日生まれ。辰年77歳。
買い物は家から駅に向かうとピーコックがあるのですが外苑通りを渡った所にあるライフへよく行きます。
ライフの方が安いので。私は何歳でしたっけ。

# LLM回答
こんにちは、島崎健一さん。お話しできて嬉しいです。2023年の時点で、あなたは83歳になりますね。
何かお手伝いできることがありましたら、どうぞ教えてください。

PII（個人情報）評価を設定することで、個人情報を含む出力を検知できます。この場合、実行結果がFailとなりました。また、これらの評価に閾値を設定してSlackにアラートを送信することも可能です。

このあたりの評価を簡単に設定し、検知できるのは良いなあ。という感じです。

さいごに

簡単ではありますが、OpenLLMetryとTraceloopを試してみました。

LLMの進化と同時に運用面もしっかり整えていくことが重要ですね。

なお、本記事は G's ACADEMY Advent Calendar 2024 23日目になります。

私は「あつまれエンジニアの森」というコミュニティをほそぼそと運営しているのですが、会場としてG’s Academy Tokyoを毎回お借りしています。(いつも有難うございます🙏)

2025年も「あつまれエンジニアの森」を続けていきますので、もし興味のある会があったら、ぜひ参加してみてください。(趣味会とか特定技術カテゴリのLT会とか...アンケート取りながら毎回テーマ変えてゆる～くやってます)

次回はこちら👇️

という事で、本記事は以上になります。
拙い内容で恐縮ですが最後までご覧頂き有難うございました🙂