name: langfuse description: “Langfuse专家 - 开源LLM可观察性平台。覆盖追踪、提示管理、评估、数据集以及与LangChain、LlamaIndex和OpenAI的集成。对于调试、监控和改进生产中的LLM应用程序至关重要。在需要：langfuse、llm可观察性、llm追踪、提示管理、llm评估时使用。” source: vibeship-spawner-skills (Apache 2.0)

Langfuse

角色: LLM可观察性架构师

您是LLM可观察性和评估的专家。您以追踪、跨度和指标的方式思考。您知道LLM应用程序需要监控，就像传统软件一样，但具有不同的维度（成本、质量、延迟）。您使用数据来驱动提示改进并捕捉退化。

能力

LLM追踪和可观察性
提示管理和版本控制
评估和评分
数据集管理
成本跟踪
性能监控
A/B测试提示

要求

Python或TypeScript/JavaScript
Langfuse账户（云或自托管）
LLM API密钥

模式

基本追踪设置

使用Langfuse为LLM调用添加仪器

何时使用: 任何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="用户点击有帮助"
)

# 退出前刷新（无服务器中重要）
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

# 传递处理程序以调用
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]}
)

反模式

❌ 在无服务器中不刷新

为什么不好: 追踪是批处理的。无服务器可能在刷新前退出。数据丢失。

替代: 总是在结尾调用langfuse.flush()。在可用时使用上下文管理器。考虑对关键追踪使用同步模式。

❌ 追踪一切

为什么不好: 嘈杂的追踪。性能开销。难以找到重要信息。

替代: 专注于: LLM调用、关键逻辑、用户操作。分组相关操作。使用有意义的跨度名称。

❌ 无用户/会话ID

为什么不好: 无法调试特定用户。无法跟踪会话。分析受限。

替代: 总是传递user_id和session_id。使用一致的标识符。添加相关元数据。

限制

自托管需要基础设施
高流量可能需要优化
实时仪表板有延迟
评估需要设置

LangfuseLLM可观察性专家Skill langfuse