LangSmith 集成会自动处理追踪。自定义插装允许你精确定义哪些函数被追踪，控制记录哪些输入和输出，并构建你的追踪层次结构，而无需重写应用程序逻辑。
前提条件
开始追踪前，请设置以下环境变量：
LANGSMITH_TRACING=true：启用追踪。设置此变量可以在不更改代码的情况下开启或关闭追踪。LANGSMITH_API_KEY：你的 LangSmith API 密钥。
要禁用追踪，请移除 LANGSMITH_TRACING 环境变量。这不会影响 RunTree 对象或直接 API 使用，它们是底层操作，不受追踪开关控制。
默认情况下，追踪记录会记录到名为 default 的项目中。要记录到其他项目，请参阅 将追踪记录到特定项目。 使用 @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-4.1-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()
run_pipeline 追踪，其中包含嵌套为子运行的 format_prompt、invoke_llm 和 parse_output。
当你用 traceable 包装一个同步函数时（例如，前面示例中的 formatPrompt），调用时请使用 await 关键字，以确保追踪被正确记录。
使用 trace 上下文管理器（仅限 Python）
在 Python 中，你可以使用 trace 上下文管理器将追踪记录到 LangSmith。这在以下情况下很有用：
- 你想为特定的代码块记录追踪。
- 你想控制追踪的输入、输出和其他属性。
- 使用装饰器或包装器不可行。
- 以上任意或全部情况。
该上下文管理器与 traceable 装饰器和 wrap_openai 包装器无缝集成，因此你可以在同一个应用程序中一起使用它们。
以下示例展示了三者一起使用的情况。wrap_openai 包装了 OpenAI 客户端，使其调用自动被追踪。my_tool 使用 @traceable 并设置 run_type="tool" 和自定义 name，以便在追踪中正确显示。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-4.1-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 对于此方法不是必需的。
此方法不推荐使用，因为在传播追踪上下文时更容易出错。
import openai
from langsmith.run_trees import RunTree
# 这可以是你的应用程序的用户输入
question = "Can you summarize this morning's meetings?"
# 创建一个顶级运行
pipeline = RunTree(
name="Chat Pipeline",
run_type="chain",
inputs={"question": question}
)
pipeline.post()
# 这可以在检索步骤中获取
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}"}
]
# 创建一个子运行
child_llm_run = pipeline.create_child(
name="OpenAI Call",
run_type="llm",
inputs={"messages": messages},
)
child_llm_run.post()
# 生成一个完成
client = openai.Client()
chat_completion = client.chat.completions.create(
model="gpt-4.1-mini", messages=messages
)
# 结束运行并记录它们
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]:
"""插装类中的所有公共方法。"""
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
# 处理 __dict__ 情况
for name in dir(cls):
if not name.startswith("_"):
try:
method = getattr(cls, name)
setattr(cls, name, wrap_method(name, method))
except AttributeError:
# 跳过无法设置的属性（例如，某些描述符）
pass
# 处理 __slots__ 情况
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:
# 跳过尚未有值的槽
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
# 查看追踪：https://smith.langchain.com/public/882f9ecf-5057-426a-ae98-0edf84fdcaf9/r
MyClass(13).combine(29)
确保所有追踪在退出前提交
LangSmith 在后台线程中执行追踪，以避免阻塞你的生产应用程序。这意味着你的进程可能在所有追踪成功发布到 LangSmith 之前就结束了。以下是确保在退出应用程序前提交所有追踪的一些选项。
使用 LangSmith SDK
如果你独立使用 LangSmith SDK，可以在退出前使用 flush 方法：
from langsmith import Client
client = Client()
@traceable(client=client)
async def my_traced_func():
# 你的代码在这里...
pass
try:
await my_traced_func()
finally:
await client.flush()
使用 LangChain
如果你正在使用 LangChain，请参考我们的 LangChain 追踪指南。
如果你更喜欢视频教程，请查看 LangSmith 入门课程中的 追踪基础视频。
