LLM 应用监控LLM 应用打包成 Docker 容器部署,但部署上线后你需要知道每次请求耗时多少token 消耗了多少LLM 调用费用如何错误率是多少系统有没有异常这就是监控的作用。核心概念类型关注点典型工具LLM 专项监控token用量、延迟、prompt/response内容、错误LangSmith、LangFuse基础设施监控CPU、内存、HTTP请求数、响应时间Prometheus GrafanaLangFuse 核心概念Trace追踪 └── Span跨度一个操作步骤 ├── GenerationLLM调用记录prompt/response/token └── Event普通事件Trace一次完整请求的全过程从用户输入到最终输出Generation专门用于记录 LLM 调用会自动统计 token 用量和成本LangChain 有 CallbackHandler可以自动把所有 LLM 调用捕获为 Trace自动模式最小示例fromlangfuse.callbackimportCallbackHandlerfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage# 初始化 LangFuse handler自动从环境变量读取 KEYhandlerCallbackHandler()llmChatOpenAI(modeldeepseek-chat,...)# 把 handler 传入 invoke自动追踪这次调用responsellm.invoke([HumanMessage(content你好)],config{callbacks:[handler]})手动模式手动模式需要用 Langfuse 客户端对象fromlangfuseimportget_client,propagate_attributesfromlangfuse.langchainimportCallbackHandler langfuseget_client()# 自动读取环境变量中的 LANGFUSE_* 配置三步操作第一步创建根 Span对应一次完整请求 / Trace用 start_as_current_observation 创建根 spanuser_id、session_id 用 propagate_attributes 挂到当前 trace 上with(langfuse.start_as_current_observation(as_typespan,namechat-pipeline)asroot_span,propagate_attributes(user_iduser-001,session_idsession-abc),):root_span.update(input你喜欢西瓜吗)# 可选整体输入# ... 下面的 span 都写在这个 with 块里 ...第二步在根 Span 下创建子 Span记录每个处理步骤子 span 同样用 with退出块时自动结束无需手动 end()# Span 1输入验证纯逻辑不调用 LLMwithlangfuse.start_as_current_observation(as_typespan,nameinput_validation)asspan1:span1.update(input{text:你喜欢西瓜吗})# ... 执行验证逻辑 ...span1.update(output{valid:True})第三步把 LangChain 的 handler 绑定到当前 Trace不用全局 handler而是用 CallbackHandler 显式指定 trace_id 和父 span# Span 2LLM 调用Langfuse 会在该 span 下记录 Generationwithlangfuse.start_as_current_observation(as_typespan,namellm_call)asspan2:handlerCallbackHandler(trace_context{trace_id:langfuse.get_current_trace_id(),parent_span_id:span2.id,})responsellm.invoke([HumanMessage(content你喜欢西瓜吗)],config{callbacks:[handler]},)root_span.update(outputresponse.content)结尾必须 flushlangfuse.flush()# 确保所有数据上报程序退出前调用示例代码contextmanager# 把普通生成器函数变成上下文管理器让你可以用 with 语法自动管理「进入 / 退出」逻辑deftrace_request(user_input:strUSER_INPUT,user_id:strUSER_ID,session_id:strSESSION_ID,):langfuseget_client()with(langfuse.start_as_current_observation(as_typespan,namechat_pipline)asroot_span,propagate_attributes(user_iduser_id,session_idsession_id),):root_span.update(input{text:user_input})# span1withlangfuse.start_as_current_observation(as_typespan,nameinput_validation)asspan1:is_validlen(user_input)0span1.update(input{valid:is_valid})# span2withlangfuse.start_as_current_observation(as_typespan,namellm_call)asspan2:handlerCallbackHandler(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):answerllm_answer(llm,user_input,handler)yieldyield 常见于生成器含义是暂停当前函数把右边的值交给调用方调用方用完后函数从 yield 之后继续执行yield 交出的值可以是任意类型若写yield handler, root_spanPython 会把它当作一个元组(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 时执行常用于收尾
