Langfuse 可观测性：开源的 LLM 监控方案
LLM 应用的可观测性困境 当你的 LLM 应用从 Demo 走向生产，一系列问题随之而来：某个请求为什么返回了错误结果？Prompt 改动后效果到底是变好还是变差？不同模型的延迟和成本如何对比？传统 APM 工具（如 Datadog）对 LLM 场景缺乏原生支持。Langfuse 作为开源的 LLM 可观测性平台，正填补了这一空白。 Langfuse 核心功能 功能矩阵 功能模块 说明 Tracing 请求级全链路追踪，含 LLM 调用、工具调用、检索流程 Prompt Management Prompt 版本管理、AB 测试、在线编辑 Analytics 延迟、Token 用量、成本统计 Evaluation 人工标注 + 自动评估（基于规则或模型裁判） User Feedback 用户点赞/点踩采集与关联 Model Usage 多模型/多项目用量看板 Tracing：全链路追踪 集成方式 Python SDK（推荐） from langfuse import Langfuse from langfuse.decorators import observe langfuse = Langfuse() @observe() # 自动创建 trace def chat_bot(user_message: str): # 嵌套的 @observe 会自动建立父子关系 retrieved_docs = retrieve(user_message) # 被 trace answer = generate_response(user_message, retrieved_docs) # 被 trace return answer @observe(as_type="generation") def generate_response(question, docs): response = openai.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"基于以下内容回答：{docs}\n问题：{question}"}], ) return response.choices[0].message.content @observe(as_type="span") def retrieve(query): # 检索逻辑 return vector_db.search(query) OpenAI SDK 集成（Drop-in） from langfuse.openai import openai # 只需替换 import，所有 OpenAI 调用自动被 trace response = openai.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Hello"}], metadata={"user_id": "user_123"}, # 附加用户信息 ) LangChain 集成 from langfuse.callback import CallbackHandler handler = CallbackHandler() chain.invoke( {"input": "什么是量子计算？"}, config={"callbacks": [handler]} ) Trace 结构 Trace (chat_bot 调用) ├── Span (retrieve - 向量检索) │ └── Generation (embedding 调用) ├── Generation (generate_response - LLM 调用) │ ├── input: "基于以下内容..." │ ├── output: "量子计算是..." │ ├── model: gpt-4o │ ├── tokens: {input: 350, output: 280} │ └── cost: $0.0091 └── Span (后处理) 自定义 Span from langfuse import Langfuse langfuse = Langfuse() trace = langfuse.trace(name="rag-pipeline", user_id="user_123") span = trace.span(name="document-retrieval", input={"query": "量子计算"}) # ... 执行检索 ... span.end(output={"docs": [...]}) gen = trace.generation( name="answer-generation", model="gpt-4o", input={"messages": [...]}, output={"content": "..."}, usage={"prompt_tokens": 350, "completion_tokens": 280}, ) Prompt 管理 版本化 Prompt from langfuse import Langfuse langfuse = Langfuse() # 获取生产环境的 Prompt（自动缓存） prompt = langfuse.get_prompt( name="summarizer", version=2, # 指定版本，或省略获取最新 ) # 编译 Prompt（变量替换） compiled = prompt.compile( text="一段需要总结的长文本...", max_length="200" ) # 直接用于 LLM 调用 response = openai.chat.completions.create( model=prompt.config["model"], messages=[{"role": "user", "content": compiled}], ) Prompt AB 测试 import random # 50% 用户用 v1，50% 用 v2 version = random.choice(["v1", "v2"]) prompt = langfuse.get_prompt(name="customer-support", version=version) # Langfuse 自动记录每个请求使用了哪个版本 trace = langfuse.trace(name="support-chat") trace.update(metadata={"prompt_version": version}) 在线编辑 Langfuse Web UI 提供 Prompt 在线编辑器： ...