Datadog LLM Observabilityやってみた（注意点あり）

AWSの監視といえばDatadog。いつもAWS Summitやre:Inventで目立ってますよね。
そんなDatadogが、生成AI向けの機能を追加しました。その名も 「Datadog LLM Observability」

プレスリリースも出てます。

公式ドキュメント

LLM Observabilityのドキュメントはこちらです。

日本語翻訳はまだのようなので、英語にして確認しましょう。

とりあえずクイックスタートでもやってみるかと思ったのですが、黄色いバナー が気になりますよね？

By using LLM Observability, you acknowledge that Datadog is authorized to share your Company's data with OpenAI Global, LLC for the purpose of providing and improving LLM Observability.

翻訳 by Claude 3.5 Sonnet

LLM Observabilityを使用することにより、あなたは、DatadogがLLM Observabilityの提供と改善を目的として、あなたの会社のデータをOpenAI Global, LLCと共有する権限を持っていることを認めたことになります。

あなたの会社のデータをOpenAI Global, LLCと共有する
あなたの会社のデータをOpenAI Global, LLCと共有する
あなたの会社のデータをOpenAI Global, LLCと共有する

え？

そんなこと気にしない人だけ続きをどうぞ。

対応状況

2024/6/28時点で、Auto Instrumentationに対応しているのは以下のフレームワークです。
ドキュメント：https://docs.datadoghq.com/llm_observability/auto_instrumentation/

• OpenAI
• Langchain
• Amazon Bedrock
• Anthropic

Auto Instrumentationとは、トレース取得対象のプログラムに手を入れずに（場合によってはちょっとだけ手をいれる）、トレースを取得して収集することを指します。

Bedrock（Boto3）でやってみた

まず、ここを参考に、DatadogのAPIキーを発行してください。

次に必要なライブラリーをインストールします。

pip install boto3 ddtrace

ddtraceがDatadogにデータを送信するライブラリーです。

Pythonのスクリプトを普通に書きます。

app.py

import boto3
import json

client = boto3.client("bedrock-runtime")

response = client.invoke_model(
    body=json.dumps(
        {
            "inputText": "Hello, how are you?",
            "textGenerationConfig": {
                "maxTokenCount": 100,
            },
        }
    ),
    modelId="amazon.titan-tg1-large",
    accept="application/json",
    contentType="application/json",
)

print(json.loads(response["body"].read().decode("utf-8")))

正しく動作するか確認しましょう。

python app.py

{'inputTextTokenCount': 6, 'results': [{'tokenCount': 12, 'outputText': '\nEverything is good! How can I assist you?', 'completionReason': 'FINISH'}]}

動作しました。
では、Datadogにトレースを送ってみましょう。

実行コマンドが変わります。

DD_LLMOBS_ENABLED=1 \
DD_LLMOBS_ML_APP=onboarding-quickstart \
DD_API_KEY={DatadogのAPIキー} \
DD_SITE=datadoghq.com \
DD_LLMOBS_AGENTLESS_ENABLED=1 \
ddtrace-run python app.py 

環境変数が色々あり見づらいですが、実行コマンドがpythonからddtrace-runに変わっています。これがDatadogにトレース情報を送信する本体です。（パラメーターとしてpython app.pyを渡しています。）

以前検証したLangfuseでは、プログラム中でトーレス情報送信の処理を書きましたが、Datadogの場合は、 プログラムの外側 でその処理が行われます。（こういう仕組み思いつく人ってすごですよね。）

実行すると、コンソールには先程と同様、printの結果が出力されます。トレースを取っているかどうかは判別できません。

{'inputTextTokenCount': 6, 'results': [{'tokenCount': 12, 'outputText': '\nEverything is good! How can I assist you?', 'completionReason': 'FINISH'}]}

Datadogの画面を確認します。

左のメニューの「LLM Observability」を選択すると、先程の結果が確認できます。1件しかありませんが、リスト表示です。

リストから1件選択すると、トレース情報が確認できます。

トークン数や呼び出したモデルのIDなども収集されていることがわかります。

BedrockのConverse APIには未対応のようで、トレースが記録されませんでした。

LangChainでやってみた

Boto3の例はLLMを一回呼び出しただけなので、あまり面白みはありませんでした。
LangChainでこちらを参考に簡単なRAGを作って検証しました。

LLM Observabilityは、LangChainのv0.2系には対応していないので、v0.1系で検証します。

pip install langchain==0.1.20 langchain-community==0.0.38 langchainhub==0.1.20 \
 langchain-aws==0.1.6 langchain-chroma==0.1.2 beautifulsoup4==4.12.3

Pythonスクリプト（参考サイトのものをBedrockに置き換えた程度です）

langchain-rag.py

import bs4
from langchain import hub
from langchain_aws import BedrockEmbeddings, ChatBedrock
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_text_splitters import RecursiveCharacterTextSplitter

llm = ChatBedrock(model_id="anthropic.claude-3-haiku-20240307-v1:0")

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0", model_kwargs={"dimensions": 256}
)

# Load, chunk and index the contents of the blog.
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings)

# Retrieve and generate using the relevant snippets of the blog.
retriever = vectorstore.as_retriever()
prompt = hub.pull("rlm/rag-prompt")


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

response = rag_chain.invoke("What is Task Decomposition?")

print(response)

先ほどと同じように、ddtrace-runを実行するのですが、以下のTypeErrorが発生します。どこがどう問題なのかは把握してませんが、2の環境変数を追加すると、正常に実行できました。

TypeError: Descriptors cannot be created directly.
If this call came from a _pb2.py file, your generated code is out of date and must be regenerated with protoc >= 3.19.0.
If you cannot immediately regenerate your protos, some other possible workarounds are:

1. Downgrade the protobuf package to 3.20.x or lower.
2. Set PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python (but this will use pure-Python parsing and will be much slower).

実行コマンド

DD_LLMOBS_ENABLED=1 \
DD_LLMOBS_ML_APP=onboarding-quickstart \
DD_API_KEY={DatadogのAPIキー} \
DD_SITE=datadoghq.com \
DD_LLMOBS_AGENTLESS_ENABLED=1 \
PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python \
ddtrace-run python langchain-rag.py

Task decomposition is the process of breaking down a complex task into smaller, more manageable sub-tasks. As described in the context, task decomposition can be done through language models using prompting techniques like "chain of thought" and "tree of thoughts", which help the model decompose a task into multiple steps and explore multiple reasoning possibilities. Task decomposition can also be done using task-specific instructions or with human inputs.

Datadogの画面を確認します。

全体

Retrieve部分
画面右下のOutputに抽出したドキュメントが確認できます。

Bedrock呼び出し部分
Retrieve部分した内容がInputに含まれたプロンプトが生成AI側にわたっていることが確認できます。

Datadogをすでに導入している方にはいいんじゃないでしょうか！