大型语言模型（LLM）应用调试：Langfuse 本地化部署与集成指南

随着 LLM 应用的不断成熟，特别是在 B 端企业场景中的逐渐落地，其不再停留在原型与验证阶段，将面临着更高的工程化要求。无论是输出的稳定性、性能、以及成本控制等，都需要实现真正的'生产就绪'。但由于大量的应用基于 LangChain、LlamaIndex 等框架开发，更多的抽象与封装使得应用难以跟踪与调试。因此，借助一个独立且侵入性较小的工程化平台来捕获 LLM 应用内部细节，帮助排障、优化与测试是很有必要的。

本文将介绍如何使用开源的 LLM 应用工程化平台 Langfuse。与在线且收费的 LangSmith 不同，Langfuse 支持完全本地化部署与使用，与应用集成也很简单。

一、快速本地化部署

Langfuse 是一个开源的 LLM 应用的工程平台，可以帮助开发者及团队进行集中、在线、协作的 LLM 应用跟踪调试、分析与测试评估。主要功能包括：

跟踪调试：跟踪应用执行过程、上下文、LLM 调用与成本、用户反馈等。
提示管理：集中的 Prompt 模板创建、维护与版本管理。
监控分析：调用统计、模型使用、tokens 成本、响应延迟、评分统计等。
测试评估：基于 LLM 与用户反馈的评估，包括质量、风格、内容安全。

方法一：Docker Compose 快速启动

适合个人开发者、对数据保存要求不高的开发团队。这种模式下会自动启动 Postgres 的独立 Docker，只需三步：

# 下载代码
git clone https://github.com/langfuse/langfuse.git

# 进入代码目录
cd langfuse

# 一键获取并启动 docker 容器
docker compose up

方法二：独立数据库 + Docker Run

适合有更高数据保存要求的开发团队，或者有现成的 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

http://你的服务器地址:3000/

常见问题排查

容器启动失败：检查网络是否连通；3000 端口是否被占用；数据库是否正常。
无法远程访问：检查安全端口是否放行；localhost 修改为 0.0.0.0 试试。
数据库无法连接：数据库端口是否放行；database_url 是否有特殊字符。

二、与 LangChain 应用集成

平台已经就绪，现在需要让应用与 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

给应用增加 Trace 功能

我们用 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 的完整会话过程。

使用 Prompts 管理功能

如果在一个团队开发中，需要对所有的 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 中支持多种评分途径：

在 Langfuse UI 中手工评分
在 Langfuse UI 中借助 LLM 做自动评估（目前仅在 cloud 版本中内测）
通过 SDK 做自定义评分或搜集用户评分

最常见的是第 3 种，其应用场景是在应用端搜集用户反馈评分，然后上报 Langfuse 平台。为了实现评分上报，需要对上述代码中的 invoke 部分做如下改造：

from langfuse.decorators import langfuse_context, observe

@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 菜单中做模型价格对齐。

三、与普通 LLM 应用的集成

除了 Langchain 框架以外，Langfuse 还支持与另外一种常用框架 LlamaIndex 的快速集成。但如果你的应用直接基于大模型的 SDK 开发，并且有着较复杂的控制流程，也能采用 Langfuse 的低层 SDK 来实现集成。这里用一个简单的与本地 Ollama 模型对话的小应用来展示集成方式：

import uuid
from langfuse.decorators import observe, langfuse_context

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 功能与 LangSmith 类似，但胜在开源可本地化部署，集成简单，且对应用性能影响很小（数据上报都是异步）。更多功能大家可自行部署并探索，有任何问题，欢迎查阅官方文档与交流。

大型语言模型（LLM）应用调试：Langfuse 本地化部署与集成指南