基于Langfuse的LLM应用可观测性实战：从部署到智能体评估

这次我们来看一个面向大模型应用开发者的实战项目：基于 Langfuse 平台的智能体评估与 LLM 观测。如果你正在开发或优化基于大语言模型（LLM）的智能体（Agent）、RAG 系统或复杂应用链，并且苦于无法有效追踪每一次 API 调用、评估生成质量、调试复杂流程，那么这个项目正是为你准备的。Langfuse 是一个开源的 LLM 应用观测与评估平台，它不是一个模型，而是一个强大的“仪表盘”和“分析工具”，能帮你把黑盒的 AI 调用过程变得透明、可度量、可优化。

它的核心价值在于，将分散的日志、评估指标和用户反馈整合到一个统一的界面中，让你能清晰地看到：每次请求的输入输出是什么、调用了哪些模型、消耗了多少 Token、花费了多少钱、生成的答案质量如何、以及整个调用链的延迟和错误情况。对于追求稳定性和成本可控的生产级应用来说，这种可观测性至关重要。

本文将带你从零开始，完成一个完整的智能体评估实战项目。我们会重点覆盖：如何快速部署 Langfuse（支持本地和云托管）、如何将你的 LLM 应用代码与 Langfuse 集成、如何设计并自动化执行评估任务、以及如何利用追踪数据来调试和优化你的智能体。整个过程不涉及复杂的模型训练，聚焦于工程化落地，目标是让你看完就能动手搭建自己的 LLM 观测体系。

1. 核心能力速览

在深入细节之前，我们先通过一个表格快速了解 Langfuse 的核心能力、技术门槛和适用场景，帮助你判断是否值得投入时间。

从表格可以看出，Langfuse 的门槛主要在于理解和集成其观测体系，而非硬件算力。它更像一个“增强版的日志系统”，专门为 AI 应用设计。

2. 适用场景与使用边界

理解一个工具适合做什么、不适合做什么，比盲目上手更重要。

Langfuse 最适合的三大场景：

1. 智能体 (Agent) 与复杂工作流调试：当一个用户问题需要经过多轮思考 (Reasoning)、工具调用 (Tool Calling) 和模型交互才能解决时，整个调用链就像一团乱麻。Langfuse 可以可视化每一步的输入、输出、耗时和 Token 消耗，让你快速定位是哪个环节的提示词出了问题，还是某个工具调用超时。
2. RAG 系统效果评估与优化：检索增强生成 (RAG) 的质量取决于检索、排序和生成多个环节。通过 Langfuse，你可以追踪每一次查询的检索结果、传递给模型的上下文、以及最终答案。结合自动化评估（如答案相关性、事实准确性），你能科学地评估不同检索策略、分块大小或重排模型的效果。
3. 生产环境监控与成本管控：当你的应用服务大量用户时，你需要知道：哪个模型 API 调用最频繁？每天消耗多少 Token，成本是多少？平均响应延迟是多少？有没有异常的失败请求？Langfuse 的仪表盘能直接给出这些洞察，帮助你优化模型选型、设置预算告警。

Langfuse 不擅长或需要谨慎使用的场景：

• 替代模型训练与微调：Langfuse 用于观测和评估模型“使用”过程，不提供模型训练能力。
• 处理极端敏感数据：虽然自托管版能保证数据不出私域，但你仍需确保数据库和服务器本身的安全。对于医疗、金融等受严格监管的数据，部署和访问控制需格外谨慎。
• 替代单元测试：它主要用于集成测试和线上监控，不能完全替代针对业务逻辑的细粒度单元测试。

合规与安全边界提醒：

• 数据隐私：如果你使用云托管版，需仔细阅读其数据协议，确认传输和存储的数据是否符合你所在地区（如 GDPR）的要求。对于敏感业务数据，强烈建议使用自托管版。
• 用户知情权：如果追踪的数据包含用户输入的隐私信息，应考虑在前端告知用户并获得同意。
• 授权使用：确保你通过 Langfuse 观测的模型 API（如 OpenAI, Anthropic）是合法授权使用的，遵守相关 API 的使用条款。

3. 环境准备与前置条件

我们将以本地 Docker 部署为例，这是最通用、数据最可控的方式。如果你希望快速体验，也可以直接注册其云服务，跳过部署步骤。

基础环境要求：

1. 操作系统：Linux (Ubuntu 20.04+ / CentOS 7+), macOS, 或 Windows 10/11 (需安装 WSL2 以获得最佳体验)。本文命令以 Linux/macOS 为例。
2. Docker 与 Docker Compose：这是运行 Langfuse 服务的核心。
   • Docker：确保已安装 Docker Engine。终端执行docker --version验证。
   • Docker Compose：确保已安装 Docker Compose V2。终端执行docker compose version验证。
3. 网络与端口：确保主机（本地机器）的3000(前端) 和9020(后端) 端口未被占用。如果需要修改，后续在配置文件中调整。
4. 磁盘空间：预留至少 2GB 的可用空间用于存储 Docker 镜像和数据库数据。
5. 开发环境（用于集成 SDK）：
   • Python：3.8 或更高版本。这是集成 Langfuse Python SDK 所必需的。
   • Node.js：16 或更高版本（如果你使用 JS/TS SDK）。

环境检查清单：在开始前，请在终端依次运行以下命令，确保基础环境就绪：

# 1. 检查 Docker docker --version # 预期输出类似：Docker version 24.0.7, build afdd53b # 2. 检查 Docker Compose docker compose version # 预期输出类似：Docker Compose version v2.23.0 # 3. 检查 Python python3 --version # 预期输出类似：Python 3.10.12 # 4. 检查 3000 和 9020 端口占用 (Linux/macOS) sudo lsof -i :3000 sudo lsof -i :9020 # 如果无输出，则表示端口空闲。

4. 安装部署与启动方式

我们将使用官方提供的docker-compose.yml文件来一键启动所有服务（前端、后端、数据库）。

步骤 1：获取部署文件在你想安装的目录下（例如~/projects/langfuse），执行以下命令：

# 创建项目目录并进入 mkdir -p ~/projects/langfuse && cd ~/projects/langfuse # 下载官方 docker-compose 配置文件 curl -o docker-compose.yml https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml

步骤 2：启动 Langfuse 服务在包含docker-compose.yml文件的目录下，运行：

# 使用 -d 参数在后台运行 docker compose up -d

这个命令会拉取必要的镜像（PostgreSQL, Langfuse 后端和前端），并启动所有容器。第一次运行可能需要几分钟下载镜像。

步骤 3：验证服务状态启动完成后，检查容器是否正常运行：

docker compose ps

你应该看到三个服务（langfuse-db,langfuse-server,langfuse-frontend）的状态都是Up。

步骤 4：访问 Web 界面在浏览器中打开http://localhost:3000。如果一切正常，你将看到 Langfuse 的注册页面。

步骤 5：创建初始账户在注册页面，输入你的邮箱和密码，创建第一个管理员账户。这个账户将用于登录和管理你的 Langfuse 实例。

至此，Langfuse 平台已经部署完成并可以访问。接下来，我们需要将其与你的 LLM 应用连接起来。

5. 功能测试与效果验证：集成与追踪

部署好平台只是第一步，核心是让你的应用数据能流入 Langfuse。我们通过一个简单的 Python 示例来演示完整的集成、追踪和评估流程。

5.1 准备测试应用环境

首先，在一个新的 Python 虚拟环境中安装必要的 SDK。

# 创建并激活虚拟环境（可选，但推荐） python3 -m venv venv source venv/bin/activate # Windows 使用 `venv\Scripts\activate` # 安装 Langfuse Python SDK 和 OpenAI SDK（用于模拟LLM调用） pip install langfuse openai

5.2 配置 Langfuse 凭证

登录 Langfuse Web 界面 (http://localhost:3000)。

1. 点击左下角个人头像，进入“Settings”。
2. 在“API Keys”页面，点击“Create new API key”。
3. 为其命名（如test-key），并复制生成的公钥 (Public Key)和私钥 (Secret Key)。页面关闭后将无法再次查看私钥，请妥善保存。

在你的 Python 代码或环境变量中配置这些凭证：

# 在终端中设置环境变量（临时） export LANGFUSE_PUBLIC_KEY="pk-lf-xxxxxx" export LANGFUSE_SECRET_KEY="sk-lf-xxxxxx" export LANGFUSE_HOST="http://localhost:3000" # 自托管地址

5.3 基础追踪：装饰器集成

这是最简单的集成方式。假设我们有一个函数，它调用 OpenAI API 来回答问题。

# test_basic_trace.py import os from langfuse.decorators import observe, langfuse_context from openai import OpenAI # 初始化 OpenAI 客户端 (你需要有自己的 OPENAI_API_KEY) client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) @observe() # 使用装饰器自动追踪此函数 def ask_ai(question: str): """一个简单的问答函数""" print(f"用户问题: {question}") # 调用 OpenAI API response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": question}], temperature=0.7, ) answer = response.choices[0].message.content print(f"AI 回答: {answer}") # 你可以手动记录一些自定义信息到当前追踪中 langfuse_context.score_current_trace( name="user_feedback", value=5, # 假设我们模拟一个用户评分，满分5分 comment="模拟反馈：回答准确" ) return answer if __name__ == "__main__": # 执行函数，自动产生追踪记录 result = ask_ai("Langfuse 是什么？它主要解决什么问题？")

运行这个脚本后，打开 Langfuse 的 Web 界面 (http://localhost:3000)，进入“Traces”页面。你应该能看到一条新的追踪记录，点击进入可以查看：

• 时间线：函数执行的开始和结束时间。
• 输入/输出：question参数和函数的返回值。
• 子步骤：自动捕获的openai.chat.completions.create调用详情，包括模型、Token 使用量、延迟。
• 评分：我们手动添加的user_feedback分数。

5.4 复杂追踪：手动插桩与智能体模拟

对于更复杂的场景（如 LangChain、LlamaIndex 或自定义的智能体流程），可以使用手动插桩来获得更精细的控制。

# test_agent_trace.py import os from langfuse import Langfuse from langfuse.callback import CallbackHandler import random # 初始化 Langfuse 客户端 langfuse = Langfuse( public_key=os.getenv("LANGFUSE_PUBLIC_KEY"), secret_key=os.getenv("LANGFUSE_SECRET_KEY"), host=os.getenv("LANGFUSE_HOST"), ) def simulate_web_search(query: str): """模拟一个网络搜索工具""" return f"关于 '{query}' 的搜索结果摘要：...（模拟数据）" def simulate_calculator(expression: str): """模拟一个计算工具""" return f"计算结果：{eval(expression)}" # 注意：生产环境勿用 eval def run_agent_workflow(user_query: str): """模拟一个简单的智能体工作流：分析问题 -> 选择工具 -> 执行 -> 总结""" # 1. 创建主追踪 (Trace) trace = langfuse.trace( name="Agent Workflow", input=user_query, metadata={"user_id": "test_user_001"} ) # 2. 第一步：分析用户意图 (Span) analysis_span = trace.span( name="Intent Analysis", input=user_query, ) # 模拟分析逻辑 if "计算" in user_query or "+" in user_query or "*" in user_query: tool_to_use = "calculator" analysis_result = "用户需要进行数学计算。" else: tool_to_use = "web_search" analysis_result = "用户需要查询信息。" analysis_span.end(output=analysis_result) # 3. 第二步：执行工具调用 (Span) tool_span = trace.span( name=f"Tool Execution: {tool_to_use}", input={"tool": tool_to_use, "query": user_query}, ) if tool_to_use == "calculator": # 简单提取计算表达式，实际应用需要更复杂的解析 import re numbers = re.findall(r'\d+', user_query) if len(numbers) >= 2: expr = f"{numbers[0]}+{numbers[1]}" tool_result = simulate_calculator(expr) else: tool_result = "无法解析计算表达式。" else: tool_result = simulate_web_search(user_query) tool_span.end(output=tool_result) # 4. 第三步：生成最终回答 (Span) generation_span = trace.generation( name="Final Answer Generation", model="simulated-llm", model_parameters={"temperature": 0.5}, input={"analysis": analysis_result, "tool_result": tool_result}, ) final_answer = f"根据分析（{analysis_result}）和工具执行结果，我的回答是：{tool_result}" generation_span.end(output=final_answer) # 5. 结束主追踪，并添加一个总体评分 trace.update(output=final_answer) langfuse.score( trace_id=trace.id, name="workflow_quality", value=random.randint(3, 5), # 模拟一个随机评分 comment="自动生成的模拟评分" ) print(f"智能体最终回答: {final_answer}") return final_answer if __name__ == "__main__": queries = ["今天北京的天气怎么样？", "请计算 25 乘以 38 等于多少？"] for q in queries: print(f"\n处理查询: {q}") run_agent_workflow(q)

运行此脚本后，再次查看 Langfuse 的“Traces”页面。这次你会看到更复杂的追踪树：

• 一个主Trace包含三个Span（分析、工具执行、生成）。
• 每个Span都有独立的输入、输出和耗时。
• 最后关联了一个Score（评分）。

这种细粒度的追踪，是调试多步骤智能体的利器。

6. 接口 API 与批量任务

Langfuse 不仅提供 Web UI，也提供了功能强大的 REST API，允许你以编程方式管理数据、执行批量操作。

6.1 核心 API 调用示例

以下示例演示如何使用 Pythonrequests库调用 Langfuse API 来创建追踪和查询数据。

# test_langfuse_api.py import requests import json import os # 配置 LANGFUSE_HOST = os.getenv("LANGFUSE_HOST", "http://localhost:3000") PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY") SECRET_KEY = os.getenv("LANGFUSE_SECRET_KEY") # API 基础路径 BASE_URL = f"{LANGFUSE_HOST}/api/public" headers = { "Authorization": f"Bearer {SECRET_KEY}", "Content-Type": "application/json" } def create_trace_via_api(): """通过 API 直接创建一条追踪记录""" url = f"{BASE_URL}/traces" payload = { "name": "API-Created-Trace", "input": {"question": "What is the capital of France?"}, "output": {"answer": "Paris"}, "metadata": {"source": "api_test"}, "sessionId": "session_api_001" } response = requests.post(url, json=payload, headers=headers) if response.status_code == 200: trace_data = response.json() print(f"追踪创建成功！Trace ID: {trace_data.get('id')}") return trace_data.get('id') else: print(f"创建失败: {response.status_code}, {response.text}") return None def get_traces(): """查询最近的追踪记录""" url = f"{BASE_URL}/traces" params = {"limit": 5} response = requests.get(url, params=params, headers=headers) if response.status_code == 200: traces = response.json().get('data', []) print(f"获取到 {len(traces)} 条追踪:") for t in traces: print(f" - {t.get('name')} (ID: {t.get('id')})") else: print(f"查询失败: {response.status_code}") def create_dataset_and_run_evaluation(): """演示如何创建数据集并关联评估（概念性步骤）""" # 1. 创建数据集 dataset_payload = { "name": "Customer_Service_QA", "description": "用于评估客服机器人效果的数据集" } # 2. 向数据集中添加样本（这里需要具体的 item 结构，请参考官方API文档） # 3. 执行批量评估（通常需要结合 SDK 或自定义脚本循环处理数据集中的每个样本） print("批量评估流程涉及多个API调用，建议结合SDK或查看官方文档。") if __name__ == "__main__": trace_id = create_trace_via_api() if trace_id: # 可以基于 trace_id 进行评分等后续操作 pass get_traces()

6.2 批量评估任务设计

批量评估是 Langfuse 的核心优势。典型流程如下：

1. 创建数据集 (Dataset)：在 Langfuse UI 中或通过 API，创建一个数据集，并上传一批测试用例（例如{“input”: “用户问题”, “expected_output”: “期望答案”}）。
2. 编写评估函数 (Evaluation Function)：定义一个 Python 函数，它接受一个测试用例，调用你的 LLM 应用，并返回一个或多个评分（如正确性、相关性、流畅度）。这个函数内部应使用 Langfuse SDK 进行追踪。
3. 执行批量运行：遍历数据集中的所有项目，对每个项目执行评估函数。Langfuse 会自动为每次运行创建追踪，并将结果与数据集项目关联。
4. 分析与比较：在 Langfuse UI 的 “Dataset” 页面，你可以看到所有测试用例的运行结果、评分对比。你可以快速识别出哪些问题你的应用处理得不好，从而针对性优化。

这种模式将评估从一次性、手动的活动，转变为可重复、可量化的自动化流程。

7. 资源占用与性能观察

由于 Langfuse 本身不运行大模型，其资源消耗主要来自 Web 服务、后端处理和数据库。

自托管版资源占用观察：

启动 Langfuse 服务后，你可以使用docker stats命令来查看容器资源使用情况：

docker stats --no-stream

在典型的小规模开发或测试场景下（日追踪量在万条以内），你可能会观察到：

• CPU：三个容器合计占用约 1-5%，大部分时间空闲。
• 内存：langfuse-server和langfuse-frontend各占用约 200-500 MB，langfuse-db(PostgreSQL) 占用约 100-300 MB。总计约 1GB 左右。
• 磁盘：镜像本身约 1GB。数据库增长取决于你存储的追踪数据量。每条追踪记录（包含所有 Spans, Generations）根据复杂度可能占用几 KB 到几十 KB。

性能影响因素：

1. SDK 集成模式：使用@observe装饰器或CallbackHandler对应用本身的性能影响极低（微秒级）。手动插桩的trace/span调用也主要是网络 I/O。
2. 网络延迟：SDK 默认是异步发送数据到 Langfuse 后端，不会阻塞你的主应用线程。但如果后端地址 (LANGFUSE_HOST) 网络不通或延迟很高，可能会在后台线程中产生错误或重试，不影响主流程但可能丢失数据。
3. 数据库压力：如果产生海量追踪数据（例如每秒上千条），PostgreSQL 可能成为瓶颈。建议定期清理旧数据，或升级数据库配置。
4. 前端响应：当单次查询需要渲染成千上万条追踪时，Web 界面可能会变慢。合理使用过滤器和分页。

优化建议：

• 生产环境部署：考虑将langfuse-db的数据卷挂载到高性能 SSD 上。
• 数据保留策略：在设置中配置自动删除旧追踪数据（如保留 30 天），或定期手动清理。
• 监控 Langfuse 自身：可以为你的 Langfuse 服务也设置基础监控（如容器健康检查、日志收集）。

8. 常见问题与排查方法

在部署和使用过程中，你可能会遇到以下问题。这里提供系统的排查思路。

通用排查命令：

# 查看 Langfuse 容器日志 docker compose logs -f langfuse-server docker compose logs -f langfuse-frontend # 重启所有服务 docker compose down docker compose up -d # 进入数据库容器执行查询（高级） docker compose exec langfuse-db psql -U postgres -d postgres

9. 最佳实践与使用建议

为了让 Langfuse 发挥最大价值，并避免常见陷阱，遵循以下最佳实践：

1. 从关键路径开始，逐步扩大追踪范围：不要一开始就在所有函数上加装饰器。先在你的核心 LLM 调用链或最复杂的智能体流程上集成，看到价值后再逐步扩展到其他部分。
2. 为追踪和会话 (Session) 设置清晰的标识：利用trace_id,session_id,user_id等字段。这能让你在 UI 中轻松过滤和查询特定用户或会话的所有交互，对于分析用户体验至关重要。
3. 善用元数据 (Metadata) 和标签 (Tags)：在创建 Trace 或 Span 时，添加有业务意义的元数据，如{“environment”: “staging”, “app_version”: “1.2.0”, “feature_flag”: “new_prompt”}。这让你能对比不同版本或配置下的应用表现。
4. 设计有意义的评估指标 (Scores)：不要只用一个“好坏”评分。针对你的场景设计多维度的评估体系。例如：
   • 事实准确性：针对 RAG 系统，评估答案是否基于提供的上下文，且事实正确。
   • 相关性：答案是否直接回答了问题。
   • 有害性：内容是否安全。
   • 风格符合度：语气、格式是否符合要求。
   • 成本与延迟：作为客观指标进行监控。
5. 建立自动化评估流水线：将你的测试数据集和评估函数脚本化，并集成到 CI/CD 流程中。每次代码或提示词更新后，自动运行评估，对比关键指标的变化，防止回归。
6. 定期审查生产环境追踪：每周或每月抽检一些生产环境的失败或低分追踪。这能帮你发现意料之外的模型行为、边缘案例或提示词缺陷。
7. 注意数据安全与合规：
   • 自托管保障数据主权：对数据敏感的项目，始终选择自托管。
   • 避免记录敏感信息：在 SDK 中配置redacted_keys，自动脱敏追踪中的密码、密钥等字段。
   • 设置访问控制：利用 Langfuse 的项目和成员管理功能，控制团队成员的数据访问权限。

10. 总结与下一步

Langfuse 将一个复杂的工程问题——LLM 应用的可观测性与评估——变成了一个可以系统化解决的方案。通过本次实战，你应该已经掌握了从零部署、集成 SDK、进行复杂追踪到设计批量评估的完整流程。

最值得尝试的下一步：

1. 将你现有的一个 LangChain 或 LlamaIndex 项目集成 Langfuse：使用CallbackHandler，这是集成最快的方式。亲眼看看一个 RAG 问答的完整检索、生成链条被可视化出来，你会立刻感受到它的价值。
2. 创建一个包含 20-30 个问题的测试数据集：涵盖你应用的典型用例和常见失败案例。运行一次批量评估，找出当前系统的薄弱环节。
3. 探索生产监控：如果你有线上应用，以低采样率（例如 1%）开启 Langfuse 追踪，监控 API 成本、延迟和错误率。设置简单的告警（如成本日环比增长超 20%）。

最容易踩的坑：

• 忽略异步发送机制：在短时运行的脚本中，数据可能因程序提前退出而丢失，记得调用flush()。
• 混淆 Host 地址：自托管时，SDK 配置的host是 Langfuse 后端地址（如http://localhost:3000），而不是前端地址。
• 过度追踪：追踪所有细节会产生大量数据，可能拖慢 UI 并增加存储成本。聚焦于核心业务逻辑。

将这个平台作为你 LLM 应用开发的“仪表盘”，持续观察、测量、实验和优化。当你能清晰地看到每一次调用、每一分成本、每一个评分时，优化方向就不再是猜测，而是数据驱动的决策。建议收藏本文，在集成和排查时作为参考。