跳转到主要内容
SimpleLLMFunc 已集成 Langfuse，可用于观测 LLM 调用、工具调用和多步执行链路。你可以把它作为生产环境中的基础可观测层，追踪输入、输出、Token 用量和调用结构。

​
能力概览

LLM 生成追踪

自动记录输入消息、输出内容、模型参数和 Token 使用统计。

工具调用观测

追踪工具调用的参数、执行结果、耗时和调用层级。

嵌套链路支持

支持复杂的多层调用链路和后续 generation 追踪。

流式响应兼容

兼容流式与非流式 LLM 响应，不改变原有开发方式。

​
安装和配置

1

安装 Langfuse 依赖

Langfuse 已包含在框架依赖中；如果你是在现有环境中补装，也可以直接安装：
poetry install
2

获取项目凭据

  1. 访问 Langfuse 并注册账户
  2. 创建新项目
  3. 获取项目的 Public Key 和 Secret Key
3

设置环境变量

export LANGFUSE_PUBLIC_KEY="your_public_key"
export LANGFUSE_SECRET_KEY="your_secret_key"
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_EXPORT_ALL_SPANS="true"
export LANGFUSE_ENABLED="true"
4

直接开始使用

SimpleLLMFunc 会自动读取环境变量并在需要时连接 Langfuse；你不需要手动初始化观测器。
from SimpleLLMFunc import llm_function

llm = ...

@llm_function(llm_interface=llm)
async def my_function(text: str) -> str:
    """功能描述"""
    pass

result = await my_function("test")
当前实现不会根据 LANGFUSE_ENABLED 自动跳过观测调用。如果你需要按环境开关启停，请在应用层自行控制是否挂载和调用相关逻辑。

​
配置方式

export LANGFUSE_PUBLIC_KEY="your_public_key"
export LANGFUSE_SECRET_KEY="your_secret_key"
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_ENABLED="true"
如需手动获取 Langfuse 客户端或读取当前配置，可使用： 
from SimpleLLMFunc.observability import langfuse_config, get_langfuse_client

config = langfuse_config
print(f"Langfuse 已启用: {config.LANGFUSE_ENABLED}")

client = get_langfuse_client()
if client:
    pass

​
使用示例

import asyncio
from SimpleLLMFunc import llm_function, OpenAICompatible

llm = OpenAICompatible.load_from_json_file("provider.json")["openai"]["gpt-3.5-turbo"]

@llm_function(llm_interface=llm)
async def analyze_text(text: str) -> str:
    """分析文本内容并提供摘要"""
    pass

async def main():
    result = await analyze_text("这是一段需要分析的文本...")
    print(result)

asyncio.run(main())

from SimpleLLMFunc import llm_function, tool

@tool(name="calculate", description="执行数学计算")
async def calculate(expression: str) -> dict:
    """执行数学表达式计算"""
    result = eval(expression)
    return {"expression": expression, "result": result}

@llm_function(
    llm_interface=llm,
    toolkit=[calculate],
    max_tool_calls=3,
)
async def math_assistant(question: str) -> str:
    """数学助手，可以回答数学问题并进行计算"""
    pass

result = await math_assistant("计算 15 * 8 + 32 的结果")

from SimpleLLMFunc import llm_chat

@llm_chat(
    llm_interface=llm,
    toolkit=[calculate],
    max_tool_calls=2,
)
async def chat_bot(message: str, history: list = None):
    """智能聊天机器人"""
    pass

history = []
async for response, updated_history in chat_bot("你好，请帮我计算一些数学题", history):
    if response.strip():
        print(response)
    history = updated_history

​
追踪数据结构

每个 LLM 调用会创建一个 Generation，通常包含：
  • 输入消息列表
  • 输出内容与工具调用
  • 模型名称和参数
  • Token 使用量与成本信息
  • 流式模式、可用工具数量等元数据
每个工具调用会创建一个 Tool 观测，包含：
  • 工具调用参数
  • 工具执行结果
  • 工具调用 ID、执行时间等元数据
Function Call (Span)
├── Initial Generation
│   ├── Input: Messages
│   ├── Output: Response + Tool Calls
│   └── Usage: Token counts
├── Tool Call 1 (Tool)
│   ├── Input: Parameters
│   └── Output: Result
├── Tool Call 2 (Tool)
│   ├── Input: Parameters
│   └── Output: Result
└── Follow-up Generation
    ├── Input: Updated Messages
    ├── Output: Final Response
    └── Usage: Token counts

​
配置选项

​
环境变量

变量名描述默认值必需
LANGFUSE_PUBLIC_KEYLangfuse 公钥-是
LANGFUSE_SECRET_KEYLangfuse 私钥-是
LANGFUSE_BASE_URLLangfuse 服务器地址https://cloud.langfuse.com否
LANGFUSE_EXPORT_ALL_SPANS导出所有 OpenTelemetry spanstrue否
LANGFUSE_ENABLED是否启用观测true否

​
Langfuse v4 注意事项

  • v4 默认会过滤非 LLM 的 OpenTelemetry spans；框架默认保持 v3 的“导出全部 spans”行为。如需使用 v4 的智能过滤，请设置 LANGFUSE_EXPORT_ALL_SPANS=false
  • v4 会将 metadata 值规范化为字符串并限制长度，框架侧已自动做字符串化处理

​
最佳实践

# 开发环境
export LANGFUSE_ENABLED="false"

# 生产环境
export LANGFUSE_ENABLED="true"
export LANGFUSE_PUBLIC_KEY="prod_public_key"
export LANGFUSE_SECRET_KEY="prod_secret_key"
当前版本不会自动读取 LANGFUSE_ENABLED 来跳过观测调用。如果你需要按环境开关，请在应用层自行判断。
import logging

logging.getLogger("SimpleLLMFunc").setLevel(logging.INFO)
框架会自动处理 Langfuse 相关错误，但建议持续观察应用日志。
  • Langfuse 调用是异步的，通常不会阻塞主业务逻辑
  • 高并发场景下，建议结合客户端配置和事件流监控一起调优
  • 发送到 Langfuse 前，请先脱敏敏感数据
  • 如果数据敏感，优先考虑自托管 Langfuse 实例

​
故障排除

常见现象：
警告: Langfuse 观测功能未启用
排查方式：
  • 检查环境变量是否正确设置
  • 确认 langfuse 包已安装
常见现象：
Langfuse 初始化失败: Connection error
排查方式：
  • 检查网络连接
  • 验证 API 密钥是否正确
  • 确认 LANGFUSE_BASE_URL 设置正确
  • 检查 Langfuse 仪表板项目设置
  • 确认使用的是正确的项目密钥
  • 等待数据同步，通常只需要几秒钟
import logging

logging.getLogger("SimpleLLMFunc").setLevel(logging.DEBUG)
logging.getLogger("langfuse").setLevel(logging.DEBUG)

​
相关示例与链接

Token 用量统计示例

查看 llm_function + 事件流 + 用量统计的完整链路。

事件流聊天示例

查看对话、工具调用与事件流协同的完整例子。

Langfuse 官方文档

阅读 Langfuse 的原生产品文档和平台能力说明。