基于langfuse api提取指定trace的日志#

背景#

在开发过程中，我需要批量的获取指定的trace的日志，但是langfuse的ui没法做到批量的下载指定trace为json文件，这里简单介绍一下如果通过langfuse api来实现。

sessionId获取#

这里需要用到sessionId来获取当时那一批session对应的全部trace。获取方法很简单，在langfuse的ui界面里，找到trace里的sessionId列，然后复制出来

获取langfuse key和BASE_URL#

在langfuseui的setting里，找到API Keys，创建apikey，这里会获得两个key，分别是PUBLIC_KEY和SECRECT_KEY，都需要记录下来

这里如果你的langfuse在公网，那么BASE_URL就是https://api.langfuse.com/api/public/，如果你的langfuse在本地，那么BASE_URL就是http://YOUR_HOST/api/public/

相关代码#

最后把sessionId，PUBLIC_KEY，SECRECT_KEY,BASE_URL写入到langfuse_api.py中，然后运行langfuse_api.py，就可以获取到指定trace的日志。

1
import requests
2
import json
3

4
# 配置
5
PUBLIC_KEY = ""
6
SECRET_KEY = ""
7
BASE_URL = "http://10.65.171.100:33000/api/public/"
8

9
# 要筛选的 message_id
10
TARGET_MESSAGE_ID = "17400353"
11
PAGE_SIZE=100
12

13
# 存储结果
14
results = 0
15
page = 1
16
has_more = True
17
target_date = "2025-08-06"
18
hour = 18
19

20

21
while has_more:
22
    print(f"📌 正在获取第 {page} 页（每页 {PAGE_SIZE} 条）...")
23

24
    url = f"{BASE_URL}/traces"
25
    response = requests.get(
26
        url,
27
        auth=(PUBLIC_KEY, SECRET_KEY),
28
        params={
29
            "page":page,
30
            "limit":PAGE_SIZE,
31
            "sessionId":"test_cases_fivedoctors_20250806185514"
32
        },
33
    )
34
    res = response.json()
35

36

37
    if response.status_code != 200:
38
        print(f"❌ 请求失败: {response.status_code} - {response.text}")
39
        break
40

41
    try:
42
        data = response.json()
43
    except json.JSONDecodeError:
44
        print("❌ 返回内容不是 JSON 格式，可能是网络问题或地址错误")
45
        print("响应内容:", response.text)
46
        break
47

48
    if isinstance(data, list):
49
        traces = data
50
        has_more = len(traces) == PAGE_SIZE  # 如果刚好满页，可能还有下一页
51
    else:
52
        traces = data.get("data", [])
53
        current_page_size = len(traces)
54
        has_more = current_page_size == PAGE_SIZE
55

56
    for trace in traces:
57
        metadata = trace.get("metadata") or {}
58
        id = metadata["message_id"]
59
        with open(f'./result/{id}.json',"w",encoding="utf-8") as f:
60
            json.dump(trace, f, indent=2, ensure_ascii=False)
61
            results+=1
62

63

64
    if len(traces) < PAGE_SIZE:
65
        has_more = False
66

67
    page += 1
68

69
print(f"✅成功导出{results}条日志")