Langfuse 是一款开源的 LLM 应用工程化平台，支持完全本地化部署。本文详细介绍了其三种部署方式（Docker Compose、独立数据库、本地开发）及健康检查步骤。内容涵盖与 LangChain 框架的深度集成方法，包括通过 CallbackHandler 实现追踪调试、利用 Prompts 管理功能进行模板版本控制、以及通过 SDK 上报用户评分以优化模型表现。此外，还展示了基础 SDK 集成方案及仪表盘数据分析功能，帮助开发者构建生产就绪的 LLM 应用系统。
随着大语言模型（LLM）应用的不断成熟，特别是在 B 端企业场景中的逐渐落地，其不再停留在原型与验证阶段，将面临着更高的工程化要求。无论是输出的稳定性、性能、以及成本控制等，都需要实现真正的'生产就绪'。但由于大量的应用基于 LangChain、LlamaIndex 等框架开发，更多的抽象与封装使得应用难以跟踪与调试。因此，借助一个独立且侵入性较小的工程化平台来捕获 LLM 应用内部细节，帮助排障、优化与测试是很有必要的。
本文将详细介绍开源的 LLM 应用工程化平台：Langfuse。与在线且收费的 LangSmith 不同，它支持完全本地化部署与使用，与应用集成也很简单。
适合个人开发者、对数据保存要求不高的开发团队。这种模式下会自动启动 Postgres 的独立 Docker，只需三步：
# 下载代码
git clone https://github.com/langfuse/langfuse.git
# 进入代码目录
cd langfuse
# 一键获取并启动 docker 容器
docker compose up
适合有更高数据保存要求的开发团队，或者有现成的 Postgres 数据库。这种模式下，需要首先自行安装并启动 Postgres 数据库。如果还没有 Postgres，可进入官网下载、安装并配置启动。记录下数据库的连接 URL。然后执行以下命令：
# 拉取最新镜像
docker pull langfuse/langfuse:latest
# 注意替换这里的 DATABASE_URL 为你的 postgres url
docker run --name langfuse \
-e DATABASE_URL=postgresql://hello \
-e NEXTAUTH_URL=http://localhost:3000 \
-e NEXTAUTH_SECRET=mysecret \
-e SALT=mysalt \
-p 3000:3000 \
-a STDOUT \
langfuse/langfuse
除非你需要对 Langfuse 做个性化定制，或者使用 Langfuse 展开商业运营，否则不建议采用这种方式。具体请参考项目中 CONTRIBUTING.md 文件说明。
完成部署后，运行如下命令进行健康检查：
# 健康测试，在本机运行
curl http://localhost:3000/api/public/health
登录 Langfuse UI 使用：访问 http://你的服务器地址:3000/，出现登录界面就大功告成。
平台已经就绪，现在需要让应用与 Langfuse 集成起来，以能够跟踪、评估与分析我们的应用。使用 Langfuse 的 SDK/API，Langfuse 能够与现有的任意应用做集成，但如果你的应用是基于 LangChain、LlamaIndex 这样的开发框架，那么使用起来会更加方便。
在开始之前，首先登录到 Langfuse UI，创建一个 Project，然后在 Settings 中生成API Keys。
然后把图中三个参数设置到本地环境变量：
import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-***"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-***"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"
最后，安装 Langfuse 的 SDK，以 Python 为例（也支持 NodeJS）：
pip install langfuse
我们用 Langchain 构建了一个简单的 RAG 应用，使用了本地的 Ollama 模型和 OpenAI 的嵌入模型。现在只需要增加红色的三行代码即可：
from langfuse.callback import CallbackHandler
import uuid
# 初始化 Handler，生成 session_id 用于组织多次 trace
langfuse_handler = CallbackHandler(session_id=str(uuid.uuid4()))
# 模型
llm = Ollama(model="qwen:14b")
embed_model = OpenAIEmbeddings(model="text-embedding-3-small")
# 构建向量索引
documents = DirectoryLoader('./data/', glob="*.txt", loader_cls=TextLoader).load()
splits = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=0).split_documents(documents)
db = FAISS.from_documents(splits, embed_model)
retriever = db.as_retriever()
# Prompt
prompt = ChatPromptTemplate.from_template("基于如下上下文:\n\n{context}\n\n请回答以下问题：\n\n{question}")
# Chain
rag_chain = (
{"context": retriever | (lambda docs: "\n\n".join(doc.page_content for doc in docs)), "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# 对话循环
while True:
user_input = input("问题：")
if user_input.lower() == "exit":
break
if user_input.lower() == "":
continue
# 在 invoke 时传入 callbacks
response = rag_chain.invoke(user_input, config={"callbacks": [langfuse_handler]})
print("AI：", response)
注意这里在每次运行时生成一个 session_id，是用来将多次 trace 组织到一次 session 中（Langfuse 跟踪的基本单位是 Session->Trace->Observation，而调用 LLM 就是 Observation 的一种类型）。
现在我们来运行程序，做 3 次对话，然后回到 Langfuse UI 观察 Tracing 菜单下的跟踪记录，可以看到 3 条 trace 记录，且有相同的 session_id。
点击一条 trace 记录，就可以追踪到详细的 RAG 运行过程，从检索到组装的 Prompt、LLM 生成，多个步骤的关系以及输入输出、延时、model 使用、tokens 等。
由于使用了 session 来组织多个 trace，我们可以在 session 菜单中查看这次 session 的完整会话过程。
如果在一个团队开发中，需要对所有的 Prompt 进行集中管理与维护，可以利用 Langfuse 的 Prompts 管理功能。这里我们通过 Langfuse UI 的 Prompts 菜单新增一个 RAG 的简单提示模板。
现在可以在应用中使用这个模版，只需要把上述代码中的创建 prompt 的部分修改为用以下方式从 Langfuse 获取模板即可：
from langfuse import Langfuse
langfuse = Langfuse()
prompt_str = langfuse.get_prompt("RAG").get_langchain_prompt()
prompt = PromptTemplate(template=prompt_str, input_variables=["context","question"])
Prompts 管理支持设置变量、版本管理与标签，可实现灵活控制。
为了对生产级的 LLM 应用做持续改进与优化，通过评分 (score) 来对应用输出质量作评估是有必要的。在 Langfuse 中支持多种评分途径：
这里的方式 1 应用场景有限，方式 2 目前仅在 cloud 版本中内测，因此这里介绍第 3 种，其最常见的场景是在应用端搜集用户反馈评分，然后上报 Langfuse 平台。为了实现评分上报，需要对上述代码中的 invoke 部分做如下改造：
from langfuse.decorators import langfuse_context, observe
session_id = str(uuid.uuid4())
@observe()
def invoke(query):
langfuse_context.update_current_trace(session_id=session_id)
langfuse_handler = langfuse_context.get_current_langchain_handler()
response = chain.invoke(query, config={"callbacks": [langfuse_handler]})
# 此处模拟搜集到用户评分，如 0.8
langfuse_context.score_current_trace(
name="feedback-on-trace",
value=0.8,
comment="用户反馈",
)
return response["result"]
这里把上述代码中的 invoke 包装成独立函数，然后增加 observe() 装饰器，以获得对 trace 控制的 langfuse_context 对象；再通过 score_current_trace 方法上报本次评分。成功后，用户的每次反馈评分都可以在 Langfuse UI 的 scores 菜单中看到，并且可以在 dashboard 看到相关统计指标，后续可以根据这些指标进行针对性优化。
Langfuse 的 Dashboard 默认展示很多有用的分析与统计指标，包括 Trace 的统计、模型成本分析、评分统计、不同类型环节的响应延迟等，非常适合用来帮助应用优化（为了能正确统计商业模型成本，注意在 models 菜单中做模型价格对齐）。
除了 Langchain 框架以外，Langfuse 还支持与另外一种常用框架 LlamaIndex 的快速集成。但如果你的应用直接基于大模型的 SDK 开发，并且有着较复杂的控制流程，也能采用 Langfuse 的低层 SDK 来实现集成。这里用一个简单的与本地 Ollama 模型对话的小应用来展示集成方式：
import uuid
from langfuse.decorators import langfuse_context, observe
session_id = str(uuid.uuid4())
# LLM 调用采用 generation 的 type，不会产生新的 trace，只会产生 observation
@observe(as_type="generation")
def call_llm(query):
response = ollama.chat(model='qwen:14b', messages=[
{
'role': 'user',
'content': query,
},
])
return response['message']['content']
# 这里不指定 type，每次调用产生一个 trace；不直接放 llm 调用逻辑是为了把 LLM 调用的 observation 分离
@observe()
def invoke(query):
# 用 session_id 把多次 trace 组织起来
langfuse_context.update_current_trace(session_id=session_id)
return call_llm(query)
def main():
while True:
user_input = input("问题：")
if user_input.lower() == "exit":
break
if user_input.lower() == "":
continue
print("AI：", invoke(user_input))
# 结束前 flush 缓存，防止漏上报
langfuse_context.flush()
if __name__ == "__main__":
main()
这里使用 @observe 装饰器来实现函数输入输出与 LLM 生成的跟踪，实际使用时，也可以使用 langfuse.trace 自行创建 trace，并上报自定义的跟踪信息，虽然较为繁琐，但控制会更灵活。具体可以参考官方 SDK 文档。
在生产环境中使用 Langfuse 时，建议遵循以下最佳实践以确保系统的稳定性和数据的准确性：
通过上述部署与集成方案，开发者可以高效地构建具备可观测性的 LLM 应用系统，显著提升调试效率与运维能力。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online