如果你曾好奇 LLM 可观测性工具是如何在下层追踪每一次 Chain、Tool 和 Retriever 调用的，那么开源的 LangSmith Python SDK 正是答案所在。虽然 LangSmith 服务端闭源，但其 langsmith 包完全开源，其中包含了生产级的追踪（Tracing）、评估（Evaluation）和数据集管理（Dataset Management）实现。本文将深入解析其源码架构、核心类（RunTree、Client、Evaluators）设计及可复用的最佳实践。

一、整体架构概览

LangSmith SDK 源码的核心目录结构如下：

langsmith/
├── client.py          # 客户端实现，连接 API
├── run_trees.py       # RunTree——追踪的核心数据结构
├── trace.py           # 追踪装饰器与上下文管理
├── evaluation/        # 评估模块
│   ├── evaluator.py   # 评估器接口
│   └── string.py      # 字符串评估器
├── datasets/          # 数据集管理
├── schemas.py         # 数据模型定义
└── utils/

SDK 整体采用分层架构：

  1. 数据模型层（schemas）：基于 Pydantic 的 BaseModel 定义所有实体
  2. 客户端层（client）：封装 HTTP 请求，提供 CRUD 操作
  3. 追踪层（run_trees / trace）：自动/手动追踪 LLM 调用
  4. 评估层（evaluation）：定义评估逻辑与运行器
  5. 数据集层（datasets）：管理示例数据与测试集

二、核心数据模型：Run 与 RunTree

2.1 Run 模型——最小追踪单元

Run 是追踪的最小单元，在 schemas.py 中定义：

class Run(BaseModel):
    id: UUID
    name: str
    run_type: str  # "llm", "chain", "tool", "retriever" 等
    inputs: dict
    outputs: Optional[dict] = None
    start_time: datetime
    end_time: Optional[datetime] = None
    error: Optional[str] = None
    parent_run_id: Optional[UUID] = None
    child_runs: List["Run"] = []
    # ... 更多字段

关键设计点：

  • 树形结构：通过 parent_run_id 和 child_runs 形成调用树，完整还原 LLM 应用的执行链路
  • 运行类型枚举：run_type 区分不同的调用来源，支持 llm、chain、tool、retriever 等
  • 时间追踪：记录每次调用的起止时间，用于延迟分析和性能 profiling

2.2 RunTree——构建嵌套调用链

RunTree 位于 run_trees.py，是 Run 的增强版，用于在客户端构建嵌套调用树：

class RunTree:
    def create_child(self, name: str, run_type: str, inputs: dict) -> "RunTree":
        child = RunTree(
            name=name, run_type=run_type, inputs=inputs,
            parent_run=self, session_name=self.session_name
        )
        self.child_runs.append(child)
        return child

    async def apost(self) -> None:
        """将当前 Run 及所有子 Run 异步提交到后端"""
        ...

核心设计模式：

  • 组合模式：RunTree 包含子 RunTree，形成树形结构
  • 延迟提交：调用 post() 时，以递归方式深度优先序列化所有子节点后一并提交
  • 上下文管理器：支持 with 语句自动记录开始/结束时间，让 API 使用非常简洁：
with RunTree(name="my_chain", run_type="chain") as parent:
    with parent.create_child("llm_call", "llm", inputs={...}) as child:
        child.outputs = {"response": "..."}

数据模型就位后，下一个问题是追踪数据如何被自动捕获。LangSmith 提供了声明式装饰器和命令式上下文管理器两种方式——下面逐一来看。

三、追踪机制：装饰器、上下文管理器与批量写入

3.1 @trace 装饰器——零侵入式插桩

trace.py 中提供了 @trace 装饰器，用于自动追踪函数调用：

def trace(
    name: Optional[str] = None,
    run_type: str = "chain",
    project_name: Optional[str] = None,
    ...
):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            run_tree = RunTree(name=name or func.__name__, run_type=run_type, ...)
            with run_tree:
                result = func(*args, **kwargs)
                run_tree.outputs = {"result": result}
                run_tree.post()
                return result
        return wrapper
    return decorator

设计亮点：

  • 零侵入式追踪：开发者只需一个装饰器即可自动记录所有 LLM 调用，无需修改业务逻辑
  • 上下文自动传递：通过 LangChainTracer 或 TracingCallbackHandler 深度集成到 LangChain/LangGraph 的回调系统，实现跨组件追踪

3.2 客户端批量写入——性能优化的关键

client.py 中的 Client 类负责将 Run 数据提交到服务端，实现了高效的批量写入：

class Client:
    def create_run(self, run: Run) -> None:
        """提交单个 Run"""
        ...

    def create_runs(self, runs: List[Run]) -> None:
        """批量提交 Run，显著减少网络开销"""
        runs_data = [run.dict() for run in runs]
        self.session.post(f"{self.api_url}/runs/batch", json=runs_data)

批量提交机制在高频追踪场景下可将网络往返次数降低一个数量级，是 SDK 面向生产环境设计的核心优化。

追踪数据收集完成后，下一步就是评估。评估模块建立在同样的 Run 模型之上，但引入了一套即插即用的评估器系统。

四、评估模块：策略模式的经典应用

评估模块位于 evaluation/ 目录，接口设计体现了**策略模式（Strategy Pattern）**的经典应用。

4.1 RunEvaluator——抽象评估器接口

class RunEvaluator(ABC):
    @abstractmethod
    def evaluate_run(self, run: Run, example: Optional[Example] = None) -> EvaluationResult:
        ...

RunEvaluator 是所有评估器的抽象基类，每个评估器只需实现 evaluate_run 方法即可。

4.2 内置评估器：Criteria、LabeledCriteria 等

# string.py
class StringEvaluator(RunEvaluator):
    def __init__(self, evaluation_name: str, ...):
        ...

class Criteria(CorrectnessStringEvaluator):
    """基于标准的评估器，如正确性、相关性、有害性等"""
    ...

class LabeledCriteria(LabeledCorrectnessStringEvaluator):
    """带参考答案的标准评估器"""
    ...

内置评估器涵盖了正确性、相关性、有害性等常见 LLM 评估维度。用户也可以通过继承 RunEvaluator 实现完全自定义的评估逻辑。

4.3 evaluate() 函数——编排数据集评估

def evaluate(
    run_on_dataset: Callable[[dict], Run],
    data: Dataset,
    evaluators: List[RunEvaluator],
    ...
) -> EvaluationResults:
    """在数据集上运行并评估"""
    for example in data:
        run = run_on_dataset(example.inputs)
        for evaluator in evaluators:
            result = evaluator.evaluate_run(run, example)
            results.append(result)
    return EvaluationResults(results=results)

这种设计将”运行”与”评估”分离——前者执行 LLM 调用生成 Run，后者对 Run 进行打分，职责清晰、易于扩展。

五、数据集管理：测试数据的 CRUD

datasets/ 模块封装了数据集的 CRUD 操作：

class Dataset:
    name: str
    description: str
    examples: List[Example]

    def create(self, client: Client) -> dict:
        ...

    def upload(self, client: Client, examples: List[dict]) -> None:
        """批量上传示例数据"""
        ...
  • Example：一条数据范例，包含 inputs（输入）和 outputs（可选参考答案）
  • Dataset：一组 Example 的集合，广泛用于回归测试和评估基准

六、客户端通信设计：会话管理、重试与鉴权

client.py 中的 Client 类是整个 SDK 与后端通信的桥梁，值得关注的设计点：

  1. 统一会话管理：使用 requests.Session 复用 TCP 连接，避免每次请求重复握手
  2. 自动重试：对 429（限流）和 5xx（服务端错误）实现指数退避重试（基于 tenacity 或类似重试库）
  3. API Key 鉴权：通过 x-api-key 请求头传递
  4. 环境变量配置：支持 LANGCHAIN_API_KEY、LANGCHAIN_ENDPOINT、LANGCHAIN_PROJECT 等环境变量
class Client:
    def __init__(self, api_key: Optional[str] = None, api_url: Optional[str] = None):
        self.api_key = api_key or os.getenv("LANGCHAIN_API_KEY")
        self.api_url = api_url or os.getenv("LANGCHAIN_ENDPOINT", "https://api.smith.langchain.com")
        self.session = requests.Session()
        self.session.headers.update({"x-api-key": self.api_key})

⚠️ 实际 API 端点以 LANGCHAIN_ENDPOINT 环境变量设置为准，以上为默认值。

七、最佳实践与架构启示

从 LangSmith SDK 源码中我们可以学到的工程智慧：

设计原则LangSmith 实现通用启示
数据结构先行用 Pydantic 明确定义 Run、Example 等模型上下游围绕数据模型编程，减少歧义
树形追踪的通用性RunTree 组合模式可复用到任何需要追踪调用链的场景
声明式 + 命令式 API同时提供 @trace 装饰器和 with RunTree()满足不同开发者的使用偏好
批量优于逐条create_runs() 批量接口减少网络 I/O 是性能优化的首要手段
策略模式解耦RunEvaluator 接口评估器可灵活扩展，无需修改评估流程

八、总结

LangSmith Python SDK 是一个设计精良的开源工程范例。它通过 RunTree 树形结构优雅地解决了 LLM 调用的追踪问题，用策略模式实现了灵活的评估系统，并通过批量写入、自动重试等机制保证了生产可用性。

无论你是在调试 Chain、构建自定义评估器，还是学习如何设计健壮的 Python SDK——研究 LangSmith 源码都将显著提升你的工程理解。

🚀 想要亲自探索？ 直接查看 GitHub 上 langchain-ai/langsmith-sdk 仓库，从 run_trees.py 开始你的源码之旅吧！