如果您决定不再跟踪您的运行,您可以删除 LANGSMITH_TRACING 环境变量。请注意,这不会影响 RunTree 对象或 API 用户,因为这些是低级别对象,不受跟踪开关的影响。
有几种方法可以将跟踪日志记录到 LangSmith。
使用 @traceable / traceable
LangSmith 通过 Python 中的 @traceable 装饰器和 TypeScript 中的 traceable 函数,让您只需对现有代码进行少量更改即可轻松记录跟踪。
即使使用 @traceable 或 traceable,也必须将 LANGSMITH_TRACING 环境变量设置为 'true' 才能将跟踪记录到 LangSmith。这允许您在不更改代码的情况下打开和关闭跟踪。此外,您还需要将 LANGSMITH_API_KEY 环境变量设置为您的 API 密钥(有关更多信息,请参阅设置)。默认情况下,跟踪将记录到名为 default 的项目。要将跟踪记录到其他项目,请参阅本节。
@traceable 装饰器是 LangSmith Python SDK 记录跟踪的简单方法。只需使用 @traceable 装饰任何函数。 请注意,当使用 traceable 包装同步函数时(例如,下面示例中的 formatPrompt),您应该在调用它时使用 await 关键字,以确保正确记录跟踪。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-4o-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()
使用 trace 上下文管理器(仅限 Python)
在 Python 中,您可以使用 trace 上下文管理器将跟踪记录到 LangSmith。这在以下情况下很有用:
- 您想要记录特定代码块的跟踪。
- 您想要控制跟踪的输入、输出和其他属性。
- 使用装饰器或包装器不可行。
- 以上任何或所有情况。
上下文管理器与 traceable 装饰器和 wrap_openai 包装器无缝集成,因此您可以在同一个应用程序中一起使用它们。
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-4o-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
# 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-4o-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)
确保在退出前提交所有跟踪
LangSmith 的跟踪是在后台线程中完成的,以避免阻塞您的生产应用程序。这意味着您的进程可能在所有跟踪成功发布到 LangSmith 之前结束。以下是一些选项,用于确保在退出应用程序之前提交所有跟踪。
使用 LangSmith SDK
如果您单独使用 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()
使用 LangChain
如果您正在使用 LangChain,请参阅我们的LangChain 跟踪指南。 如果您喜欢视频教程,请查看 LangSmith 入门课程中的跟踪基础视频。