---

一、开篇导读

欢迎来到《LangSmith从入门到精通》专栏的第4节课。在前三节课中，我们完成了LangSmith认知框架的搭建、账号注册与计费规则的理解，以及开发环境的完整配置。现在，你的本地环境已经可以顺畅地将追踪数据发送到LangSmith云端了。

不过，当你打开LangSmith的Web控制台，面对“Traces”“Runs”“Projects”等术语时，是否感到一丝迷茫？Trace和Run到底是什么关系？Project和Session有什么区别？Tag和Metadata应该怎么用？这些概念看似简单，却是LangSmith数据模型的基石——不理解它们，就无法真正掌控LangSmith的强大功能。

本节课将系统拆解LangSmith的五大核心概念：Trace、Run、Project、Session、Tag。我们会从官方定义出发，结合原理剖析和大量实战代码，让你彻底厘清这些概念的内涵、外延以及它们之间的关联。本课结束时，你不仅能在代码中熟练运用这些概念，还能在LangSmith面板中高效地组织和检索数据。

本节课你将收获：

• 深刻理解Trace和Run的本质差异及嵌套关系
• 掌握Project的组织原则和多环境管理策略
• 学会使用Session追踪多轮对话
• 熟练运用Tag和Metadata进行业务维度的数据标注
• 理解Run类型的分类及其在UI中的差异化呈现
• 掌握通过Python SDK查询和过滤Runs的方法
• 学会在实际项目中应用这些概念解决真实问题

二、知识前置铺垫

2.1 从可观测性到大模型应用的数据建模

在深入LangSmith的具体概念之前，我们需要理解一个根本问题：LangSmith为什么要设计这样一套数据模型？

大模型应用与传统软件最大的不同在于其非确定性和链路复杂性。传统程序的一条请求路径是确定的——A函数调用B函数，B函数调用C函数，逻辑清晰。而大模型应用可能包含Prompt渲染、LLM调用、输出解析、工具调用、检索增强等多个环节，且每次执行都可能因为模型输出的不同而走不同的分支。

为了给这样的应用建立可观测性，LangSmith借鉴了分布式追踪领域的OpenTelemetry标准。你可以在LangSmith文档中找到相关的设计理念参考。正如分布式系统中一个请求经过多个微服务会产生一个Trace，LangSmith中一个用户请求经过多个处理环节会产生一个Trace。这两种设计思路是相通的。

LangSmith的数据模型建立在“Trace-Project-Run”三层结构之上：Project是顶层容器，Trace是一次完整请求的记录，Run是Trace中的单个步骤。此外，通过Session可以将多个Trace串联为对话线程，通过Tag和Metadata可以为数据附加业务维度的标记。

2.2 与OpenTelemetry的类比

如果你了解OpenTelemetry标准，理解LangSmith的概念会事半功倍。LangSmith官方的说明是：Trace相当于OpenTelemetry中的Trace，Run相当于Span。

在OpenTelemetry中：

• Trace：一次请求在分布式系统中的完整传播路径，由多个Span组成，共享同一个Trace ID。
• Span：单个操作单元，包含开始时间、结束时间、操作名称、属性、状态等信息。

在LangSmith中：

• Trace：一次应用调用的完整记录，包含多个Run，共享同一个Trace ID。
• Run：单个操作单元，可以是LLM调用、Chain执行、Tool调用等。

这种类比并非偶然。LangSmith在设计之初就充分参考了业界的可观测性标准，这让熟悉分布式追踪的开发者能够快速上手。

2.3 数据模型的三个核心设计目标

理解了设计来源后，我们再来看LangSmith数据模型要实现的核心目标：

1. 完整还原调用链路
通过嵌套的Run结构，LangSmith能够精确地重建每一次调用的调用栈。从根Run（Trace入口）到叶子Run（最小操作单元），开发者可以逐层下钻，定位问题发生的具体位置。

2. 支持多维度检索
通过Tag和Metadata，开发者可以从业务视角（用户ID、实验版本、环境等）对追踪数据进行分类和检索，而不只是从技术视角（时间、耗时等）。

3. 保持高效存储和查询
LangSmith对单个Trace最多支持25000个Run。这个上限足以覆盖绝大多数复杂链路（如ReAct Agent的多次思考-行动循环），同时也防止了个别超大Trace拖垮系统性能。

三、核心概念精讲

3.1 Trace（追踪）

Trace是LangSmith中最核心的顶层概念。LangChain官方文档将其定义为“应用程序为完成一个操作而执行的一系列步骤”。

一个Trace包含的关键信息：

• 唯一Trace ID：用于标识这一次完整的请求
• 所有Run的列表：包含根Run和所有嵌套子Run
• 总耗时：从第一个Run开始到最后一个Run结束
• 总体状态：成功或失败
• 附加元数据：Project信息、Session信息等

Trace的特性：

• 一个Trace解决一个“任务” ：从用户输入到系统输出的一次完整往返。例如用户问“今天天气怎么样”，系统返回天气预报，整个过程就是一个Trace。
• Trace内部有层次结构：一个Trace包含一个根Run（Root Run）和若干子Run，形成树形结构。
• Trace之间相互独立：不同用户请求或同用户的不同轮次，会产生不同的Trace，各Trace通过各自的Trace ID区分。

为什么需要Trace？

在缺乏可视化的开发模式下，你只能通过print或者日志来理解程序执行流程。当多个用户同时使用时，日志会交错在一起，很难将某次执行的日志完整提取出来。Trace通过Trace ID将所有相关日志聚合在一起，形成一个完整视图。

3.2 Run（运行）

Run是Trace中的最小单位。LangChain官方将其定义为“LLM应用中的一个工作单元”：对LLM的调用、Prompt格式化步骤、检索调用或其他离散操作。

一个Run包含的关键信息：

• 唯一Run ID：标识该Run
• Trace ID：标识该Run所属的Trace
• Parent Run ID：指向父Run（NULL表示是根Run）
• Run Type：llm、chain、tool、retriever等
• 名称：Run的可读标识
• 输入和输出数据
• 执行时间：start_time和end_time
• 耗时：自动计算
• Token用量：仅对llm类型有效
• 错误信息：如果执行失败
• Tags和Metadata

Run的层次结构示例：

Trace (用户查询“今天的天气”)
├── Run1: Chain (generate_weather_report)
│   ├── Run1.1: Tool (fetch_weather_api)
│   │   └── Run1.1.1: Retriever (缓存查询)
│   └── Run1.2: LLM (生成自然语言回答)
└── Run2: Parser (提取关键信息)

Run Type的作用：不同Run Type在LangSmith UI中会有不同的渲染方式。llm类型的Run会突出显示模型名称、Token用量；retriever类型的Run会突出显示检索到的文档列表。这个分类体系让UI能够针对性地呈现信息，而不是统一用同一套模板。

3.3 Project（项目）

Project是组织中所有Trace和Run的容器。LangChain官方定义：Project将相关的Trace和Run组织成逻辑分组。值得一提的是，在某些LangSmith API文档中，Project也被称为“TracerSession”或“Session”。这意味着SDK中的TracerSession类和Web UI中的Project概念是同一回事。

Project的特点：

• 容纳多Trace：一个Project可以包含成千上万个Trace
• 数据隔离：不同Project之间的数据完全隔离，既可按环境（dev/staging/prod）划分，也可按应用划分
• 可管理：支持创建、删除、重命名、分享Project

如何决定创建几个Project？

• 按环境：为开发环境、测试环境、生产环境分别建Project
• 按应用：每个AI应用独立Project
• 按实验：进行A/B测试时，为不同实验版本建临时Project
• 默认Project：若未指定，数据发送到名为default的Project

3.4 Session（会话）

Session（又名Thread，线程）是将多个Trace串联成对话序列的机制。LangChain官方定义：Thread是代表单一对话的Trace序列。在多轮对话中，每一轮是一个单独的Trace，但它们通过共享标识被连接在一起。

Session的设计初衷：在多轮对话（如聊天机器人）场景中，用户的每次提问都会产生一个独立的Trace。若同一个用户连续提问10次，就会产生10个独立的Trace。若没有Session，你无法从这10个Trace中知道它们属于同一个对话。Session通过给这10个Trace打上相同的session_id来标识它们同属一个对话。

Session标识符：LangSmith支持使用sesssion_id、thread_id或conversation_id作为分组键。

Session与Project的区别：

• Project是静态的容器划分（按环境、按应用）
• Session是动态的对话聚合（按对话线程）

3.5 Tag（标签）

Tag是附加到Run上的字符串标识，用于分类、筛选和分组。

Tag的定义和用法：

• Tag是字符串形式，不是键值对（那是Metadata的职责）
• 一个Run可以附带多个Tag
• Tag通常用于低基数（low-cardinality）属性的标记，如production、debug、v1.0、experiment-a

Tagvs. Metadata：

维度	Tag	Metadata
数据类型	字符串	键值对
基数	建议低（如环境、版本）	高基数也可（如user_id）
查询效率	高效率	相对较低
典型用途	分类聚合	存储额外上下文

LangSmith官方的说明也从侧面印证了这种分工：Metadata可存储版本、环境等上下文信息，Tag和Metadata均可在UI中用于筛选和分组。

四、原理底层剖析

4.1 Run的嵌套实现原理

理解了核心概念后，我们深入到底层实现层面。Run之间的父子关系是其最为关键的特性。

当一个Trace开始时，LangSmith会创建根Run（Root Run），其parent_run_id为NULL。当程序执行过程中调用子组件时，LangSmith会创建子Run并将当前Run的ID作为parent_run_id传入。这个机制与函数调用栈的工作方式极其相似——栈帧嵌套对应Run嵌套。

LangSmith SDK会在内部维护一个上下文栈。当进入一个被追踪的函数时，SDK将当前Run压入栈顶；当函数退出时，将Run弹出栈。SDK正是通过这个栈来确定新建Run的父Run归属。

为什么需要嵌套结构？

嵌套Run的价值体现在以下几个方面：

• 精准的耗时归因：通过parent_run_id可以精确计算每个父Run的总耗时，并且可以看到各子Run的时间比例。若某个子Run耗时占父Run的90%，就找到了主要瓶颈。
• 故障传播路径分析：当异常发生时，嵌套结构能让你沿着父子链路逐级追踪，定位到具体是LLM调用失败还是Tool调用失败。
• 上下文的自动继承：子Run会自动继承父Run的上下文（如Project信息、Tags的合并规则等），避免了重复配置。

4.2 Trace与Run的生命周期

LangSmith中Trace和Run的生命周期是有严格顺序的：

一个Trace的开始结束都关联着相应的回调事件。当chain.invoke()执行时，LangSmith依次触发on_chain_start（创建根Run）、深度优先遍历执行子组件（每个子组件的start/end触发对应的回调）、on_chain_end（结束根Run）。最后将这些Run数据异步上报到LangSmith云端。

异步上报机制：LangSmith为了不影响主业务的执行性能，采用异步队列上报。每个Run的数据先暂存在本地内存队列中，后台线程持续消费队列并通过HTTP POST请求批量发送到LangSmith API。即使LangSmith云端短暂不可用，数据也不会立即丢失（会在队列中等待重试）。

4.3 Session（Thread）的数据组织

Session在底层是如何存储的？答案是通过元数据。

当你调用LangSmith的API时，可以在请求中添加名为session_id的特殊元数据。LangSmith后端会识别这个键，并在数据库中将所有具有相同session_id的Trace进行逻辑关联。

在UI中查看时，你可以按session_id进行筛选和分组，看到同一个对话的完整历史轨迹。这为分析Agent行为的时序演变、调试多轮对话的状态管理问题提供了极大便利。

4.4 Tag与Metadata的存储与索引

Tag和Metadata存储在Run对象的附加字段中。LangSmith为Tag建立了索引来获得高效查询性能。正是因为Tag有索引支持，它们通常用于高频筛选场景。

而Metadata作为更自由的键值对，虽然也支持查询，但由于其键值不是预定义的，查询性能相对较低。文档中也提到，LangSmith提供了配套的filter query language支持更复杂的查询。

在成本层面，Tag和Metadata也会带来存储开销。在大规模生产环境中，如果给每个Run都附带大量高基数的Metadata，存储成本会显著上升。一个常见的设计原则是——Tag用于频繁筛选的低维度标签（env、version、region），Metadata用于辅助调试的附加信息（user_id、request_id、traceparent）。

五、环境配置手把手实战

提醒一下，以下实战默认你已经完成了前三节课的环境配置，即：

• Python 3.10+ 虚拟环境已创建并激活
• langchain、langchain-openai、langsmith、python-dotenv已安装
• .env文件中正确配置了LANGCHAIN_API_KEY、OPENAI_API_KEY等变量

若尚未完成，请先回头学习第3课。

5.1 验证基础配置

创建test_config.py，验证环境配置是否就绪：

import os
from dotenv import load_dotenv

load_dotenv()

def check_config():
    required = ["LANGCHAIN_API_KEY", "OPENAI_API_KEY"]
    for var in required:
        if not os.getenv(var):
            print(f"❌ 缺失 {var}")
            return False
    print("✅ 环境配置检查通过")
    print(f"📁 LangSmith Project: {os.getenv('LANGCHAIN_PROJECT', 'default')}")
    return True

if __name__ == "__main__":
    check_config()

5.2 创建第一个带自定义配置的Project

Python SDK中，Client类是与LangSmith服务交互的入口。我们来实战创建一个Project：

# 文件名: manage_projects.py
from langsmith import Client
from dotenv import load_dotenv

load_dotenv()

client = Client()

# 获取或创建Project
project_name = "core-concepts-demo"
try:
    # 尝试读取Project
    project = client.read_project(project_name=project_name)
    print(f"📁 Project已存在: {project.name}")
except Exception:
    # Project不存在则创建
    project = client.create_project(
        project_name=project_name,
        description="演示Trace、Run、Session、Tag概念",
        metadata={"env": "demo", "tutorial": "lesson4"}
    )
    print(f"✅ 创建新Project: {project.name}")

print(f"📋 Project ID: {project.id}")
print(f"📝 描述: {project.description}")

5.3 配置环境变量指向新Project

# 方式一: 临时export
export LANGCHAIN_PROJECT="core-concepts-demo"

# 方式二: 永久写入.env
echo 'LANGCHAIN_PROJECT="core-concepts-demo"' >> .env

六、完整可运行代码案例

6.1 基础Trace——理解和区分Trace与Run

从一个最简单的LCEL链开始，在LangSmith中观察Trace和Run的对应关系。

# 文件名: trace_and_run_demo.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# 显式指定Project名称
os.environ["LANGCHAIN_PROJECT"] = "core-concepts-demo"

print("=" * 60)
print("Trace & Run 概念演示")
print("=" * 60)

# 构建一个简单的链: 只有Prompt、LLM、Parser三个组件
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，请用{language}回答问题。"),
    ("human", "{input}")
])
output_parser = StrOutputParser()

chain = prompt | llm | output_parser

print("\n🚀 开始执行链式调用...")
result = chain.invoke({
    "role": "AI助手",
    "language": "中文",
    "input": "请用一句话解释什么是LangSmith"
})

print(f"\n📝 最终回答:\n{result}")
print("\n✨ 调用完成！")
print("📊 请登录LangSmith控制台查看Trace和Run的关系")

6.2 嵌套Run——Trace内部的层级结构

使用@traceable装饰器和RunnableLambda构建嵌套调用，在LangSmith中观察Run层级。

# 文件名: nested_runs_demo.py
import os
from dotenv import load_dotenv
from langsmith import traceable
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "core-concepts-demo"

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

@traceable(run_type="tool", name="数据清洗")
def clean_input(raw_text: str) -> str:
    """模拟数据清洗步骤"""
    cleaned = raw_text.strip().lower()
    print(f"  🧹 清洗前: '{raw_text}' -> 清洗后: '{cleaned}'")
    return cleaned

@traceable(run_type="llm", name="调用GPT")
def call_llm(cleaned_input: str) -> str:
    """LLM调用步骤"""
    response = llm.invoke(f"请简洁回答: {cleaned_input}")
    return response.content

@traceable(run_type="chain", name="主流程")
def main_pipeline(user_input: str) -> str:
    """主流程编排"""
    cleaned = clean_input(user_input)
    answer = call_llm(cleaned)
    return answer

print("=" * 60)
print("嵌套Run演示")
print("=" * 60)

test_input = "  朗史米斯是什么？  "
print(f"\n📥 输入: {test_input}")

result = main_pipeline(test_input)
print(f"\n📤 输出: {result}")
print("\n🔗 请在LangSmith中查看Run的嵌套层级")

6.3 多种Run类型对比

通过代码一次性创建chain、llm、tool、retriever四种类型的Run。

# 文件名: run_types_demo.py
import os
from dotenv import load_dotenv
from langsmith import traceable
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "core-concepts-demo"

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

@traceable(run_type="tool", name="calculator", metadata={"operation": "add"})
def add_numbers(a: int, b: int) -> int:
    """模拟加法工具"""
    return a + b

@traceable(run_type="retriever", name="knowledge_base")
def retrieve_info(query: str) -> str:
    """模拟知识库检索"""
    fake_db = {
        "langsmith": "LangSmith是LangChain官方可观测性平台，用于调试和监控LLM应用",
        "langchain": "LangChain是构建LLM应用的开发框架"
    }
    result = fake_db.get(query.lower(), "未找到相关信息")
    print(f"  📚 检索到: {result[:50]}...")
    return result

@traceable(run_type="chain", name="query_chain")
def process_query(topic: str) -> str:
    """整合各组件的主链"""
    # Step 1: 计算 (这里仅作演示,实际计算与topic无关)
    _ = add_numbers(5, 3)  
    # Step 2: 检索
    context = retrieve_info(topic)
    # Step 3: LLM生成
    prompt = PromptTemplate.from_template(
        "基于以下信息回答问题: {context}\n问题: 什么是{topic}?"
    )
    chain = prompt | llm
    response = chain.invoke({"context": context, "topic": topic})
    return response.content

print("=" * 60)
print("Run类型对比演示")
print("=" * 60)

topics = ["langsmith", "langchain"]
for topic in topics:
    print(f"\n📌 处理话题: {topic}")
    result = process_query(topic)
    print(f"📝 回答: {result[:100]}...")

print("\n✅ 演示完成！")
print("🎨 在LangSmith UI中观察不同类型Run的渲染差异:")
print("   - 'llm'类型: 显示Token用量和模型名称")
print("   - 'tool'类型: 突出工具输入输出")
print("   - 'retriever'类型: 展示检索到的文档列表")

6.4 Session——追踪多轮对话

# 文件名: session_demo.py
import os
import uuid
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "core-concepts-demo"

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的AI助手，记住对话上下文。"),
    ("human", "{input}")
])
chain = prompt | llm

# 模拟一个对话会话
session_id = f"chat_session_{uuid.uuid4().hex[:8]}"

def chat(user_input: str, session_id: str) -> str:
    config = RunnableConfig(
        configurable={"session_id": session_id},
        metadata={"session_id": session_id}  # LangSmith识别session_id
    )
    response = chain.invoke({"input": user_input}, config=config)
    return response.content

print("=" * 60)
print("Session多轮对话演示")
print(f"📱 会话ID: {session_id}")
print("=" * 60)

conversation = [
    "你好，我想了解一下LangSmith",
    "它能做什么？",
    "它能帮我调试Agent吗？",
    "谢谢，我明白了"
]

for i, user_msg in enumerate(conversation, 1):
    print(f"\n👤 用户: {user_msg}")
    assistant_reply = chat(user_msg, session_id)
    print(f"🤖 助手: {assistant_reply[:80]}...")

print(f"\n🔗 在LangSmith中按session_id筛选，可查看完整对话历史")

6.5 Tag和Metadata实战

# 文件名: tag_metadata_demo.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "core-concepts-demo"

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

def analyze_sentiment(text: str, user_id: str, version: str) -> str:
    """情感分析函数，演示Tag和Metadata的使用"""
    prompt = PromptTemplate.from_template(
        "分析以下文本的情感倾向(积极/消极/中性): {text}"
    )
    chain = prompt | llm
    
    # 通过RunnableConfig传递Tags和Metadata
    config = RunnableConfig(
        tags=["sentiment-analysis", f"v{version}", "production"],
        metadata={
            "user_id": user_id,
            "version": version,
            "environment": "demo",
            "request_id": os.urandom(4).hex()
        }
    )
    
    result = chain.invoke({"text": text}, config=config)
    return result.content

print("=" * 60)
print("Tag和Metadata实战演示")
print("=" * 60)

test_samples = [
    ("这个产品太棒了，我非常满意！", "user_001", "2.1"),
    ("服务态度很差，体验不好。", "user_002", "2.1"),
]

for text, user_id, version in test_samples:
    print(f"\n📝 文本: {text}")
    sentiment = analyze_sentiment(text, user_id, version)
    print(f"🎯 情感分析: {sentiment}")

6.6 通过SDK查询Runs

# 文件名: query_runs.py
import os
from datetime import datetime, timedelta
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()
client = Client()

def demonstrate_queries():
    """演示通过SDK查询Runs的各种方式"""
    project_name = "core-concepts-demo"
    
    print("=" * 60)
    print("通过LangSmith SDK查询Runs")
    print("=" * 60)
    
    # 1. 获取Project的所有Runs
    print("\n📊 1. 获取最近24小时的Runs:")
    yesterday = datetime.now() - timedelta(days=1)
    recent_runs = list(client.list_runs(
        project_name=project_name,
        start_time=yesterday
    ))
    print(f"   共 {len(recent_runs)} 条Runs")
    
    # 2. 按run_type筛选
    print("\n🤖 2. 筛选'llm'类型的Runs:")
    llm_runs = list(client.list_runs(
        project_name=project_name,
        run_type="llm"
    ))
    print(f"   共 {len(llm_runs)} 条LLM调用记录")
    
    # 3. 筛选失败的Runs
    print("\n❌ 3. 筛选执行失败的Runs:")
    error_runs = list(client.list_runs(
        project_name=project_name,
        error=True
    ))
    print(f"   共 {len(error_runs)} 条失败记录")
    if error_runs:
        print(f"   示例: {error_runs[0].name} -> {error_runs[0].error}")
    
    # 4. 按Tag筛选
    print("\n🏷️ 4. 按Tag筛选(production):")
    try:
        tagged_runs = list(client.list_runs(
            project_name=project_name,
            filter='has(tags, "production")'
        ))
        print(f"   共 {len(tagged_runs)} 条")
    except Exception as e:
        print(f"   ⚠️ Tag筛选需要新版SDK支持: {e}")
    
    # 5. 按Metadata筛选
    print("\n🔑 5. 按Metadata筛选(environment=demo):")
    try:
        metadata_runs = list(client.list_runs(
            project_name=project_name,
            filter='eq(metadata["environment"], "demo")'
        ))
        print(f"   共 {len(metadata_runs)} 条")
    except Exception as e:
        print(f"   ⚠️ 若需复杂查询，建议使用list_runs + filter参数")
    
    # 6. 获取根Run
    print("\n🌳 6. 获取根Run:")
    root_runs = list(client.list_runs(
        project_name=project_name,
        is_root=True
    ))
    print(f"   共 {len(root_runs)} 条根Run (代表独立Trace)")
    
    return recent_runs

if __name__ == "__main__":
    runs = demonstrate_queries()
    print("\n✨ 查询完成！")

七、代码逐行详解

7.1 trace_and_run_demo.py关键代码

chain = prompt | llm | output_parser

这里是通过|操作符将PromptTemplate、ChatOpenAI、StrOutputParser拼接成一个RunnableSequence。在LangSmith中，这会生成一个Chain类型的根Run，内部嵌套LLM Run。通过控制台你就能逐个查看每个组件的输入输出。

7.2 nested_runs_demo.py关键注解

@traceable(run_type="tool", name="数据清洗")
def clean_input(raw_text: str) -> str:

@traceable装饰器将普通函数包装为LangSmith追踪的Run。run_type决定LangSmith UI中的渲染方式。

cleaned = clean_input(user_input)
answer = call_llm(cleaned)

两个子函数调用自动成为当前函数的子Run，形成三层嵌套结构。

7.3 session_demo.py中的Session传递

config = RunnableConfig(
    configurable={"session_id": session_id},
    metadata={"session_id": session_id}
)

这里在RunnableConfig的metadata中放入session_id，LangSmith后端会识别并分组。多轮对话即可通过相同session_id追溯。

7.4 tag_metadata_demo.py中的元数据设计

metadata={
    "user_id": user_id,
    "version": version,
    "environment": "demo",
    "request_id": request_id
}

user_id（高基数）和environment（低基数）并存。Tag适合可枚举的低基数属性，Metadata适合记录上下文细节。

八、常见坑点与避坑指南

8.1 Trace和Run的概念混淆

坑点：误认为Trace就是Run，或分不清二者关系。

厘清：Trace是一个集合，Run是集合中的元素。Trace代表一次完整请求，Run代表其中的步骤。一个Trace至少包含一个根Run。

8.2 Session ID不生效

现象：设置了session_id，但LangSmith中无法按对话分组。

原因：通过RunnableConfig传递时，session_id被LangSmith识别但需放在metadata中；或使用了错误的环境变量名。

解法：

config = RunnableConfig(metadata={"session_id": session_id})
result = chain.invoke(input_data, config=config)

8.3 Run类型设置错误

现象：检索步骤被当作普通Chain渲染，UI中看不到文档详情。

原理：LangSmith UI对retriever、llm等类型有专属优化面板。

解法：

@traceable(run_type="retriever", name="vector_search")
def my_retriever(query): ...

8.4 Trace中Run过多导致UI卡顿

原理：单个Trace最多支持25000个Run。如果Agent陷入死循环或递归调用，会产生非常多的子Run。

解法：

• 检查Agent的max_iterations参数设置上限
• 分析LangSmith中的Run树，定位循环是在哪个环节发生的

8.5 异步调用中追踪不完整

现象：使用async/await调用时，LangSmith只捕获了部分Run。

原因：异步上下文管理需要特殊处理——LangSmith的上下文栈在线程间传递时可能丢失。

解法：使用atraceable装饰器替代，或使用trace上下文管理器。

九、企业级落地最佳实践

9.1 Project的多环境隔离

企业应将不同环境的追踪数据存入独立的Project，避免污染：

import os

ENV = os.getenv("APP_ENV", "development")
PROJECT_MAP = {
    "development": "myapp-dev",
    "staging": "myapp-staging", 
    "production": "myapp-prod"
}
os.environ["LANGCHAIN_PROJECT"] = PROJECT_MAP[ENV]

9.2 Tag命名规范

制定统一Tag规范，示例命名模式：{category}:{value}

tags = [
    f"env:{ENV}",
    f"version:{APP_VERSION}",
    f"region:{REGION}",
    f"feature:{FEATURE_FLAG}"
]

规范化的Tag可直接用于LangSmith监控图表的分组筛选。

9.3 Session在客服系统中的应用

def chat_with_user(user_message: str, user_id: str):
    session_id = f"user_{user_id}"
    # 每次调用都传入session_id
    config = RunnableConfig(metadata={"session_id": session_id})
    return chain.invoke({"input": user_message}, config=config)

9.4 通过Trace ID做问题追溯

from langsmith import Client

client = Client()
run_id = "a36092d2-4ad5-4fb4-9c0d-0dba9a2ed836"
run = client.read_run(run_id)

print(f"输入: {run.inputs}")
print(f"输出: {run.outputs}")
print(f"错误: {run.error}")

当你从用户反馈、日志或告警中获得Trace ID后，立即可用这段代码追溯。

9.5 Metadata自动注入中间件

from contextvars import ContextVar

request_id_var = ContextVar("request_id", default="")

@app.middleware("http")
async def add_langsmith_metadata(request, call_next):
    request_id = str(uuid4())
    request_id_var.set(request_id)
    
    # 全局配置
    with ls.tracing_context(metadata={"request_id": request_id}):
        response = await call_next(request)
    return response

企业生产中，最好建立一个自动注入的中间件，避免在代码各处手动添加标识。

9.6 处理高基数Tag

原则是Tag用低基数属性，Metadata用高基数属性。如果需要按高基数维度快速过滤，可以在SDK查询层做二次过滤：

# 全量拉取后本地过滤
all_runs = client.list_runs(project_name="myapp-prod")
filtered = [run for run in all_runs 
            if run.metadata.get("user_id") == "specific_user"]

十、本节知识点总结

Trace（追踪）

• 一次从输入到输出的完整操作记录
• 包含多个Run
• 拥有唯一Trace ID
• 单Trace最多支持25000个Run

Run（运行）

• Trace中的单个步骤
• 支持类型：llm、chain、tool、retriever
• 通过parent_run_id形成嵌套关系
• 记录输入输出、耗时、Token、错误等信息

Project（项目）

• Trace和Run的逻辑容器，对应TracerSession
• 数据隔离的基本单位
• 通过LANGCHAIN_PROJECT指定
• 建议按环境或应用划分

Session（会话）

• 多轮对话的场景下将多个Trace串联
• 通过metadata中的session_id、thread_id或conversation_id实现
• Agent应用可按thread_id聚合

Tag（标签）

• 字符串标识，用于筛选、分组
• 建议用于低基数维度
• 支持在配置中使用tags参数传入

Metadata（元数据）

• 键值对形式的附加信息
• 支持存储高基数数据
• 通过metadata参数传入

Run类型体系

• 决定了LangSmith UI中的渲染方式
• 通过@traceable的run_type参数指定

十一、课后思考练习题

练习题1：理论理解

1.1 一个RAG应用在生成答案前需要：Query改写 → 向量检索 → LLM生成。请问这个过程会产生几个Trace？几个Run？画出Run的嵌套结构图。

1.2 假设你想为同一服务增加环境隔离（dev、staging、prod）和A/B测试对比（baseline、experiment），如何设计Project和Tag既能完成环境隔离，又能方便地横向对比？

1.3 Session与Project的数据隔离策略有何不同？对于一个多租户的客服机器人，建议用Session还是Project区分不同客户公司？为什么？

练习题2：动手实践

2.1 修改run_types_demo.py，在process_query中插入两个新的子Run：一个retriever和一个parser，用嵌套结构在LangSmith中观察树形视图变化。

2.2 运行session_demo.py后在LangSmith UI中：

• 验证session_id确实将多轮对话串联
• 使用筛选功能找出该Session的所有Traces

2.3 编写脚本export_runs.py，将core-concepts-demo项目中所有llm类型的Runs导出为JSON文件。

练习题3：场景设计

3.1 设计一个方案：为不同租户分配独立的LangSmith Project，且部分租户可联合看板。需满足：

• 租户A只能看到自己的Project
• 平台管理员能看到所有Project

3.2 团队提出Tag和Metadata的命名规范，要求统一键名和取值范围，并预留未来扩展空间。请给出方案。

3.3 改进错误处理逻辑：当某个Run出现异常时，自动在根Run上记录自定义Tag（has_error=true），方便快速筛选失败请求。

练习题4：源码探索（选做）

4.1 阅读LangSmith Python SDK中@traceable装饰器源码，理解它如何在不破坏原函数签名的前提下收集输入输出。

---

下节课预告：

第5节课我们将正式进入快速入门实战——从零搭建第一个完整的LangSmith链路追踪项目。你会综合运用Trace、Run、Project、Tag等概念，构建一个包含LLM调用、Tool调用、输出解析的完整应用，并学会在LangSmith控制台进行可视化调试。

请确保完成本节课的所有代码练习，尤其要熟悉@traceable装饰器和RunnableConfig的用法，下节课我们将在这些基础上构建更复杂的链路。

下一节课见！

---

🔗《20节课 LangSmith 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！
💡 如果本文对您有所启发欢迎：
👍 点赞📌 收藏 📤 分享给更多需要的伙伴。
🗣️ 期待在评论区看到您的想法, 共同进步。
🔔 关注我，持续获取更多干货内容～
🤗 我们下篇文章见～