Skip to main content
直接在代码中添加检测可以让您精确控制应用程序追踪哪些函数、记录哪些输入和输出,以及如何构建您的追踪层次结构。三种核心检测方法是: 本页还涵盖: 对于 LangChain(Python 或 JS/TS),请参阅 LangChain 特定说明
如果您使用的是内置 LangSmith 集成的 LLM 提供商或代理框架,请参阅集成概述

前提条件

在追踪之前,请设置以下环境变量:
  • LANGSMITH_TRACING=true:启用追踪。设置此变量可以在不更改代码的情况下切换追踪的开启和关闭。
    LANGSMITH_TRACING 控制 @traceable 装饰器和 trace 上下文管理器。要在运行时为 @traceable 覆盖此设置而不更改环境变量,请使用 tracing_context(enabled=True/False)(Python)或直接向 traceable 传递 tracingEnabled(JS/TS)。RunTree 对象不受这些控制中的任何一个影响;它们在发布时总是向 LangSmith 发送数据。
  • LANGSMITH_API_KEY:您的 LangSmith API 密钥
  • 默认情况下,LangSmith 将追踪记录到名为 default 的项目中。要记录到不同的项目,请设置 LANGSMITH_PROJECT。更多详情,请参阅将追踪记录到特定项目

使用 @traceable / traceable

@traceable(Python)或 traceable(TypeScript)应用于任何函数,使其成为被追踪的运行。LangSmith 会自动处理嵌套调用之间的上下文传播。 以下示例追踪一个简单的管道:run_pipeline 调用 format_prompt 构建消息,调用 invoke_llm 调用模型,然后调用 parse_output 提取结果。 每个函数都被单独追踪,并且由于它们是从 run_pipeline(也被追踪)内部调用的,LangSmith 会自动将它们嵌套为子运行。invoke_llm 使用 run_type="llm" 将其标记为 LLM 调用,以便 LangSmith 可以正确渲染令牌计数和延迟:
from langsmith import traceable
from openai import Client

openai = Client()

@traceable
def format_prompt(subject):
  return [
      {
          "role": "system",
          "content": "You are a helpful assistant.",
      },
      {
          "role": "user",
          "content": f"What's a good name for a store that sells {subject}?"
      }
  ]

@traceable(run_type="llm")
def invoke_llm(messages):
  return openai.chat.completions.create(
      messages=messages, model="gpt-5.4-mini", temperature=0
  )

@traceable
def parse_output(response):
  return response.choices[0].message.content

@traceable
def run_pipeline():
  messages = format_prompt("colorful socks")
  response = invoke_llm(messages)
  return parse_output(response)

run_pipeline()
UI 中,您将找到一个 run_pipeline 追踪,其中 format_promptinvoke_llmparse_output 作为嵌套的子运行。
当您用 traceable 包装同步函数(例如,前面示例中的 formatPrompt)时,请在调用它时使用 await 关键字,以确保追踪被正确记录。

使用 trace 上下文管理器(仅限 Python)

在 Python 中,您可以使用 trace 上下文管理器将追踪记录到 LangSmith。这在以下情况下很有用:
  1. 您想为特定代码块记录追踪。
  2. 您想控制追踪的输入、输出和其他属性。
  3. 使用装饰器或包装器不可行。
  4. 以上任何或全部情况。
该上下文管理器与 traceable 装饰器和 wrap_openai 包装器无缝集成,因此您可以在同一应用程序中一起使用它们。 以下示例展示了三者一起使用的情况。wrap_openai 包装 OpenAI 客户端,使其调用被自动追踪。my_tool 使用带有 run_type="tool" 和自定义 name@traceable,以便在追踪中正确显示。chat_pipeline 本身没有被装饰;相反,ls.trace 包装了调用,允许您显式传递项目名称和输入,并通过 rt.end() 手动设置输出:
import openai
import langsmith as ls
from langsmith.wrappers import wrap_openai

client = wrap_openai(openai.Client())

@ls.traceable(run_type="tool", name="Retrieve Context")
def my_tool(question: str) -> str:
    return "During this morning's meeting, we solved all world conflict."

def chat_pipeline(question: str):
    context = my_tool(question)
    messages = [
        { "role": "system", "content": "You are a helpful assistant. Please respond to the user's request only based on the given context." },
        { "role": "user", "content": f"Question: {question}\nContext: {context}"}
    ]
    chat_completion = client.chat.completions.create(
        model="gpt-5.4-mini", messages=messages
    )
    return chat_completion.choices[0].message.content

app_inputs = {"input": "Can you summarize this morning's meetings?"}

with ls.trace("Chat Pipeline", "chain", project_name="my_test", inputs=app_inputs) as rt:
    output = chat_pipeline("Can you summarize this morning's meetings?")
    rt.end(outputs={"output": output})

使用 RunTree API

另一种将追踪记录到 LangSmith 的更显式的方法是通过 RunTree API。此 API 允许您对追踪进行更多控制。您可以手动创建运行和子运行来组装您的追踪。您仍然需要设置 LANGSMITH_API_KEY,但此方法不需要 LANGSMITH_TRACING 此方法不推荐用于大多数用例;与自动处理上下文传播的 @traceable 相比,手动管理追踪上下文容易出错。
import openai
from langsmith.run_trees import RunTree

# This can be a user input to your app
question = "Can you summarize this morning's meetings?"

# Create a top-level run
pipeline = RunTree(
  name="Chat Pipeline",
  run_type="chain",
  inputs={"question": question}
)
pipeline.post()

# This can be retrieved in a retrieval step
context = "During this morning's meeting, we solved all world conflict."
messages = [
  { "role": "system", "content": "You are a helpful assistant. Please respond to the user's request only based on the given context." },
  { "role": "user", "content": f"Question: {question}\nContext: {context}"}
]

# Create a child run
child_llm_run = pipeline.create_child(
  name="OpenAI Call",
  run_type="llm",
  inputs={"messages": messages},
)
child_llm_run.post()

# Generate a completion
client = openai.Client()
chat_completion = client.chat.completions.create(
  model="gpt-5.4-mini", messages=messages
)

# End the runs and log them
child_llm_run.end(outputs=chat_completion)
child_llm_run.patch()
pipeline.end(outputs={"answer": chat_completion.choices[0].message.content})
pipeline.patch()

示例用法

您可以扩展上一节中解释的实用程序来追踪任何代码。以下代码显示了一些示例扩展。 追踪类中的任何公共方法:
from typing import Any, Callable, Type, TypeVar

T = TypeVar("T")

def traceable_cls(cls: Type[T]) -> Type[T]:
    """Instrument all public methods in a class."""
    def wrap_method(name: str, method: Any) -> Any:
        if callable(method) and not name.startswith("__"):
            return traceable(name=f"{cls.__name__}.{name}")(method)
        return method

    # Handle __dict__ case
    for name in dir(cls):
        if not name.startswith("_"):
            try:
                method = getattr(cls, name)
                setattr(cls, name, wrap_method(name, method))
            except AttributeError:
                # Skip attributes that can't be set (e.g., some descriptors)
                pass

    # Handle __slots__ case
    if hasattr(cls, "__slots__"):
        for slot in cls.__slots__:  # type: ignore[attr-defined]
            if not slot.startswith("__"):
                try:
                    method = getattr(cls, slot)
                    setattr(cls, slot, wrap_method(slot, method))
                except AttributeError:
                    # Skip slots that don't have a value yet
                    pass

    return cls

@traceable_cls
class MyClass:
    def __init__(self, some_val: int):
        self.some_val = some_val

    def combine(self, other_val: int):
        return self.some_val + other_val

# See trace: https://smith.langchain.com/public/882f9ecf-5057-426a-ae98-0edf84fdcaf9/r
MyClass(13).combine(29)

指定自定义运行 ID

默认情况下,LangSmith 为每个运行分配一个随机 ID。当您需要提前知道运行 ID 时(例如,在运行后立即附加反馈),可以覆盖此 ID,将 LangSmith 运行与外部系统的 ID 关联,或使用确定性 ID 使运行幂等。
自定义运行 ID 请使用 UUID v7。UUIDv7 嵌入时间戳,这保留了追踪中运行的正确时间顺序。LangSmith SDK 导出了一个 uuid7 辅助函数(Python v0.4.43+,JS v0.3.80+):
  • Pythonfrom langsmith import uuid7
  • JS/TSimport { uuid7 } from 'langsmith'
任何 UUID v7 字符串都被接受——您可以使用 SDK 辅助函数,或者如果您系统已经使用 UUID v7 标识符,也可以使用您自己的。
使用以下方法之一:
  • @traceable:在调用 @traceable 函数时,在 langsmith_extra 中传递 run_id(Python),或在传递给 traceable 的配置对象中传递 id(TypeScript):
    from langsmith import traceable, uuid7
    
    @traceable
    def my_pipeline(question: str) -> str:
        return "answer"
    
    run_id = uuid7()
    my_pipeline("What is the capital of France?", langsmith_extra={"run_id": run_id})
    
    # run_id can now be used to attach feedback, query the run, etc.
    
  • trace 上下文管理器(仅限 Python):直接向 trace 上下文管理器构造函数传递 run_id
    Python
    from langsmith import trace, uuid7
    
    run_id = uuid7()
    
    with trace("my-pipeline", run_id=run_id) as run:
        result = "answer"
        run.end(outputs={"result": result})
    
    # run_id can now be used to attach feedback, query the run, etc.
    

确保在退出前提交所有追踪

LangSmith 在后台线程中执行追踪,以避免阻碍您的生产应用程序。这意味着您的进程可能在所有追踪成功发布到 LangSmith 之前结束。请参阅以下选项:
  • 如果您使用的是 LangChain,请参阅 LangChain 追踪指南
  • 如果您独立使用 LangSmith SDK,可以在退出前使用 flush 方法:
    from langsmith import Client
    
    client = Client()
    
    @traceable(client=client)
    async def my_traced_func():
    # Your code here...
    pass
    
    try:
    await my_traced_func()
    finally:
    await client.flush()
    

相关内容