企业官网建设流程全解析

LLM 应用监控

LLM 应用打包成 Docker 容器部署,但部署上线后，你需要知道：

• 每次请求耗时多少？
• token 消耗了多少？LLM 调用费用如何？
• 错误率是多少？
• 系统有没有异常？

这就是监控的作用。

核心概念

LangFuse 核心概念

Trace（追踪） └── Span（跨度，一个操作步骤） ├── Generation（LLM调用，记录prompt/response/token） └── Event（普通事件）

• Trace：一次完整请求的全过程（从用户输入到最终输出）
• Generation：专门用于记录 LLM 调用，会自动统计 token 用量和成本
• LangChain 有 CallbackHandler，可以自动把所有 LLM 调用捕获为 Trace

自动模式最小示例

fromlangfuse.callbackimportCallbackHandlerfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage# 初始化 LangFuse handler（自动从环境变量读取 KEY）handler=CallbackHandler()llm=ChatOpenAI(model="deepseek-chat",...)# 把 handler 传入 invoke，自动追踪这次调用response=llm.invoke([HumanMessage(content="你好")],config={"callbacks":[handler]})

手动模式

手动模式需要用 Langfuse 客户端对象：

fromlangfuseimportget_client,propagate_attributesfromlangfuse.langchainimportCallbackHandler langfuse=get_client()# 自动读取环境变量中的 LANGFUSE_* 配置

三步操作

第一步：创建根 Span（对应一次完整请求 / Trace）

用 start_as_current_observation 创建根 span；user_id、session_id 用 propagate_attributes 挂到当前 trace 上：

with(langfuse.start_as_current_observation(as_type="span",name="chat-pipeline")asroot_span,propagate_attributes(user_id="user-001",session_id="session-abc"),):root_span.update(input="你喜欢西瓜吗？")# 可选：整体输入# ... 下面的 span 都写在这个 with 块里 ...

第二步：在根 Span 下创建子 Span（记录每个处理步骤）

子 span 同样用 with，退出块时自动结束，无需手动 end()：

# Span 1：输入验证（纯逻辑，不调用 LLM）withlangfuse.start_as_current_observation(as_type="span",name="input_validation")asspan1:span1.update(input={"text":"你喜欢西瓜吗？"})# ... 执行验证逻辑 ...span1.update(output={"valid":True})

第三步：把 LangChain 的 handler 绑定到当前 Trace

不用全局 handler，而是用 CallbackHandler 显式指定 trace_id 和父 span：

# Span 2：LLM 调用（Langfuse 会在该 span 下记录 Generation）withlangfuse.start_as_current_observation(as_type="span",name="llm_call")asspan2:handler=CallbackHandler(trace_context={"trace_id":langfuse.get_current_trace_id(),"parent_span_id":span2.id,})response=llm.invoke([HumanMessage(content="你喜欢西瓜吗？")],config={"callbacks":[handler]},)root_span.update(output=response.content)

结尾：必须 flush

langfuse.flush()# 确保所有数据上报，程序退出前调用

示例代码

@contextmanager# 把普通生成器函数变成上下文管理器，让你可以用 with 语法自动管理「进入 / 退出」逻辑deftrace_request(user_input:str=USER_INPUT,user_id:str=USER_ID,session_id:str=SESSION_ID,):langfuse=get_client()with(langfuse.start_as_current_observation(as_type="span",name="chat_pipline")asroot_span,propagate_attributes(user_id=user_id,session_id=session_id),):root_span.update(input={"text":user_input})# span1withlangfuse.start_as_current_observation(as_type="span",name="input_validation")asspan1:is_valid=len(user_input)>0span1.update(input={"valid":is_valid})# span2withlangfuse.start_as_current_observation(as_type="span",name="llm_call")asspan2:handler=CallbackHandler(trace_context={"trace_id":langfuse.get_current_trace_id(),"parent_span_id":span2.id,})yieldhandler,root_span# 交给 app 使用，span 保持打开

withtrace_request(user_input)as(handler,root_span):answer=llm_answer(llm,user_input,handler)

yield

yield 常见于生成器，含义是：

• 暂停当前函数
• 把右边的值交给调用方
• 调用方用完后，函数从 yield 之后继续执行

yield 交出的值可以是任意类型；若写yield handler, root_span，Python 会把它当作一个元组(handler, root_span)交出，as可以用元组解包直接拆开，例如as (handler, root_span)。

@contextmanager

把带 yield 的生成器函数变成上下文管理器，让你可以用 with 语法自动管理「进入 / 退出」逻辑。

python 的 with 语法不会直接调用普通函数，它要求右侧对象实现上下文管理器协议，即提供：

• enter()：进入 with 时调用，返回值赋给 as 后面的变量
• exit()：离开 with 时调用（包括异常），负责清理资源

@contextmanager会按下面方式映射到上述协议：

• yield之前的代码：相当于enter（进入 with 时执行）
• yield右边的值：赋给as后面的变量
• yield之后的代码：相当于exit（离开 with 时执行，常用于收尾）