背景
日本でオープンソースの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と比べるとやはり少し微妙