如果您决定不再跟踪运行,可以删除 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
另一种更明确的方式是通过 RunTree API 将跟踪记录到 LangSmith。此 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-4o-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
# 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 课程介绍中的跟踪基础视频。