企业官网建设流程全解析

使用 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 API；label-studio-sdk用于创建 Label Studio 项目与导入任务；langfuse与langchain/langgraph用于第 3 节生成示例 Trace；anthropic用于驱动带扩展思考（extended thinking）的 Claude 模型；python-dotenv用于加载环境变量。

环境变量配置

在仓库根目录（或与 notebook 同目录）创建.env文件：

# Label Studio Enterprise LABEL_STUDIO_HOST=http://localhost:8080 # 或你的 LS Enterprise 实例地址 LABEL_STUDIO_API_KEY=your_label_studio_api_key # Langfuse LANGFUSE_BASE_URL=https://cloud.langfuse.com # 或你的自托管 Langfuse 地址 LANGFUSE_PUBLIC_KEY=your_langfuse_public_key LANGFUSE_SECRET_KEY=your_langfuse_secret_key LANGFUSE_PROJECT=your_project_name # 项目名称（仅用于在 Label Studio 中展示） # Anthropic（仅第 3 节生成示例 Trace 时需要） ANTHROPIC_API_KEY=your_anthropic_api_key

注意：Langfuse 的 API Key（public + 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(override=True) load_dotenv(os.path.join(os.path.dirname(os.getcwd()), '.env'), override=True) # 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 包含三个面板：

标注模型：每轮捕获什么

• Verdict（判定）—— Pass 或 Fail；
• Issue tags（问题标签）—— 覆盖 5 大类的分类体系：Accuracy & Faithfulness（准确性与忠实性）、Tool & Retrieval（工具与检索）、Reasoning & Planning（推理与规划）、Response Quality（回答质量）、Safety & Compliance（安全与合规）；
• Severity（严重程度）—— Critical / Major / Minor / Suggestion；
• Expected behavior（期望行为）—— 自由文本：Agent 本应怎么做？
• Comments（评论）—— 任何补充说明。

底部栏另有trace 级判定（Pass / Fail / Mixed），用于评估整段对话的整体质量，独立于单轮判定。

标注配置 XML 结构

ReactCode 的标注配置在 XML 中声明。原文档给出的核心结构如下（_TEMPLATE_JS是内联的完整 ~40KB React 组件，此处以占位示意）：

_TEMPLATE_JS = r"""function TraceAnnotator({ React, addRegion, regions, data }) { // 736 行 React 组件，定义三面板 Trace 审阅 UI。 // 面板：Turns 列表（左）| Turn details（中）| Annotation 表单（右） // 底部栏：轮次统计 + trace 级判定（Pass / Fail / Mixed） // ... 完整实现见 notebook ... }""" LABEL_CONFIG_XML = ( '<View>\n' ' <ReactCode style="height: 95vh" name="trace" toName="trace"' ' 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自身（本配置中二者均为trace）；
• CDATA 包装：复杂的 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 签发限定用户与项目的短期 JWT（TTL 默认 3600 秒，可配置 60～86400 秒），后者以该 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_TRACES=False.') langfuse = get_client() @tool def calculator(expression: str) -> str: """Evaluate a math expression.""" try: return str(eval(expression)) except Exception as e: return f"Error: {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 f"No 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(), f"Weather data not available for {city}") # Claude 开启扩展思考——产生更丰富的 Trace，暴露模型的推理过程 llm = ChatAnthropic( model='claude-sonnet-4-5-20250929', max_tokens=16000, 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, what's our 12-month total?"], ["I'm 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, what's 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(content=msg_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 f'Basic {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, params=params, timeout=60) 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}', timeout=60) 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}, timeout=60) 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 Auth：Langfuse 公共 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 标准化为统一 Schema

Langfuse 将 Trace 存储为带类型的观测（GENERATION、TOOL、SPAN、CHAIN）。本单元格提取相关观测类型，映射为一个扁平的轮次（turn）序列——同一套 Schema 也被 Braintrust、LangSmith 集成共用——因此 ReactCode UI 无需关心 Trace 来自哪个平台。

每一轮携带：role、content、tool_name、tool_input、tool_calls、model、usage（Token 计数）、duration_ms、thinking（Claude 扩展思考块，若存在）。

import json as _json def _to_str(x): if x is None: return '' if isinstance(x, str): return x try: return _json.dumps(x, indent=2, default=str) 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, key=lambda 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': f'turn_{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, timestamp=ts) 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, timestamp=ts, model=obs.get('model') or obs.get('providedModelName'), usage=_normalize_usage(obs), tool_calls=tool_calls if tool_calls else None, duration_ms=duration, thinking=thinking) 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, timestamp=ts, tool_name=tool_name, tool_input=_to_str(inp) if inp else '', duration_ms=duration) if not turns: if trace_input := _extract_content(trace.get('input')): add_turn('user', trace_input, timestamp=trace.get('timestamp') or '') if trace_output := _extract_content(trace.get('output')): add_turn('assistant', trace_output, timestamp=trace.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_seconds=120) def create_project(ls_host: str, api_key: str, title: str, label_config: str) -> int: client = LabelStudio(base_url=ls_host, api_key=api_key) project = client.projects.create(title=title, label_config=label_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_url=ls_host, api_key=api_key) return client.projects.import_tasks(id=project_id, request=tasks, return_task_ids=True) 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(limit=20, page=1) 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(f'Fetched {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_host=LABEL_STUDIO_HOST, api_key=LABEL_STUDIO_API_KEY, title=f'Langfuse Trace Review ({LANGFUSE_PROJECT})', label_config=LABEL_CONFIG_XML, ) print(f'Created project: {project_id}') resp = import_tasks(LABEL_STUDIO_HOST, LABEL_STUDIO_API_KEY, project_id, tasks) print(f'Imported {len(tasks)} tasks') print(f'\nDone! Open your project: {LABEL_STUDIO_HOST.rstrip("/")}/projects/{project_id}')

流程说明：

1. 拉取：lf.list_traces(limit=20, page=1)获取 Trace 列表（同时兼容data与traces两种响应字段命名）；
2. 逐条补齐：对每个 Trace ID 调用get_trace与list_observations_v2获取完整数据；
3. 标准化：normalize_langfuse_trace输出统一 schema，仅当存在至少一个轮次时才构造成任务（{'data': normalized}）——每个标准化后的 Trace 成为 Label Studio 中的一个任务；
4. 创建项目：通过 Label Studio SDK 的client.projects.create(title=..., label_config=LABEL_CONFIG_XML)创建项目，title中带上LANGFUSE_PROJECT以便区分来源；
5. 导入任务：client.projects.import_tasks(id=project_id, request=tasks, return_task_ids=True)批量导入；
6. 打开项目：输出项目链接{LABEL_STUDIO_HOST}/projects/{project_id}，即可进入 ReactCode 三面板界面开始审阅。

后续步骤

• 开始标注：打开上方项目链接，在 ReactCode UI 中逐条审阅 Trace；
• 邀请 SME 协作：将领域专家加入你的 Label Studio 项目，进行协作评估；
• 增量同步：周期性重跑第 4～6 节，拉取新增 Trace；
• 导出标注：使用 Label Studio SDK 或 REST API 拉取结构化标注，用于下游分析或微调（API 用法见 api.md 与 sdk.md）；
• 自定义分类体系：编辑标注配置单元格中的_TEMPLATE_JS变量，加入你所在领域特有的 failure modes；
• 其他观测平台：本教程属于「LLM 观测平台 + Label Studio 评估」系列，配套教程见 Braintrust 版 与 LangSmith 版。

总结

本教程演示了从 Langfuse Trace 到专家评估的完整工作流：

1. ✅ 配置 Langfuse 与 Label Studio Enterprise 环境；
2. ✅ 定义基于 ReactCode 的三面板标注 UI（Enterprise 功能）；
3. ✅ 运行带多工具的 ReAct Agent 并开启 Claude 扩展思考——Langfuse 通过CallbackHandler捕获 Trace；
4. ✅ 使用 Basic Auth 通过 REST API 从 Langfuse 拉取 Trace；
5. ✅ 将带类型观测（GENERATION、TOOL）标准化为统一 Trace schema；
6. ✅ 创建 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