Langfuse

维护一个生产级的 LLM 应用，我们需要做什么

各种指标监控与统计：访问记录、响应时长、Token 用量、计费等等
调试 Prompt
测试/验证系统的相关评估指标
数据集管理（便于回归测试）
Prompt 版本管理（便于升级/回滚）

针对以上需求，目前有两个生产级 LLM App 维护平台

LangFuse: 开源 + SaaS（免费/付费），LangSmith 平替，可集成 LangChain 也可直接对接 OpenAI API；
LangSmith: LangChain 的官方平台，SaaS 服务（免费/付费），非开源，企业版支持私有部署；

根据自己的技术栈，选择：

LangFuse：开源平台，支持 LangChain 和原生 OpenAI API
LangSmith: LangChain 的原始管理平台
Prompt Flow：开源平台，支持 Semantic Kernel

LangFuse

开源，支持 LangChain 集成或原生 OpenAI API 集成

官方网站：https://langfuse.com/

项目地址：https://github.com/langfuse

文档地址：https://langfuse.com/docs

API文档：https://api.reference.langfuse.com/

Python SDK:
https://python.reference.langfuse.com/
JS SDK:
https://js.reference.langfuse.com/
通过官方云服务使用：
1. 注册: cloud.langfuse.com
2. 创建 API Key

1 2	LANGFUSE_SECRET_KEY="sk-lf-..." LANGFUSE_PUBLIC_KEY="pk-lf-..."

通过 Docker 本地部署

# Clone repository
git clone https://github.com/langfuse/langfuse.git
cd langfuse

# Run server and db
docker compose up -d

# 在自己部署的系统中生成上述两个 KEY
# 并在环境变量中指定服务地址

LANGFUSE_SECRET_KEY="sk-lf-..."
LANGFUSE_PUBLIC_KEY="pk-lf-.."
LANGFUSE_HOST="http://localhost:3000"

几个基本概念

Trace 一般表示用户与系统的一次交互，其中记录输入、输出，也包括自定义的 metadata 比如用户名、session id 等；
一个 trace 内部可以包含多个子过程，这里叫 observarions；
Observation 可以是多个类型：
1. Event 是最基本的单元，用于记录一个 trace 中的每个事件；
2. Span 表一个 trace 中的一个”耗时”的过程；
3. Generation 是用于记录与 AI 模型交互的 span，例如：调用 embedding 模型、调用 LLM。
Observation 可以嵌套使用。

通过装饰器记录（上报）

observe() 装饰器的参数

def observe(
    self,
    *,
    name: Optional[str] = None, # Trace 或 Span 的名称，默认为函数名
    as_type: Optional[Literal['generation']] = None, # 将记录定义为 Observation (LLM 调用）
    capture_input: bool = True, # 记录输入
    capture_output: bool = True, # 记录输出
    transform_to_string: Optional[Callable[[Iterable], str]] = None # 将输出转为 string
) -> Callable[[~F], ~F]

from langfuse.decorators import observe
from langfuse.openai import openai # OpenAI integration

@observe()
def run():
    return openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
          {"role": "user", "content": "对我说Hello, World!"}
        ],
    ).choices[0].message.content

print(run())

通过 langfuse_context 记录 User ID、Metadata 等

from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai # OpenAI integration

@observe()
def run():
    langfuse_context.update_current_trace(
        name="HelloWorld",
        user_id="wzr",
        metadata={"test":"test value"}
    )
    return openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
          {"role": "user", "content": "对我说Hello, World!"}
        ],
    ).choices[0].message.content

print(run())

通过 LangChain 的回调集成

from langfuse.decorators import langfuse_context, observe

@observe()
def run():
    langfuse_context.update_current_trace(
            name="LangChainDemo",
            user_id="wzr",
        )
    
    # 获取当前 LangChain 回调处理器
    langfuse_handler = langfuse_context.get_current_langchain_handler()
    
    return chain.invoke(input="AGIClass", config={"callbacks": [langfuse_handler]})

print(run())

用 Trace 记录一个多次调用 LLM 的过程

import uuid
from langfuse.decorators import langfuse_context, observe

# 主流程
@observe()
def verify_question(
    question: str,
    outlines: str,
    question_list: list,
    user_id: str,
) -> bool:
    langfuse_context.update_current_trace(
            name="AGIClassAssistant",
            user_id=user_id,
        )
    langfuse_handler = langfuse_context.get_current_langchain_handler()
    # 判断是否需要回答
    if need_answer_chain.invoke(
        {"user_input": question, "outlines": outlines},
        config={"callbacks": [langfuse_handler]}
    ) == 'Y':
        # 判断是否为重复问题
        if is_duplicated_chain.invoke(
            {"user_input": question,
                "question_list": "\n".join(question_list)},
            config={"callbacks": [langfuse_handler]}
        ) == 'N':
            question_list.append(question)
            return True
    return False

用 Session 记录一个用户的多轮对话

@observe()
def chat_one_turn(user_input, user_id, turn_id):
    langfuse_context.update_current_trace(
        name=f"ChatTurn{turn_id}",
        user_id=user_id,
        session_id="chat-"+now.strftime("%d/%m/%Y %H:%M:%S")
    )
    langfuse_handler = langfuse_context.get_current_langchain_handler()
    messages.append(HumanMessage(content=user_input))
    response = llm.invoke(messages, config={"callbacks": [langfuse_handler]})
    messages.append(response)
    return response.content

数据集与测试

在线标注(在平台上进行标注)

上传已有数据集

定义评估函数

运行测试

Prompt 调优与回归测试

Prompt 版本管理

目前只支持 Langfuse 自己的 SDK

# 按名称加载
prompt = langfuse.get_prompt("need_answer_v1")

# 按名称和版本号加载
prompt = langfuse.get_prompt("need_answer_v1", version=2)

# 对模板中的变量赋值
compiled_prompt = prompt.compile(input="老师好", outlines="test")

print(compiled_prompt)

# 获取 config

prompt = langfuse.get_prompt("need_answer_v1", version=5)

print(prompt.config)

如何比较两个句子的相似性：一些经典 NLP 的评测方法（选）

用途：比较llm 返回值和预期值，从而进行打分计算

编辑距离：也叫莱文斯坦距离(Levenshtein),是针对二个字符串的差异程度的量化量测，量测方式是看至少需要多少次的处理才能将一个字符串变成另一个字符串。
1. 具体计算过程是一个动态规划算法：https://zhuanlan.zhihu.com/p/164599274
2. 衡量两个句子的相似度时，可以以词为单位计算
BLEU Score:
1. 计算输出与参照句之间的 n-gram 准确率（n=1…4）
2. 对短输出做惩罚
3. 在整个测试集上平均下述值
4. 函数库：https://www.nltk.org/_modules/nltk/translate/bleu_score.html
Rouge Score:
1. Rouge-N：将模型生成的结果和标准结果按 N-gram 拆分后，只计算召回率；
2. Rouge-L: 利用了最长公共子序列（Longest Common Sequence）
3. 函数库：https://pypi.org/project/rouge-score/
4. 对比 BLEU 与 ROUGE：
  1. BLEU 能评估流畅度，但指标偏向于较短的翻译结果（brevity penalty 没有想象中那么强）
  2. ROUGE 不管流畅度，所以只适合深度学习的生成模型：结果都是流畅的前提下，ROUGE 反应参照句中多少内容被生成的句子包含（召回）
METEOR: 另一个从机器翻译领域借鉴的指标。与 BLEU 相比，METEOR 考虑了更多的因素，如同义词匹配、词干匹配、词序等，因此它通常被认为是一个更全面的评价指标。
1. 对语言学和语义词表有依赖，所以对语言依赖强。

此类方法常用于对文本生成模型的自动化评估。实际使用中，我们通常更关注相对变化而不是绝对值（调优过程中指标是不是在变好）。

基于 LLM 的测试方法

LangFuse 集成了一些原生的基于 LLM 的自动测试标准。

具体参考：https://langfuse.com/docs/scores/model-based-evals

划重点：此类方法，对于用于评估的 LLM 自身能力有要求。需根据具体情况选择使用。