langfuse

Langfuse

角色：大模型可观测性架构师

你是大模型可观测性和评估方面的专家。你从跟踪、跨度和指标的角度思考。你知道大模型应用和传统软件一样需要监控，但监控维度有所不同（成本、质量、延迟）。你会用数据推动提示词改进并发现回归问题。

能力

• 大模型跟踪与可观测性
• 提示词管理与版本控制
• 评估与评分
• 数据集管理
• 成本跟踪
• 性能监控
• 提示词 A/B 测试

要求

• Python 或 TypeScript/JavaScript
• Langfuse 账号（云端或可自托管）
• 大模型 API 密钥

模式

基础跟踪设置

使用 Langfuse 为大模型调用添加监控

适用场景：任何大模型应用

from langfuse import Langfuse

# Initialize client
langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"  # or self-hosted URL
)

# Create a trace for a user request
trace = langfuse.trace(
    name="chat-completion",
    user_id="user-123",
    session_id="session-456",  # Groups related traces
    metadata={"feature": "customer-support"},
    tags=["production", "v2"]
)

# Log a generation (LLM call)
generation = trace.generation(
    name="gpt-4o-response",
    model="gpt-4o",
    model_parameters={"temperature": 0.7},
    input={"messages": [{"role": "user", "content": "Hello"}]},
    metadata={"attempt": 1}
)

# Make actual LLM call
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)

# Complete the generation with output
generation.end(
    output=response.choices[0].message.content,
    usage={
        "input": response.usage.prompt_tokens,
        "output": response.usage.completion_tokens
    }
)

# Score the trace
trace.score(
    name="user-feedback",
    value=1,  # 1 = positive, 0 = negative
    comment="User clicked helpful"
)

# Flush before exit (important in serverless)
langfuse.flush()

OpenAI 集成

使用 OpenAI SDK 自动跟踪

适用场景：基于 OpenAI 的应用

from langfuse.openai import openai

# Drop-in replacement for OpenAI client
# All calls automatically traced

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    # Langfuse-specific parameters
    name="greeting",  # Trace name
    session_id="session-123",
    user_id="user-456",
    tags=["test"],
    metadata={"feature": "chat"}
)

# Works with streaming
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="")

# Works with async
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

# Create Langfuse callback handler
langfuse_handler = CallbackHandler(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com",
    session_id="session-123",
    user_id="user-456"
)

# Use with any LangChain component
llm = ChatOpenAI(model="gpt-4o")

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

chain = prompt | llm

# Pass handler to invoke
response = chain.invoke(
    {"input": "Hello"},
    config={"callbacks": [langfuse_handler]}
)

# Or set as default
import langchain
langchain.callbacks.manager.set_handler(langfuse_handler)

# Then all calls are traced
response = chain.invoke({"input": "Hello"})

# Works with agents, retrievers, etc.
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]}
)

反模式

❌ 无服务器环境中未刷新数据

问题所在：跟踪数据会批量处理。 无服务器环境可能在刷新前退出。 数据会丢失。

建议做法：始终在末尾调用 langfuse.flush()。 在可用时使用上下文管理器。 对于关键跟踪数据，可考虑使用同步模式。

❌ 跟踪所有内容

问题所在：跟踪数据噪声过多。 会产生性能开销。 难以找到重要信息。

建议做法：重点跟踪：大模型调用、关键逻辑、用户操作。 将相关操作分组。 使用有意义的跨度名称。

❌ 未设置用户/会话 ID

问题所在：无法调试特定用户的问题。 无法跟踪会话。 分析能力受限。

建议做法：始终传入 user_id 和 session_id。 使用一致的标识符。 添加相关元数据。

局限性

• 可自托管版本需要基础设施
• 高流量场景可能需要优化
• 实时仪表板存在延迟
• 评估需要进行设置

相关技能

可与以下工具配合使用：langgraph、crewai、structured-output、autonomous-agents