添加 插桩 直接添加到代码中,让你可以精确控制应用程序追踪哪些函数、记录哪些输入和输出,以及如何构建你的 追踪 层级结构。三种核心插桩方法是:
- -
@traceable装饰器:适用于大多数场景 - -
trace上下文管理器:仅适用于 Python - -
RunTreeAPI:显式的底层控制
本页还涵盖:
- - 指定自定义运行 ID,这对于在运行后立即附加反馈或与外部系统关联非常有用。
- - 确保在进程退出前提交所有追踪 。
For LangChain (Python or JS/TS), refer to the LangChain 特定说明.
前提条件
在开始追踪之前,请设置以下环境变量:
- -
LANGSMITH_TRACING=true:启用追踪。设置此变量可开启或关闭追踪,而无需更改代码。
- -
LANGSMITH_API_KEY:你的 LangSmith API 密钥. - - 默认情况下,LangSmith 将追踪记录到名为
default的项目中。若要记录到其他项目,请设置LANGSMITH_PROJECT。有关更多详情,请参阅 将追踪记录到特定项目.
使用 @traceable / traceable
将 @traceable(Python)、 traceable (TypeScript)、 traceable (Kotlin)或 Tracing.traceFunction (Java)应用到任何函数,使其成为追踪的运行。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()
const openai = new OpenAI();
const formatPrompt = traceable((subject: string) => {
return [
{
role: "system" as const,
content: "You are a helpful assistant.",
},
{
role: "user" as const,
content: `What's a good name for a store that sells ${subject}?`,
},
];
},{ name: "formatPrompt" });
const invokeLLM = traceable(
async ({ messages }: { messages: { role: string; content: string }[] }) => {
return openai.chat.completions.create({
model: "gpt-5.4-mini",
messages: messages,
temperature: 0,
});
},
{ run_type: "llm", name: "invokeLLM" }
);
const parseOutput = traceable(
(response: any) => {
return response.choices[0].message.content;
},
{ name: "parseOutput" }
);
const runPipeline = traceable(
async () => {
const messages = await formatPrompt("colorful socks");
const response = await invokeLLM({ messages });
return parseOutput(response);
},
{ name: "runPipeline" }
);
await runPipeline();
在 UI中,您会找到一个 run_pipeline 追踪,包含 format_prompt, invoke_llm和 parse_output 作为嵌套的子运行。
使用 trace 上下文管理器(仅限 Python)
在 Python 中,您可以使用 trace 上下文管理器将追踪记录到 LangSmith。这在以下情况下很有用:
- 您想为特定代码块记录追踪。
- 您想控制追踪的输入、输出和其他属性。
- 使用装饰器或包装器不可行。
- 以上任意或全部。
上下文管理器与 traceable 装饰器和 wrap_openai 包装器无缝集成,因此您可以在同一应用程序中一起使用它们。
以下示例展示了三种方法一起使用的情况。 wrap_openai 包装了 OpenAI 客户端,因此其调用会被自动追踪。 my_tool 使用 @traceable 和 run_type="tool" 以及自定义 name 以在追踪中正确显示。 chat_pipeline 本身没有装饰;相反, ls.trace 包装了调用,允许您显式传递项目名称和输入,并通过 rt.end():
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相比,手动管理追踪上下文更容易出错,后者自动处理上下文传播。
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()
// This can be a user input to your app
const question = "Can you summarize this morning's meetings?";
const pipeline = new RunTree({
name: "Chat Pipeline",
run_type: "chain",
inputs: { question }
});
await pipeline.postRun();
// This can be retrieved in a retrieval step
const context = "During this morning's meeting, we solved all world conflict.";
const 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: `Question: ${question}Context: ${context}` }
];
// Create a child run
const childRun = await pipeline.createChild({
name: "OpenAI Call",
run_type: "llm",
inputs: { messages },
});
await childRun.postRun();
// Generate a completion
const client = new OpenAI();
const chatCompletion = await client.chat.completions.create({
model: "gpt-5.4-mini",
messages: messages,
});
// End the runs and log them
childRun.end(chatCompletion);
await childRun.patchRun();
pipeline.end({ outputs: { answer: chatCompletion.choices[0].message.content } });
await pipeline.patchRun();
Java 和 Kotlin 示例使用自定义根运行 ID 和专用执行器。关闭执行器并等待终止可确保后台运行提交在进程退出前完成。
示例用法
您可以扩展上一节中解释的实用程序来追踪任何代码。以下代码展示了一些扩展示例。
追踪类中的任何公共方法:
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 时(例如,在运行后立即附加 反馈 )、将 LangSmith 运行与外部系统的 ID 关联,或使用确定性 ID 使运行具有幂等性时,您可以覆盖此设置。
使用以下方法之一:
- -
@traceable:传递run_id在内部langsmith_extra调用@traceable函数(Python),或传递id在传递给traceable(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.
const runId = uuid7();
const myPipeline = traceable(
async (question: string) => {
return "answer";
},
{ name: "my-pipeline", id: runId }
);
await myPipeline("What is the capital of France?");
// runId can now be used to attach feedback, query the run, etc.
- -
trace上下文管理器(仅Python):传递run_id直接传递给 trace 上下文管理器构造函数:
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()
const langsmithClient = new Client({});
const myTracedFunc = traceable(async () => {
// Your code here...
},{ client: langsmithClient });
try {
await myTracedFunc();
} finally {
await langsmithClient.flush();
}
相关
- - 可观测性概念:运行、追踪和LangSmith数据模型的背景知识
- - 运行(跨度)数据格式:运行字段(包括
dotted_order,trace_id和parent_run_id - - 使用SDK记录用户反馈:预先指定运行ID的常见用例
- - 在追踪函数内访问当前运行(跨度):从追踪内部读取或修改活动运行
- - 将追踪记录到特定项目:将追踪路由到命名项目而不是
default - - 使用API进行追踪:SDK的低级REST API替代方案
- - 追踪基础视频 来自LangSmith入门课程