你开发了一个基于大模型的智能客服助手上线后用户反馈时好时坏。有的用户夸它“回答专业、反应迅速”有的用户却抱怨“答非所问、逻辑混乱”。作为开发者你看着后台的日志和API调用记录却像面对一个黑盒你只知道它调用了多少次API、花了多少钱但完全不知道它内部到底“想”了什么为什么同一个问题在不同时间会给出不同质量的答案。这种“盲人摸象”的困境正是当前AI应用开发尤其是智能体Agent开发中最普遍的痛点。我们投入大量精力设计提示词、编排工作流、集成工具链但最终应用的表现却难以量化、难以调试、难以持续优化。传统的日志监控只能告诉你“发生了什么”却无法解释“为什么发生”。今天要介绍的Langfuse正是为解决这一核心痛点而生。它不是一个简单的日志工具而是一个专为大模型应用设计的可观测性Observability平台。你可以把它理解为AI应用开发的“X光机”和“仪表盘”它能让你清晰地看到智能体内部的每一次思考、每一次工具调用、每一次成本消耗并将这些数据转化为可评估、可优化的具体指标。本文将带你从零开始基于Langfuse平台实战构建一套完整的智能体评估与观测体系。这不是一个简单的工具介绍而是一个贯穿追踪Tracing、调试Debugging、评估Evaluation、优化Optimization全流程的工程指南。读完本文你将能彻底告别黑盒为你的AI应用装上“眼睛”清晰洞察每一次请求的完整生命周期。建立量化评估体系告别主观感觉用数据指标成本、延迟、质量客观衡量智能体表现。高效定位与修复问题快速定位导致回答质量下降的环节是提示词问题还是工具调用错误。实现持续迭代优化基于观测数据科学地优化提示词、调整工作流、控制成本。我们从一个最经典的智能体场景——联网搜索问答助手开始一步步拆解如何利用Langfuse将其从一个“不可控的黑盒”变成一个“透明、可度量、可优化的系统”。1. 为什么你需要Langfuse智能体开发从“炼丹”到“工程”的跨越在深入代码之前我们必须先理解问题的本质。为什么传统的开发监控手段在AI应用面前失效了想象一下你开发一个传统的微服务。一个请求进来经过鉴权、业务逻辑、数据库查询、返回结果。每个环节都有清晰的输入输出、状态码和错误信息。你可以轻松地通过链路追踪如Jaeger看到请求流经了哪些服务通过日志如ELK看到关键信息通过指标如Prometheus监控吞吐量和延迟。然而当你开发一个基于大模型的智能体时情况完全不同过程非确定性同样的输入大模型可能产生不同的输出受温度temperature、随机种子等参数影响。内部状态复杂一个智能体的执行可能包含多轮思考Chain-of-Thought、多次工具调用如搜索、计算、以及自我反思Self-Reflection。这些复杂的内部状态传统的print日志或简单日志框架难以清晰、结构化地记录。评估主观性强回答的“好坏”没有像HTTP状态码那样明确的标准需要结合业务场景设计评估指标如相关性、事实准确性、有害性。成本敏感每一次API调用都直接产生费用不同模型、不同输入输出长度成本差异巨大。你需要精确知道钱花在了哪里哪些环节可以优化以降低成本。Langfuse的核心价值就是为AI应用这个特殊领域重建了“可观测性”的三支柱链路追踪Traces、日志Logs、指标Metrics并将其与评估Evaluation深度集成。它通过一个简单的SDK以非侵入或低侵入的方式自动捕获你应用中大模型调用、工具执行、用户反馈等所有关键事件并以清晰的可视化界面呈现。这让你能从“凭感觉和运气调整提示词”的“炼丹”模式转向“基于数据驱动决策”的工程化开发模式。2. Langfuse核心概念全景图理解观测的维度开始实战前我们需要统一语言。Langfuse围绕几个核心概念构建理解它们就理解了整个观测体系。概念通俗解释类比在智能体中的作用Trace追踪一次完整的AI应用请求生命周期。一次完整的“患者就诊流程”挂号、问诊、检查、开药、离院。代表用户的一次提问到获得最终答案的完整过程。Span跨度Trace中的一个具体操作步骤。就诊流程中的“医生问诊”或“抽血检查”环节。可以是“生成提示词”、“调用大模型”、“执行搜索工具”、“解析结果”等具体步骤。Generation生成一个特殊类型的Span特指调用大模型LLM生成内容的操作。“医生根据检查单思考并撰写诊断书”。记录LLM的输入提示词、输出、使用的模型、token消耗、耗时等核心元数据。Event事件在Trace中标记的一个有意义的时间点。在就诊流程中标记“开始缴费”、“取药完成”。可用于标记“用户收到答案”、“开始工具调用”等关键节点。Observation观测Span和Generation的统称。即所有被记录的操作。所有被记录的医疗环节。-Score评分对Trace或Observation的人工或自动评价。患者对“就诊态度”或“治疗效果”的打分。可以是用户反馈的“ thumbs up/down”也可以是自动评估模型对“答案相关性”的打分0-5分。一个典型的智能体Trace结构如下一次用户问答 (Trace) ├── 接收用户问题 (Event) ├── 规划思考步骤 (Span) ├── 调用搜索工具 (Span) │ └── 执行搜索API (Generation) ├── 合成最终提示词 (Span) ├── 调用LLM生成答案 (Generation) └── 返回答案给用户 (Event)Langfuse的UI会以时间线或树形结构直观展示这个Trace让你一目了然。3. 环境准备三分钟搭建观测平台Langfuse提供了云托管版和自托管版。对于个人学习和中小型项目强烈建议直接从云托管版开始它免费额度充足免去运维烦恼。我们以云托管版为例。3.1 注册与项目创建访问 Langfuse 官网 并注册账号。登录后点击 “Create new project”输入项目名称例如AI-Customer-Support-Agent。创建成功后进入项目设置在API Keys页面你会看到三组关键凭证LANGFUSE_PUBLIC_KEYLANGFUSE_SECRET_KEYLANGFUSE_HOST(云托管版通常是https://cloud.langfuse.com)请妥善保存这些信息我们将在代码中使用。3.2 安装SDKLangfuse支持Python、Node.js等多种语言。本文以Python为例。确保你的Python环境在3.8以上。# 安装Langfuse Python SDK pip install langfuse3.3 初始化客户端在你的应用入口文件如app.py或agent.py中初始化Langfuse客户端。# file: observability/setup.py from langfuse import Langfuse # 从环境变量读取密钥这是安全的最佳实践 import os langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST, https://cloud.langfuse.com) # 默认云主机 ) # 测试连接 try: langfuse.auth_check() print(✅ Langfuse 客户端初始化成功连接正常。) except Exception as e: print(f❌ Langfuse 连接失败: {e})将你的密钥设置为环境变量# Linux/macOS export LANGFUSE_PUBLIC_KEYpk-lf-... export LANGFUSE_SECRET_KEYsk-lf-... export LANGFUSE_HOSThttps://cloud.langfuse.com # Windows (PowerShell) $env:LANGFUSE_PUBLIC_KEYpk-lf-... $env:LANGFUSE_SECRET_KEYsk-lf-... $env:LANGFUSE_HOSThttps://cloud.langfuse.com安全提醒永远不要将密钥硬编码在代码中或提交到版本控制系统如Git。4. 实战为联网搜索问答智能体注入可观测性现在我们构建一个简单的智能体用户提问智能体自动联网搜索并基于搜索结果生成答案。我们将使用langchain框架和Tavily搜索工具并用Langfuse进行全链路追踪。4.1 基础智能体实现无观测首先看看一个没有观测的“黑盒”版本# file: agent/vanilla_agent.py from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain.agents import initialize_agent, AgentType from langchain.memory import ConversationBufferMemory import os # 1. 设置API密钥示例请使用环境变量 os.environ[OPENAI_API_KEY] your-openai-key os.environ[TAVILY_API_KEY] your-tavily-key # 2. 初始化组件 llm ChatOpenAI(modelgpt-4o-mini, temperature0) search_tool TavilySearchResults(max_results3) memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) # 3. 创建智能体 agent initialize_agent( tools[search_tool], llmllm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 支持对话和工具调用 memorymemory, verboseTrue, # LangChain自带的简单日志信息有限 ) # 4. 运行 question 2024年巴黎奥运会中国代表团获得了多少枚金牌 result agent.run(question) print(result)这个版本能工作但verboseTrue只能输出简单的文本日志无法结构化存储、无法可视化、无法进行后续分析评估。4.2 集成Langfuse进行追踪现在我们使用langfuse的LangchainCallbackHandler来无缝集成追踪。# file: agent/observed_agent.py from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain.agents import initialize_agent, AgentType from langchain.memory import ConversationBufferMemory from langfuse.callback import CallbackHandler # Langfuse的回调处理器 import os # 初始化Langfuse回调处理器每个请求一个独立的handler def create_agent_with_observation(user_input: str): langfuse_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) llm ChatOpenAI(modelgpt-4o-mini, temperature0, callbacks[langfuse_handler]) # 将handler注入LLM search_tool TavilySearchResults(max_results3) memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent initialize_agent( tools[search_tool], llmllm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, memorymemory, callbacks[langfuse_handler], # 将handler注入Agent verboseFalse, # 可以关闭LangChain自带的verbose避免日志干扰 ) # 运行前可以为本次Trace设置一些属性便于后续筛选 langfuse_handler.langfuse.trace.update( nameWeb-Search-QA, user_iduser_123, inputuser_input, # 记录原始输入 metadata{environment: testing, agent_version: 1.0} ) try: result agent.run(user_input) # 运行成功后更新Trace的输出和状态 langfuse_handler.langfuse.trace.update(outputresult) return result except Exception as e: # 如果运行失败记录错误信息 langfuse_handler.langfuse.trace.update(status_messagestr(e)) raise e finally: # 确保所有数据被刷新到Langfuse服务器 langfuse_handler.flush() # 执行智能体 if __name__ __main__: question 2024年巴黎奥运会中国代表团获得了多少枚金牌 answer create_agent_with_observation(question) print(f问题: {question}) print(f答案: {answer})执行这段代码后打开Langfuse项目的Dashboard你就能看到一次完整的Trace记录。4.3 在Langfuse UI中查看追踪结果进入你的Langfuse项目。点击左侧菜单的Traces。你会看到一条名为“Web-Search-QA”的Trace。点击进入详情页。在这里你将获得前所未有的洞察力时间线视图清晰展示整个请求的耗时分布一眼看出是LLM调用慢还是工具执行慢。树形结构视图展开Trace你能看到agent.run下包含的所有子步骤LLM步骤记录了发送给模型的完整提示词Prompt和模型返回的完整回答Completion以及使用的模型、消耗的Token、耗时。Tool步骤记录了搜索工具的执行详情包括搜索查询词和返回的原始搜索结果。Chain步骤展示了LangChain内部的决策逻辑。输入输出在Trace顶部直接看到用户的原始问题input和智能体的最终答案output。元数据可以看到我们设置的user_id,environment等信息方便过滤和分组查询。至此你的智能体已经实现了基础的可观测性。但这只是开始Langfuse更强大的能力在于评估。5. 核心进阶构建自动化评估体系观测是为了评估评估是为了优化。Langfuse允许你为Trace或其中的某个步骤Observation打上“分数”Score。这个分数可以来自人工反馈用户在UI上点“赞/踩”或提供一个1-5分的评分。模型自动评估用另一个LLM如GPT-4作为“裁判”根据既定标准相关性、事实性、有害性等对输出进行评分。我们重点讲解自动化评估这是实现规模化评估和持续集成的关键。5.1 设计评估标准与提示词假设我们要评估智能体答案的事实准确性Factual Correctness。我们设计一个“裁判”LLM的提示词# file: evaluation/prompts.py FACTUALITY_EVALUATION_PROMPT 你是一个严谨的事实核查员。请根据提供的“参考信息”来评估“待评估答案”的事实准确性。 参考信息 {reference} 待评估答案 {answer} 请从以下维度进行评分1-5分 1分答案与参考信息完全矛盾或包含关键事实错误。 2分答案部分错误或遗漏了关键信息。 3分答案基本正确但有一些不精确或模糊的表述。 4分答案准确清晰传达了参考信息中的关键事实。 5分答案完全准确、精炼并且可能补充了相关的上下文。 请只输出一个JSON对象格式如下 {{ score: 1-5的整数, reason: 简要的评分理由指出具体正确或错误之处 }} 5.2 实现自动化评估并记录Score我们在智能体生成答案后自动调用评估流程并将结果记录到Langfuse中。# file: evaluation/auto_evaluator.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser import json from langfuse.callback import CallbackHandler import os def evaluate_factuality_and_score(trace_id: str, reference: str, answer: str): 自动化评估答案的事实准确性并将评分记录到指定的Trace上。 :param trace_id: Langfuse中本次对话的Trace ID :param reference: 用于比对的参考信息如搜索得到的事实片段 :param answer: 待评估的智能体答案 # 初始化一个独立的Langfuse handler用于评估环节 eval_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) # 初始化“裁判”LLM judge_llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 创建评估链 prompt_template ChatPromptTemplate.from_template(FACTUALITY_EVALUATION_PROMPT) evaluation_chain prompt_template | judge_llm | StrOutputParser() try: # 执行评估 eval_result_str evaluation_chain.invoke( {reference: reference, answer: answer}, config{callbacks: [eval_handler]} # 评估过程本身也可被追踪 ) eval_result json.loads(eval_result_str) # 将评估结果作为Score记录到原始的Trace上 # 注意这里需要通过Langfuse Python SDK直接创建Score from langfuse import Langfuse langfuse_client Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) langfuse_client.score( trace_idtrace_id, # 关联到原始Trace namefactual_correctness, # 评分名称 valueeval_result[score], # 分数值 commenteval_result[reason], # 评分理由 observation_idNone, # 可以为Trace整体打分也可以为某个具体Generation打分 ) print(f✅ 自动化评估完成。事实准确性得分: {eval_result[score]}/5) return eval_result except json.JSONDecodeError: print(❌ 评估LLM返回了非JSON格式结果。) return None finally: eval_handler.flush()5.3 改造智能体集成自动化评估现在我们需要修改智能体使其在获取搜索结果并生成答案后自动调用评估函数。# file: agent/agent_with_eval.py # ... (前面的导入和初始化代码与 observed_agent.py 类似) ... from your_project.evaluation.auto_evaluator import evaluate_factuality_and_score def create_agent_with_observation_and_eval(user_input: str): langfuse_handler CallbackHandler(...) # 初始化handler # ... 初始化llm, tool, memory, agent ... (与之前相同) # 关键在创建Trace后获取其ID langfuse_handler.langfuse.trace.update(nameWeb-Search-QA-Eval, inputuser_input) current_trace_id langfuse_handler.get_trace_id() # 获取当前Trace的ID try: # 为了评估我们需要捕获工具搜索返回的原始结果作为“参考信息” # 这里需要自定义Agent的执行来获取中间结果简化示例 result agent.run(user_input) # 假设我们通过其他方式拿到了本次执行中的搜索原始结果 search_raw_data # 在实际中你可能需要自定义Callback或解析Trace来获取。 search_raw_data 2024巴黎奥运会中国代表团获得40枚金牌27枚银牌24枚铜牌... # 示例数据 # 记录最终输出 langfuse_handler.langfuse.trace.update(outputresult) # 核心执行自动化评估并记录Score evaluate_factuality_and_score( trace_idcurrent_trace_id, referencesearch_raw_data, answerresult ) return result except Exception as e: langfuse_handler.langfuse.trace.update(status_messagestr(e)) raise e finally: langfuse_handler.flush()运行这个智能体后在Langfuse UI中查看Trace你会在详情页的“Scores”部分看到一个名为factual_correctness的评分。点击图表图标你还可以看到所有Trace的评分分布。6. 从数据到洞察利用Langfuse Dashboard驱动优化收集了追踪和评估数据后Langfuse的Dashboard是你的作战指挥中心。6.1 核心数据分析面板Traces Table列表显示所有请求支持按时间、属性如user_id、评分、耗时、成本等筛选和排序。快速定位表现不佳的对话。Analytics可视化图表。延迟分析查看LLM调用、工具调用的P50、P95、P99延迟发现性能瓶颈。成本分析按模型、按天统计Token消耗和费用需配置模型价格精准控制预算。评分趋势观察factual_correctness等评分随时间的变化验证优化措施是否有效。Prompt Management重量级功能。自动从Traces中提取出使用过的提示词版本并进行对比测试A/B测试。你可以直接修改提示词并基于历史对话数据批量回测用数据选择最优提示词。6.2 典型优化工作流发现问题在Dashboard中看到最近一小时factual_correctness平均分从4.2降到了3.5。定位问题筛选出评分低的Traces点击进入详情。发现低分答案都源于一个特定的搜索工具调用返回了过时信息。提出假设可能是搜索查询词不够精确或者需要过滤更近期的结果。实施优化修改提示词在Langfuse的Prompt管理中找到Agent中用于生成搜索查询的提示词创建一个优化版本如加入“请使用2024年后的信息”。修改工具参数在代码中调整TavilySearchResults的max_results或search_depth。测试验证单条测试在Langfuse的Playground中直接用新旧提示词对比测试几个典型问题。批量回测使用Prompt Management的批量评测功能用历史问题集自动运行并对比新旧版本的评分。部署与监控将验证有效的优化部署上线并在Dashboard中持续监控评分和成本指标。7. 常见问题与排查指南在集成和使用Langfuse过程中你可能会遇到以下问题问题现象可能原因排查步骤解决方案Trace数据没有出现在Langfuse UI中1. API密钥错误或权限不足。2. 网络问题导致数据发送失败。3. 代码中未执行handler.flush()或程序提前退出。1. 检查环境变量LANGFUSE_PUBLIC_KEY和LANGFUSE_SECRET_KEY是否正确。2. 在代码中添加try-catch捕获初始化异常。3. 确保在程序结束前调用了flush()方法。1. 在Langfuse项目设置中重新生成密钥。2. 使用langfuse.auth_check()测试连接。3. 将flush()放在finally块中。Trace树形结构中缺少某些步骤如工具调用使用的框架或库可能没有被Langfuse的回调处理器完全覆盖。1. 检查是否将所有必要的对象llm,agent,tools都传入了callbacks[handler]参数。2. 查看LangChain等框架的文档确认其回调支持度。1. 确保初始化时在所有可能产生事件的地方传入callback handler。2. 考虑使用Langfuse的装饰器observe()手动包装自定义函数。自动化评估LLM调用失败或返回格式错误1. 评估提示词设计有歧义导致LLM未返回合规JSON。2. 评估LLM的API密钥或网络有问题。1. 打印出评估LLM的原始输出检查是否符合JSON格式。2. 简化评估提示词加入更严格的格式指令。1. 在评估提示词中强化输出格式要求例如使用“必须输出JSON”等字眼。2. 在代码中添加json.loads的异常处理并设置默认评分。Dashboard中成本数据显示为0未正确配置模型定价信息。进入Langfuse项目设置检查Model Prices配置。在设置中为使用的模型如gpt-4o-mini添加对应的输入/输出每百万Token价格。数据来自OpenAI官网。追踪数据延迟高影响应用性能SDK默认是异步批量上报数据通常不影响应用性能。如果担心可能是网络延迟。检查应用所在网络到cloud.langfuse.com的连接。1. Langfuse SDK是异步的主流程不会阻塞等待网络IO。2. 对于极高并发场景可考虑调整SDK的批量发送参数或使用自托管版部署在内网。8. 生产环境最佳实践当你准备将集成了Langfuse的智能体部署到生产环境时请遵循以下建议密钥管理使用安全的秘密管理服务如AWS Secrets Manager, HashiCorp Vault或云平台的环境变量注入功能切勿硬编码。采样率控制在生产环境中可能不需要记录100%的请求。Langfuse SDK支持设置采样率只记录一部分请求以控制成本和数据量。langfuse_handler CallbackHandler( ..., sample_rate0.1, # 只记录10%的请求 )区分环境为开发、测试、生产环境创建不同的Langfuse项目并使用不同的API密钥。通过metadata或Trace的name字段区分环境。用户隐私与数据安全避免在input、output或metadata中记录个人身份信息PII、密码等敏感数据。Langfuse支持数据脱敏规则。自定义追踪不要局限于框架自动追踪的内容。使用observe()装饰器或SDK的trace、span方法手动记录业务关键节点。from langfuse.decorators import observe observe() # 自动将此函数记录为一个Span def complex_business_logic(data): # ... 你的业务逻辑 ... return result建立评估基线在项目启动初期就定义好核心评估指标如事实准确性、响应速度、成本并收集一段时间的数据建立基线。后续所有优化都应与基线对比。告警集成利用Langfuse的webhook功能或定期查询API当关键指标如平均评分骤降、成本异常升高出现异常时发送告警到你的监控系统如Slack, PagerDuty。通过将Langfuse深度集成到你的AI应用开发流程中你构建的将不再是一个神秘的黑盒而是一个透明、可度量、可持续进化的智能系统。从追踪每一次调用开始到建立全面的评估体系最终实现数据驱动的快速迭代这正是AI工程化落地的关键一步。 Langfuse实战：构建智能体可观测性，告别AI应用黑盒调试
你开发了一个基于大模型的智能客服助手上线后用户反馈时好时坏。有的用户夸它“回答专业、反应迅速”有的用户却抱怨“答非所问、逻辑混乱”。作为开发者你看着后台的日志和API调用记录却像面对一个黑盒你只知道它调用了多少次API、花了多少钱但完全不知道它内部到底“想”了什么为什么同一个问题在不同时间会给出不同质量的答案。这种“盲人摸象”的困境正是当前AI应用开发尤其是智能体Agent开发中最普遍的痛点。我们投入大量精力设计提示词、编排工作流、集成工具链但最终应用的表现却难以量化、难以调试、难以持续优化。传统的日志监控只能告诉你“发生了什么”却无法解释“为什么发生”。今天要介绍的Langfuse正是为解决这一核心痛点而生。它不是一个简单的日志工具而是一个专为大模型应用设计的可观测性Observability平台。你可以把它理解为AI应用开发的“X光机”和“仪表盘”它能让你清晰地看到智能体内部的每一次思考、每一次工具调用、每一次成本消耗并将这些数据转化为可评估、可优化的具体指标。本文将带你从零开始基于Langfuse平台实战构建一套完整的智能体评估与观测体系。这不是一个简单的工具介绍而是一个贯穿追踪Tracing、调试Debugging、评估Evaluation、优化Optimization全流程的工程指南。读完本文你将能彻底告别黑盒为你的AI应用装上“眼睛”清晰洞察每一次请求的完整生命周期。建立量化评估体系告别主观感觉用数据指标成本、延迟、质量客观衡量智能体表现。高效定位与修复问题快速定位导致回答质量下降的环节是提示词问题还是工具调用错误。实现持续迭代优化基于观测数据科学地优化提示词、调整工作流、控制成本。我们从一个最经典的智能体场景——联网搜索问答助手开始一步步拆解如何利用Langfuse将其从一个“不可控的黑盒”变成一个“透明、可度量、可优化的系统”。1. 为什么你需要Langfuse智能体开发从“炼丹”到“工程”的跨越在深入代码之前我们必须先理解问题的本质。为什么传统的开发监控手段在AI应用面前失效了想象一下你开发一个传统的微服务。一个请求进来经过鉴权、业务逻辑、数据库查询、返回结果。每个环节都有清晰的输入输出、状态码和错误信息。你可以轻松地通过链路追踪如Jaeger看到请求流经了哪些服务通过日志如ELK看到关键信息通过指标如Prometheus监控吞吐量和延迟。然而当你开发一个基于大模型的智能体时情况完全不同过程非确定性同样的输入大模型可能产生不同的输出受温度temperature、随机种子等参数影响。内部状态复杂一个智能体的执行可能包含多轮思考Chain-of-Thought、多次工具调用如搜索、计算、以及自我反思Self-Reflection。这些复杂的内部状态传统的print日志或简单日志框架难以清晰、结构化地记录。评估主观性强回答的“好坏”没有像HTTP状态码那样明确的标准需要结合业务场景设计评估指标如相关性、事实准确性、有害性。成本敏感每一次API调用都直接产生费用不同模型、不同输入输出长度成本差异巨大。你需要精确知道钱花在了哪里哪些环节可以优化以降低成本。Langfuse的核心价值就是为AI应用这个特殊领域重建了“可观测性”的三支柱链路追踪Traces、日志Logs、指标Metrics并将其与评估Evaluation深度集成。它通过一个简单的SDK以非侵入或低侵入的方式自动捕获你应用中大模型调用、工具执行、用户反馈等所有关键事件并以清晰的可视化界面呈现。这让你能从“凭感觉和运气调整提示词”的“炼丹”模式转向“基于数据驱动决策”的工程化开发模式。2. Langfuse核心概念全景图理解观测的维度开始实战前我们需要统一语言。Langfuse围绕几个核心概念构建理解它们就理解了整个观测体系。概念通俗解释类比在智能体中的作用Trace追踪一次完整的AI应用请求生命周期。一次完整的“患者就诊流程”挂号、问诊、检查、开药、离院。代表用户的一次提问到获得最终答案的完整过程。Span跨度Trace中的一个具体操作步骤。就诊流程中的“医生问诊”或“抽血检查”环节。可以是“生成提示词”、“调用大模型”、“执行搜索工具”、“解析结果”等具体步骤。Generation生成一个特殊类型的Span特指调用大模型LLM生成内容的操作。“医生根据检查单思考并撰写诊断书”。记录LLM的输入提示词、输出、使用的模型、token消耗、耗时等核心元数据。Event事件在Trace中标记的一个有意义的时间点。在就诊流程中标记“开始缴费”、“取药完成”。可用于标记“用户收到答案”、“开始工具调用”等关键节点。Observation观测Span和Generation的统称。即所有被记录的操作。所有被记录的医疗环节。-Score评分对Trace或Observation的人工或自动评价。患者对“就诊态度”或“治疗效果”的打分。可以是用户反馈的“ thumbs up/down”也可以是自动评估模型对“答案相关性”的打分0-5分。一个典型的智能体Trace结构如下一次用户问答 (Trace) ├── 接收用户问题 (Event) ├── 规划思考步骤 (Span) ├── 调用搜索工具 (Span) │ └── 执行搜索API (Generation) ├── 合成最终提示词 (Span) ├── 调用LLM生成答案 (Generation) └── 返回答案给用户 (Event)Langfuse的UI会以时间线或树形结构直观展示这个Trace让你一目了然。3. 环境准备三分钟搭建观测平台Langfuse提供了云托管版和自托管版。对于个人学习和中小型项目强烈建议直接从云托管版开始它免费额度充足免去运维烦恼。我们以云托管版为例。3.1 注册与项目创建访问 Langfuse 官网 并注册账号。登录后点击 “Create new project”输入项目名称例如AI-Customer-Support-Agent。创建成功后进入项目设置在API Keys页面你会看到三组关键凭证LANGFUSE_PUBLIC_KEYLANGFUSE_SECRET_KEYLANGFUSE_HOST(云托管版通常是https://cloud.langfuse.com)请妥善保存这些信息我们将在代码中使用。3.2 安装SDKLangfuse支持Python、Node.js等多种语言。本文以Python为例。确保你的Python环境在3.8以上。# 安装Langfuse Python SDK pip install langfuse3.3 初始化客户端在你的应用入口文件如app.py或agent.py中初始化Langfuse客户端。# file: observability/setup.py from langfuse import Langfuse # 从环境变量读取密钥这是安全的最佳实践 import os langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST, https://cloud.langfuse.com) # 默认云主机 ) # 测试连接 try: langfuse.auth_check() print(✅ Langfuse 客户端初始化成功连接正常。) except Exception as e: print(f❌ Langfuse 连接失败: {e})将你的密钥设置为环境变量# Linux/macOS export LANGFUSE_PUBLIC_KEYpk-lf-... export LANGFUSE_SECRET_KEYsk-lf-... export LANGFUSE_HOSThttps://cloud.langfuse.com # Windows (PowerShell) $env:LANGFUSE_PUBLIC_KEYpk-lf-... $env:LANGFUSE_SECRET_KEYsk-lf-... $env:LANGFUSE_HOSThttps://cloud.langfuse.com安全提醒永远不要将密钥硬编码在代码中或提交到版本控制系统如Git。4. 实战为联网搜索问答智能体注入可观测性现在我们构建一个简单的智能体用户提问智能体自动联网搜索并基于搜索结果生成答案。我们将使用langchain框架和Tavily搜索工具并用Langfuse进行全链路追踪。4.1 基础智能体实现无观测首先看看一个没有观测的“黑盒”版本# file: agent/vanilla_agent.py from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain.agents import initialize_agent, AgentType from langchain.memory import ConversationBufferMemory import os # 1. 设置API密钥示例请使用环境变量 os.environ[OPENAI_API_KEY] your-openai-key os.environ[TAVILY_API_KEY] your-tavily-key # 2. 初始化组件 llm ChatOpenAI(modelgpt-4o-mini, temperature0) search_tool TavilySearchResults(max_results3) memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) # 3. 创建智能体 agent initialize_agent( tools[search_tool], llmllm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 支持对话和工具调用 memorymemory, verboseTrue, # LangChain自带的简单日志信息有限 ) # 4. 运行 question 2024年巴黎奥运会中国代表团获得了多少枚金牌 result agent.run(question) print(result)这个版本能工作但verboseTrue只能输出简单的文本日志无法结构化存储、无法可视化、无法进行后续分析评估。4.2 集成Langfuse进行追踪现在我们使用langfuse的LangchainCallbackHandler来无缝集成追踪。# file: agent/observed_agent.py from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain.agents import initialize_agent, AgentType from langchain.memory import ConversationBufferMemory from langfuse.callback import CallbackHandler # Langfuse的回调处理器 import os # 初始化Langfuse回调处理器每个请求一个独立的handler def create_agent_with_observation(user_input: str): langfuse_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) llm ChatOpenAI(modelgpt-4o-mini, temperature0, callbacks[langfuse_handler]) # 将handler注入LLM search_tool TavilySearchResults(max_results3) memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent initialize_agent( tools[search_tool], llmllm, agentAgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, memorymemory, callbacks[langfuse_handler], # 将handler注入Agent verboseFalse, # 可以关闭LangChain自带的verbose避免日志干扰 ) # 运行前可以为本次Trace设置一些属性便于后续筛选 langfuse_handler.langfuse.trace.update( nameWeb-Search-QA, user_iduser_123, inputuser_input, # 记录原始输入 metadata{environment: testing, agent_version: 1.0} ) try: result agent.run(user_input) # 运行成功后更新Trace的输出和状态 langfuse_handler.langfuse.trace.update(outputresult) return result except Exception as e: # 如果运行失败记录错误信息 langfuse_handler.langfuse.trace.update(status_messagestr(e)) raise e finally: # 确保所有数据被刷新到Langfuse服务器 langfuse_handler.flush() # 执行智能体 if __name__ __main__: question 2024年巴黎奥运会中国代表团获得了多少枚金牌 answer create_agent_with_observation(question) print(f问题: {question}) print(f答案: {answer})执行这段代码后打开Langfuse项目的Dashboard你就能看到一次完整的Trace记录。4.3 在Langfuse UI中查看追踪结果进入你的Langfuse项目。点击左侧菜单的Traces。你会看到一条名为“Web-Search-QA”的Trace。点击进入详情页。在这里你将获得前所未有的洞察力时间线视图清晰展示整个请求的耗时分布一眼看出是LLM调用慢还是工具执行慢。树形结构视图展开Trace你能看到agent.run下包含的所有子步骤LLM步骤记录了发送给模型的完整提示词Prompt和模型返回的完整回答Completion以及使用的模型、消耗的Token、耗时。Tool步骤记录了搜索工具的执行详情包括搜索查询词和返回的原始搜索结果。Chain步骤展示了LangChain内部的决策逻辑。输入输出在Trace顶部直接看到用户的原始问题input和智能体的最终答案output。元数据可以看到我们设置的user_id,environment等信息方便过滤和分组查询。至此你的智能体已经实现了基础的可观测性。但这只是开始Langfuse更强大的能力在于评估。5. 核心进阶构建自动化评估体系观测是为了评估评估是为了优化。Langfuse允许你为Trace或其中的某个步骤Observation打上“分数”Score。这个分数可以来自人工反馈用户在UI上点“赞/踩”或提供一个1-5分的评分。模型自动评估用另一个LLM如GPT-4作为“裁判”根据既定标准相关性、事实性、有害性等对输出进行评分。我们重点讲解自动化评估这是实现规模化评估和持续集成的关键。5.1 设计评估标准与提示词假设我们要评估智能体答案的事实准确性Factual Correctness。我们设计一个“裁判”LLM的提示词# file: evaluation/prompts.py FACTUALITY_EVALUATION_PROMPT 你是一个严谨的事实核查员。请根据提供的“参考信息”来评估“待评估答案”的事实准确性。 参考信息 {reference} 待评估答案 {answer} 请从以下维度进行评分1-5分 1分答案与参考信息完全矛盾或包含关键事实错误。 2分答案部分错误或遗漏了关键信息。 3分答案基本正确但有一些不精确或模糊的表述。 4分答案准确清晰传达了参考信息中的关键事实。 5分答案完全准确、精炼并且可能补充了相关的上下文。 请只输出一个JSON对象格式如下 {{ score: 1-5的整数, reason: 简要的评分理由指出具体正确或错误之处 }} 5.2 实现自动化评估并记录Score我们在智能体生成答案后自动调用评估流程并将结果记录到Langfuse中。# file: evaluation/auto_evaluator.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser import json from langfuse.callback import CallbackHandler import os def evaluate_factuality_and_score(trace_id: str, reference: str, answer: str): 自动化评估答案的事实准确性并将评分记录到指定的Trace上。 :param trace_id: Langfuse中本次对话的Trace ID :param reference: 用于比对的参考信息如搜索得到的事实片段 :param answer: 待评估的智能体答案 # 初始化一个独立的Langfuse handler用于评估环节 eval_handler CallbackHandler( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) # 初始化“裁判”LLM judge_llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 创建评估链 prompt_template ChatPromptTemplate.from_template(FACTUALITY_EVALUATION_PROMPT) evaluation_chain prompt_template | judge_llm | StrOutputParser() try: # 执行评估 eval_result_str evaluation_chain.invoke( {reference: reference, answer: answer}, config{callbacks: [eval_handler]} # 评估过程本身也可被追踪 ) eval_result json.loads(eval_result_str) # 将评估结果作为Score记录到原始的Trace上 # 注意这里需要通过Langfuse Python SDK直接创建Score from langfuse import Langfuse langfuse_client Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST), ) langfuse_client.score( trace_idtrace_id, # 关联到原始Trace namefactual_correctness, # 评分名称 valueeval_result[score], # 分数值 commenteval_result[reason], # 评分理由 observation_idNone, # 可以为Trace整体打分也可以为某个具体Generation打分 ) print(f✅ 自动化评估完成。事实准确性得分: {eval_result[score]}/5) return eval_result except json.JSONDecodeError: print(❌ 评估LLM返回了非JSON格式结果。) return None finally: eval_handler.flush()5.3 改造智能体集成自动化评估现在我们需要修改智能体使其在获取搜索结果并生成答案后自动调用评估函数。# file: agent/agent_with_eval.py # ... (前面的导入和初始化代码与 observed_agent.py 类似) ... from your_project.evaluation.auto_evaluator import evaluate_factuality_and_score def create_agent_with_observation_and_eval(user_input: str): langfuse_handler CallbackHandler(...) # 初始化handler # ... 初始化llm, tool, memory, agent ... (与之前相同) # 关键在创建Trace后获取其ID langfuse_handler.langfuse.trace.update(nameWeb-Search-QA-Eval, inputuser_input) current_trace_id langfuse_handler.get_trace_id() # 获取当前Trace的ID try: # 为了评估我们需要捕获工具搜索返回的原始结果作为“参考信息” # 这里需要自定义Agent的执行来获取中间结果简化示例 result agent.run(user_input) # 假设我们通过其他方式拿到了本次执行中的搜索原始结果 search_raw_data # 在实际中你可能需要自定义Callback或解析Trace来获取。 search_raw_data 2024巴黎奥运会中国代表团获得40枚金牌27枚银牌24枚铜牌... # 示例数据 # 记录最终输出 langfuse_handler.langfuse.trace.update(outputresult) # 核心执行自动化评估并记录Score evaluate_factuality_and_score( trace_idcurrent_trace_id, referencesearch_raw_data, answerresult ) return result except Exception as e: langfuse_handler.langfuse.trace.update(status_messagestr(e)) raise e finally: langfuse_handler.flush()运行这个智能体后在Langfuse UI中查看Trace你会在详情页的“Scores”部分看到一个名为factual_correctness的评分。点击图表图标你还可以看到所有Trace的评分分布。6. 从数据到洞察利用Langfuse Dashboard驱动优化收集了追踪和评估数据后Langfuse的Dashboard是你的作战指挥中心。6.1 核心数据分析面板Traces Table列表显示所有请求支持按时间、属性如user_id、评分、耗时、成本等筛选和排序。快速定位表现不佳的对话。Analytics可视化图表。延迟分析查看LLM调用、工具调用的P50、P95、P99延迟发现性能瓶颈。成本分析按模型、按天统计Token消耗和费用需配置模型价格精准控制预算。评分趋势观察factual_correctness等评分随时间的变化验证优化措施是否有效。Prompt Management重量级功能。自动从Traces中提取出使用过的提示词版本并进行对比测试A/B测试。你可以直接修改提示词并基于历史对话数据批量回测用数据选择最优提示词。6.2 典型优化工作流发现问题在Dashboard中看到最近一小时factual_correctness平均分从4.2降到了3.5。定位问题筛选出评分低的Traces点击进入详情。发现低分答案都源于一个特定的搜索工具调用返回了过时信息。提出假设可能是搜索查询词不够精确或者需要过滤更近期的结果。实施优化修改提示词在Langfuse的Prompt管理中找到Agent中用于生成搜索查询的提示词创建一个优化版本如加入“请使用2024年后的信息”。修改工具参数在代码中调整TavilySearchResults的max_results或search_depth。测试验证单条测试在Langfuse的Playground中直接用新旧提示词对比测试几个典型问题。批量回测使用Prompt Management的批量评测功能用历史问题集自动运行并对比新旧版本的评分。部署与监控将验证有效的优化部署上线并在Dashboard中持续监控评分和成本指标。7. 常见问题与排查指南在集成和使用Langfuse过程中你可能会遇到以下问题问题现象可能原因排查步骤解决方案Trace数据没有出现在Langfuse UI中1. API密钥错误或权限不足。2. 网络问题导致数据发送失败。3. 代码中未执行handler.flush()或程序提前退出。1. 检查环境变量LANGFUSE_PUBLIC_KEY和LANGFUSE_SECRET_KEY是否正确。2. 在代码中添加try-catch捕获初始化异常。3. 确保在程序结束前调用了flush()方法。1. 在Langfuse项目设置中重新生成密钥。2. 使用langfuse.auth_check()测试连接。3. 将flush()放在finally块中。Trace树形结构中缺少某些步骤如工具调用使用的框架或库可能没有被Langfuse的回调处理器完全覆盖。1. 检查是否将所有必要的对象llm,agent,tools都传入了callbacks[handler]参数。2. 查看LangChain等框架的文档确认其回调支持度。1. 确保初始化时在所有可能产生事件的地方传入callback handler。2. 考虑使用Langfuse的装饰器observe()手动包装自定义函数。自动化评估LLM调用失败或返回格式错误1. 评估提示词设计有歧义导致LLM未返回合规JSON。2. 评估LLM的API密钥或网络有问题。1. 打印出评估LLM的原始输出检查是否符合JSON格式。2. 简化评估提示词加入更严格的格式指令。1. 在评估提示词中强化输出格式要求例如使用“必须输出JSON”等字眼。2. 在代码中添加json.loads的异常处理并设置默认评分。Dashboard中成本数据显示为0未正确配置模型定价信息。进入Langfuse项目设置检查Model Prices配置。在设置中为使用的模型如gpt-4o-mini添加对应的输入/输出每百万Token价格。数据来自OpenAI官网。追踪数据延迟高影响应用性能SDK默认是异步批量上报数据通常不影响应用性能。如果担心可能是网络延迟。检查应用所在网络到cloud.langfuse.com的连接。1. Langfuse SDK是异步的主流程不会阻塞等待网络IO。2. 对于极高并发场景可考虑调整SDK的批量发送参数或使用自托管版部署在内网。8. 生产环境最佳实践当你准备将集成了Langfuse的智能体部署到生产环境时请遵循以下建议密钥管理使用安全的秘密管理服务如AWS Secrets Manager, HashiCorp Vault或云平台的环境变量注入功能切勿硬编码。采样率控制在生产环境中可能不需要记录100%的请求。Langfuse SDK支持设置采样率只记录一部分请求以控制成本和数据量。langfuse_handler CallbackHandler( ..., sample_rate0.1, # 只记录10%的请求 )区分环境为开发、测试、生产环境创建不同的Langfuse项目并使用不同的API密钥。通过metadata或Trace的name字段区分环境。用户隐私与数据安全避免在input、output或metadata中记录个人身份信息PII、密码等敏感数据。Langfuse支持数据脱敏规则。自定义追踪不要局限于框架自动追踪的内容。使用observe()装饰器或SDK的trace、span方法手动记录业务关键节点。from langfuse.decorators import observe observe() # 自动将此函数记录为一个Span def complex_business_logic(data): # ... 你的业务逻辑 ... return result建立评估基线在项目启动初期就定义好核心评估指标如事实准确性、响应速度、成本并收集一段时间的数据建立基线。后续所有优化都应与基线对比。告警集成利用Langfuse的webhook功能或定期查询API当关键指标如平均评分骤降、成本异常升高出现异常时发送告警到你的监控系统如Slack, PagerDuty。通过将Langfuse深度集成到你的AI应用开发流程中你构建的将不再是一个神秘的黑盒而是一个透明、可度量、可持续进化的智能系统。从追踪每一次调用开始到建立全面的评估体系最终实现数据驱动的快速迭代这正是AI工程化落地的关键一步。