追踪器 API

LangSmith 链路追踪与调试

概述

LangSmith 追踪系统让你完整记录 LangChain 应用的执行过程,支持调试、性能分析和质量评估。

graph LR
    A[LangChain 应用] --> B[设置环境变量]
    B --> C[自动追踪]
    C --> D[LangSmith 平台]

    D --> E[查看执行链路]
    D --> F[性能分析]
    D --> G[对比版本]

    E --> H[调试]
    F --> I[优化]
    G --> J[A/B 测试]

    style B fill:#e1f5fe
    style D fill:#c8e6c9

配置

环境变量配置

设置环境变量启用自动追踪。

bash
# 设置环境变量
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="lsv2_your-api-key-here"
export LANGCHAIN_PROJECT="my-awesome-project"

# 或在代码中设置
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_your-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

可选环境变量

变量 说明 默认值
LANGCHAIN_TRACING_V2 是否启用追踪 false
LANGCHAIN_API_KEY LangSmith API 密钥 -
LANGCHAIN_ENDPOINT API 端点 URL https://api.smith.langchain.com
LANGCHAIN_PROJECT 项目名称 default
LANGCHAIN_SESSION 会话名称 -
LANGCHAIN_HIDE_INPUTS 是否隐藏输入(隐私保护) false

Runnable 追踪

配置参数

from langchain_core.runnables import RunnableConfig

# 追踪配置
config = RunnableConfig(
    callbacks=[callback_handler],  # 回调处理器
    tags=["production", "api"],      # 标签
    metadata={                        # 元数据
        "user_id": "user_123",
        "version": "1.0"
    },
    run_name="my_chain_run"            # 运行名称
)

result = chain.invoke(input_data, config=config)

使用回调追踪

from langchain_core.callbacks import BaseCallbackHandler

class TracingCallback(BaseCallbackHandler):
    """自定义追踪回调"""

    def __init__(self, project_name: str):
        self.project_name = project_name

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[{self.project_name}] Chain 开始: {inputs}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[{self.project_name}] Chain 结束: {outputs}")

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[{self.project_name}] LLM 开始")

    def on_llm_end(self, response, **kwargs):
        print(f"[{self.project_name}] LLM 结束")

# 使用
handler = TracingCallback("my_project")
result = chain.invoke(input, config={"callbacks": [handler]})

LangSmith 集成

from langsmith import Client

# 初始化客户端
client = Client(
    api_key="lsv2_your-key",
    api_url="https://api.smith.langchain.com"  # 可选
)

# 创建运行
run = client.create_run(
    project_name="my-project",
    name="test_run",
    inputs={"query": "test"}
)

# 记录事件
run.create_event(
    event_type="llm",
    inputs={"prompt": "hello"},
    outputs={"response": "hi there"}
)

# 结束运行
run.end(outputs={"result": "success"})

# ========== 查询运行 ==========
# 列出运行
runs = client.list_runs(
    project_name="my-project",
    execution_order="DESC",  # 排序
    limit=10
)

# 获取特定运行
run = client.read_run(run_id="run_id_here")

# ========== 对比运行 ==========
from langsmith import evaluators

result = evaluators.create_comparison(
    runs=[run_id_1, run_id_2]
)

使用示例

python
# ========== 示例1: 基础自动追踪 ==========
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 启用追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_your-key"
os.environ["LANGCHAIN_PROJECT"] = "demo"

# 正常使用 - 自动追踪
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的故事")
llm = ChatOpenAI(model="gpt-4o")
chain = prompt | llm

result = chain.invoke({"topic": "Python"})
# 自动记录到 LangSmith

# ========== 示例2: 添加自定义追踪 ==========
from langchain_core.callbacks import BaseCallbackHandler

class CustomTracer(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM 输入: {prompts[0][:50]}...")

    def on_llm_end(self, response, **kwargs):
        if hasattr(response, "llm_output"):
            usage = response.llm_output.get("token_usage", {})
            print(f"Token 使用: {usage}")

result = chain.invoke(
    {"topic": "Python"},
    config={"callbacks": [CustomTracer()]}
)

# ========== 示例3: 多标签和元数据 ==========
result = chain.invoke(
    {"topic": "Python"},
    config={
        "tags": ["production", "v1.0"],
        "metadata": {
            "user_id": "user_123",
            "environment": "production",
            "model": "gpt-4o"
        },
        "run_name": "story_generation"
    }
)

# ========== 示例4: 查询运行历史 ==========
from langsmith import Client

client = Client(api_key="lsv2_your-key")

# 列出最近运行
runs = client.list_runs(
    project_name="demo",
    limit=10
)

for run in runs:
    print(f"Run: {run.name}, ID: {run.id}")

# ========== 示例5: 对比两个运行 ==========
run_id_1 = "first_run_id"
run_id_2 = "second_run_id"

comparison = client.compare_runs(
    run_a=run_id_1,
    run_b=run_id_2
)

# ========== 示例6: 获取性能数据 ==========
run = client.read_run(run_id="some_id")

# 计算耗时
start_time = run.start_time
end_time = run.end_time
duration = (end_time - start_time).total_seconds()

# Token 统计
if run.outputs:
    llm_results = run.outputs.get("llm_output", {})
    tokens = llm_results.get("token_usage", {})
    print(f"耗时: {duration}s, Tokens: {tokens}")

# ========== 示例7: 本地追踪(不使用 LangSmith)==========
from langchain.callbacks import StdOutCallbackHandler

handler = StdOutCallbackHandler()
result = chain.invoke(
    {"topic": "Python"},
    config={"callbacks": [handler]}
)
# 打印到控制台

相关 API