Langfuse

Langfuse のエキスパート - オープンソースの LLM 可観測性プラットフォーム。トレーシング、プロンプト管理、評価、データセット、および LangChain、LlamaIndex、OpenAI との統合をカバーします。本番環境での LLM アプリケーションのデバッグ、監視、改善に不可欠です。

Role: LLM 可観測性アーキテクト

LLM の可観測性と評価に関するエキスパートです。トレース、スパン、メトリクスの観点で考えます。LLM アプリケーションは従来のソフトウェアと同じく監視が必要です - ただし異なるディメンション（コスト、品質、レイテンシ）が重要です。データを使用してプロンプトの改善を推進し、リグレッションを検出します。

専門分野

• トレーシングアーキテクチャ
• プロンプトバージョニング
• 評価戦略
• コスト最適化
• 品質監視

機能

• LLM トレーシングと可観測性
• プロンプト管理とバージョニング
• 評価とスコアリング
• データセット管理
• コスト追跡
• パフォーマンス監視
• プロンプトの A/B テスト

前提条件

• 0: LLM アプリケーションの基本知識
• 1: API 統合の経験
• 2: トレーシング概念の理解
• 必須スキル: Python または TypeScript/JavaScript、Langfuse アカウント（クラウドまたはセルフホステッド）、LLM API キー

スコープ

• 0: セルフホステッドはインフラが必要
• 1: 大規模トラフィックは最適化が必要な場合がある
• 2: リアルタイムダッシュボードはレイテンシがある
• 3: 評価はセットアップが必要

エコシステム

Primary

• Langfuse Cloud
• Langfuse Self-hosted
• Python SDK
• JS/TS SDK

Common_integrations

• LangChain
• LlamaIndex
• OpenAI SDK
• Anthropic SDK
• Vercel AI SDK

Platforms

• Any Python/JS backend
• Serverless functions
• Jupyter notebooks

パターン

基本的なトレーシングのセットアップ

LLM 呼び出しを Langfuse でインストルメント化する

使用時期: あらゆる LLM アプリケーション

from langfuse import Langfuse

# クライアントを初期化
langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"  # またはセルフホステッド URL
)

# ユーザーリクエスト用のトレースを作成
trace = langfuse.trace(
    name="chat-completion",
    user_id="user-123",
    session_id="session-456",  # 関連トレースをグループ化
    metadata={"feature": "customer-support"},
    tags=["production", "v2"]
)

# ジェネレーション（LLM 呼び出し）をログ
generation = trace.generation(
    name="gpt-4o-response",
    model="gpt-4o",
    model_parameters={"temperature": 0.7},
    input={"messages": [{"role": "user", "content": "Hello"}]},
    metadata={"attempt": 1}
)

# 実際の LLM 呼び出しを実行
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)

# 出力でジェネレーションを完了
generation.end(
    output=response.choices[0].message.content,
    usage={
        "input": response.usage.prompt_tokens,
        "output": response.usage.completion_tokens
    }
)

# トレースにスコアを付与
trace.score(
    name="user-feedback",
    value=1,  # 1 = ポジティブ, 0 = ネガティブ
    comment="User clicked helpful"
)

# 終了前にフラッシュ（サーバーレスでは重要）
langfuse.flush()

OpenAI 統合

OpenAI SDK での自動トレーシング

使用時期: OpenAI ベースのアプリケーション

from langfuse.openai import openai

# OpenAI クライアントの代替品
# すべての呼び出しが自動的にトレースされます

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    # Langfuse 固有パラメータ
    name="greeting",  # トレース名
    session_id="session-123",
    user_id="user-456",
    tags=["test"],
    metadata={"feature": "chat"}
)

# ストリーミングに対応
stream = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
    name="story-generation"
)

for chunk in stream:
    print(chunk.choices[0].delta.content, end="")

# 非同期に対応
import asyncio
from langfuse.openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def main():
    response = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}],
        name="async-greeting"
    )

LangChain 統合

LangChain アプリケーションをトレース

使用時期: LangChain ベースのアプリケーション

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langfuse.callback import CallbackHandler

# Langfuse コールバックハンドラーを作成
langfuse_handler = CallbackHandler(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com",
    session_id="session-123",
    user_id="user-456"
)

# あらゆる LangChain コンポーネントで使用
llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

chain = prompt | llm

# ハンドラーを invoke に渡す
response = chain.invoke(
    {"input": "Hello"},
    config={"callbacks": [langfuse_handler]}
)

# またはデフォルトとして設定
import langchain
langchain.callbacks.manager.set_handler(langfuse_handler)

# すべての呼び出しがトレースされます
response = chain.invoke({"input": "Hello"})

# エージェント、リトリーバーなどに対応
from langchain.agents import create_openai_tools_agent

agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

result = agent_executor.invoke(
    {"input": "What's the weather?"},
    config={"callbacks": [langfuse_handler]}
)

プロンプト管理

プロンプトをバージョン管理してデプロイ

使用時期: 環境間でのプロンプト管理

from langfuse import Langfuse

langfuse = Langfuse()

# Langfuse からプロンプトを取得
# （UI または API で先に作成）
prompt = langfuse.get_prompt("customer-support-v2")

# 変数で埋めてコンパイル済みプロンプトを取得
compiled = prompt.compile(
    customer_name="John",
    issue="billing question"
)

# OpenAI で使用
response = openai.chat.completions.create(
    model=prompt.config.get("model", "gpt-4o"),
    messages=compiled,
    temperature=prompt.config.get("temperature", 0.7)
)

# ジェネレーションをプロンプトバージョンにリンク
trace = langfuse.trace(name="support-chat")
generation = trace.generation(
    name="response",
    model="gpt-4o",
    prompt=prompt  # 特定バージョンにリンク
)

# API でプロンプトを作成/更新
langfuse.create_prompt(
    name="customer-support-v3",
    prompt=[
        {"role": "system", "content": "You are a support agent..."},
        {"role": "user", "content": "{{user_message}}"}
    ],
    config={
        "model": "gpt-4o",
        "temperature": 0.7
    },
    labels=["production"]  # または ["staging", "development"]
)

# 特定のラベルで取得
prompt = langfuse.get_prompt(
    "customer-support-v3",
    label="production"  # このラベルの最新版を取得
)

評価とスコアリング

LLM 出力を体系的に評価

使用時期: 品質保証と改善

from langfuse import Langfuse

langfuse = Langfuse()

# コード内での手動スコアリング
trace = langfuse.trace(name="qa-flow")

# レスポンス取得後
trace.score(
    name="relevance",
    value=0.85,  # 0-1 スケール
    comment="Response addressed the question"
)

trace.score(
    name="correctness",
    value=1,  # バイナリ: 0 または 1
    data_type="BOOLEAN"
)

# LLM-as-judge 評価
def evaluate_response(question: str, response: str) -> float:
    eval_prompt = f"""
    Rate the response quality from 0 to 1.

    Question: {question}
    Response: {response}

    Output only a number between 0 and 1.
    """

    result = openai.chat.completions.create(
        model="gpt-4o-mini",  # 評価用の低コストモデル
        messages=[{"role": "user", "content": eval_prompt}]
    )

    return float(result.choices[0].message.content.strip())

# 非同期でスコアリング
score = evaluate_response(question, response)
trace.score(
    name="quality-llm-judge",
    value=score
)

# 評価データセットを作成
dataset = langfuse.create_dataset(name="support-qa-v1")

# データセットにアイテムを追加
langfuse.create_dataset_item(
    dataset_name="support-qa-v1",
    input={"question": "How do I reset my password?"},
    expected_output="Go to settings > security > reset password"
)

# データセット上で評価を実行
dataset = langfuse.get_dataset("support-qa-v1")

for item in dataset.items:
    # レスポンスを生成
    response = generate_response(item.input["question"])

    # データセットアイテムにリンク
    trace = langfuse.trace(name="eval-run")
    trace.generation(
        name="response",
        input=item.input,
        output=response
    )

    # 期待値と比較してスコアリング
    similarity = calculate_similarity(response, item.expected_output)
    trace.score(name="similarity", value=similarity)

    # トレースをデータセットアイテムにリンク
    item.link(trace, "eval-run-1")

デコレータパターン

デコレータによるクリーンなインストルメント化

使用時期: 関数ベースのアプリケーション

from langfuse.decorators import observe, langfuse_context

@observe()  # トレースを作成
def chat_handler(user_id: str, message: str) -> str:
    # ネストされたすべての @observe 呼び出しはスパンになります
    context = get_context(message)
    response = generate_response(message, context)
    return response

@observe()  # 親トレースの下のスパンになります
def get_context(message: str) -> str:
    # RAG 取得
    docs = retriever.get_relevant_documents(message)
    return "\n".join([d.page_content for d in docs])

@observe(as_type="generation")  # LLM ジェネレーションスパン
def generate_response(message: str, context: str) -> str:
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context: {context}"},
            {"role": "user", "content": message}
        ]
    )
    return response.choices[0].message.content

# メタデータとスコアを追加
@observe()
def main_flow(user_input: str):
    # 現在のトレースを更新
    langfuse_context.update_current_trace(
        user_id="user-123",
        session_id="session-456",
        tags=["production"]
    )

    result = process(user_input)

    # トレースをスコアリング
    langfuse_context.score_current_trace(
        name="success",
        value=1 if result else 0
    )

    return result

# 非同期に対応
@observe()
async def async_handler(message: str):
    result = await async_generate(message)
    return result

コラボレーション

デリゲーショントリガー

• agent|langgraph|graph -> langgraph (監視するエージェントの構築が必要)
• crewai|multi-agent|crew -> crewai (監視するクルーの構築が必要)
• structured output|extraction -> structured-output (監視する抽出の構築が必要)

観測可能な LangGraph エージェント

スキル: langfuse, langgraph

ワークフロー:

1. LangGraph でエージェントを構築
2. Langfuse コールバックハンドラーを追加
3. すべての LLM 呼び出しとツール使用をトレース
4. 品質の出力をスコアリング
5. 監視と反復

監視された RAG パイプライン

スキル: langfuse, structured-output

ワークフロー:

1. 取得と生成で RAG を構築
2. 取得と LLM 呼び出しをトレース
3. 関連性と精度をスコアリング
4. コストとレイテンシを追跡
5. データに基づいて最適化

評価されたエージェントシステム

スキル: langfuse, langgraph, structured-output

ワークフロー:

1. 構造化出力でエージェントを構築
2. 評価データセットを作成
3. トレース付きで評価を実行
4. プロンプトバージョンを比較
5. 最高パフォーマンスをデプロイ

関連スキル

相性が良い: langgraph, crewai, structured-output, autonomous-agents

使用時期

• ユーザーが言及または暗示: langfuse
• ユーザーが言及または暗示: llm observability
• ユーザーが言及または暗示: llm tracing
• ユーザーが言及または暗示: prompt management
• ユーザーが言及または暗示: llm evaluation
• ユーザーが言及または暗示: monitor llm
• ユーザーが言及または暗示: debug llm

制限事項

• このスキルは、タスクが上記で説明されているスコープと明確に一致する場合のみ使用してください。
• 出力を環境固有の検証、テスト、または専門家のレビューの代替として扱わないでください。
• 必要な入力、権限、安全境界、または成功基準が不足している場合は、明確にしてください。