LangChain 官方中文文档

学习如何直接使用 LangSmith API 追踪你的 LLM 应用程序。

强烈建议使用我们的 Python 或 TypeScript SDK 将追踪数据发送到 LangSmith。我们设计的这些 SDK 包含了批处理和后台处理等优化，以确保发送追踪数据不会影响应用程序的性能。但是，如果你无法使用我们的 SDK，可以使用 LangSmith REST API 来发送追踪数据。如果在应用程序中同步发送追踪数据，性能可能会受到影响。本指南将展示如何使用 LangSmith REST API 追踪一个请求。请查看我们的 API 文档以获取完整的端点列表以及请求/响应模式。

基础追踪

记录运行（run）的最简单方法是通过 POST 和 PATCH /runs 端点。这些路由期望最少的关于树结构的上下文信息。

使用 LangSmith REST API 时，你需要在请求头中以 "x-api-key" 的形式提供你的 API 密钥。

如果你的 API 密钥关联到多个工作区，你需要在请求头中使用 "x-tenant-id" 指定要使用的工作区。

在简单的示例中，你不需要在请求体中设置 dotted_order 或 trace_id 字段。这些字段将由系统自动生成。虽然这更简单，但在 LangSmith 中速度较慢且速率限制较低。

以下示例展示了如何在 Python 中直接利用我们的 API。同样的原理也适用于其他语言。

python

import openai
import os
import requests
from datetime import datetime, timezone
from langsmith import uuid7

# 在请求头中发送你的 API 密钥
headers = {
    "x-api-key": os.environ["LANGSMITH_API_KEY"],
    "x-tenant-id": os.environ["LANGSMITH_WORKSPACE_ID"]
}

def post_run(run_id, name, run_type, inputs, parent_id=None):
    """向 API 发布新运行的函数。"""
    data = {
        "id": run_id.hex,
        "name": name,
        "run_type": run_type,
        "inputs": inputs,
        "start_time": datetime.utcnow().isoformat(),
        # "session_name": "project-name",  # 要追踪到的项目名称
        # "session_id": "project-id",  # 要追踪到的项目 ID。指定 session_name 或 session_id 中的一个
    }
    if parent_id:
        data["parent_run_id"] = parent_id.hex

    requests.post(
        "https://api.smith.langchain.com/runs",  # 对于自托管安装或欧盟区域，请适当更新
        json=data,
        headers=headers
    )

def patch_run(run_id, outputs):
    """用输出修补运行的函数。"""
    requests.patch(
        f"https://api.smith.langchain.com/runs/{run_id}",
        json={
            "outputs": outputs,
            "end_time": datetime.now(timezone.utc).isoformat(),
        },
        headers=headers,
    )

# 这可以是你的应用程序的用户输入
question = "Can you summarize this morning's meetings?"

# 这可以在检索步骤中获取
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}"}
]

# 创建父运行
parent_run_id = uuid7()
post_run(parent_run_id, "Chat Pipeline", "chain", {"question": question})

# 创建子运行
child_run_id = uuid7()
post_run(child_run_id, "OpenAI Call", "llm", {"messages": messages}, parent_run_id)

# 生成完成
client = openai.Client()
chat_completion = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages
)

# 结束运行
patch_run(child_run_id, chat_completion.dict())
patch_run(parent_run_id, {"answer": chat_completion.choices[0].message.content})

有关更多信息，请参阅运行（span）数据格式文档。

批量摄取

为了更快地摄取运行并获得更高的速率限制，你可以使用 POST /runs/multipart 链接端点。下面是一个示例。它需要 orjson（用于快速 JSON 处理）、requests-toolbelt 和 uuid-utils 才能运行。

python

import json
import os
import uuid
from datetime import datetime, timezone
from typing import Dict, List
import requests
from requests_toolbelt import MultipartEncoder
from uuid_utils.compat import uuid7

def create_dotted_order(
    start_time: datetime | None = None,
    run_id: uuid.UUID | None = None
) -> str:
    """为运行排序和层次结构创建点序字符串。

    点序用于建立运行之间的顺序和关系。
    它结合了时间戳和唯一标识符，以确保正确的排序和追踪。
    """
    st = start_time or datetime.now(timezone.utc)
    id_ = run_id or uuid7()
    return f"{st.strftime('%Y%m%dT%H%M%S%fZ')}{id_}"

def create_run_base(
    name: str,
    run_type: str,
    inputs: dict,
    start_time: datetime
) -> dict:
    """创建运行的基础结构。"""
    run_id = uuid7()
    return {
        "id": str(run_id),
        "trace_id": str(run_id),
        "name": name,
        "start_time": start_time.isoformat(),
        "inputs": inputs,
        "run_type": run_type,
    }

def construct_run(
    name: str,
    run_type: str,
    inputs: dict,
    parent_dotted_order: str | None = None,
) -> dict:
    """使用给定参数构造一个运行字典。

    此函数创建一个具有唯一 ID 和点序的运行，如果它是子运行，则建立其在追踪层次结构中的位置。
    """
    start_time = datetime.now(timezone.utc)
    run = create_run_base(name, run_type, inputs, start_time)
    current_dotted_order = create_dotted_order(start_time, uuid.UUID(run["id"]))

    if parent_dotted_order:
        current_dotted_order = f"{parent_dotted_order}.{current_dotted_order}"
        run["trace_id"] = parent_dotted_order.split(".")[0].split("Z")[1]
        run["parent_run_id"] = parent_dotted_order.split(".")[-1].split("Z")[1]

    run["dotted_order"] = current_dotted_order
    return run

def serialize_run(operation: str, run_data: dict) -> List[tuple]:
    """为多部分请求序列化一个运行。

    此函数将运行数据分离为多个部分，以便高效传输和存储。
    主要的运行数据和可选字段（inputs, outputs, events）被分别序列化。
    """
    run_id = run_data.get("id", str(uuid7()))

    # 分离可选字段
    inputs = run_data.pop("inputs", None)
    outputs = run_data.pop("outputs", None)
    events = run_data.pop("events", None)

    parts = []

    # 序列化主要运行数据
    run_data_json = json.dumps(run_data).encode("utf-8")
    parts.append(
        (
            f"{operation}.{run_id}",
            (
                None,
                run_data_json,
                "application/json",
                {"Content-Length": str(len(run_data_json))},
            ),
        )
    )

    # 序列化可选字段
    for key, value in [("inputs", inputs), ("outputs", outputs), ("events", events)]:
        if value:
            serialized_value = json.dumps(value).encode("utf-8")
            parts.append(
                (
                    f"{operation}.{run_id}.{key}",
                    (
                        None,
                        serialized_value,
                        "application/json",
                        {"Content-Length": str(len(serialized_value))},
                    ),
                )
            )

    return parts

def batch_ingest_runs(
    api_url: str,
    api_key: str,
    posts: list[dict] | None = None,
    patches: list[dict] | None = None,
) -> None:
    """在单个批量请求中摄取多个运行。

    此函数处理创建新运行（posts）和更新现有运行（patches）。
    与单个 API 调用相比，这对于摄取多个运行更高效。
    """
    boundary = uuid.uuid4().hex
    all_parts = []

    for operation, runs in zip(("post", "patch"), (posts, patches)):
        if runs:
            all_parts.extend(
                [part for run in runs for part in serialize_run(operation, run)]
            )

    encoder = MultipartEncoder(fields=all_parts, boundary=boundary)
    headers = {"Content-Type": encoder.content_type, "x-api-key": api_key}

    try:
        response = requests.post(
            f"{api_url}/runs/multipart",
            data=encoder,
            headers=headers
        )
        response.raise_for_status()
        print("Successfully ingested runs.")
    except requests.RequestException as e:
        print(f"Error ingesting runs: {e}")
        # 在生产环境中，你可能希望记录此错误或更稳健地处理它

# 配置 API URL 和密钥
# 对于生产使用，请考虑使用配置文件或环境变量
api_url = "https://api.smith.langchain.com"
api_key = os.environ.get("LANGSMITH_API_KEY")

if not api_key:
    raise ValueError("LANGSMITH_API_KEY environment variable is not set")

# 创建一个父运行
parent_run = construct_run(
    name="Parent Run",
    run_type="chain",
    inputs={"main_question": "Tell me about France"},
)

# 创建一个子运行，链接到父运行
child_run = construct_run(
    name="Child Run",
    run_type="llm",
    inputs={"question": "What is the capital of France?"},
    parent_dotted_order=parent_run["dotted_order"],
)

# 首先，发布运行以创建它们
posts = [parent_run, child_run]
batch_ingest_runs(api_url, api_key, posts=posts)

# 然后，用结束时间和任何输出更新运行
child_run_update = {
    **child_run,
    "end_time": datetime.now(timezone.utc).isoformat(),
    "outputs": {"answer": "Paris is the capital of France."},
}

parent_run_update = {
    **parent_run,
    "end_time": datetime.now(timezone.utc).isoformat(),
    "outputs": {"summary": "Discussion about France, including its capital."},
}

patches = [parent_run_update, child_run_update]
batch_ingest_runs(api_url, api_key, patches=patches)

# 注意：此示例需要 `requests` 和 `requests_toolbelt` 库。
# 你可以使用 pip 安装它们：
# pip install requests requests_toolbelt

使用 API 进行追踪

基础追踪 ​

批量摄取 ​

基础追踪

批量摄取