背景

日本でオープンソースのLLM監視と分析ツールだったらLangfuseがほぼ一強の状況だけど、海外にはいくつかのツールがあります。その中にあまり日本で知られていないOpikというツールを使ってみたいと思います。

概要

Comet発のOpik：

近代のオープンソースツールと同じく、SaaSとセルフホスト型もあって、セルフホストの方法がやや複雑だけど、LLMの中身を全部監視するので、セキュリティ要件的にSaaS使えないケースが多いでしょう。

一般的なユースケースが懸念なくカバーされています。

ログトレース
コスト追跡
データセット管理
プロンプト管理
エージェント評価
ガードレール
MCP

また、対応するLLMも非常に多くて、この部分一番安心感があります。

（サポートLLMの一部）

使ってみる

ログトレース

早速使ってみて、まず簡単なプロンプトでテストします。

import os
import dotenv

from opik.integrations.openai import track_openai
from openai import OpenAI

dotenv.load_dotenv()

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
openai_client = track_openai(openai_client)

response = openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What is the capital of France?"}]
)

print(response.choices[0].message.content)

ツール上で結果を見てみます。

UI的にわかりやすくて、知りたい情報も揃っています。

LangGraph連携

上記あくまで簡単な例で、実際のユースケースは多くの場合がLangGraphなどエージェントを構築して使うので、こちらでLangGraph連携の場合でどうなるかを試します。

import opik
import dotenv

from langgraph.graph import StateGraph, END
from typing import TypedDict, Optional

dotenv.load_dotenv()
opik.configure(use_local=False)

# Define the graph state
class GraphState(TypedDict):
    question: Optional[str] = None
    classification: Optional[str] = None
    response: Optional[str] = None

# Create the node functions
def classify(question: str) -> str:
    return "greeting" if question.startswith("Hello") else "search"

def classify_input_node(state):
    question = state.get("question", "").strip()
    classification = classify(question)
    return {"classification": classification}

def handle_greeting_node(state):
    return {"response": "Hello! How can I help you today?"}

def handle_search_node(state):
    question = state.get("question", "").strip()
    search_result = f"Search result for '{question}'"
    return {"response": search_result}

# Create the workflow
workflow = StateGraph(GraphState)
workflow.add_node("classify_input", classify_input_node)
workflow.add_node("handle_greeting", handle_greeting_node)
workflow.add_node("handle_search", handle_search_node)

# Add conditional routing
def decide_next_node(state):
    return (
        "handle_greeting"
        if state.get("classification") == "greeting"
        else "handle_search"
    )

workflow.add_conditional_edges(
    "classify_input",
    decide_next_node,
    {"handle_greeting": "handle_greeting", "handle_search": "handle_search"},
)

workflow.set_entry_point("classify_input")
workflow.add_edge("handle_greeting", END)
workflow.add_edge("handle_search", END)

app = workflow.compile()

from opik.integrations.langchain import OpikTracer

# Create the OpikTracer with graph visualization
tracer = OpikTracer(graph=app.get_graph(xray=True))

# Execute the workflow
inputs = {"question": "Hello, how are you?"}
result = app.invoke(inputs, config={"callbacks": [tracer]})
print(result)

ソースコードを見ると長くてわかりにくいけど、実行結果の画面を見るとわかりやすくて、デバッグの時にこれがあると非常に協力になります。

プロンプト管理

せっかくなので、もう1個重要な機能を試していきます。

プロンプトの追加はソースコード上でできます。その結果は画面上で確認できます。

import opik
import dotenv

dotenv.load_dotenv()
opik.configure(use_local=False)

PROMPT_TEXT = "Rewrite the following text: {{text}}"

prompt = opik.Prompt(
    name="prompt-summary",
    prompt=PROMPT_TEXT,
    metadata={"environment": "production"}
)

print(prompt.prompt)

print(prompt.format(text="Hello, world!"))

ツールの画面を見ると、バーションで複数のプロンプトを管理することができます。またメタデータの設定でバーション以外の属性で管理することもできます。

もちろん追加したプロンプトの管理と確認はソースコード上でもできます。

import opik

client = opik.Opik()

history = client.get_prompt_history(name="prompt-summary")
for version in history:
    print(version.commit, version.prompt)

で、その出力結果は：

803efbf8 Rewrite the following text: {{text}}
fefb23c4 Write a summary of the following text: {{text}}

サマリー

使い方は簡単で、ラッパーだけで完結、高度な細かい制御の場合はローレベルSDKもある
機能アップデートが頻繁、ちゃんと本番利用できえうver 1.Xになっている
日本でのコミュニティーがないので（GitHubに14kの星があっても）、Langfuseと比べるとやはり少し微妙