使用 Label Studio 与 ReactCode 审阅 Langfuse 追踪数据的完整实战指南【免费下载链接】label-studioLabel Studio is a multi-type data labeling and annotation tool with standardized output format项目地址: https://gitcode.com/GitHub_Trending/la/label-studio本指南面向 LLM 应用开发者与评估工程师将 Langfuse 中捕获的 LLM 追踪Traces拉取到 Label Studio Enterprise使用自定义 ReactCode 三面板标注界面让领域专家对每一轮 Agent 对话进行逐轮人工评估最终产出结构化标注结果直接服务于质量报告、Prompt 优化与 LLM-as-a-judge 流水线。读完本文你将掌握从「Langfuse 观测数据」到「Label Studio 专家评估任务」的完整落地流程并理解其背后的数据标准化设计与 ReactCode 接口原理。本教程基于开源仓库 label-studio 文档目录中的 how_to_review_langfuse_traces_with_label_studio.md 编写仓库源码与文档可作为进一步研究依据。0. 前置要求Label Studio Enterprise 与 ReactCode本教程的核心界面依赖ReactCode 模板这是Label Studio Enterprise 专属功能。ReactCode 允许你在标注界面中嵌入完全自定义的 React 组件——本案例中即一个三面板的 Trace 审阅 UI。从仓库文档 docs/source/tags/reactcode.md 可以看到ReactCode标签让你在 Label Studio 内嵌入自定义标注 UI同时将输出保存为标准的 Label Studio 标注region/result格式因此可以继续使用 Label Studio 的标注管理、审核工作流与数据导出能力。在完成第 2 节之前你需要准备一个运行中的 Label Studio Enterprise 实例从账户设置中生成一个 API Key生成方式见 access_tokens.md。1. 安装与配置依赖安装在 Python 环境中安装以下依赖对应原文档第 1 节!pip -q install requests label-studio-sdk python-dotenv langfuse langchain langchain-anthropic anthropic langgraph各依赖用途requests用于直接调用 Langfuse REST APIlabel-studio-sdk用于创建 Label Studio 项目与导入任务langfuse与langchain/langgraph用于第 3 节生成示例 Traceanthropic用于驱动带扩展思考extended thinking的 Claude 模型python-dotenv用于加载环境变量。环境变量配置在仓库根目录或与 notebook 同目录创建.env文件# Label Studio Enterprise LABEL_STUDIO_HOSThttp://localhost:8080 # 或你的 LS Enterprise 实例地址 LABEL_STUDIO_API_KEYyour_label_studio_api_key # Langfuse LANGFUSE_BASE_URLhttps://cloud.langfuse.com # 或你的自托管 Langfuse 地址 LANGFUSE_PUBLIC_KEYyour_langfuse_public_key LANGFUSE_SECRET_KEYyour_langfuse_secret_key LANGFUSE_PROJECTyour_project_name # 项目名称仅用于在 Label Studio 中展示 # Anthropic仅第 3 节生成示例 Trace 时需要 ANTHROPIC_API_KEYyour_anthropic_api_key注意Langfuse 的 API Keypublic secret 密钥对本身已限定到特定项目因此LANGFUSE_PROJECT仅作为展示名使用无需解析项目 ID。Langfuse 配置在 Langfuse 平台创建账户后生成 API Key 对并记录你的项目 Base URL。Label Studio 配置安装 Label Studio 后在账户设置中生成 API Token获取方式见 access_tokens.md。在 Python 中加载这些变量import os from dotenv import load_dotenv load_dotenv(overrideTrue) load_dotenv(os.path.join(os.path.dirname(os.getcwd()), .env), overrideTrue) # Label Studio Enterprise LABEL_STUDIO_HOST os.getenv(LABEL_STUDIO_HOST, http://localhost:8080) LABEL_STUDIO_API_KEY os.getenv(LABEL_STUDIO_API_KEY, ) # Langfuse LANGFUSE_BASE_URL os.getenv(LANGFUSE_BASE_URL, https://cloud.langfuse.com) LANGFUSE_PUBLIC_KEY os.getenv(LANGFUSE_PUBLIC_KEY, ) LANGFUSE_SECRET_KEY os.getenv(LANGFUSE_SECRET_KEY, ) LANGFUSE_PROJECT os.getenv(LANGFUSE_PROJECT, ) # Anthropic仅第 3a 节生成示例 Trace 时使用 ANTHROPIC_API_KEY os.getenv(ANTHROPIC_API_KEY, ) print(LABEL_STUDIO_HOST:, LABEL_STUDIO_HOST) print(LANGFUSE_BASE_URL:, LANGFUSE_BASE_URL) print(LANGFUSE_PROJECT:, LANGFUSE_PROJECT or (not set — will fetch all traces)) print(Has LABEL_STUDIO_API_KEY?, bool(LABEL_STUDIO_API_KEY)) print(Has LANGFUSE keys?, bool(LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY)) print(Has ANTHROPIC_API_KEY?, bool(ANTHROPIC_API_KEY))评估管线总览Langfuse 观测 Label Studio 专家评估本教程将 Langfuse 面向工程师的观测工具与 Label Studio 面向专家的评估界面连接起来形成两条互补的链路第 1 步Langfuse 中的 Trace 收集Langfuse 将 LLM Trace 捕获为带类型的观测GENERATION、TOOL、SPAN、CHAIN提供面向工程师的调试、评分与迭代界面项目级 API Key 让认证非常简单无需查找项目 ID。第 2 步Label Studio 中的专家评估将 Langfuse 中的 Trace 导入 Label Studio形成结构化标注任务领域专家使用自定义 ReactCode UI 逐轮评估支持协作流程多名 SME领域专家可标注同一批 Trace结构化输出可直接汇入质量报告、Prompt 改进与 LLM-as-a-judge 流水线。2. Label Studio ReactCode 配置跳过配置直接克隆项目原文档提供了「Open in Label Studio」按钮可一键将包含完整三面板 ReactCode 标注界面的预配置项目克隆到你的 Enterprise 账户中直接跳到第 4 节导入 Trace。如果希望以编程方式配置请继续阅读本节。三面板标注 UI本教程使用ReactCode标注配置——这是 Label Studio Enterprise 功能允许你嵌入自定义 React 组件作为标注界面。该 UI 包含三个面板面板用途Turns左可滚动的全部轮次列表。支持按角色过滤、按内容搜索。每张卡片展示角色、工具徽章、延迟以及标注后的判定结果。Turn Details中完整内容、工具调用输入/输出、Token 用量、延迟以及 Claude 的扩展思考内容若存在。Annotation右用于评估每一轮的结构化表单——具体标注模型见下。标注模型每轮捕获什么Verdict判定—— Pass 或 FailIssue tags问题标签—— 覆盖 5 大类的分类体系Accuracy Faithfulness准确性与忠实性、Tool Retrieval工具与检索、Reasoning Planning推理与规划、Response Quality回答质量、Safety Compliance安全与合规Severity严重程度—— Critical / Major / Minor / SuggestionExpected behavior期望行为—— 自由文本Agent 本应怎么做Comments评论—— 任何补充说明。底部栏另有trace 级判定Pass / Fail / Mixed用于评估整段对话的整体质量独立于单轮判定。标注配置 XML 结构ReactCode 的标注配置在 XML 中声明。原文档给出的核心结构如下_TEMPLATE_JS是内联的完整 ~40KB React 组件此处以占位示意_TEMPLATE_JS rfunction TraceAnnotator({ React, addRegion, regions, data }) { // 736 行 React 组件定义三面板 Trace 审阅 UI。 // 面板Turns 列表左| Turn details中| Annotation 表单右 // 底部栏轮次统计 trace 级判定Pass / Fail / Mixed // ... 完整实现见 notebook ... } LABEL_CONFIG_XML ( View\n ReactCode styleheight: 95vh nametrace toNametrace outputs\{trace_id:string,turn_id:string,turn_role:string, verdict:string,failure_modes:array,severity:string, expected_behavior:string,comments:string}\\n ![CDATA[\n ) _TEMPLATE_JS ( \n ]]\n /ReactCode\n /View ) print(LABEL_CONFIG_XML[:300] \n...)关键点解读对照仓库文档 docs/source/tags/reactcode.md自引用标签与其他对象标签不同ReactCode可单独使用此时toName必须指向name自身本配置中二者均为traceCDATA 包装复杂的 JS 代码尤其含、、字符时必须用![CDATA[和]]包裹避免被 XML 解析器误解析无 JSX组件内必须使用React.createElement()不支持 JSX 语法outputs 参数定义标注输出的 JSON Schema用于校验与数据导出。本配置声明了trace_id、turn_id、turn_role、verdict、failure_modes数组、severity、expected_behavior、comments等字段——实际标注 JSON 总是存于value.reactcode中。从源码层面看ReactCode标签在后端有完整的配套实现仓库 label_studio/io_storages/react_code_proxy.py 提供了ReactCodeTokenView与ReactCodeResolveView两个接口——前者为 ReactCode iframe 签发限定用户与项目的短期 JWTTTL 默认 3600 秒可配置 6086400 秒后者以该 JWT 代替会话 Cookie 代理存储 URI 的解析。这是因为沙箱 iframe 具有不透明 origin、无法携带 Cookie必须通过 JWT 完成鉴权详见 urls.py 中的路由挂载。这也解释了为什么 ReactCode 界面能够安全地读取任务数据包括云存储与本地上传文件。3. 生成示例 Trace可选如果你已经在 Langfuse 中拥有 Trace请跳过本节——设置GENERATE_TRACES False后直接进入第 4 节。否则本单元格创建一个带多个工具的 ReAct Agent并使用开启扩展思考extended thinking的 Claude运行 4 段多轮对话在你的 Langfuse 项目中产生逼真的 Trace。此步骤需要ANTHROPIC_API_KEY。扩展思考让 Claude 在作答前逐步推理复杂、模糊的问题。思考内容会被捕获进 Trace并显示在 Label Studio UI 中——这正是人工评估 Agent 推理质量的关键素材。GENERATE_TRACES True # 如果已有 Trace设为 False if GENERATE_TRACES: from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langchain_anthropic import ChatAnthropic from langchain.agents import create_agent from langfuse.langchain import CallbackHandler from langfuse import get_client if not ANTHROPIC_API_KEY: raise RuntimeError(ANTHROPIC_API_KEY is required. Set it in your .env or set GENERATE_TRACESFalse.) langfuse get_client() tool def calculator(expression: str) - str: Evaluate a math expression. try: return str(eval(expression)) except Exception as e: return fError: {e} tool def search_knowledge_base(query: str) - str: Search an internal knowledge base for company policies, products, or procedures. kb { refund: Refund policy: Full refund within 30 days. After 30 days, store credit only. Damaged items: full refund at any time with photo evidence., shipping: Standard (5-7 days, free over $50), Express (2-3 days, $12.99), Overnight ($24.99)., warranty: 1-year limited warranty. 2-year extended warranty available for $29.99., pricing: Base $99/mo (10 users), Pro $249/mo (50 users), Enterprise custom. Annual billing saves 20%., } results [v for k, v in kb.items() if k in query.lower()] return results[0] if results else fNo results found for: {query} tool def get_weather(city: str) - str: Get current weather for a city. weather_data { new york: New York: 72°F, Partly Cloudy, Humidity 65%, Wind 8 mph SW, london: London: 58°F, Overcast, Humidity 80%, Wind 12 mph W, tokyo: Tokyo: 82°F, Clear, Humidity 55%, Wind 5 mph NE, paris: Paris: 63°F, Light Rain, Humidity 75%, Wind 10 mph NW, } return weather_data.get(city.lower(), fWeather data not available for {city}) # Claude 开启扩展思考——产生更丰富的 Trace暴露模型的推理过程 llm ChatAnthropic( modelclaude-sonnet-4-5-20250929, max_tokens16000, thinking{type: enabled, budget_tokens: 5000}, ) agent create_agent(llm, [calculator, search_knowledge_base, get_weather]) # 4 段特意设计来触发扩展思考的多轮对话 conversations [ [I bought a product 37 days ago with a manufacturing defect and an extended warranty. What are all my options?, The item costs $289. Can I use store credit toward a new extended warranty while keeping the original warranty claim open?], [We have 60 employees — 40 need full access, 20 need read-only. How do we minimize cost?, If we commit to annual billing and add 15 more full-access users next quarter, whats our 12-month total?], [Im planning a 20-person client retreat. Compare Tokyo, London, and New York on weather and logistics., 12 attendees are in New York, 8 in London. Re-evaluate the three options for minimal travel disruption.], [I ordered 3 items for $180 with express shipping. One arrived damaged — I need a replacement urgently., If I return the damaged item and pay for express shipping on the replacement, whats my net out-of-pocket?], ] # Langfuse 通过传给 Agent 的 CallbackHandler 对每段对话进行埋点 for i, conv_messages in enumerate(conversations, 1): print(f\n--- Conversation {i} ---) handler CallbackHandler() chat_history [] for msg_text in conv_messages: print(f User: {msg_text[:80]}...) chat_history.append(HumanMessage(contentmsg_text)) result agent.invoke({messages: chat_history}, config{callbacks: [handler]}) chat_history result[messages] reply result[messages][-1].content if isinstance(reply, list): reply .join(b.get(text, ) for b in reply if isinstance(b, dict) and b.get(type) text) print(f Assistant: {str(reply)[:100]}...) langfuse.flush() print(f\n✓ Generated {len(conversations)} traces. Proceed to Section 4.) else: print(Skipped trace generation. Proceed to Section 4.)这些对话被刻意设计为「模糊、多约束、需要逐步推理」的场景退款 延保叠加、成本最小化、多城市对比、退货 加急补发以充分触发 Claude 的扩展思考与多工具调用产生值得专家逐轮评估的高质量 Trace。4. Langfuse API 客户端本节从 Langfuse REST API 拉取 Trace 与观测。你的 API Key 已限定到具体 Langfuse 项目因此返回的所有 Trace 都属于该项目。import base64 from typing import Any, Dict, List, Optional import requests def _basic_auth(public_key: str, secret_key: str) - str: token base64.b64encode(f{public_key}:{secret_key}.encode(utf-8)).decode(utf-8) return fBasic {token} class LangfuseClient: def __init__(self, base_url: str, public_key: str, secret_key: str): self.base_url base_url.rstrip(/) self.s requests.Session() self.s.headers.update({ Authorization: _basic_auth(public_key, secret_key), Content-Type: application/json }) def list_traces(self, limit: int 20, page: int 1, from_ts: Optional[str] None, to_ts: Optional[str] None) - Dict[str, Any]: url f{self.base_url}/api/public/traces params: Dict[str, Any] {limit: limit, page: page, fields: core} if from_ts: params[fromTimestamp] from_ts if to_ts: params[toTimestamp] to_ts r self.s.get(url, paramsparams, timeout60) r.raise_for_status() return r.json() def get_trace(self, trace_id: str) - Dict[str, Any]: r self.s.get(f{self.base_url}/api/public/traces/{trace_id}, timeout60) r.raise_for_status() return r.json() def list_observations_v2(self, trace_id: str, limit: int 200) - List[Dict[str, Any]]: 通过 v1 API 获取 trace 的观测避免 v2 parseIoAsJson 400。 url f{self.base_url}/api/public/observations out: List[Dict[str, Any]] [] page, page_size 1, min(max(limit, 1), 100) while True: r self.s.get(url, params{traceId: trace_id, page: page, limit: page_size}, timeout60) r.raise_for_status() data r.json().get(data) or [] out.extend(data) if len(data) page_size: break page 1 return out lf LangfuseClient(LANGFUSE_BASE_URL, LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY) print(Langfuse client ready)实现要点Basic AuthLangfuse 公共 API 使用public_key:secret_key的 Base64 Basic Auth无需 OAuth 流程list_traces分页拉取 Trace 列表支持fromTimestamp/toTimestamp时间窗过滤fields: core只取核心字段以减小负载get_trace按 ID 获取单个 Trace 的完整详情list_observations_v2按traceId分页拉取该 Trace 的全部观测。原文档特别注明此函数刻意走 v1 观测 API以规避 v2parseIoAsJson的 400 错误——这是实践踩坑后的经验之谈值得保留。5. 将 Langfuse Trace 标准化为统一 SchemaLangfuse 将 Trace 存储为带类型的观测GENERATION、TOOL、SPAN、CHAIN。本单元格提取相关观测类型映射为一个扁平的轮次turn序列——同一套 Schema 也被 Braintrust、LangSmith 集成共用——因此 ReactCode UI 无需关心 Trace 来自哪个平台。每一轮携带role、content、tool_name、tool_input、tool_calls、model、usageToken 计数、duration_ms、thinkingClaude 扩展思考块若存在。import json as _json def _to_str(x): if x is None: return if isinstance(x, str): return x try: return _json.dumps(x, indent2, defaultstr) except: return str(x) def _extract_content(obj): if obj is None: return if isinstance(obj, str): return obj if isinstance(obj, dict): for key in (content, text, input, output, result): if isinstance(obj.get(key), str) and obj[key].strip(): return obj[key] return _to_str(obj) if isinstance(obj, list): parts [_extract_content(item) for item in obj if _extract_content(item).strip()] return \n.join(parts) if parts else _to_str(obj) return str(obj) def _normalize_usage(obs): Extract token usage from a Langfuse observation. raw obs.get(usageDetails) or obs.get(usage) if not isinstance(raw, dict): return None return { input_tokens: raw.get(inputTokens) or raw.get(input_tokens) or raw.get(input) or 0, output_tokens: raw.get(outputTokens) or raw.get(output_tokens) or raw.get(output) or 0, } def _duration_ms(start_str, end_str): if not start_str or not end_str: return None try: from datetime import datetime def _parse(s): return datetime.fromisoformat(str(s).replace(Z, 00:00)) return int((_parse(end_str) - _parse(start_str)).total_seconds() * 1000) except: return None def _split_thinking(content): Split Anthropic extended-thinking content blocks into (text, thinking). if isinstance(content, str): return content, None if isinstance(content, list): text_parts, thinking_parts [], [] for block in content: if isinstance(block, dict): if block.get(type) thinking: thinking_parts.append(block.get(thinking, )) elif block.get(type) text: text_parts.append(block.get(text, )) elif isinstance(block, str): text_parts.append(block) return \n\n.join(text_parts), \n\n.join(thinking_parts) or None return str(content) if content else , None def normalize_langfuse_trace(trace, observations): 将 Langfuse trace observations 转换为统一 schema。 观测类型 - GENERATION → 从 input 提取用户消息 从 output 提取助手回复 - TOOL → 将工具执行提取为一个 tool turn - CHAIN, SPAN, AGENT → 跳过结构包装层 trace_id trace.get(id) or trace.get(traceId) obs_sorted sorted(observations, keylambda o: o.get(startTime) or o.get(createdAt) or ) turns [] turn_counter 0 seen_user_messages set() def add_turn(role, content, **kwargs): nonlocal turn_counter if not content or not content.strip(): return turn {turn_id: fturn_{turn_counter}, role: role, content: content.strip(), timestamp: kwargs.get(timestamp, )} for k in (model, usage, tool_calls, tool_name, tool_input, duration_ms, thinking): if kwargs.get(k) is not None: turn[k] kwargs[k] turns.append(turn) turn_counter 1 for obs in obs_sorted: otype (obs.get(type) or ).upper() ts obs.get(startTime) or obs.get(createdAt) or duration _duration_ms(obs.get(startTime) or obs.get(createdAt), obs.get(endTime)) inp, out obs.get(input), obs.get(output) if otype GENERATION: if isinstance(inp, list): for msg in inp: if isinstance(msg, dict) and msg.get(role) user: content msg.get(content, ) if isinstance(content, list): content .join(p.get(text, ) if isinstance(p, dict) else str(p) for p in content) if content and content.strip(): msg_key content[:200] if msg_key not in seen_user_messages: seen_user_messages.add(msg_key) add_turn(user, content, timestampts) if isinstance(out, dict): raw_content out.get(content, ) tool_calls [] for tc in out.get(tool_calls, []): if isinstance(tc, dict): tool_calls.append({tool_name: tc.get(name, unknown), input: _to_str(tc.get(args, tc.get(input, ))), call_id: tc.get(id, )}) assistant_content, thinking _split_thinking(raw_content) if assistant_content and assistant_content.strip(): add_turn(assistant, assistant_content, timestampts, modelobs.get(model) or obs.get(providedModelName), usage_normalize_usage(obs), tool_callstool_calls if tool_calls else None, duration_msduration, thinkingthinking) elif otype TOOL: tool_name obs.get(name) or unknown tool_output _extract_content(out) if out else if tool_output: add_turn(tool, tool_output, timestampts, tool_nametool_name, tool_input_to_str(inp) if inp else , duration_msduration) if not turns: if trace_input : _extract_content(trace.get(input)): add_turn(user, trace_input, timestamptrace.get(timestamp) or ) if trace_output : _extract_content(trace.get(output)): add_turn(assistant, trace_output, timestamptrace.get(timestamp) or ) return { trace_id: str(trace_id), session_id: str(trace.get(sessionId) or trace_id), metadata: { name: trace.get(name), source: langfuse, tags: trace.get(tags) or [], start_time: trace.get(timestamp) or trace.get(createdAt) or , }, turns: turns, } print(✓ Normalization functions defined)标准化逻辑的关键设计按时间排序先按startTime/createdAt排序观测保证轮次顺序与真实对话一致GENERATION双向提取从input消息列表中提取role user的消息用内容前缀去重避免同一用户消息在多次补全中重复出现从output中提取助手回复并拆分tool_calls与扩展思考内容_split_thinking将 Anthropic content blocks 中type thinking与type text分开TOOL提取将工具执行提取为tool轮次记录工具名、输入与输出结构包装层跳过CHAIN、SPAN、AGENT等仅作结构包装直接忽略兜底逻辑若观测中提取不到任何轮次则回退到 Trace 自身的input/output字段保证数据不丢失usage归一化兼容 Langfuse 中usageDetails与usage两种字段命名及inputTokens/input_tokens/input等多种键名提升健壮性。6. 拉取、标准化并导入 Label Studio本步完成完整闭环从 Langfuse 拉取 Trace → 标准化 → 使用 ReactCode 配置创建 Label Studio 项目 → 导入标注任务。from label_studio_sdk import LabelStudio from label_studio_sdk.core.request_options import RequestOptions from typing import Any, Dict, List _REQUEST_OPTS RequestOptions(timeout_in_seconds120) def create_project(ls_host: str, api_key: str, title: str, label_config: str) - int: client LabelStudio(base_urlls_host, api_keyapi_key) project client.projects.create(titletitle, label_configlabel_config, request_options_REQUEST_OPTS) return int(project.id) def import_tasks(ls_host: str, api_key: str, project_id: int, tasks: List[Dict[str, Any]]) - Any: client LabelStudio(base_urlls_host, api_keyapi_key) return client.projects.import_tasks(idproject_id, requesttasks, return_task_idsTrue) if not LABEL_STUDIO_API_KEY: raise RuntimeError(Missing LABEL_STUDIO_API_KEY — set it in your .env file.) # 1) 从 Langfuse 拉取 Trace traces_payload lf.list_traces(limit20, page1) traces_list traces_payload.get(data) or traces_payload.get(traces) or [] if not traces_list: raise RuntimeError(No traces returned. Run Section 3 to generate sample traces.) print(fFetched {len(traces_list)} traces from Langfuse) # 2) 标准化每条 Trace tasks: List[Dict[str, Any]] [] for t in traces_list: tid t.get(id) or t.get(traceId) if not tid: continue full_trace lf.get_trace(str(tid)) obs lf.list_observations_v2(str(tid)) normalized normalize_langfuse_trace(full_trace, obs) if normalized[turns]: tasks.append({data: normalized}) print(f Trace {tid[:12]}... - {len(normalized[turns])} turns f({sum(1 for t in normalized[turns] if t[role]user)} user, f{sum(1 for t in normalized[turns] if t[role]assistant)} assistant, f{sum(1 for t in normalized[turns] if t[role]tool)} tool)) print(f\nPrepared {len(tasks)} tasks for import) # 3) 创建项目并导入 project_id create_project( ls_hostLABEL_STUDIO_HOST, api_keyLABEL_STUDIO_API_KEY, titlefLangfuse Trace Review ({LANGFUSE_PROJECT}), label_configLABEL_CONFIG_XML, ) print(fCreated project: {project_id}) resp import_tasks(LABEL_STUDIO_HOST, LABEL_STUDIO_API_KEY, project_id, tasks) print(fImported {len(tasks)} tasks) print(f\nDone! Open your project: {LABEL_STUDIO_HOST.rstrip(/)}/projects/{project_id})流程说明拉取lf.list_traces(limit20, page1)获取 Trace 列表同时兼容data与traces两种响应字段命名逐条补齐对每个 Trace ID 调用get_trace与list_observations_v2获取完整数据标准化normalize_langfuse_trace输出统一 schema仅当存在至少一个轮次时才构造成任务{data: normalized}——每个标准化后的 Trace 成为 Label Studio 中的一个任务创建项目通过 Label Studio SDK 的client.projects.create(title..., label_configLABEL_CONFIG_XML)创建项目title中带上LANGFUSE_PROJECT以便区分来源导入任务client.projects.import_tasks(idproject_id, requesttasks, return_task_idsTrue)批量导入打开项目输出项目链接{LABEL_STUDIO_HOST}/projects/{project_id}即可进入 ReactCode 三面板界面开始审阅。后续步骤开始标注打开上方项目链接在 ReactCode UI 中逐条审阅 Trace邀请 SME 协作将领域专家加入你的 Label Studio 项目进行协作评估增量同步周期性重跑第 46 节拉取新增 Trace导出标注使用 Label Studio SDK 或 REST API 拉取结构化标注用于下游分析或微调API 用法见 api.md 与 sdk.md自定义分类体系编辑标注配置单元格中的_TEMPLATE_JS变量加入你所在领域特有的 failure modes其他观测平台本教程属于「LLM 观测平台 Label Studio 评估」系列配套教程见 Braintrust 版 与 LangSmith 版。总结本教程演示了从 Langfuse Trace 到专家评估的完整工作流✅ 配置 Langfuse 与 Label Studio Enterprise 环境✅ 定义基于 ReactCode 的三面板标注 UIEnterprise 功能✅ 运行带多工具的 ReAct Agent 并开启 Claude 扩展思考——Langfuse 通过CallbackHandler捕获 Trace✅ 使用 Basic Auth 通过 REST API 从 Langfuse 拉取 Trace✅ 将带类型观测GENERATION、TOOL标准化为统一 Trace schema✅ 创建 Label Studio 项目并将 Trace 导入为标注任务。核心结论Langfuse 在开发期擅长带类型的观测存储与项目级 API 访问Label Studio Enterprise 则提供协作式、专家驱动的评估框架——ReactCode 界面让领域专家获得直观的逐轮审阅体验。两者在整个 AI 开发生命周期中互补工程侧的观测数据与业务侧的专家判断在这里汇合为结构化的评估结果为质量监控、Prompt 迭代与模型微调提供真实可信的标注依据。如需深入了解相关能力可继续阅读仓库内的 ReactCode 标签文档、ReactCode 后端代理实现 及 Label Studio SDK 用法。【免费下载链接】label-studioLabel Studio is a multi-type data labeling and annotation tool with standardized output format项目地址: https://gitcode.com/GitHub_Trending/la/label-studio创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考