ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangChain智能体追踪数据批量导出:LangSmith API 本地化实践

LangChain智能体追踪数据批量导出:LangSmith API 本地化实践 接手过 LangChain 智能体项目的朋友应该都有同感跑通流程只是第一步真正让人头疼的是后面那堆说不清的追踪数据。LangChain 本身会把每次智能体调用的轨迹、事件、Token 消耗、耗时全部记录在 LangSmith 上开发调试的时候后台点点很方便但一旦进入联调、回归、成本核算或者合规审计阶段你就得面对一个现实问题——怎么把这一大堆追踪数据批量、干净、可复用地把导出本地。我这次写的就是这件事围绕 LangChain 智能体开发过程中如何通过 LangSmith 的 API 把项目里的运行追踪数据批量捞下来整理成结构化文件再做本地分析和归档。整个过程是基于 LangChain LangSmith 的常见实践写出来的适合那些已经用 LangSmith 做追踪、但想把数据拿到手里的开发者和算法工程师。1. 整体思路拆解追踪数据的价值与导出方案选型1.1 LangChain 追踪数据到底长什么样先花点时间把追踪数据这个词说明白。我用 LangChain 开发智能体时每跑一次 Agent 的完整链路LangSmith 都会自动生成一个 Run Object。这个对象记录的不只是调用了哪个模型而是一条完整的调用链外部输入被拆成了哪些内部步骤每一步进了哪个 LLM、用了哪些工具、工具返回了什么最后又是怎么组装成最终结果的。它更像一份智能体运行解剖报告。字段层面包括 run_id、session会话ID、name节点名称、run_type是链、模型、工具还是检索器、inputs/outputs每一步的输入输出、start_time/end_time时间戳、duration耗时、token_usage输入/输出 Token 数以及 error 信息。这些数据的价值在于它们能支撑很多日常开发很难量化判断的事排查这一步为什么慢——从 trace 里能看到具体卡在哪个工具或者模型调用上。成本拆分——按项目、按会话、按模型维度汇总 Token 消耗给老板和客户一个交代。回归对比——改了 Prompt 或者工具逻辑之后跑一批历史数据对比效果有没有变差。合规存档——有些行业要求保留不可变的调用日志LangSmith 后台的保留期可能不够必须本地归档。所以追踪数据不是给 LangSmith 看的它是我们自己的资产。未导出之前所有数据都存在别人家的数据库里这本身就有点被动。1.2 导出需求背后的常见场景什么场景下你会产生批量导出这个念头我根据自己的项目经验总结出了三个典型场景。第一类是平台迁移或者说云成本控制。LangSmith 的免费额度对个人开发和小项目还够用但团队项目跑一两个月数据量大留存策略一变你就会想与其什么都依赖后台不如定期把数据拉回来放自己机器上。第二类是本地化分析。很多人有自定义的 BI 或者监控看板需求LangSmith 自带的筛选器不够灵活而导出的数据可以进 Pandas、Excel、甚至思源笔记做额外的分析、统计、可视化。这个优势我后面在实操里会演示。第三类是质量评估与数据集构建。LangChain 智能体测试的时候经常需要一批真实的调用记录来挑选正负样本或者为评估集做素材。导出追踪数据就是现成的爬楼材料可以直接用来构造 few-shot 示例或者给自动评估器当基准答案。这也是我把方案最终锁定在LangSmith API 批量拉取 本地清洗落盘的原因。1.3 方案选型为什么不用浏览器抓包我知道有朋友会说既然 LangSmith 后台有 UI那我能不能直接用浏览器 F12 抓接口或者写个爬虫按页面翻着把数据抠下来理论上能但我强烈不建议作为常规方案。第一前端接口参数是内部的LangSmith 页面迭代很快一改版爬虫就废。第二前端分页和懒加载机制会严重限制单次拿到的数据量翻几百条数据后台卡得不行做批量导出效率极低。第三内部接口没有官方的鉴权和限流保障用多了可能导致你的账号被风控得不偿失。所以最可靠、最可持续的路径就是走 LangChain 官方维护的 langsmith SDK调用其提供的 list_runs 等公开方法。虽然没有哪个方法叫 export_all但组合分页接口加本地文件追加写入完全可以拼出一个可靠的批量导出工具。2. 核心细节解析Run 对象结构、导出格式与批量拉取策略2.1 Run 对象的关键字段与解析重点写导出脚本之前你得先认识我们要处理的货长什么样。LangSmith 的 Run 对象看似简单实际嵌套很深这也是很多人导出了但看不懂数据的原因。第一层是你一眼能看懂的扁平字段例如字段类型说明idstrRun 的唯一 ID去重和断点续传的钥匙namestr节点名字通常等于 Chain 名、Model 名或工具名run_typestrchain / llm / tool / retriever 等start_time / end_timedatetime开始与结束时间点durationfloat以毫秒为单位的耗时inputsdict该步骤的输入参数outputsdict该步骤的输出内容token_usagedict部分模型节点会带 Token 使用情况errorstr / None节点异常时的报错信息child_runslist它内部调用的子节点列表这才是重头戏最容易被忽略的是 child_runs。一个智能体的完整运行往往是一个嵌套结构。顶层 Run 是 Agent 的执行入口它下面可能挂着一串 llm 和 tool 类型的子 Run。如果你只导出了顶层就没法还原每一次工具调用的输入输出那导出材料价值起码丢一半。这里我额外提醒一点不同版本的 langsmith SDK 在不同模型提供商的 traces 上token_usage 字段的标准程度并不一致。以 OpenAI 系为主的项目一般数据比较全本地向量库或自宿主模型项目的 token_usage 常常是空的。导出之后做报表时需要考虑空值处理不能硬算。2.2 时间窗口、分页与批量拉取策略LangSmith 的 API 本质上是面向交互式查询的不是为全量导出设计的。所以你想一瓶倒完不太现实必须分页、分时间片地慢慢拉。我把它总结成三个原则。第一时间永远按窗口切。LangSmith 的 list_runs 允许你传 start_time 和 end_time但一次查询跨度太长服务端响应会很慢也容易超时或返回不完整。常见做法是把你要导出的区间按天切片比如从 2025-01-01 到 2025-03-01那就循环 60 天每天作为一个查询窗口。如果某一天的运行量特别大比如压测日再精细到小时级别。第二游标分页必须接住。list_runs 支持的并不是简单的 offset 翻页而是 cursor 游标。你每次调用会拿到一个下一页的游标等游标为空或返回条数小于 limit才表示这页拉完了。有人图省事直接用页码做偏移结果数据重复或者漏数据最后对账的时候会非常痛苦。第三量力而行分批落盘而不是全部攒在内存里一次写文件。LangSmith 大项目的 run 数量以十万计全部塞进列表再序列化内存先撑不住。更稳的思路是拉一批、解析一批、追加写一批。实际代码我放在第 3 章。2.3 导出格式怎么选JSON、CSV 还是两者都留导出格式的选择本质上取决于你接下来要拿数据干什么。JSON 是信息无损的格式适合做归档、二次程序处理和迁移。Run 对象本来就是 JSON 形态导出的 JSON 可以完整保留 child_runs、inputs、outputs 等嵌套结构。缺点是文件大、人看起来不直观。CSV 适合数据分析、Excel 处理和给非技术人员看。把关键字段拍平比如每一行是一条 run带时间、类型、耗时、Token 数、输入输出摘要做透视表和可视化非常方便。缺点很明显嵌套的 child_runs 表达不了必须做展平取舍。我的建议是归档场景导出 JSON分析场景导出 CSV。实际操作时脚本可以同时产两套反正核心的解析逻辑就写一遍。下面第 3 节我会直接给出能跑起来的核心代码。3. 实操复现用 LangSmith API 写批量导出脚本3.1 环境准备与鉴权正式写代码之前先做环境准备工作。你至少需要一个 Python 3.9 的环境和一个能访问目标 LangSmith 项目的 API Key注意这个 Key 需要有读取权限。安装依赖就一条命令pip install langchain langsmith pandas tqdm这里重点说下 API Key 的管理。如果你是在自己的开发机调试可以直接从后台复制如果是团队项目建议把 key 放到环境变量里不要硬编码进脚本因为这玩意儿一旦泄漏相当于把整个项目的追踪日志全暴露了。import os os.environ[LANGCHAIN_API_KEY] ls_... # 推荐在 .env 或环境变量里配接下来初始化 Client 对象from langsmith import Client client Client()3.2 查询项目并确认项目 ID导出的单位是项目。你需要先找到目标项目的名称或者 ID。常见做法是列出所有项目然后按名称筛选from datetime import datetime, timedelta, timezone projects client.list_projects() for p in projects: if p.name my-agent: project_id p.id print(找到项目, project_id) break这里有一个细节list_projects 返回的可能是迭代器要看具体 SDK 版本的实现。老版本是直接返回列表新版本可能变成了分页迭代器处理方法略有不同但核心都是拿到 project_id 或者 project_name 供后续查询。如果你的项目名里带有中文或特殊字符注意编码问题Py 3 下一般没问题。如果你同时要导出多个项目可以用一个简单的列表来组织配置TARGET_PROJECTS [ (my-agent, 2025-01-01, 2025-03-01), (my-agent-v2, 2025-02-01, 2025-03-01), ]我个人建议时间范围写成参数化配置而不是散落在循环里后面调整会省心很多。3.3 核心代码分批拉取与追加写入这一步是整个脚本的心脏。我把它拆成三个函数来设计fetch_runs(project_id, start_time, end_time)负责按时间窗口和游标分页拉取一次返回一批运行记录。parse_run_safely(run)负责把 Run 对象转成便于分析的字典遇到空值或异常不报错中断。write_batch_to_jsonl(records, output_file)负责把一批数据追加写入本地 JSON Lines 文件。from langsmith.schemas import Run def fetch_runs(project_id, start_time, end_time, page_size50): runs [] cursor None while True: result client.list_runs( project_idproject_id, start_timestart_time, end_timeend_time, limitpage_size, cursorcursor, ) batch [r for r in result] runs.extend(batch) if not batch or len(batch) page_size: break # 这里的 next_cursor 是分页游标的关键 cursor getattr(result, next_cursor, None) if not cursor: break return runs这里要额外解释一下 cursor 的获取。不同版本的 langsmith SDK 返回结构会有差异我自己的经验是用getattr(result, next_cursor, None)来容错有的版本直接从 list_runs 的返回值里拿到next_cursor有的版本藏在元组或响应对象的属性里。万一你用的版本拿不到游标退而求其次的临时方案是直接用offset循环但那样重复和漏数据的风险就得自己兜住了。写完上面这个基础的 fetch 函数你会发现一个实际问题如果某一天运行记录特别多直接把当天所有 run 一口气扩展到内存再返回依然有压力。所以我实际更常用的版本是把拉取和落盘绑在一起拉到一批就写出到文件def stream_runs_to_jsonl(project_id, start_time, end_time, output_file): cursor None total 0 while True: result client.list_runs( project_idproject_id, start_timestart_time, end_timeend_time, limit50, cursorcursor, ) batch [r for r in result] if not batch: break with open(output_file, a, encodingutf-8) as f: for run in batch: f.write(run.json() \n) total len(batch) print(f已写入 {len(batch)} 条累计 {total} 条) cursor getattr(result, next_cursor, None) if len(batch) 50 or not cursor: break这样的好处是一旦脚本中途崩了已经导出的数据全在磁盘上不会因为内存清空而白跑。3.4 数据整理从 JSONL 到 DataFrame 再到 CSVJSONL 适合程序读但做分析还是得进 DataFrame。下面这段代码负责把 JSONL 文件读进来、展平关键字段、输出 CSVimport pandas as pd import json def jsonl_to_dataframe(file_path, limitNone): rows [] with open(file_path, r, encodingutf-8) as f: for i, line in enumerate(f): if limit and i limit: break data json.loads(line) # 只取扁平字段嵌套的 child_runs 先略过 rows.append({ id: data.get(id), name: data.get(name), run_type: data.get(run_type), start_time: data.get(start_time), end_time: data.get(end_time), duration_ms: data.get(duration), status: data.get(status), prompt_tokens: data.get(token_usage, {}).get(prompt_tokens, 0), completion_tokens: data.get(token_usage, {}).get(completion_tokens, 0), input_preview: str(data.get(inputs))[:200], output_preview: str(data.get(outputs))[:200], error: data.get(error), }) return pd.DataFrame(rows)用 Pandas 的好处是可以用一行代码做成本汇总比如按 run_type 统计 Token 使用情况df jsonl_to_dataframe(my-agent_traces.jsonl) cost_summary df.groupby(run_type)[[prompt_tokens, completion_tokens]].sum() print(cost_summary)这在智能体优化中特别好用——你会直观看到每次 Agent 跑下来浪费的 Token 到底花在哪些多余节点上然后决定要不要裁剪工具描述或改路由。那 CSV 怎么生成在 DataFrame 后加一行df.to_csv(traces_flat.csv, indexFalse, encodingutf-8-sig)就完事了。注意用utf-8-sig而不是utf-8是为了照顾 Windows 下 Excel 打开 CSV 会乱码的问题这个坑我踩过很多次。3.5 加上断点续传和增量导出既然标题叫批量导出量大了之后脚本持续跑几个小时的场景很常见。这时候最怕的是跑到一半网络抖了一下整个流程前功尽弃。我的办法是维护一个已导出 run_id的红黑集合文件。每次写入一条 Run 时顺带把它压进一个集合每次进入新批次的时候先检查这批 run 的 id 是否已经在集合里。这样即使脚本中断下次重跑同一区间已导出的记录会被直接跳过。import os exp_file exported_ids.txt exp_ids set() if os.path.exists(exp_file): with open(exp_file, r, encodingutf-8) as f: exp_ids set(line.strip() for line in f) # 在循环里 needed [r for r in batch if r.id not in exp_ids] # 处理完后 with open(exp_file, a, encodingutf-8) as f: for r in needed: f.write(r.id \n)增量导出的思路也一样记录下上次导出的时间戳下次只从那之后拉。LangSmith 的查询按新记录生成时间过滤跟项目本身的 run 逻辑天然吻合。4. 常见问题与排查技巧实录4.1 游标分页导致数据漏拉或者重复这是用 LangSmith API 导出时最典型的问题。症状是后台 UI 显示某天有 1200 条运行记录你导出来却发现只有 1100 条或者明明导出了一批结果重新跑一遍发现多了几条重复的。排查思路要区分两个层面如果漏的是整体数量优先怀疑时间跨度太大服务端返回不全把时间窗调小到当天甚至当小时再试如果漏的是某一类 run_type比如特别短的工具调用很可能是 query 条件里误加了 filter导致某些记录被过滤掉了重复的问题则基本可以认定是游标没有正确传递也可能是因为你同时用了 offset 之后又把游标混在一起服务端状态错乱了。我把排查路径总结成了一张速查表现象优先级排查点解决建议导出数量明显偏少时间窗口跨度过大按天拆窗口重试抽检有缺漏记录过滤条件太严去掉不必要的 filter 后再拉一轮对比重复记录游标未接住或混用 offset确认 next_cursor 每次都更新游标取不到SDK 版本不一致升级或降级到稳定版本别硬凑我的个人习惯是拉完全量数据后用一个 SQL 或者 Pandas 的nunique()去重统计以 run_id 为维度做一个完整性校验。宁可多拉一点重复数据也别漏数据。去重放在后面清洗时做。4.2 导出的追踪数据里子事件丢失有次我导出完之后做效果分析发现所有顶层 Run 都齐了但打开一看child_runs 列表是空的导致工具调用细节全没了。排查才发现list_runs 默认返回的 Run 对象是浅层版本不自动带上子运行需要显式指定或者在拉取之后做二次展开。LangSmith 的 API 支持查询时带 filter 或者参数去包含完整 trace具体参数名在不同 SDK 版本可能叫include_stats、include_child_runs或通过传递trace方式。我试验下来最稳的方案是先按项目把顶层 Run 都拉回来再对每个 Run 的 id 调用client.read_run去获取完整子结构。虽然多了一些请求次数但数据完整性有保证。如果你的环境对请求频率有严格限制也可以只在某些特定会话上二次展开不必全量做。这里提供一段二次展开的示例代码def get_full_trace(client, run_id): try: return client.read_run(run_id) except Exception as e: print(f读取 {run_id} 失败{e}) return None这个函数写得很简单但实际用起来能解决百分之八十的子事件消失问题。你还可以顺手把read_run拿到的不完整 Run 缓存进内存字典避免同一个 run_id 被重复读两次。4.3 导出太慢怎么优化当数据量到几十万条单线程拉取会慢到让人怀疑人生。但实际上 LangSmith API 的瓶颈主要在网络往返而不在服务端计算。所以提升效率最直观的办法是并发。我们可以用ThreadPoolExecutor把时间窗口切片后并发拉取from concurrent.futures import ThreadPoolExecutor, as_completed day_windows [(2025-01-01, 2025-01-02), (2025-01-02, 2025-01-03)] def fetch_day(win): start, end win return fetch_runs(project_id, start, end, page_size50) with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(fetch_day, w) for w in day_windows] for fut in as_completed(futures): batch fut.result() write_batch(batch)有一点要提醒并发数不建议开太高LangSmith 对单个账号的 API 是有速率限制的。如果 429 报错频繁建议退回单线程或者把并发数降到 2、3并加上退避重试。另外一个隐藏的坑是并发写入同一个文件时要用加锁或者让每个线程写独立文件最后再合并否则会出现文件内容互相覆盖。如果是追求极致速度的场景也可以考虑通过后台导出功能来做全量导出然后程序只需要下载归档文件即可。不过这个功能在不同版本的后台里位置不一样我这里不展开。4.4 敏感信息导出后如何脱敏我相信很多智能体项目内部会处理用户的隐私信息比如手机号、地址、甚至账号凭证。导出追踪数据时如果不加处理等于把一批敏感数据直接搬进了本地文本文件。个人项目还好企业项目这就是信息安全事故。我的建议是在写 JSONL 或者 CSV 之前先做一道脱敏过滤器。这个过滤器最基本要做两件事一是把明确是密钥/口令的字段替换成掩码比如用正则把形如sk-[A-Za-z0-9]的字符串替换成sk-***二是对常见字段如手机号、邮箱做局部打码。import re MASK_PATTERNS [ (re.compile(rsk-[A-Za-z0-9_-]), sk-***), (re.compile(r\b1[3-9]\d{9}\b), 138****0000), (re.compile(r\b[\w.][\w.-]\.\w\b), ******.com), ] def mask_text(text): if text is None: return text for pattern, repl in MASK_PATTERNS: text pattern.sub(repl, text) return text再提醒一点脱敏的位置要选对。input_preview 和 output_preview 这类预览字段必须脱敏如果你导出的还是完整 JSON那么所有嵌套的 inputs、outputs 里也要递归脱敏。我个人的经验是先写一个递归函数处理整个 dict再把结果落盘这样最稳妥。写在最后的实操体会这套脚本我已经在多个智能体项目里跑过了从最初的单项目手工翻页到现在配置化批量导出的过程踩过的坑主要集中在三处游标没接住导致数据不全、子事件没展开导致分析失真、以及本地落盘格式不统一导致后续处理要反复洗数据。如今我的习惯是先按天切片预跑一遍小范围测试确认数量对得上再全量跑导出完成后立刻对 run_id 做去重校验然后用 Pandas 看一份基础统计快照确认项目名、时间范围、节点类型分布都正常这份数据才敢拿来用。如果后续你的项目需要持续监控智能体质量变差或者 Token 成本漂移这套数据就是很好的底座甚至可以把它接到你自己的指标系统里做按天的趋势报警。别嫌这一步基础做扎实了整个智能体开发流程的底气和效率都会完全不一样。
返回列表