---

一、开篇导读

欢迎来到《LangSmith从入门到精通》专栏的第7节课。在前六节课中，我们已经完成了LangSmith从认知到环境搭建，再到核心概念的完整学习。现在，我们要真正深入到LangChain的应用开发中，利用LangSmith对Prompt、LLM、Chain进行全链路的监控和调试。

当我刚开始接触LangChain时，常常遇到这样的困境：一个看似简单的Chain跑起来之后，对内部执行逻辑缺乏了解。当应用表现异常（如回答不准确、响应慢、出现错误）时，无法精准定位问题——是Prompt设计不合理、Chain中LLM调用出错还是网络超时？这种不确定性严重影响开发进度和迭代优化。

LangSmith通过挂钩LangChain应用执行流程，自动捕获每步操作，将信息转化为有序、可视化的数据流，帮助开发者清晰掌握执行细节。它的追踪能力是实现可观察性的关键——每次应用完整执行会被记录为一个Trace（调用链），Trace内部由层级分明的Run（运行步骤）构成，形成可无限展开的“追踪树”。

本节课你将收获：

• 全链路监控体系：掌握LangChain中Prompt、LLM、Chain三类核心组件的追踪方式
• LCEL链路追踪：深入理解LCEL自动追踪机制，学会在复杂链路中定位瓶颈
• 回调系统深度剖析：理解LangChain Callback机制原理，学习消费追踪数据
• 企业级实战案例：构建一个完整的RAG搜索问答应用，实现全链路可观测
• 性能诊断与优化：学会利用追踪数据分析耗时和Token消耗

二、知识前置铺垫

2.1 LangChain执行链路的数据模型

LangChain的任何一次执行，本质上都是一个Runnable序列的编排与运行。当我们定义一个Chain（无论是通过LCEL的|操作符，还是传统的LLMChain），LangChain内部会构建一个执行图（Execution Graph）。这个图是由多个Runnable节点串联而成的有向无环图（DAG）。

在LCEL中，每个Runnable组件都是一个执行节点，组件之间的数据通过管道传递，这正是LangSmith能够精确追踪每条链路的原因。代码逻辑结构与追踪可视化结构同构，开发者使用LangSmith调试时，不仅能查看日志，还能直观验证代码逻辑是否符合预期。

2.2 三种追踪方式的比较

LangSmith提供了三种灵活的实现方式，你可以根据使用场景选择合适的方案：

1. 环境变量全自动模式
设置LANGSMITH_TRACING=true（或旧版本LANGCHAIN_TRACING_V2=true），所有LangChain组件都会自动上报追踪数据。这是零成本改造成本最低的方案，适合初次接入和快速原型开发。

2. 回调处理器手动模式
通过RunnableConfig中的callbacks字段手动传入LangChainTracer实例。这种方式提供了更精细的控制粒度，比如可以针对不同的Chain使用不同的Tracer，或者只在特定环境下开启追踪。

3. 选择性子调用追踪
使用tracing_context上下文管理器进行选择性子调用追踪。这种方式适用于需要区分开发版和生产版追踪的场景，或者需要在一个应用中对不同的请求采取不同的追踪策略。

对比总览：

追踪方式	粒度	代码侵入	灵活性	推荐场景
环境变量	全局	零侵入	低	快速入门、本地开发
回调处理器	组件级	低	中	需要精细控制时
上下文管理器	调用级	低	高	混合监控策略
@traceable装饰器	函数级	中	高	封装函数追踪

三、核心概念精讲

3.1 全链路监控的三个层次

真实的大模型链路通常涉及多个组件（LLM、检索器、工具、分发器等），LangSmith通过精巧的设计实现了这些组件的全链路可观测性。

层次一：Prompt层监控
核心关注点是Prompt模板的渲染结果、变量替换的准确性、Prompt长度对Token消耗的影响。LangSmith会记录PromptTemplate组件的渲染前变量（inputs）和渲染后完整Prompt（outputs），你可以清晰地看到变量填充是否正确地嵌入了模板。

层次二：LLM层监控
LLM调用是最核心也是最昂贵的环节。LangSmith会记录模型名称（如gpt-3.5-turbo）、temperature等参数、输入消息、输出内容、Token消耗（prompt_tokens + completion_tokens）、响应耗时、错误信息（如超时、认证失败）。

层次三：Chain层监控
Chain层面的核心价值在于链路归因。当一个Chain包含多个子Chain时，整个链路在LangSmith中会展开为一棵清晰的树形结构，任何子节点的耗时异常都能被快速定位。

3.2 LCEL与LangSmith的深度集成

LCEL（LangChain Expression Language）是LangChain的核心灵魂，通过重写|操作符实现了链式调用。LCEL与LangSmith有一个天然契合点：所有LCEL链都是Runnable的子类，每个Runnable在invoke时都会触发标准回调事件，而LangSmith正是通过监听这些事件完成自动追踪的。

当你用prompt | llm | output_parser构建一个简单的LCEL链时，LangSmith会自动捕获三个Runs：

• ChatPromptTemplate类型的Run（渲染Prompt）
• ChatOpenAI类型的Run（模型调用）
• StrOutputParser类型的Run（结果解析）

3.3 RunnableConfig：追踪配置的传递枢纽

在LCEL中，RunnableConfig是一个核心配置对象，用于统一所有Runnable组件的执行配置。LangSmith的追踪信息（如tags、metadata）正是通过RunnableConfig在所有Runnable之间传递的。

RunnableConfig核心字段（与追踪高度相关）：

• tags： 用于给当前调用及其子调用打标签，可在LangSmith面板中做筛选和分组
• metadata： 存储结构化补充信息的字典（如user_id、session_id），会在LangSmith的元数据区域完整展示
• run_name： 自定义Run的可读名称
• callbacks： 自定义回调处理器列表，包括LangChainTracer

企业级应用建议：通过RunnableConfig统一传递user_id、session_id等元数据，将业务维度的筛选能力与全链路追踪深度结合，是生产级项目的必备能力。

3.4 Thead聚合与多轮对话追踪

在多轮对话场景中，LangSmith通过thread_id、session_id或conversation_id将这些Trace聚合成一个可视化线程。这统一使用RunnableConfig的metadata字段传递thread_id即可自动完成聚合。在多租户系统中，可以用复合thread_id来隔离数据，如tenant_{tenant_id}:user_{user_id}:session_{session_token}。

四、原理底层剖析

4.1 Callback机制：拉通组件与监控的桥梁

LangChain的Callback机制是整个监控系统实现的基础。LangChain在各个组件的关键生命周期节点都定义了回调钩子。

生命周期钩子包括：

• on_llm_start / on_llm_end / on_llm_error：模型调用的开始、结束和异常
• on_chain_start / on_chain_end / on_chain_error：Chain执行的开始、结束和异常
• on_tool_start / on_tool_end / on_tool_error：工具调用的类似钩子

LangSmith的LangChainTracer是BaseCallbackHandler的子类，自动注册后，LangChain在执行每个Runnable时都会调用所有已注册的回调处理器中的对应方法，触发追踪数据的采集。

4.2 从回调事件到Run数据的转换流程

当你调用chain.invoke()时，背后发生的流程：

1. 根Run创建：触发on_chain_start，LangChainTracer据此创建根Run，记录start_time
2. 深度优先遍历执行：PromptTemplate执行（输入：占位符变量；输出：完整Prompt）→ 触发on_chain_end → 创建新的Run记录
3. LLM调用：触发on_llm_start → 发送HTTP请求到OpenAI → 收到响应后触发on_llm_end，记录Token消耗和耗时
4. OutputParser执行：类似地完成解析
5. 根Run完成：触发on_chain_end，计算总耗时，结束本次Trace

4.3 异步与非阻塞上报

LangSmith SDK采用异步队列上报机制，所有Run先写入内存队列，后台线程定期批量发送到LangSmith API，使得——即使LangSmith服务短暂不可用，数据也不会立即丢失；业务主线程的执行不受网络I/O的影响。

4.4 @traceable装饰器的运作原理

@traceable装饰器可以将任意普通函数包装为LangSmith链路中的一个Run节点。当你用@traceable(run_type="chain", name="custom_func")封装函数后，装饰器会在函数开始执行时调用追踪系统创建Run节点，附带tags和metadata，执行后根据执行结果上报成功或异常，最终形成多级嵌套的子Run。

五、环境配置手把手实战

由于前六节课已经完成环境配置，本节只做快速回顾和关键补充。

5.1 配置LangSmith环境变量与项目

# .env 关键配置
LANGCHAIN_TRACING_V2=true            # 启用追踪（最新版官方推荐使用LANGSMITH_TRACING）
LANGSMITH_TRACING=true               # 最新版推荐的环境变量名称
LANGCHAIN_API_KEY=lsv2_pt_xxxxx
LANGCHAIN_PROJECT=lesson07-fullstack-trace
OPENAI_API_KEY=sk-xxxxx

# 推荐额外配置
# LANGSMITH_OTEL_ENABLED=true        # 启用OpenTelemetry标准追踪
# LANGSMITH_ENDPOINT=https://api.smith.langchain.com

5.2 验证追踪配置

编写通用验证脚本check_tracing.py：

import os
from dotenv import load_dotenv
load_dotenv()  # 必须在所有导入之前

def check_tracing_config():
    """验证LangSmith追踪相关环境变量"""
    required = ["LANGCHAIN_API_KEY", "OPENAI_API_KEY"]
    tracing_opts = [
        ("LANGCHAIN_TRACING_V2", "旧版追踪开关"),
        ("LANGSMITH_TRACING", "新版追踪开关"),
        ("LANGCHAIN_PROJECT", "项目名称"),
        ("LANGSMITH_ENDPOINT", "API端点")
    ]
    
    for var, desc in tracing_opts:
        val = os.getenv(var)
        if val:
            print(f"✅ {var}={val}")
        else:
            print(f"⚠️ {var} 未设置 ({desc})")
    
    # 注意：LangSmith追踪需要两者至少一个为true
    if not (os.getenv("LANGCHAIN_TRACING_V2") == "true" or 
            os.getenv("LANGSMITH_TRACING") == "true"):
        print("❌ 追踪未启用！请设置 LANGCHAIN_TRACING_V2=true 或 LANGSMITH_TRACING=true")

if __name__ == "__main__":
    check_tracing_config()

关键提醒：新旧版本存在环境变量命名变化，LANGSMITH_TRACING是新版标准，为了兼容性建议同时设置两者。

六、完整可运行代码案例

6.1 Prompt层监控：精细捕捉模板渲染

本示例展示LangSmith中PromptTemplate的执行追踪，让你清晰对比“渲染前的变量”与“渲染后的完整Prompt”。

# prompt_tracing.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()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-prompt-trace"

def demo_prompt_tracing():
    """演示Prompt层的LangSmith追踪"""
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
    
    # 构建包含system和human消息的复杂模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是{role}专家，请用{language}回答问题"),
        ("human", "{question}")
    ])
    
    chain = prompt | llm | StrOutputParser()
    
    # 第一次调用：普通参数
    result1 = chain.invoke({
        "role": "AI",
        "language": "中文",
        "question": "什么是LangSmith？请用一句话解释"
    })
    print("第一次响应:", result1)
    
    # 第二次调用：改变role和language
    result2 = chain.invoke({
        "role": "资深工程师",
        "language": "英文",
        "question": "What are the key features of LangSmith?"
    })
    print("第二次响应:", result2)

if __name__ == "__main__":
    demo_prompt_tracing()

运行后登录LangSmith，点击任意Trace查看详情——展开ChatPromptTemplate节点，对比inputs（渲染前变量字典）和outputs（渲染后的完整Prompt）的差异。

6.2 LLM监控：捕获模型调用与Token消耗

# llm_tracing.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-llm-trace"

def demo_llm_tracing():
    """演示LLM层的LangSmith追踪"""
    # 配置不同的temperature值
    creative_llm = ChatOpenAI(
        model="gpt-3.5-turbo", 
        temperature=0.9,
        max_tokens=100
    )
    
    precise_llm = ChatOpenAI(
        model="gpt-3.5-turbo", 
        temperature=0.1,
        max_tokens=50
    )
    
    question = "请用三个词描述人工智能"
    
    print("创造性模式 (temperature=0.9):")
    creative_response = creative_llm.invoke(question)
    print(creative_response.content)
    
    print("\n精准模式 (temperature=0.1):")
    precise_response = precise_llm.invoke(question)
    print(precise_response.content)
    
    print("\n完成！请在LangSmith中对比两次调用的Token消耗和模型参数差异。")

if __name__ == "__main__":
    demo_llm_tracing()

6.3 Chain与LCEL全链路监控（核心案例）

本案例构建一个自定义的多步骤处理链，包含函数包装、自定义前置处理和后置处理，完整演示LCEL全链路监控。

# lcel_fullchain_trace.py
import os
from dotenv import load_dotenv
from langsmith import traceable
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-lcel-fullchain"

# 自定义前置预处理函数（使用@traceable精细化包装）
@traceable(run_type="chain", name="题目预处理")
def preprocess_topic(topic: str) -> str:
    """清理和增强（标准化输入）"""
    topic = topic.strip().lower()
    if topic.startswith("什么是"):
        topic = topic.replace("什么是", "")
    return f"[已预处理] {topic}"

# 自定义后置处理函数（用于字段抽取）
@traceable(run_type="chain", name="答案净化")
def extract_first_sentence(response: str) -> str:
    """提取第一句作为回答"""
    # 动态提取句号或问号结尾的句子
    import re
    match = re.search(r'^[^。？！]*[。？！]', response.strip())
    return match.group(0) if match else response[:100]

def demo_lcel_fullchain():
    """演示LCEL链的全链路LangSmith追踪"""
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)
    
    # 构建LCEL链（链式调用）
    chain = (
        {"raw_topic": RunnablePassthrough()}  # 步骤1: 透传原始输入
        | RunnableLambda(lambda x: preprocess_topic(x["raw_topic"]))  # 步骤2: 预处理
        | PromptTemplate.from_template("请简洁易懂地介绍：{topic}")  # 步骤3: 构建提示
        | llm  # 步骤4: LLM调用
        | StrOutputParser()  # 步骤5: 解析输出
        | RunnableLambda(lambda x: extract_first_sentence(x))  # 步骤6: 答案净化
    )
    
    topics = ["什么是LangSmith", "LangChain框架", "人工智能"]
    for topic in topics:
        print(f"\n处理题目: {topic}")
        result = chain.invoke(topic)
        print(f"最终答案: {result}")

if __name__ == "__main__":
    demo_lcel_fullchain()

运行后在LangSmith控制台中，你将看到完整的调用树从输入透传到答案净化的6个Run，完整展现了每个组件的输入输出及执行耗时。

6.4 RunnableConfig注入业务元数据实战

通过RunnableConfig为每次调用赋予业务维度信息，将用户维度、AB测试信息带入LangSmith。

# config_metadata_trace.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
from langchain_core.output_parsers import StrOutputParser

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-metadata-trace"

def inject_metadata_to_chain():
    """演示通过RunnableConfig注入业务元数据"""
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "产品专家，用{style}风格回答问题。"),
        ("human", "{question}")
    ])
    chain = prompt | llm | StrOutputParser()
    
    # 模拟请求队列业务场景
    requests = [
        ("LangSmith的优势是什么？", "详细", "user_001"),
        ("你能做什么？", "简洁", "user_002")
    ]
    
    for i, (question, style, user_id) in enumerate(requests):
        # 生成唯一请求标识符
        request_id = str(uuid.uuid4())
        
        # 通过RunnableConfig注入业务元数据
        config = RunnableConfig(
            tags=[f"user_tier:premium", f"request_priority:high", f"batch:demo"],
            metadata={
                "user_id": user_id,
                "request_id": request_id,
                "style": style,
                "ab_test_group": "group_B",         # AB测试分组
                "environment": os.getenv("APP_ENV", "development"),
                "experiment_id": "exp_2025_001"
            }
        )
        
        print(f"\n第{i+1}次调用 - 用户:{user_id} 请求ID:{request_id[:8]}")
        result = chain.invoke({"style": style, "question": question}, config=config)
        print(f"结果预览: {result[:80]}...")

if __name__ == "__main__":
    inject_metadata_to_chain()

6.5 RAG问答应用全链路监控

构建最贴近生产环境的RAG应用，完整监控包括Embedding、向量检索、最终问答的全链路。

# rag_tracing.py
import os
from dotenv import load_dotenv
from langsmith import traceable
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableConfig
from langchain_core.output_parsers import StrOutputParser

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-rag-solution"

# 模拟文档库
DOCUMENTS = [
    "LangSmith是LangChain官方推出的可观测性平台，支持调试、测试、监控大语言模型应用的完整生命流程。",
    "LangSmith提供调试专用控制台，可以可视化查看追踪树，有效降低调试难度。",
    "LCEL（LangChain表达式语言）能够通过管道符快速串联多个组件，构建LLM工作流。",
    "LangSmith的自动评测功能支持快速集成测试用例，实现Prompt版本的持续迭代。"
]

def setup_vectorstore():
    """构建向量数据库"""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_texts(DOCUMENTS, embeddings, persist_directory="./chroma_db")
    return vectorstore

@traceable(run_type="retriever", name="智能检索")
def retrieve_docs(vectorstore, query: str, k: int = 2):
    """检索相关的文档片段"""
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})
    docs = retriever.invoke(query)
    print(f"检索到{len(docs)}个相关文档片段")
    for idx, doc in enumerate(docs, 1):
        print(f"  文档{idx}: {doc.page_content[:50]}...")
    return docs

def build_rag_chain(vectorstore):
    """构建RAG链"""
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.2)
    prompt = ChatPromptTemplate.from_template("""
    基于以下背景知识回答用户的问题。
    如果背景知识不包含答案，直接说明“资料库中没有该信息，无法回答”。
    
    背景知识: {context}
    用户问题: {question}
    
    回答:
    """)
    
    def format_docs(docs):
        return "\n\n".join([doc.page_content for doc in docs])
    
    rag_chain = (
        {"context": lambda q: format_docs(retrieve_docs(vectorstore, q)), 
         "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return rag_chain

def rag_scenario():
    """全链路的RAG问答场景"""
    vectorstore = setup_vectorstore()
    rag_chain = build_rag_chain(vectorstore)
    
    questions = [
        "LangSmith是什么？有什么主要功能？",
        "LCEL有什么优势？如何构建链？",
        "今天天气怎么样？"
    ]
    
    for idx, q in enumerate(questions, 1):
        print(f"\n[{idx}]问题: {q}")
        config = RunnableConfig(
            metadata={"scenario": "RAG问答", "question_id": idx, "model": "gpt3.5"}
        )
        answer = rag_chain.invoke(q, config=config)
        print(f"答: {answer}\n")

if __name__ == "__main__":
    rag_scenario()

6.6 错误注入与全链路排障

通过模拟各种运行时错误，掌握LangSmith在排障中的定位能力。

# error_tracing.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
os.environ["LANGCHAIN_PROJECT"] = "lesson07-error-trace"

def simulate_openai_errors():
    """模拟OpenAI API调用可能出现的错误"""
    
    # 场景1: 无效模型名称
    print("=== 场景1: 无效模型名称 ===")
    try:
        wrong_model = ChatOpenAI(model="gpt-5.0")
        wrong_model.invoke("测试调用")
    except Exception as e:
        print(f"错误: {type(e).__name__}")
    
    # 场景2: 无效的API Key
    print("\n=== 场景2: 无效的API Key ===")
    try:
        original_key = os.getenv("OPENAI_API_KEY")
        os.environ["OPENAI_API_KEY"] = "invalid-key"
        bad_key_llm = ChatOpenAI()
        bad_key_llm.invoke("测试调用")
    except Exception as e:
        print(f"错误: {type(e).__name__}")
    finally:
        os.environ["OPENAI_API_KEY"] = original_key or ""
    
    # 场景3: 超时模拟（可选）
    print("\n=== 场景3: 超时配置 ===")
    slow_llm = ChatOpenAI(request_timeout=0.001)
    try:
        slow_llm.invoke("快速回答")
    except Exception as e:
        print(f"错误: {type(e).__name__}")

if __name__ == "__main__":
    simulate_openai_errors()

七、代码逐行详解

7.1 从Prompt到LLM的完整追踪链路

以prompt_tracing.py为例：

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}专家，请用{language}回答问题"),
    ("human", "{question}")
])

ChatPromptTemplate支持system和human等多角色消息，在LangSmith追踪中会展示messages数组。

chain = prompt | llm | StrOutputParser()

LCEL编织出三层嵌套运行结构，在LangSmith中对应形成清晰的追踪树。

result1 = chain.invoke({"role": "AI", "language": "中文", "question": "什么是LangSmith？"})

这里invoke内部包含了最完整的Run生命周期。

7.2 LCEL链的执行与追踪细节

chain = (
    {"raw_topic": RunnablePassthrough()}
    | RunnableLambda(lambda x: preprocess_topic(x["raw_topic"]))
    | ...

RunnablePassthrough作为数据传递的基础组件，其重要作用是维持数据流向。RunnableLambda将普通函数提升为Runnable组件，LangSmith会自动追踪其输入输出。

7.3 RunnableConfig的多字段与元数据功能剖析

config = RunnableConfig(
    tags=[f"user_tier:premium", "request_priority:high"],
    metadata={"user_id": user_id, "request_id": request_id, "style": style}
)

RunnableConfig解析中，tags和metadata不仅作用于当前Run，而且会自动传播到所有子Run，被广泛应用于业务维度的端到端追踪。

八、常见坑点与避坑指南

8.1 环境变量设置顺序不当导致追踪失效

典型错误：在导入LangChain组件后才加载环境变量。

解决方案：

# ❌ 错误
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()

# ✅ 正确
from dotenv import load_dotenv
load_dotenv()  # 必须在所有LangChain导入之前
from langchain_openai import ChatOpenAI

因为LangSmith的回调系统在LangChain导入时就会根据当前环境变量决定是否初始化追踪。

8.2 自定义函数未被追踪

问题：在LCEL链中调用了普通Python函数，但在LangSmith控制台里看不到该函数的详细追踪信息。

解决方案：使用@traceable装饰器包装：

from langsmith import traceable

@traceable(run_type="chain", name="我的自定义函数")
def my_custom_func(input_data):
    return process_data(input_data)

这样LangSmith就会将该函数作为Chain类型节点加入到追踪树中。

8.3 metadata中的高基数键导致UI卡顿

问题：将user_id等极高基数的字段作为metadata键，导致LangSmith UI筛选列表臃肿、响应缓慢。

原则：Tag命名使用低基数属性（env、version、region），高基数属性保留在metadata中。

8.4 异步代码追踪不全

问题：在async/await环境中，LangSmith的某些回调事件缺失或被丢失。

解决方案：优先使用兼容异步的atraceable装饰器，或确保在调用时传递config配置。

8.5 流式输出事件追踪

问题：启用streaming=True后，LangSmith只记录了最终完整响应，未保留中间Token。

解决方案：在回调处理器中实现on_llm_new_token钩子，或在LangSmith配置中开启流式记录。

8.6 跨节点链路中断

问题：追踪树在某一个步骤后断了，无法追溯后续调用。

这个情况在LangGraph、多进程或复杂工具调用时容易出现。解决方案是确保回调处理器在整个执行上下文中保持一致。

九、企业级落地最佳实践

9.1 统一的Project环境隔离策略

企业可将LANGCHAIN_PROJECT命名为{app_name}-{env}，通过CI/CD动态注入不同阶段的值，实现环境级别的数据隔离。

9.2 标准化Tag与Metadata

建立统一的Tag和Metadata命名规范，便于跨团队协作和后期筛选：

• Tag命名规范：env:prod、service:rag-service、version:2.1.0
• Metadata规范：request_id、traceparent、user_tier、geo_region

9.3 生产环境降采样与配额控制

使用确定性采样策略控制免费版配额，均匀分布采样确保代表性。

def deterministic_sample(user_id: str, sample_rate: float = 0.1) -> bool:
    import hashlib
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
    return hash_val < sample_rate * 100

在app.main中调用该函数，动态设置LANGCHAIN_TRACING_V2环境变量。

9.4 性能与耗时分析

利用Trace视图快速定位性能瓶颈，通过对比不同Runs的耗时比例找出耗时占比异常的子Run，调整对应组件的配置（如增加检索k值或更换模型）。

9.5 自定义回调处理器扩展

在LangSmith之上扩展自定义监控，创建继承BaseCallbackHandler的子类用于记录敏感操作或转发数据到内部审计系统。

9.6 敏感信息脱敏

对PII（个人身份信息）或API Key等敏感信息，LangSmith提供两种脱敏方案：

• 全局脱敏：设置LANGCHAIN_HIDE_INPUTS=true、LANGCHAIN_HIDE_OUTPUTS=true
• 函数级脱敏：在@traceable中设置capture_input=False参数

十、本节知识点总结

追踪层级	组件类型	记录的关键信息	业务价值
Prompt层	PromptTemplate、ChatPromptTemplate	渲染变量、完整Prompt	检查变量是否错漏、Prompt长度控制
LLM层	ChatOpenAI、BaseChatModel	模型参数、输入输出、Token消耗、耗时	定位模型选型和成本
Chain层	RunnableSequence、LCEL	执行图结构、嵌套Run树	链路归因、定位瓶颈组件
工具/检索层	Tool、Retriever	工具输入输出、检索相关度	工具调用失效跟踪

核心技术链条回顾：
环境变量LANGSMITH_TRACING=true → LangChain自动注册回调处理器 → 捕获生命周期事件（start/end/error） → 组装成Run对象 → 异步上报到LangSmith云端 → Web UI展示可视化追踪树。

LCEL与LangSmith的天然契合：LCEL链由一连串Runnable组件构成，每个Runnable都会触发回调事件，LangSmith通过监听这些事件自动捕获每一步的执行细节。

企业能力扩展：

• RunnableConfig支持多租户追踪和业务元数据注入
• @traceable让自定义代码块也可被可视化追踪
• Thread聚合实现多轮对话视图
• 分层Project进行环境隔离

十一、课后思考练习题

练习题1：理论理解

1.1 阐述LangChain的Callback机制和LangSmith SDK的@traceable装饰器之间的区别，各在什么场景下是最优选择。

1.2 在企业级LLM应用中，假如一套RAG链路每天被调用5000次，使用免费版LangSmith必然超配额。请设计一套工程方案，实现10%的可控随机采样追踪。

1.3 RunnableConfig中的tags和metadata字段在LangSmith中是如何体现分层的？如果给根Run设置tags=["A"]，子Run中的标签是["B"]，LangSmith会如何展示？

练习题2：动手实践

2.1 改造lcel_fullchain_trace.py，新增一个RunnableLambda步骤——在“答案净化”之后增加“字数统计”，统计答案的字数作为元数据记录到对应的Run中，验证LangSmith中能否看到额外信息。

2.2 编写一个实验对比Prompt质量的小脚本：构建两个ChatPromptTemplate模板（一个风格“正式”，另一个“幽默”），在Loop中调用同一个LLM 5次。在LangSmith控制台中使用tag筛选出两个实验组，对比各组响应速度和质量差异。

2.3 从rag_tracing.py中导出一次完整问答的所有Runs数据，并分析retrieve_docs步骤的平均耗时。若耗时超过整体链路的50%，你会从哪些方面进行优化？

练习题3：场景设计

3.1 你的RAG应用在线上突然变慢，你怀疑是某个检索步骤返回了过多冗余文档。请设计一套基于LangSmith追踪树的排查方案，包括如何定位慢节点、如何通过tags精准筛选慢查询。

3.2 公司有自己开发的内部Tool（计算器API），需要通过LangSmith进行追踪和性能监控。请给出三种技术集成方案，包括面向Python SDK的@traceable方式、通过RunnableConfig传递元数据的方式、以及基于OpenTelemetry标准监控的方式，并说明各自优缺点。

练习题4：源码分析（选做）

4.1 阅读LangChain源码中的BaseCallbackHandler的实现（位于langchain_core/callbacks/base.py），以及LangSmith的LangChainTracer（位于langsmith/wrappers/langchain.py），梳理LangSmith是如何把回调事件转换为最终上传到API的Run数据结构的。

4.2 探索LangSmith SDK中tracing_context上下文管理器如何跨async/await环境传播追踪信息，总结异步环境追踪的最佳实践。

---

下节课预告：

第8节课《LCEL表达式语言结合LangSmith：链路拆解、可视化调试与流程排查》，我们将系统化学习LCEL的高级模式——链的拆解与可视化、并行RunnableMap的时序分析，以及利用LangSmith高效排查性能瓶颈，帮你进一步提升链路级别的问题定位效率。

下一节见！

---

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

去订阅

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