实施分布式跟踪

有时，您需要跨多个服务跟踪请求。

LangSmith 支持开箱即用的分布式跟踪，使用上下文传播标头 （langsmith-trace和可选baggage用于元数据/标签）。

客户端-服务器设置示例：

• 跟踪从客户端开始
• 在服务器上继续

Python 中的分布式跟踪

# client.py
from langsmith.run_helpers import get_current_run_tree, traceable
import httpx

@traceable
async def my_client_function():
    headers = {}
    async with httpx.AsyncClient(base_url="...") as client:
        if run_tree := get_current_run_tree():
            # add langsmith-id to headers
            headers.update(run_tree.to_headers())
        return await client.post("/my-route", headers=headers)

然后，服务器（或其他服务）可以通过适当地处理标头来继续跟踪。 如果您使用的是 asgi 应用程序 Starlette 或 FastAPI，则可以使用 LangSmith 的TracingMiddleware.

信息

这TracingMiddleware类已添加到langsmith==0.1.133.

使用 FastAPI 的示例：

from langsmith import traceable
from langsmith.middleware import TracingMiddleware
from fastapi import FastAPI, Request


app = FastAPI()  # Or Flask, Django, or any other framework
app.add_middleware(TracingMiddleware)

@traceable
async def some_function():
    ...

@app.post("/my-route")
async def fake_route(request: Request):
  return await some_function()

或者在 Starlette 中：

from starlette.applications import Starlette
from starlette.middleware import Middleware
from langsmith.middleware import TracingMiddleware

routes = ...

middleware = [
    Middleware(TracingMiddleware),
]

app = Starlette(..., middleware=middleware)

如果您使用的是其他服务器框架，则始终可以通过langsmith_extra:

# server.py
from langsmith import traceable
from langsmith.run_helpers import tracing_context
from fastapi import FastAPI, Request


@traceable
async def my_application():
    ...


app = FastAPI()  # Or Flask, Django, or any other framework


@app.post("/my-route")
async def fake_route(request: Request):
    # request.headers:  {"langsmith-trace": "..."}
    # as well as optional metadata/tags in `baggage`
    with tracing_context(parent=request.headers):
        return await my_application()

上面的示例使用了tracing_context上下文管理器。您还可以直接在langsmith_extra参数@traceable.

from langsmith.run_helpers import traceable, trace
# ... same as above
@app.post("/my-route")
async def fake_route(request: Request):
    # request.headers:  {"langsmith-trace": "..."}
    my_application(langsmith_extra={"parent": request.headers})

TypeScript 中的分布式跟踪

注意

TypeScript 中的分布式跟踪需要langsmith版本>=0.1.31

首先，我们从客户端获取当前运行树并将其转换为langsmith-trace和baggageheader 值，我们可以将其传递给服务器：

// client.mts
import { getCurrentRunTree, traceable } from "langsmith/traceable";

const client = traceable(
  async () => {
    const runTree = getCurrentRunTree();
    return await fetch("...", {
      method: "POST",
      headers: runTree.toHeaders(),
    }).then((a) => a.text());
  },
  { name: "client" }
);

await client();

然后，服务器将标头转换回运行树，用于进一步继续跟踪。

要将新创建的 run tree 传递给可跟踪函数，我们可以使用withRunTreehelper，这将确保 Run 树在可跟踪的调用中传播。

Express.JS

// server.mts
import { RunTree } from "langsmith";
import { traceable, withRunTree } from "langsmith/traceable";

import express from "express";
import bodyParser from "body-parser";

const server = traceable(
  (text: string) => `Hello from the server! Received "${text}"`,
  { name: "server" }
);

const app = express();

app.use(bodyParser.text());

app.post("/", async (req, res) => {
  const runTree = RunTree.fromHeaders(req.headers);
  const result = await withRunTree(runTree, () => server(req.body));
  res.send(result);
});

穗

// server.mts
import { RunTree } from "langsmith";
import { traceable, withRunTree } from "langsmith/traceable";

import { Hono } from "hono";

const server = traceable(
  (text: string) => `Hello from the server! Received "${text}"`,
  { name: "server" }
);

const app = new Hono();

app.post("/", async (c) => {
  const body = await c.req.text();
  const runTree = RunTree.fromHeaders(c.req.raw.headers);
  const result = await withRunTree(runTree, () => server(body));
  return c.body(result);
});