# Langfuse实战：如何用Python SDK快速搭建LLM评估系统（附避坑指南） 在构建基于大语言模型（LLM）的应用时，评估提示词效果和模型输出质量是开发过程中最关键的环节之一。Langfuse作为开源的LLM工程平台，提供了一套完整的解决方案来帮助开发者系统化地解决这些问题。本文将带你从零开始，使用Langfuse的Python SDK快速搭建一个LLM评估系统，并分享在实际项目中积累的实战经验。 ## 1. 环境准备与SDK集成 在开始之前，确保你的开发环境满足以下条件： - Python 3.8或更高版本 - 已安装pip包管理工具 - 访问Langfuse Cloud或自建Langfuse实例的权限 首先安装必要的Python包： ```bash pip install langfuse openai ``` Langfuse提供了两种主要的集成方式：直接使用SDK或通过装饰器自动追踪。对于评估系统，我们推荐结合使用这两种方式以获得最大的灵活性。 **配置Langfuse客户端**： ```python from langfuse import Langfuse # 初始化客户端 langfuse = Langfuse( public_key="pk-lf-your-public-key", secret_key="sk-lf-your-secret-key", host="https://cloud.langfuse.com" # 或自建实例地址 ) ``` > 提示：建议将敏感信息存储在环境变量中，而不是直接硬编码在代码里。可以使用`os.environ`来管理这些配置。 ## 2. 数据集创建与管理 评估系统的核心是高质量的数据集。Langfuse允许你通过Web界面或API创建和管理数据集。以下是通过Python SDK创建数据集的完整示例： ```python # 定义数据集项 capital_items = [ {"input": {"country": "France"}, "expected_output": "Paris"}, {"input": {"country": "Germany"}, "expected_output": "Berlin"}, # 添加更多测试用例... ] # 创建数据集 dataset_name = "world-capitals-v1" for item in capital_items: langfuse.create_dataset_item( dataset_name=dataset_name, input=item["input"], expected_output=item["expected_output"], # 可选：添加metadata用于分类或过滤 metadata={"category": "geography", "difficulty": "easy"} ) ``` **数据集管理最佳实践**： - 为不同测试场景创建独立的数据集（如"basic-facts"、"creative-writing"等） - 使用metadata字段标记数据项的属性和特征 - 定期更新数据集以覆盖更多边缘案例 - 保持输入输出结构的统一性，便于批量处理 ## 3. 自定义评估函数设计 Langfuse的强大之处在于支持灵活的自定义评估逻辑。评估函数可以根据你的具体需求设计，从简单的精确匹配到复杂的LLM-as-judge模式。 **基础评估函数示例**： ```python def exact_match_evaluation(output: str, expected_output: str) -> bool: """精确匹配评估""" return output.strip().lower() == expected_output.strip().lower() def contains_keywords_evaluation(output: str, keywords: list) -> float: """关键词包含评估，返回匹配比例""" output_words = output.lower().split() matches = sum(1 for kw in keywords if kw.lower() in output_words) return matches / len(keywords) if keywords else 0.0 ``` **高级LLM-as-judge评估**： ```python from openai import OpenAI client = OpenAI() def llm_judge_evaluation(prompt: str, output: str, criteria: str) -> float: """使用LLM作为裁判进行评分""" evaluation_prompt = f""" 请根据以下标准评估回答质量，给出1-5分的评分： 评估标准：{criteria} 问题：{prompt} 回答：{output} 请只返回评分数字，不要包含其他内容。 """ response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": evaluation_prompt}], temperature=0, max_tokens=1 ) try: return float(response.choices[0].message.content) except: return 0.0 # 评估失败时返回最低分 ``` ## 4. 完整评估流程实现 结合上述组件，我们可以构建一个完整的评估流程。以下代码展示了如何遍历数据集、调用LLM并记录评估结果： ```python def run_evaluation_pipeline(dataset_name: str, system_prompt: str): """执行完整的评估流程""" dataset = langfuse.get_dataset(dataset_name) for item in dataset.items: # 获取Langchain回调处理器 handler = item.get_langchain_handler(run_name="capital-quiz-eval") try: # 调用LLM获取实际输出 actual_output = call_llm_with_prompt( input=item.input["country"], system_message=system_prompt, callback_handler=handler ) # 执行评估 score_value = exact_match_evaluation( output=actual_output, expected_output=item.expected_output ) # 记录评估结果 langfuse.score( name="exact_match", value=score_value, trace_id=handler.get_trace_id(), comment=f"评估国家：{item.input['country']}" ) except Exception as e: print(f"评估{item.input['country']}时出错：{str(e)}") continue def call_llm_with_prompt(input: str, system_message: str, callback_handler): """使用Langchain调用LLM""" from langchain.schema import HumanMessage, SystemMessage from langchain.chat_models import ChatOpenAI messages = [ SystemMessage(content=system_message), HumanMessage(content=input), ] model = ChatOpenAI( model="gpt-3.5-turbo", temperature=0.7, callbacks=[callback_handler] ) return model.invoke(messages).content ``` ## 5. 常见问题与解决方案 在实际使用Langfuse构建评估系统时，开发者可能会遇到一些典型问题。以下是我们在项目中总结的经验和解决方案： **问题1：handler.root_span为None** 这是Langchain集成中最常见的问题之一。新版本中推荐使用以下替代方案： ```python # 旧版问题代码 # handler.root_span.score(name="exact_match", value=score_value) # 新版解决方案 langfuse.score( name="exact_match", value=score_value, trace_id=handler.get_trace_id() ) ``` **问题2：评估结果不一致** 当使用LLM-as-judge时，评估结果可能出现波动。解决方法包括： - 设置temperature=0确保确定性输出 - 提供更详细的评估标准和示例 - 对同一项目进行多次评估取平均值 **问题3：大规模评估性能瓶颈** 评估大量数据项时，可以考虑： ```python from concurrent.futures import ThreadPoolExecutor def batch_evaluate(dataset_name: str, workers=4): """使用多线程加速评估""" dataset = langfuse.get_dataset(dataset_name) items = list(dataset.items) # 转换为列表避免多次API调用 with ThreadPoolExecutor(max_workers=workers) as executor: futures = [ executor.submit(evaluate_item, item) for item in items ] for future in futures: try: future.result() except Exception as e: print(f"评估失败：{str(e)}") def evaluate_item(item): """评估单个数据项""" handler = item.get_langchain_handler(run_name="batch-eval") # ...其余评估逻辑... ``` ## 6. 评估结果分析与可视化 Langfuse提供了丰富的分析工具来解读评估结果。通过Python SDK，我们可以提取数据并生成自定义报告： ```python def get_evaluation_stats(dataset_name: str): """获取评估统计数据""" traces = langfuse.get_traces(name="capital-quiz-eval") scores = [] for trace in traces: for score in trace.scores: if score.name == "exact_match": scores.append(score.value) if not scores: return None return { "dataset": dataset_name, "total_items": len(scores), "accuracy": sum(scores) / len(scores), "perfect_count": sum(1 for s in scores if s == 1), "failed_count": sum(1 for s in scores if s == 0) } ``` 对于更复杂的分析，可以将数据导出到Pandas DataFrame： ```python import pandas as pd def export_evaluation_data(run_name: str) -> pd.DataFrame: """导出评估数据到DataFrame""" traces = langfuse.get_traces(name=run_name) records = [] for trace in traces: record = { "trace_id": trace.id, "input": trace.input.get("country", ""), "output": trace.output, "latency_ms": trace.latency, "cost": trace.cost } # 添加所有评分项 for score in trace.scores: record[score.name] = score.value records.append(record) return pd.DataFrame(records) ``` ## 7. 高级技巧与最佳实践 在多个生产级项目中，我们总结了以下提升评估系统效能的经验： **提示词版本控制** ```python # 在评估时记录使用的提示词版本 langfuse.score( name="prompt_version", value=1.0, trace_id=handler.get_trace_id(), observation_id=handler.get_trace_id() # 关联到同一trace ) ``` **多维度评估** ```python # 同时执行多个维度的评估 def evaluate_multidimensional(output, expected): return { "exact_match": exact_match_evaluation(output, expected), "contains_keywords": contains_keywords_evaluation(output, ["capital"]), "length_appropriate": 1 if 10 < len(output) < 50 else 0, "llm_judge": llm_judge_evaluation( prompt="What is the capital?", output=output, criteria="准确性和简洁性" ) } ``` **自动化评估流水线** ```python from prefect import flow, task @task def prepare_dataset(dataset_name): return langfuse.get_dataset(dataset_name) @task def evaluate_item(item, system_prompt): handler = item.get_langchain_handler(run_name="auto-eval") # ...评估逻辑... return results @flow(name="LLM Evaluation Pipeline") def evaluation_flow(dataset_name: str, system_prompt: str): dataset = prepare_dataset(dataset_name) for item in dataset.items: evaluate_item.submit(item, system_prompt) ``` 在实际项目中，我们发现将Langfuse与CI/CD管道集成可以显著提升迭代效率。例如，可以在每次代码提交后自动运行评估，只有当关键指标超过阈值时才允许合并。