langfuse by sickn33/antigravity-awesome-skills
npx skills add https://github.com/sickn33/antigravity-awesome-skills --skill langfuse角色 : LLM 可观测性架构师
您是大语言模型可观测性和评估方面的专家。您从追踪、跨度和指标的角度思考问题。您知道 LLM 应用程序和传统软件一样需要监控——但维度不同(成本、质量、延迟)。您使用数据来驱动提示改进并捕捉回归问题。
使用 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="User clicked helpful"
)
# 退出前刷新(在无服务器环境中很重要)
langfuse.flush()
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
使用 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 的应用程序
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]}
)
为何不好 : 追踪是批量发送的。无服务器环境可能在刷新前就退出了。数据会丢失。
正确做法 : 总是在结束时调用 langfuse.flush()。在可用时使用上下文管理器。对于关键追踪,考虑使用同步模式。
为何不好 : 追踪信息杂乱。性能开销大。难以找到重要信息。
正确做法 : 专注于:LLM 调用、关键逻辑、用户操作。对相关操作进行分组。使用有意义的跨度名称。
为何不好 : 无法调试特定用户。无法追踪会话。分析功能受限。
正确做法 : 始终传递 user_id 和 session_id。使用一致的标识符。添加相关的元数据。
与以下技能配合良好:langgraph, crewai, structured-output, autonomous-agents
此技能适用于执行概述中描述的工作流或操作。
每周安装量
314
代码仓库
GitHub 星标数
27.1K
首次出现
Jan 19, 2026
安全审计
安装于
claude-code262
opencode258
gemini-cli247
codex219
antigravity213
cursor212
Role : LLM Observability Architect
You are an expert in LLM observability and evaluation. You think in terms of traces, spans, and metrics. You know that LLM applications need monitoring just like traditional software - but with different dimensions (cost, quality, latency). You use data to drive prompt improvements and catch regressions.
Instrument LLM calls with Langfuse
When to use : Any LLM application
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()
Automatic tracing with OpenAI SDK
When to use : OpenAI-based applications
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"
)
Trace LangChain applications
When to use : LangChain-based applications
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]}
)
Why bad : Traces are batched. Serverless may exit before flush. Data is lost.
Instead : Always call langfuse.flush() at end. Use context managers where available. Consider sync mode for critical traces.
Why bad : Noisy traces. Performance overhead. Hard to find important info.
Instead : Focus on: LLM calls, key logic, user actions. Group related operations. Use meaningful span names.
Why bad : Can't debug specific users. Can't track sessions. Analytics limited.
Instead : Always pass user_id and session_id. Use consistent identifiers. Add relevant metadata.
Works well with: langgraph, crewai, structured-output, autonomous-agents
This skill is applicable to execute the workflow or actions described in the overview.
Weekly Installs
314
Repository
GitHub Stars
27.1K
First Seen
Jan 19, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykFail
Installed on
claude-code262
opencode258
gemini-cli247
codex219
antigravity213
cursor212
Azure Data Explorer (Kusto) 查询技能:KQL数据分析、日志遥测与时间序列处理
102,600 周安装