当您从另一个服务调用 Agent Server 时，您可以传播追踪上下文，使整个请求在 LangSmith 中显示为单一的统一追踪。这使用 LangSmith 的 分布式追踪 功能，通过 HTTP 头传播上下文。

工作原理

分布式追踪使用上下文传播头链接跨服务的运行：

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

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

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

配置服务器

要接受分布式追踪上下文，您的图必须从配置中读取追踪头并设置追踪上下文。头通过 configurable 字段作为 langsmith-trace 和 langsmith-project.

from langgraph.graph import StateGraph, MessagesState

# Define your graph
builder = StateGraph(MessagesState)
# ... add nodes and edges ...
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")
    # If you want to also include metadata and tags from the client
    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

导出此 graph 函数在您的 langgraph.json:

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

从客户端连接

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

remote_graph = RemoteGraph(
    "agent",
    url="",
    distributed_tracing=True,  # Enable trace propagation
)

def subgraph_node(query: str):
    # Trace context is automatically propagated
    return remote_graph.invoke({
        "messages": [{"role": "user", "content": query}]
    })['messages'][-1]['content']

# The RemoteGraph is called in the context of some on going work.
# This could be a parent LangGraph agent, code traced with `@ls.traceable`,
# or any other instrumented code.
graph = (
        StateGraph(str)
            .add_node(subgraph_node)
            .add_edge("__start__", "subgraph_node")
            .compile()
)
# The remote graph's execution will appear as a child of this trace
result = graph.invoke("What's the weather in SF?")

from langgraph_sdk import get_client


client = get_client(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 trace headers
    ):
        pass
    return chunk

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

相关

• - 分布式追踪：通用分布式追踪概念和模式
• - RemoteGraph：使用 RemoteGraph 与部署交互的完整指南