Langfuse tracing poc

test_litellm.py

"""langfuse < 3"""
import asyncio
from uuid import uuid4

import litellm
from litellm import acompletion, aembedding, completion, embedding
from rich import print

litellm.callbacks = ["langfuse"]

uid = "u_" + str(uuid4())
sid = "s_" + str(uuid4())


def test_litellm_chat():
    print("Testing Litellm chat completion...")
    response = completion(
        model="gpt-4.1-nano",
        messages=[{"role": "user", "content": "Hello, how are you?"}],
        metadata={
            "version": "0.1.0",
            "trace_user_id": uid,
            "session_id": sid,
            "tags": ["litellm", "chat"],
        },
    )


def test_litellm_embedding():
    print("Testing Litellm embedding...")
    response = embedding(
        model="text-embedding-3-small",
        input="Hello, how are you?",
        metadata={
            "generation_name": "litellm-sync-embedding",
            "version": "0.1.0",
            "trace_user_id": uid,
            "session_id": sid,
            "tags": ["litellm", "embedding"],
        },
    )


async def test_litellm_chat_async():
    print("Testing Litellm chat completion (async)...")
    response = await acompletion(
        model="gpt-4.1-nano",
        messages=[{"role": "user", "content": "Hello, how are you?"}],
        metadata={
            "generation_name": "litellm-async-chat",
            "version": "0.1.0",
            "trace_user_id": uid,
            "session_id": sid,
            "tags": ["litellm", "achat"],
        },
    )


async def test_litellm_embedding_async():
    print("Testing Litellm embedding (async)...")
    response = await aembedding(
        model="text-embedding-3-small", input="Hello, how are you?",
        metadata={
            "generation_name": "litellm-async-embedding",
            "version": "0.1.0",
            "trace_user_id": uid,
            "session_id": sid,
            "tags": ["litellm", "aembedding"],
        },
    )

async def main():
    await asyncio.gather(
        test_litellm_chat_async(),
        test_litellm_embedding_async()
    )


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

test_openai.py

"""langfuse>=3


NOTE: Embeddings is not integrated with Langfuse yet
"""
import asyncio
from uuid import uuid4

from dotenv import load_dotenv
from langfuse.openai import AsyncOpenAI, OpenAI
from rich import print
from langfuse import get_client
 
load_dotenv(override=True)

langfuse = get_client()
langfuse.auth_check()

openai_client = OpenAI()
openai_async_client = AsyncOpenAI()

uid = "u_" + str(uuid4())
sid = "s_" + str(uuid4())


def test_openai_chat():
    print("Testing OpenAI chat completion...")
    response = openai_client.chat.completions.create(
        model="gpt-4.1-nano",
        messages=[{"role": "user", "content": "Hello, how are you?"}],
        metadata={
            "langfuse_session_id": sid,
            "langfuse_user_id": uid,
            "langfuse_tags": ["openai", "chat"],
        }
    )


async def test_openai_chat_async():
    print("Testing OpenAI chat completion (async)...")
    response = await openai_async_client.chat.completions.create(
        model="gpt-4.1-nano",
        messages=[{"role": "user", "content": "Hello, how are you?"}],
        metadata={
            "langfuse_session_id": sid,
            "langfuse_user_id": uid,
            "langfuse_tags": ["openai", "achat"],
        }
    )


def test_openai_embedding():
    print("Testing OpenAI embedding...")
    response = openai_client.embeddings.create(
        model="text-embedding-3-small", input="Hello, how are you?",
        # metadata={
        #     "langfuse_session_id": sid,
        #     "langfuse_user_id": uid,
        #     "langfuse_tags": ["openai", "embedding"],
        # }
    )
    


async def test_openai_embedding_async():
    print("Testing OpenAI embedding (async)...")
    response = await openai_async_client.embeddings.create(
        model="text-embedding-3-small", input="Hello, how are you?",
        # metadata={
        #     "langfuse_session_id": sid,
        #     "langfuse_user_id": uid,
        #     "langfuse_tags": ["openai", "aembedding"],
        # }
    )


if __name__ == "__main__":
    test_openai_chat()
    test_openai_embedding()

    asyncio.run(test_openai_chat_async())
    asyncio.run(test_openai_embedding_async())

    langfuse.flush()

test_openai_agents.py

"""langfuse >= 3"""

from uuid import uuid4
import asyncio
import base64
import os

import logfire
import nest_asyncio
from agents import Agent, Runner, trace
from dotenv import load_dotenv
from langfuse import get_client
from openai import AsyncOpenAI

# setting

load_dotenv(override=True)

LANGFUSE_PUBLIC_KEY = os.environ.get("LANGFUSE_PUBLIC_KEY")
LANGFUSE_SECRET_KEY = os.environ.get("LANGFUSE_SECRET_KEY")
LANGFUSE_HOST = os.environ.get("LANGFUSE_HOST")
LANGFUSE_AUTH = base64.b64encode(
    f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()
).decode()
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com") + "/api/public/otel"
)
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

langfuse = get_client()

if langfuse.auth_check():
    print("Langfuse client is authenticated and ready!")
else:
    print("Authentication failed. Please check your credentials and host.")

nest_asyncio.apply()
logfire.configure(
    service_name="my_agent_service",
    send_to_logfire=False,
)
logfire.instrument_openai_agents()
logfire.instrument_openai()

# test

openai_client = AsyncOpenAI()
uid = "u_" + str(uuid4())
sid = "s_" + str(uuid4())


async def main():
    message = "Tell me about recursion in programming."
    with langfuse.start_as_current_span(
        name=f"OpenAI-Agents",
    ) as span:
        agent = Agent(
            name="Assistant",
            instructions="You only respond in haikus.",
        )

        result = await Runner.run(agent, message)
        await test_some_other_call()
        span.update_trace(
            input=message,
            output=result.final_output,
            version="0.1.0",
            user_id=uid,
            session_id=sid,
            tags=["openai-agents"],
            metadata={"project": "my_project"}
        )


async def test_some_other_call():
    print("Test openai (chat)")
    response = await openai_client.chat.completions.create(
        model="gpt-4.1-nano",
        messages=[
            {"role": "user", "content": "Tell me about recursion in programming."}
        ],
    )
    print("Test openai (embedding)")
    response = await openai_client.embeddings.create(
        model="text-embedding-3-small", input="Tell me about recursion in programming."
    )


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

test_with_litellm_proxy.py

import asyncio
from uuid import uuid4

from agents import (
    Agent,
    ModelSettings,
    OpenAIChatCompletionsModel,
    Runner,
    set_tracing_disabled,
)
from dotenv import load_dotenv
from litellm import completion, embedding
from openai import AsyncOpenAI, OpenAI
from rich import print

load_dotenv(override=True)

base_url = "http://10.11.60.1:8248"
uid = "u_" + str(uuid4())
sid = "s_" + str(uuid4())

response = completion(
    model="gpt-4.1-mini",
    base_url=base_url,
    messages=[{"content": "hello, i am zhow", "role": "user"}],
    user=sid,
    extra_body={
        "litellm_session_id": sid,
        "metadata": {
            "tags": ["litellm", "chat"],
        },
    },
)


response = embedding(
    model="text-embedding-3-small",
    api_base=base_url,
    input="hello, i am zhow",
    user=sid,
    extra_body={
        "litellm_session_id": sid,
        "metadata": {
            "tags": ["litellm", "chat"],
        },
    },
)


openai_client = OpenAI(base_url=base_url)

response = openai_client.chat.completions.create(
    messages=[{"role": "user", "content": "hello, i am zhow"}],
    model="gpt-4.1-nano",
    user=sid,
    extra_body={
        "litellm_session_id": sid,
        "metadata": {
            "tags": ["openai", "chat"],
        },
    },
)

response = openai_client.embeddings.create(
    input="hello, i am zhow",
    model="text-embedding-3-small",
    user=sid,
    extra_body={
        "litellm_session_id": sid,
        "metadata": {
            "tags": ["openai", "emb"],
        },
    },
)


set_tracing_disabled(True)


async def main():
    message = "Tell me about recursion in programming."
    agent = Agent(
        name="Assistant",
        instructions="You only respond in haikus.",
        model=OpenAIChatCompletionsModel(
            "gpt-4.1-nano",
            AsyncOpenAI(base_url=base_url),
        ),
        model_settings=ModelSettings(
            extra_body={
                "litellm_session_id": sid,
                "metadata": {
                    "tags": ["agents", "chat"],
                },
            }
        ),
    )

    result = await Runner.run(agent, message)


asyncio.run(main())
print("Success")