在实际的大模型应用开发中我们经常面临一个核心挑战如何系统性地追踪、评估和优化智能体Agent或LLM应用的性能模型输出不稳定、成本不可控、调试过程黑盒化是开发者最常遇到的痛点。Langfuse 作为一个开源的 LLM 观测平台正是为了解决这些问题而生。它提供了从数据收集、追踪、评估到可视化的完整链路让开发者能够像观测传统软件系统一样观测大模型应用。本文面向正在或计划构建基于大模型的智能体、RAG系统或复杂链式应用的开发者。我们将从一个实战项目出发带你从零开始基于 Langfuse 平台完成对一个智能体应用的追踪、调试和评估优化。你将学会如何搭建观测环境如何将你的应用与 Langfuse 集成如何定义和运行评估指标并最终通过数据驱动的方式优化你的模型应用。整个过程将覆盖从环境准备、代码集成、数据追踪到评估分析的全流程确保你能够将理论转化为可落地的工程实践。1. 理解 Langfuse 的核心概念与工作机制在开始动手之前我们需要先厘清 Langfuse 是什么以及它如何帮助我们解决大模型应用开发中的观测难题。1.1 Langfuse 是什么它能解决什么问题Langfuse 是一个专为 LLM 应用设计的开源可观测性平台。你可以把它理解为 LLM 世界的“APM”应用性能监控工具。传统软件开发中我们有日志、指标和链路追踪来监控服务状态。但对于大模型应用一次调用可能涉及多次模型调用、工具调用、上下文检索等复杂步骤传统的监控手段难以清晰地描绘出完整的执行链路和成本构成。Langfuse 的核心价值在于端到端追踪自动记录一次用户请求例如一个智能体问答背后所有步骤的细节包括每次 LLM 调用、工具执行、提示词、输入输出、耗时、Token 消耗和成本。集中化调试在一个统一的 UI 界面中回放和分析每次请求的完整执行过程快速定位问题步骤比如是提示词问题、工具调用失败还是模型生成不佳。量化评估与对比通过定义评估指标如相关性、正确性、有害性并批量运行测试量化评估不同模型、不同提示词版本或不同参数配置下的应用表现为优化提供数据依据。成本与性能分析清晰展示每个项目、每个模型、每个用户的 Token 消耗和成本帮助进行预算控制和性能优化。1.2 Langfuse 的核心组件与数据流理解 Langfuse 的架构有助于我们更好地使用它。其核心组件包括SDK/集成你需要在你的大模型应用代码中集成 Langfuse SDK支持 Python、JS/TS 等。SDK 负责收集追踪数据并发送到 Langfuse 后端。Langfuse Server这是数据存储和处理的核心可以自托管Docker或使用其云服务。它接收 SDK 发送的数据并存储在数据库中。Langfuse UI一个 Web 界面用于查看追踪数据、调试单次请求、创建数据集、运行评估和查看分析报告。数据流非常简单你的应用运行时SDK 会将追踪事件Trace, Span, Generation, Event发送到 Langfuse Server随后你便可以在 Langfuse UI 中查看和分析这些数据。1.3 关键术语Trace, Span, Generation为了有效使用 Langfuse必须理解其数据模型中的几个关键概念Trace代表一次最高级别的操作或工作流。例如处理一次用户查询的完整智能体会话。一个 Trace 包含多个子步骤。Span代表 Trace 中的一个逻辑单元或步骤。例如在 RAG 流程中“检索文档”和“生成答案”可以分别是两个 Span。Span 可以嵌套。Generation一种特殊的 Span专用于记录一次 LLM 的调用。它包含了模型名称、输入提示词、输出结果、Token 使用量、耗时和成本等详细信息。通过这种层级结构Langfuse 能够清晰地展示复杂工作流的执行树这是进行有效调试和评估的基础。2. 环境准备与 Langfuse 部署我们将采用自托管的方式部署 Langfuse以便完全掌控数据。生产环境也可以选择其云服务。2.1 基础环境要求确保你的开发环境满足以下要求操作系统Linux, macOS 或 WSL2 (Windows)。Docker 与 Docker Compose这是部署 Langfuse 最简单的方式。请确保已安装最新稳定版。# 检查 Docker 和 Docker Compose 版本 docker --version docker-compose --versionPython 环境我们将用 Python 编写示例智能体应用。推荐使用 Python 3.10 和虚拟环境如 venv 或 conda。大模型 API 密钥示例中将使用 OpenAI API你需要准备一个有效的OPENAI_API_KEY。2.2 部署 Langfuse ServerLangfuse 官方提供了 Docker Compose 配置文件可以一键启动所有依赖服务包括 Postgres 数据库。下载配置文件# 创建一个项目目录 mkdir langfuse-agent-eval cd langfuse-agent-eval # 下载官方的 docker-compose.yml 文件 curl -o docker-compose.yml https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml可选配置环境变量编辑docker-compose.yml你可以修改数据库密码、服务端口等。对于初次体验使用默认配置即可。注意文件中 Langfuse 服务的端口映射3000:3000。启动服务docker-compose up -d这个命令会在后台启动 Postgres 和 Langfuse 服务。首次启动会拉取镜像并初始化数据库可能需要一两分钟。验证部署访问http://localhost:3000如果修改了端口请对应调整。你应该能看到 Langfuse 的注册/登录界面。首次使用需要创建一个账号。登录后进入 Dashboard界面应该是空的因为我们还没有发送任何数据。注意自托管版本的数据将持久化在本地 Docker 卷中。生产环境部署需要考虑数据备份、服务高可用、网络安全性如配置反向代理和 HTTPS以及资源限制。2.3 创建 Langfuse 项目与获取密钥在 Langfuse UI 中我们需要创建一个项目来接收来自我们智能体应用的数据。登录 Langfuse UI (http://localhost:3000)。点击页面上的 “Create new project”输入项目名称例如Agent-Eval-Demo。项目创建成功后进入项目设置Project Settings。在 “API Keys” 部分你会看到PUBLIC_KEY和SECRET_KEY。记录下它们。同时页面上会显示你的 Langfuse Server 地址LANGFUSE_HOST自托管情况下就是http://localhost:3000或你的服务器地址。这三个值LANGFUSE_HOST,PUBLIC_KEY,SECRET_KEY是 SDK 连接服务器的凭证我们将在下一步的代码中用到。3. 构建一个可追踪的简易智能体应用为了演示评估流程我们需要一个被观测的对象。让我们构建一个简单的“研究助手”智能体它的功能是根据用户提出的学术概念先通过工具模拟搜索相关论文然后总结核心观点。3.1 项目初始化与依赖安装在我们的项目目录下创建 Python 虚拟环境并安装依赖。# 在项目根目录 (langfuse-agent-eval) 下 python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # .\venv\Scripts\activate # 安装必要的包 pip install langfuse openai python-dotenvlangfuse: Langfuse 的 Python SDK。openai: OpenAI 官方库用于调用 GPT 模型。python-dotenv: 用于从.env文件加载环境变量。3.2 配置环境变量与 Langfuse 客户端创建.env文件来安全地存储密钥# .env 文件内容 OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_HOSThttp://localhost:3000 LANGFUSE_PUBLIC_KEYyour-langfuse-public-key LANGFUSE_SECRET_KEYyour-langfuse-secret-key创建app.py并初始化 Langfuse 客户端和 OpenAI 客户端# app.py import os from dotenv import load_dotenv from langfuse import Langfuse from openai import OpenAI # 加载环境变量 load_dotenv() # 初始化 Langfuse 客户端 langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) # 初始化 OpenAI 客户端 openai_client OpenAI(api_keyos.getenv(OPENAI_API_KEY))3.3 实现智能体工作流并集成追踪我们将使用 Langfuse SDK 的装饰器observe()或langfuse.trace来轻松地包装函数实现自动追踪。这里我们模拟一个包含两个步骤的工作流搜索和总结。# app.py (续) from langfuse.decorators import observe import time import random # 模拟一个搜索工具 observe() # 使用装饰器自动创建 Span def search_papers(topic: str): 模拟搜索学术论文返回模拟结果 time.sleep(0.5) # 模拟网络延迟 # 模拟返回一些论文标题和摘要 mock_papers [ f《{topic}的理论基础研究综述》, f基于深度学习的{topic}应用进展, f{topic}在跨学科领域中的挑战与机遇 ] return mock_papers # 核心的智能体函数使用 langfuse.trace 包装整个流程 langfuse.trace(nameresearch_assistant_agent) def research_assistant_agent(user_query: str): 研究助手智能体主函数。 1. 搜索相关论文 2. 调用 LLM 总结核心观点 print(f用户查询: {user_query}) # 步骤1: 搜索论文 papers search_papers(user_query) print(f搜索到论文: {papers}) # 构建提示词 prompt f 用户想了解的概念是{user_query} 以下是一些相关的学术论文标题 {chr(10).join(papers)} 请基于这些信息用简洁的语言总结关于“{user_query}”的2-3个核心学术观点或研究方向。 # 步骤2: 调用 LLM 生成总结 # 使用 langfuse.generation 来详细记录这次 LLM 调用 with langfuse.generation(namesummarize_with_gpt, modelgpt-3.5-turbo) as generation: response openai_client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.7, max_tokens300 ) summary response.choices[0].message.content # 将输入输出和元数据记录到 generation 中 generation.input prompt generation.output summary generation.metadata { model: response.model, usage_tokens: response.usage.total_tokens, temperature: 0.7 } print(f生成总结: {summary}) return summary # 主程序入口 if __name__ __main__: # 模拟处理一个用户查询 query 联邦学习 result research_assistant_agent(query) print(\n智能体处理完成。) # 确保所有追踪数据被发送 langfuse.flush()关键代码解释langfuse.trace(name...): 这个装饰器将整个research_assistant_agent函数包装为一个Trace。所有函数内部通过 Langfuse SDK 记录的活动如search_papers的 Span 和generation都会成为这个 Trace 的子节点。observe(): 装饰在search_papers函数上会自动将其执行记录为一个Span包含开始时间、结束时间和任何输出的记录。with langfuse.generation(...) as generation:: 这是一个上下文管理器专门用于记录GenerationLLM调用。我们在其中执行 OpenAI 调用并在调用结束后将输入、输出和模型元数据如 Token 使用量赋值给generation对象。这是记录成本和分析模型性能的关键。langfuse.flush(): SDK 默认是异步批量发送数据以提高性能。flush()会强制将缓存的所有数据立即发送到服务器在脚本结束时调用确保数据不丢失。3.4 运行应用并验证数据上报运行我们的智能体应用python app.py如果一切正常控制台会输出搜索和总结的结果。此时打开 Langfuse UI (http://localhost:3000)进入你的项目。在Dashboard或Traces页面你应该能看到一条新的 Trace名称是research_assistant_agent。点击这条 Trace进入详情页。你会看到一个清晰的执行树根节点是research_assistant_agent(Trace)。其下有一个search_papers(Span)。再其下有一个summarize_with_gpt(Generation)。点击summarize_with_gpt右侧面板会展示这次 LLM 调用的所有细节完整的提示词Input、模型回复Output、使用的模型、Token 数量、耗时和计算出的成本。至此我们已经成功将一个简单的智能体应用与 Langfuse 集成并实现了端到端的执行追踪。4. 在 Langfuse 中评估与优化智能体追踪和调试是第一步而系统性的评估则是优化迭代的指南针。Langfuse 提供了强大的评估功能允许我们定义评估标准对多次运行结果进行批量和自动化评分。4.1 创建数据集Dataset评估需要有输入和期望输出Ground Truth作为基准。我们首先在 Langfuse UI 中创建一个数据集。在 Langfuse UI 左侧导航栏点击“Dataset”。点击“Create new dataset”命名为Research-Agent-Test。点击创建好的数据集进入详情页。这里我们可以手动添加测试用例也可以从已有的 Trace 中导入。点击“Add item”我们手动添加两个测试用例Item 1:Input:{query: 联邦学习}Expected Output:一个关于联邦学习的核心观点总结应包含隐私保护、分布式训练等关键词。Item 2:Input:{query: 注意力机制}Expected Output:一个关于注意力机制的核心观点总结应提及其在Transformer中的关键作用。注意Expected Output 可以是一个精确答案也可以是一个描述性的评估标准。在复杂场景下后者更实用。4.2 批量运行测试并生成 Traces接下来我们需要编写一个脚本读取数据集中的输入运行我们的智能体并将结果Trace关联到这个数据集项。这样每次运行都成为了一个可评估的“实验”。创建run_evaluation.py脚本# run_evaluation.py import os from dotenv import load_dotenv from langfuse import Langfuse from app import research_assistant_agent # 导入我们之前写的智能体 import json load_dotenv() langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) # 定义我们的测试用例在实际中可以从 Langfuse API 或 UI 导出 dataset_items [ { id: item_1, # 可以自定义一个ID input: {query: 联邦学习}, expected_output: 一个关于联邦学习的核心观点总结应包含隐私保护、分布式训练等关键词。 }, { id: item_2, input: {query: 注意力机制}, expected_output: 一个关于注意力机制的核心观点总结应提及其在Transformer中的关键作用。 } ] for item in dataset_items: print(f\n处理测试用例: {item[input][query]}) # 关键在 trace 中关联 dataset item id trace langfuse.trace( namefeval_{item[input][query]}, inputitem[input], metadata{dataset_item_id: item[id]} # 关联数据集项 ) # 运行智能体并将本次执行关联到上面创建的 trace with trace.generation(nameagent_execution): result research_assistant_agent(item[input][query]) # 可以将实际输出也记录到 trace 的 output 字段 trace.output result print(f结果已记录Trace ID: {trace.id}) langfuse.flush() print(\n评估运行完成请在 Langfuse UI 的 Dataset 页面查看结果。)运行此脚本python run_evaluation.py。完成后回到 Langfuse UI。进入Dataset-Research-Agent-Test。你会看到两个数据集项每个项下面现在都关联了一次运行Trace。点击任意一个项下的 “View Trace”可以跳转到该次执行的详细追踪页面。4.3 定义并执行评估Scores现在我们需要为每次运行打分。评估可以是手动的也可以是自动的通过代码调用 LLM 或规则判断。我们演示两种方式。方式一在 UI 中手动评分在 Trace 详情页找到output字段附近或页面上的 “Add Score” 按钮。你可以添加一个分数例如Name:relevance(相关性)Value:8(0-10分)Comment: “总结涵盖了隐私和分布式但未提及通信效率。”方式二通过 SDK 自动评分编程化评估我们可以编写一个评估函数调用另一个 LLM或使用规则来给输出打分。创建auto_evaluate.py# auto_evaluate.py import os from dotenv import load_dotenv from langfuse import Langfuse from openai import OpenAI import json load_dotenv() langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) openai_client OpenAI() def evaluate_with_llm(trace_id, query, expected, actual): 使用 GPT-4 作为评估器对输出进行评分 evaluation_prompt f 你是一个严谨的评估专家。请评估以下AI助手的回答质量。 用户问题{query} 期望回答应具备的特征{expected} 助手实际回答{actual} 请从“相关性”回答是否紧扣问题和“完整性”是否覆盖期望特征两个维度评分每个维度满分10分。 只返回一个JSON对象格式如{{relevance: 8, completeness: 7, comment: ...}} try: response openai_client.chat.completions.create( modelgpt-4, messages[{role: user, content: evaluation_prompt}], temperature0, max_tokens200 ) eval_result json.loads(response.choices[0].message.content) # 将评分提交到 Langfuse langfuse.score( trace_idtrace_id, namerelevance, valueeval_result[relevance], commenteval_result.get(comment, ) ) langfuse.score( trace_idtrace_id, namecompleteness, valueeval_result[completeness], commenteval_result.get(comment, ) ) print(fTrace {trace_id} 评估完成: {eval_result}) except Exception as e: print(f评估 Trace {trace_id} 时出错: {e}) # 假设我们获取到需要评估的 Trace ID 列表这里手动模拟 # 实际应用中可以通过 Langfuse API 获取特定数据集的 Trace trace_ids_to_evaluate [trace_id_1, trace_id_2] # 替换为实际的 Trace ID for tid in trace_ids_to_evaluate: # 这里需要根据 trace_id 查询到对应的 input 和 output # 简化演示假设我们已经有了这些数据 sample_query 联邦学习 sample_expected 应包含隐私保护、分布式训练等关键词。 sample_actual 联邦学习是一种分布式机器学习框架能在保护数据隐私的前提下进行模型训练。 evaluate_with_llm(tid, sample_query, sample_expected, sample_actual) langfuse.flush()4.4 分析与优化基于数据做出决策完成评分后Langfuse 的分析能力就派上用场了。在 Trace 列表筛选和排序在 Traces 页面你可以根据分数如relevance 7进行筛选快速找到表现不佳的案例进行调试。对比不同版本如果你修改了提示词例如在提示词中明确要求“列出三点”并重新运行了评估脚本生成了新的 Traces。你可以在 Dataset 页面直接对比同一个问题在不同 Trace 下的输出、耗时和成本直观看到优化效果。查看成本分析在Analytics或Dashboard页面Langfuse 提供了按模型、按时间、按项目统计的 Token 消耗和成本图表。如果你发现gpt-4成本过高但gpt-3.5-turbo分数相差不大这就是一个明确的优化方向在非关键步骤降级模型。定位性能瓶颈通过分析 Trace 中各个 Span 和 Generation 的耗时可以轻易发现是检索工具慢还是 LLM 生成慢从而有针对性地进行优化如缓存、并行化、模型选择。5. 生产环境最佳实践与常见问题排查将 Langfuse 用于生产环境时需要考虑更多工程化因素。5.1 生产环境部署与配置清单事项学习/开发环境生产环境建议部署方式Docker Compose 单机Kubernetes 集群部署或使用 Langfuse Cloud数据持久化Docker 卷配置持久化存储并建立定期备份机制网络与安全localhost 访问配置域名、HTTPS (SSL/TLS)、防火墙规则、访问控制列表 (ACL)认证简单账号密码建议集成 SSO (如 OAuth 2.0)SDK 集成同步flush()使用异步 SDK并配置合理的批处理大小和发送间隔错误处理简单打印日志SDK 集成应具备重试和降级逻辑避免影响主业务数据采样全量记录配置采样率在高流量下只记录部分 Trace 以控制成本和负载5.2 集成与代码层面的最佳实践为 Trace 和 Span 设置有意义的名称不要使用默认或模糊的名称。使用如rag_retrieval、customer_support_agent、sql_generation等能清晰反映业务逻辑的名称。丰富 Metadata 和 Tags利用metadata和tags字段记录业务上下文如用户ID、会话ID、功能模块、模型版本、提示词版本等。这为后续的筛选和分析提供了巨大便利。trace langfuse.trace( nameprocess_order, metadata{user_id: 123, order_id: 456, app_version: 2.1.0}, tags[production, checkout_flow] )分离敏感信息确保不会将密码、密钥、个人身份信息PII记录到input/output或metadata中。可以在 Langfuse Server 配置数据脱敏规则或在 SDK 发送前进行清洗。监控 Langfuse 服务本身监控其健康状态、资源使用率和队列长度确保观测系统自身稳定可靠。5.3 常见问题排查清单当 Langfuse 没有按预期工作时可以按照以下清单进行排查问题现象可能原因检查与解决步骤UI 中看不到 Trace1. SDK 未正确初始化或密钥错误。2. 数据未发送异步队列未刷新。3. 网络不通。1. 检查.env文件中的LANGFUSE_HOST,PUBLIC_KEY,SECRET_KEY是否正确。2. 在代码末尾调用langfuse.flush()并等待片刻。3. 检查 SDK 日志设置LANGFUSE_DEBUGtrue环境变量。4. 使用curl测试LANGFUSE_HOST的网络连通性。Trace 数据不完整1. 代码异常导致 Trace/Span 未正常结束。2. 在异步代码中Trace 上下文丢失。1. 使用try...finally确保trace.update()或上下文管理器退出被调用。2. 在异步框架中确保正确传递 Langfuse 的上下文。参考官方异步集成文档。Generation 中没有成本信息1. 使用的模型不在 Langfuse 内置定价列表中。2. 未正确设置generation.model参数。1. 检查generation.model字段是否设置为准确的模型标识符如gpt-4-0125-preview。2. 对于自定义或本地模型需要在 Langfuse 项目设置中手动配置单价。评估分数未显示1.trace_id在评分时填写错误。2. 评分 API 调用失败。1. 确认评分时使用的trace_id与 UI 中看到的完全一致。2. 检查评分 API 的返回值或错误日志。UI 加载缓慢1. 追踪数据量过大。2. 服务器资源不足。1. 考虑为 Trace 设置合理的保留策略TTL。2. 在生产环境确保为数据库和 Langfuse 服务分配足够的 CPU 和内存。5.4 扩展方向从观测到自动化运维当你熟练使用 Langfuse 的基础功能后可以探索更高级的用法构建更强大的 LLM 运维体系告警集成通过 Langfuse 的 webhook 或查询 API监控特定错误模式、成本超阈值或评估分数下降并发送告警到 Slack、钉钉或 PagerDuty。自动化评估流水线将run_evaluation.py和auto_evaluate.py脚本集成到 CI/CD 流程中每次代码或提示词变更后自动运行回归测试确保核心指标不下降。A/B 测试与实验管理利用metadata中的version字段标记不同的提示词或模型版本在 Langfuse UI 中轻松对比不同实验组的平均响应时间、成本和评估分数实现数据驱动的决策。深入分析将 Langfuse 数据导出到你的数据仓库如 Snowflake, BigQuery与业务数据结合进行更深层次的关联分析例如分析不同用户群体的回答满意度。通过本指南你不仅学会了如何部署和集成 Langfuse更重要的是掌握了构建可观测、可评估、可优化的大模型应用的系统方法。真正的优化始于测量现在你可以用数据代替直觉来驱动你的智能体迭代了。
基于Langfuse的LLM应用可观测性实战：从追踪到评估优化
在实际的大模型应用开发中我们经常面临一个核心挑战如何系统性地追踪、评估和优化智能体Agent或LLM应用的性能模型输出不稳定、成本不可控、调试过程黑盒化是开发者最常遇到的痛点。Langfuse 作为一个开源的 LLM 观测平台正是为了解决这些问题而生。它提供了从数据收集、追踪、评估到可视化的完整链路让开发者能够像观测传统软件系统一样观测大模型应用。本文面向正在或计划构建基于大模型的智能体、RAG系统或复杂链式应用的开发者。我们将从一个实战项目出发带你从零开始基于 Langfuse 平台完成对一个智能体应用的追踪、调试和评估优化。你将学会如何搭建观测环境如何将你的应用与 Langfuse 集成如何定义和运行评估指标并最终通过数据驱动的方式优化你的模型应用。整个过程将覆盖从环境准备、代码集成、数据追踪到评估分析的全流程确保你能够将理论转化为可落地的工程实践。1. 理解 Langfuse 的核心概念与工作机制在开始动手之前我们需要先厘清 Langfuse 是什么以及它如何帮助我们解决大模型应用开发中的观测难题。1.1 Langfuse 是什么它能解决什么问题Langfuse 是一个专为 LLM 应用设计的开源可观测性平台。你可以把它理解为 LLM 世界的“APM”应用性能监控工具。传统软件开发中我们有日志、指标和链路追踪来监控服务状态。但对于大模型应用一次调用可能涉及多次模型调用、工具调用、上下文检索等复杂步骤传统的监控手段难以清晰地描绘出完整的执行链路和成本构成。Langfuse 的核心价值在于端到端追踪自动记录一次用户请求例如一个智能体问答背后所有步骤的细节包括每次 LLM 调用、工具执行、提示词、输入输出、耗时、Token 消耗和成本。集中化调试在一个统一的 UI 界面中回放和分析每次请求的完整执行过程快速定位问题步骤比如是提示词问题、工具调用失败还是模型生成不佳。量化评估与对比通过定义评估指标如相关性、正确性、有害性并批量运行测试量化评估不同模型、不同提示词版本或不同参数配置下的应用表现为优化提供数据依据。成本与性能分析清晰展示每个项目、每个模型、每个用户的 Token 消耗和成本帮助进行预算控制和性能优化。1.2 Langfuse 的核心组件与数据流理解 Langfuse 的架构有助于我们更好地使用它。其核心组件包括SDK/集成你需要在你的大模型应用代码中集成 Langfuse SDK支持 Python、JS/TS 等。SDK 负责收集追踪数据并发送到 Langfuse 后端。Langfuse Server这是数据存储和处理的核心可以自托管Docker或使用其云服务。它接收 SDK 发送的数据并存储在数据库中。Langfuse UI一个 Web 界面用于查看追踪数据、调试单次请求、创建数据集、运行评估和查看分析报告。数据流非常简单你的应用运行时SDK 会将追踪事件Trace, Span, Generation, Event发送到 Langfuse Server随后你便可以在 Langfuse UI 中查看和分析这些数据。1.3 关键术语Trace, Span, Generation为了有效使用 Langfuse必须理解其数据模型中的几个关键概念Trace代表一次最高级别的操作或工作流。例如处理一次用户查询的完整智能体会话。一个 Trace 包含多个子步骤。Span代表 Trace 中的一个逻辑单元或步骤。例如在 RAG 流程中“检索文档”和“生成答案”可以分别是两个 Span。Span 可以嵌套。Generation一种特殊的 Span专用于记录一次 LLM 的调用。它包含了模型名称、输入提示词、输出结果、Token 使用量、耗时和成本等详细信息。通过这种层级结构Langfuse 能够清晰地展示复杂工作流的执行树这是进行有效调试和评估的基础。2. 环境准备与 Langfuse 部署我们将采用自托管的方式部署 Langfuse以便完全掌控数据。生产环境也可以选择其云服务。2.1 基础环境要求确保你的开发环境满足以下要求操作系统Linux, macOS 或 WSL2 (Windows)。Docker 与 Docker Compose这是部署 Langfuse 最简单的方式。请确保已安装最新稳定版。# 检查 Docker 和 Docker Compose 版本 docker --version docker-compose --versionPython 环境我们将用 Python 编写示例智能体应用。推荐使用 Python 3.10 和虚拟环境如 venv 或 conda。大模型 API 密钥示例中将使用 OpenAI API你需要准备一个有效的OPENAI_API_KEY。2.2 部署 Langfuse ServerLangfuse 官方提供了 Docker Compose 配置文件可以一键启动所有依赖服务包括 Postgres 数据库。下载配置文件# 创建一个项目目录 mkdir langfuse-agent-eval cd langfuse-agent-eval # 下载官方的 docker-compose.yml 文件 curl -o docker-compose.yml https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml可选配置环境变量编辑docker-compose.yml你可以修改数据库密码、服务端口等。对于初次体验使用默认配置即可。注意文件中 Langfuse 服务的端口映射3000:3000。启动服务docker-compose up -d这个命令会在后台启动 Postgres 和 Langfuse 服务。首次启动会拉取镜像并初始化数据库可能需要一两分钟。验证部署访问http://localhost:3000如果修改了端口请对应调整。你应该能看到 Langfuse 的注册/登录界面。首次使用需要创建一个账号。登录后进入 Dashboard界面应该是空的因为我们还没有发送任何数据。注意自托管版本的数据将持久化在本地 Docker 卷中。生产环境部署需要考虑数据备份、服务高可用、网络安全性如配置反向代理和 HTTPS以及资源限制。2.3 创建 Langfuse 项目与获取密钥在 Langfuse UI 中我们需要创建一个项目来接收来自我们智能体应用的数据。登录 Langfuse UI (http://localhost:3000)。点击页面上的 “Create new project”输入项目名称例如Agent-Eval-Demo。项目创建成功后进入项目设置Project Settings。在 “API Keys” 部分你会看到PUBLIC_KEY和SECRET_KEY。记录下它们。同时页面上会显示你的 Langfuse Server 地址LANGFUSE_HOST自托管情况下就是http://localhost:3000或你的服务器地址。这三个值LANGFUSE_HOST,PUBLIC_KEY,SECRET_KEY是 SDK 连接服务器的凭证我们将在下一步的代码中用到。3. 构建一个可追踪的简易智能体应用为了演示评估流程我们需要一个被观测的对象。让我们构建一个简单的“研究助手”智能体它的功能是根据用户提出的学术概念先通过工具模拟搜索相关论文然后总结核心观点。3.1 项目初始化与依赖安装在我们的项目目录下创建 Python 虚拟环境并安装依赖。# 在项目根目录 (langfuse-agent-eval) 下 python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # .\venv\Scripts\activate # 安装必要的包 pip install langfuse openai python-dotenvlangfuse: Langfuse 的 Python SDK。openai: OpenAI 官方库用于调用 GPT 模型。python-dotenv: 用于从.env文件加载环境变量。3.2 配置环境变量与 Langfuse 客户端创建.env文件来安全地存储密钥# .env 文件内容 OPENAI_API_KEYsk-your-openai-api-key-here LANGFUSE_HOSThttp://localhost:3000 LANGFUSE_PUBLIC_KEYyour-langfuse-public-key LANGFUSE_SECRET_KEYyour-langfuse-secret-key创建app.py并初始化 Langfuse 客户端和 OpenAI 客户端# app.py import os from dotenv import load_dotenv from langfuse import Langfuse from openai import OpenAI # 加载环境变量 load_dotenv() # 初始化 Langfuse 客户端 langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) # 初始化 OpenAI 客户端 openai_client OpenAI(api_keyos.getenv(OPENAI_API_KEY))3.3 实现智能体工作流并集成追踪我们将使用 Langfuse SDK 的装饰器observe()或langfuse.trace来轻松地包装函数实现自动追踪。这里我们模拟一个包含两个步骤的工作流搜索和总结。# app.py (续) from langfuse.decorators import observe import time import random # 模拟一个搜索工具 observe() # 使用装饰器自动创建 Span def search_papers(topic: str): 模拟搜索学术论文返回模拟结果 time.sleep(0.5) # 模拟网络延迟 # 模拟返回一些论文标题和摘要 mock_papers [ f《{topic}的理论基础研究综述》, f基于深度学习的{topic}应用进展, f{topic}在跨学科领域中的挑战与机遇 ] return mock_papers # 核心的智能体函数使用 langfuse.trace 包装整个流程 langfuse.trace(nameresearch_assistant_agent) def research_assistant_agent(user_query: str): 研究助手智能体主函数。 1. 搜索相关论文 2. 调用 LLM 总结核心观点 print(f用户查询: {user_query}) # 步骤1: 搜索论文 papers search_papers(user_query) print(f搜索到论文: {papers}) # 构建提示词 prompt f 用户想了解的概念是{user_query} 以下是一些相关的学术论文标题 {chr(10).join(papers)} 请基于这些信息用简洁的语言总结关于“{user_query}”的2-3个核心学术观点或研究方向。 # 步骤2: 调用 LLM 生成总结 # 使用 langfuse.generation 来详细记录这次 LLM 调用 with langfuse.generation(namesummarize_with_gpt, modelgpt-3.5-turbo) as generation: response openai_client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.7, max_tokens300 ) summary response.choices[0].message.content # 将输入输出和元数据记录到 generation 中 generation.input prompt generation.output summary generation.metadata { model: response.model, usage_tokens: response.usage.total_tokens, temperature: 0.7 } print(f生成总结: {summary}) return summary # 主程序入口 if __name__ __main__: # 模拟处理一个用户查询 query 联邦学习 result research_assistant_agent(query) print(\n智能体处理完成。) # 确保所有追踪数据被发送 langfuse.flush()关键代码解释langfuse.trace(name...): 这个装饰器将整个research_assistant_agent函数包装为一个Trace。所有函数内部通过 Langfuse SDK 记录的活动如search_papers的 Span 和generation都会成为这个 Trace 的子节点。observe(): 装饰在search_papers函数上会自动将其执行记录为一个Span包含开始时间、结束时间和任何输出的记录。with langfuse.generation(...) as generation:: 这是一个上下文管理器专门用于记录GenerationLLM调用。我们在其中执行 OpenAI 调用并在调用结束后将输入、输出和模型元数据如 Token 使用量赋值给generation对象。这是记录成本和分析模型性能的关键。langfuse.flush(): SDK 默认是异步批量发送数据以提高性能。flush()会强制将缓存的所有数据立即发送到服务器在脚本结束时调用确保数据不丢失。3.4 运行应用并验证数据上报运行我们的智能体应用python app.py如果一切正常控制台会输出搜索和总结的结果。此时打开 Langfuse UI (http://localhost:3000)进入你的项目。在Dashboard或Traces页面你应该能看到一条新的 Trace名称是research_assistant_agent。点击这条 Trace进入详情页。你会看到一个清晰的执行树根节点是research_assistant_agent(Trace)。其下有一个search_papers(Span)。再其下有一个summarize_with_gpt(Generation)。点击summarize_with_gpt右侧面板会展示这次 LLM 调用的所有细节完整的提示词Input、模型回复Output、使用的模型、Token 数量、耗时和计算出的成本。至此我们已经成功将一个简单的智能体应用与 Langfuse 集成并实现了端到端的执行追踪。4. 在 Langfuse 中评估与优化智能体追踪和调试是第一步而系统性的评估则是优化迭代的指南针。Langfuse 提供了强大的评估功能允许我们定义评估标准对多次运行结果进行批量和自动化评分。4.1 创建数据集Dataset评估需要有输入和期望输出Ground Truth作为基准。我们首先在 Langfuse UI 中创建一个数据集。在 Langfuse UI 左侧导航栏点击“Dataset”。点击“Create new dataset”命名为Research-Agent-Test。点击创建好的数据集进入详情页。这里我们可以手动添加测试用例也可以从已有的 Trace 中导入。点击“Add item”我们手动添加两个测试用例Item 1:Input:{query: 联邦学习}Expected Output:一个关于联邦学习的核心观点总结应包含隐私保护、分布式训练等关键词。Item 2:Input:{query: 注意力机制}Expected Output:一个关于注意力机制的核心观点总结应提及其在Transformer中的关键作用。注意Expected Output 可以是一个精确答案也可以是一个描述性的评估标准。在复杂场景下后者更实用。4.2 批量运行测试并生成 Traces接下来我们需要编写一个脚本读取数据集中的输入运行我们的智能体并将结果Trace关联到这个数据集项。这样每次运行都成为了一个可评估的“实验”。创建run_evaluation.py脚本# run_evaluation.py import os from dotenv import load_dotenv from langfuse import Langfuse from app import research_assistant_agent # 导入我们之前写的智能体 import json load_dotenv() langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) # 定义我们的测试用例在实际中可以从 Langfuse API 或 UI 导出 dataset_items [ { id: item_1, # 可以自定义一个ID input: {query: 联邦学习}, expected_output: 一个关于联邦学习的核心观点总结应包含隐私保护、分布式训练等关键词。 }, { id: item_2, input: {query: 注意力机制}, expected_output: 一个关于注意力机制的核心观点总结应提及其在Transformer中的关键作用。 } ] for item in dataset_items: print(f\n处理测试用例: {item[input][query]}) # 关键在 trace 中关联 dataset item id trace langfuse.trace( namefeval_{item[input][query]}, inputitem[input], metadata{dataset_item_id: item[id]} # 关联数据集项 ) # 运行智能体并将本次执行关联到上面创建的 trace with trace.generation(nameagent_execution): result research_assistant_agent(item[input][query]) # 可以将实际输出也记录到 trace 的 output 字段 trace.output result print(f结果已记录Trace ID: {trace.id}) langfuse.flush() print(\n评估运行完成请在 Langfuse UI 的 Dataset 页面查看结果。)运行此脚本python run_evaluation.py。完成后回到 Langfuse UI。进入Dataset-Research-Agent-Test。你会看到两个数据集项每个项下面现在都关联了一次运行Trace。点击任意一个项下的 “View Trace”可以跳转到该次执行的详细追踪页面。4.3 定义并执行评估Scores现在我们需要为每次运行打分。评估可以是手动的也可以是自动的通过代码调用 LLM 或规则判断。我们演示两种方式。方式一在 UI 中手动评分在 Trace 详情页找到output字段附近或页面上的 “Add Score” 按钮。你可以添加一个分数例如Name:relevance(相关性)Value:8(0-10分)Comment: “总结涵盖了隐私和分布式但未提及通信效率。”方式二通过 SDK 自动评分编程化评估我们可以编写一个评估函数调用另一个 LLM或使用规则来给输出打分。创建auto_evaluate.py# auto_evaluate.py import os from dotenv import load_dotenv from langfuse import Langfuse from openai import OpenAI import json load_dotenv() langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) openai_client OpenAI() def evaluate_with_llm(trace_id, query, expected, actual): 使用 GPT-4 作为评估器对输出进行评分 evaluation_prompt f 你是一个严谨的评估专家。请评估以下AI助手的回答质量。 用户问题{query} 期望回答应具备的特征{expected} 助手实际回答{actual} 请从“相关性”回答是否紧扣问题和“完整性”是否覆盖期望特征两个维度评分每个维度满分10分。 只返回一个JSON对象格式如{{relevance: 8, completeness: 7, comment: ...}} try: response openai_client.chat.completions.create( modelgpt-4, messages[{role: user, content: evaluation_prompt}], temperature0, max_tokens200 ) eval_result json.loads(response.choices[0].message.content) # 将评分提交到 Langfuse langfuse.score( trace_idtrace_id, namerelevance, valueeval_result[relevance], commenteval_result.get(comment, ) ) langfuse.score( trace_idtrace_id, namecompleteness, valueeval_result[completeness], commenteval_result.get(comment, ) ) print(fTrace {trace_id} 评估完成: {eval_result}) except Exception as e: print(f评估 Trace {trace_id} 时出错: {e}) # 假设我们获取到需要评估的 Trace ID 列表这里手动模拟 # 实际应用中可以通过 Langfuse API 获取特定数据集的 Trace trace_ids_to_evaluate [trace_id_1, trace_id_2] # 替换为实际的 Trace ID for tid in trace_ids_to_evaluate: # 这里需要根据 trace_id 查询到对应的 input 和 output # 简化演示假设我们已经有了这些数据 sample_query 联邦学习 sample_expected 应包含隐私保护、分布式训练等关键词。 sample_actual 联邦学习是一种分布式机器学习框架能在保护数据隐私的前提下进行模型训练。 evaluate_with_llm(tid, sample_query, sample_expected, sample_actual) langfuse.flush()4.4 分析与优化基于数据做出决策完成评分后Langfuse 的分析能力就派上用场了。在 Trace 列表筛选和排序在 Traces 页面你可以根据分数如relevance 7进行筛选快速找到表现不佳的案例进行调试。对比不同版本如果你修改了提示词例如在提示词中明确要求“列出三点”并重新运行了评估脚本生成了新的 Traces。你可以在 Dataset 页面直接对比同一个问题在不同 Trace 下的输出、耗时和成本直观看到优化效果。查看成本分析在Analytics或Dashboard页面Langfuse 提供了按模型、按时间、按项目统计的 Token 消耗和成本图表。如果你发现gpt-4成本过高但gpt-3.5-turbo分数相差不大这就是一个明确的优化方向在非关键步骤降级模型。定位性能瓶颈通过分析 Trace 中各个 Span 和 Generation 的耗时可以轻易发现是检索工具慢还是 LLM 生成慢从而有针对性地进行优化如缓存、并行化、模型选择。5. 生产环境最佳实践与常见问题排查将 Langfuse 用于生产环境时需要考虑更多工程化因素。5.1 生产环境部署与配置清单事项学习/开发环境生产环境建议部署方式Docker Compose 单机Kubernetes 集群部署或使用 Langfuse Cloud数据持久化Docker 卷配置持久化存储并建立定期备份机制网络与安全localhost 访问配置域名、HTTPS (SSL/TLS)、防火墙规则、访问控制列表 (ACL)认证简单账号密码建议集成 SSO (如 OAuth 2.0)SDK 集成同步flush()使用异步 SDK并配置合理的批处理大小和发送间隔错误处理简单打印日志SDK 集成应具备重试和降级逻辑避免影响主业务数据采样全量记录配置采样率在高流量下只记录部分 Trace 以控制成本和负载5.2 集成与代码层面的最佳实践为 Trace 和 Span 设置有意义的名称不要使用默认或模糊的名称。使用如rag_retrieval、customer_support_agent、sql_generation等能清晰反映业务逻辑的名称。丰富 Metadata 和 Tags利用metadata和tags字段记录业务上下文如用户ID、会话ID、功能模块、模型版本、提示词版本等。这为后续的筛选和分析提供了巨大便利。trace langfuse.trace( nameprocess_order, metadata{user_id: 123, order_id: 456, app_version: 2.1.0}, tags[production, checkout_flow] )分离敏感信息确保不会将密码、密钥、个人身份信息PII记录到input/output或metadata中。可以在 Langfuse Server 配置数据脱敏规则或在 SDK 发送前进行清洗。监控 Langfuse 服务本身监控其健康状态、资源使用率和队列长度确保观测系统自身稳定可靠。5.3 常见问题排查清单当 Langfuse 没有按预期工作时可以按照以下清单进行排查问题现象可能原因检查与解决步骤UI 中看不到 Trace1. SDK 未正确初始化或密钥错误。2. 数据未发送异步队列未刷新。3. 网络不通。1. 检查.env文件中的LANGFUSE_HOST,PUBLIC_KEY,SECRET_KEY是否正确。2. 在代码末尾调用langfuse.flush()并等待片刻。3. 检查 SDK 日志设置LANGFUSE_DEBUGtrue环境变量。4. 使用curl测试LANGFUSE_HOST的网络连通性。Trace 数据不完整1. 代码异常导致 Trace/Span 未正常结束。2. 在异步代码中Trace 上下文丢失。1. 使用try...finally确保trace.update()或上下文管理器退出被调用。2. 在异步框架中确保正确传递 Langfuse 的上下文。参考官方异步集成文档。Generation 中没有成本信息1. 使用的模型不在 Langfuse 内置定价列表中。2. 未正确设置generation.model参数。1. 检查generation.model字段是否设置为准确的模型标识符如gpt-4-0125-preview。2. 对于自定义或本地模型需要在 Langfuse 项目设置中手动配置单价。评估分数未显示1.trace_id在评分时填写错误。2. 评分 API 调用失败。1. 确认评分时使用的trace_id与 UI 中看到的完全一致。2. 检查评分 API 的返回值或错误日志。UI 加载缓慢1. 追踪数据量过大。2. 服务器资源不足。1. 考虑为 Trace 设置合理的保留策略TTL。2. 在生产环境确保为数据库和 Langfuse 服务分配足够的 CPU 和内存。5.4 扩展方向从观测到自动化运维当你熟练使用 Langfuse 的基础功能后可以探索更高级的用法构建更强大的 LLM 运维体系告警集成通过 Langfuse 的 webhook 或查询 API监控特定错误模式、成本超阈值或评估分数下降并发送告警到 Slack、钉钉或 PagerDuty。自动化评估流水线将run_evaluation.py和auto_evaluate.py脚本集成到 CI/CD 流程中每次代码或提示词变更后自动运行回归测试确保核心指标不下降。A/B 测试与实验管理利用metadata中的version字段标记不同的提示词或模型版本在 Langfuse UI 中轻松对比不同实验组的平均响应时间、成本和评估分数实现数据驱动的决策。深入分析将 Langfuse 数据导出到你的数据仓库如 Snowflake, BigQuery与业务数据结合进行更深层次的关联分析例如分析不同用户群体的回答满意度。通过本指南你不仅学会了如何部署和集成 Langfuse更重要的是掌握了构建可观测、可评估、可优化的大模型应用的系统方法。真正的优化始于测量现在你可以用数据代替直觉来驱动你的智能体迭代了。
