Google ADKで作ったエージェントに Langfuseのトレースにプロンプトを紐付ける方法

from openinference.instrumentation.google_adk import GoogleADKInstrumentorGoogleADKInstrumentor().instrument()

pip install langfuse google-adk openinference-instrumentation-google-adk opentelemetry-sdk python-dotenv

import asyncio
from contextvars import ContextVar
from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from langfuse import get_client
from openinference.instrumentation.google_adk import GoogleADKInstrumentor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, SpanProcessor

# ============================================================
# Langfuse プロンプト紐付け用コード
# ============================================================

prompt_info_var = ContextVar("prompt_info", default=None)

class LangfusePromptProcessor(SpanProcessor):
    """call_llm スパンにプロンプト情報を付与する"""

    def on_start(self, span, parent_context=None):
        if hasattr(span, "name") and span.name == "call_llm":
            prompt_info = prompt_info_var.get()
            if prompt_info:
                span.set_attribute("langfuse.prompt.name", prompt_info["name"])
                span.set_attribute("langfuse.prompt.version", prompt_info["version"])

    def on_end(self, span):
        pass

    def shutdown(self):
        pass

    def force_flush(self, timeout_millis=30000):
        return True

# ============================================================
# アプリケーションコード
# ============================================================

async def main():
    load_dotenv()
    langfuse = get_client()

    # 1. TracerProvider を作成し、SpanProcessor を登録
    provider = TracerProvider()
    provider.add_span_processor(LangfusePromptProcessor())
    trace.set_tracer_provider(provider)

    # 2. Google ADK のトレースを有効化
    GoogleADKInstrumentor().instrument()

    # 3. instruction 関数を定義（ここでプロンプトを紐付け）
    def get_instruction(ctx):
        prompt = langfuse.get_prompt("my_agent_instruction")
        prompt_info_var.set({"name": prompt.name, "version": prompt.version})
        return prompt.compile()

    # 4. Agent を作成
    agent = Agent(
        name="my_agent",
        model="gemini-2.5-flash",
        instruction=get_instruction,  # 関数を渡す
        tools=[],
    )

    # 5. セッションを作成して実行
    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name="my_app", user_id="user-1", session_id="session-1"
    )
    runner = Runner(agent=agent, app_name="my_app", session_service=session_service)

    user_msg = types.Content(role="user", parts=[types.Part(text="Hello")])
    for event in runner.run(user_id="user-1", session_id="session-1", new_message=user_msg):
        if event.is_final_response():
            print(event.content.parts[0].text)

    # 6. トレースデータを送信
    langfuse.flush()

if __name__ == "__main__":
    asyncio.run(main())

LANGFUSE_PUBLIC_KEY=pk-lf-xxx
LANGFUSE_SECRET_KEY=sk-lf-xxx
LANGFUSE_BASE_URL=https://xxx
GOOGLE_API_KEY=xxx

provider = TracerProvider()
provider.add_span_processor(LangfusePromptProcessor())  # 先に登録
trace.set_tracer_provider(provider)
GoogleADKInstrumentor().instrument()  # 後から計装

GoogleADKInstrumentor().instrument()
provider.add_span_processor(...)  # 既に計装済みで反映されない

def get_instruction(ctx):
    prompt = langfuse.get_prompt("my_prompt")
    prompt_info_var.set({"name": prompt.name, "version": prompt.version})
    return prompt.compile()

agent = Agent(instruction=get_instruction, ...)

agent = Agent(instruction="You are a helpful assistant", ...)