使用 Agent Server 进行分布式追踪

当你从另一个服务调用已部署的 Agent Server 时，可以传播追踪上下文，使整个请求在 LangSmith 中显示为一条统一的追踪记录。这利用了 LangSmith 的分布式追踪功能，该功能通过 HTTP 头传播上下文。

工作原理

分布式追踪通过上下文传播头将不同服务中的运行记录关联起来：

客户端 从当前运行中推断出追踪上下文，并将其作为 HTTP 头发送。
服务器 读取这些头，并将它们作为 langsmith-trace 和 langsmith-project 的可配置值添加到运行的配置和元数据中。当你的代理被使用时，你可以选择使用这些值来设置特定运行的追踪上下文。

使用的头包括：

langsmith-trace：包含追踪的点分顺序。
baggage：指定 LangSmith 项目以及其他可选的标签和元数据。

要启用分布式追踪，客户端和服务器都需要选择启用。

配置服务器

要接受分布式追踪上下文，你的图必须从配置中读取追踪头并设置追踪上下文。这些头通过 configurable 字段以 langsmith-trace 和 langsmith-project 的形式传递。

分布式追踪头（langsmith-trace、baggage）被视为受信任的追踪上下文。仅当部署由受信任的内部服务调用时，才配置服务器应用入站追踪上下文。如果你的 Agent Server 直接接收来自不受信任的第三方或公共互联网的请求，请勿将这些头传播到追踪上下文中：应在网关或代理处剥离它们。信任来自外部调用者的 baggage 会让他们影响你的运行记录方式。

import contextlib
import langsmith as ls
from langgraph.graph import StateGraph, MessagesState

# 定义你的图
builder = StateGraph(MessagesState)
# ... 添加节点和边 ...
my_graph = builder.compile()

@contextlib.contextmanager
async def graph(config):
    configurable = config.get("configurable", {})
    parent_trace = configurable.get("langsmith-trace")
    parent_project = configurable.get("langsmith-project")
    # 如果你还想包含来自客户端的元数据和标签
    metadata = configurable.get("langsmith-metadata")
    tags = configurable.get("langsmith-tags")
    with ls.tracing_context(parent=parent_trace, project_name=parent_project, metadata=metadata, tags=tags):
        yield my_graph

在你的 langgraph.json 中导出这个 graph 函数：

{
  "graphs": {
    "agent": "./src/agent.py:graph"
  }
}

从客户端连接

RemoteGraph
SDK

初始化 RemoteGraph 时设置 distributed_tracing=True。这会自动在所有请求上传播追踪头。

from langgraph.graph import StateGraph
from langgraph.pregel.remote import RemoteGraph

remote_graph = RemoteGraph(
    "agent",
    url="<DEPLOYMENT_URL>",
    distributed_tracing=True,  # 启用追踪传播
)

def subgraph_node(query: str):
    # 追踪上下文会自动传播
    return remote_graph.invoke({
        "messages": [{"role": "user", "content": query}]
    })['messages'][-1]['content']

# RemoteGraph 在某个正在进行的工作上下文中被调用。
# 这可以是父 LangGraph 代理、使用 `@ls.traceable` 追踪的代码，
# 或任何其他已插桩的代码。
graph = (
        StateGraph(str)
            .add_node(subgraph_node)
            .add_edge("__start__", "subgraph_node")
            .compile()
)
# 远程图的执行将显示为这个追踪的子项
result = graph.invoke("What's the weather in SF?")

如果你直接使用 LangGraph SDK，请使用 run_tree.to_headers() 手动传播追踪头：

from langgraph_sdk import get_client
import langsmith as ls

client = get_client(url="<DEPLOYMENT_URL>")

with ls.trace("call_remote_agent", inputs={"query": query}) as rt:
    headers = rt.to_headers()
    async for chunk in client.runs.stream(
        thread_id=None,
        assistant_id="agent",
        input={"messages": [{"role": "user", "content": query}]},
        stream_mode="values",
        headers=headers,  # 传递追踪头
    ):
        pass
    return chunk

result = await call_remote_agent("What's the weather in SF?")

Develop your application

Capabilities

How to build

Auth & access control

Server customization

使用 Agent Server 进行分布式追踪

工作原理

配置服务器

从客户端连接

相关

​工作原理

​配置服务器

​从客户端连接

​相关

工作原理

配置服务器

从客户端连接

相关