Agent Server와 분산 트레이싱

배포된 Agent Server를 다른 서비스에서 호출할 때 분산 트레이싱으로 트레이스 컨텍스트를 전파하면, 전체 요청이 LangSmith에서 하나의 통합된 트레이스로 보이게 할 수 있어요. Agent Server를 RemoteGraph나 SDK로 호출할 때 특히 유용합니다.

출처: 문서

본문

배포된 Agent Server를 다른 서비스에서 호출할 때, 트레이스 컨텍스트를 전파해서 전체 요청이 LangSmith에서 하나의 통합된 트레이스로 나타나게 할 수 있습니다. 이는 HTTP 헤더를 통해 컨텍스트를 전파하는 LangSmith의 분산 트레이싱 기능을 사용합니다.

작동 방식

분산 트레이싱은 컨텍스트 전파 헤더를 사용해 서비스 간 런을 연결합니다:

클라이언트는 현재 런에서 트레이스 컨텍스트를 유추해 HTTP 헤더로 보냅니다.
서버는 헤더를 읽어 런의 config와 metadata에 langsmith-trace 및 langsmith-project configurable 값으로 추가합니다. 에이전트가 사용될 때 특정 런의 트레이싱 컨텍스트를 설정하는 데 이 값을 사용할 수 있습니다.

사용되는 헤더는 다음과 같습니다:

langsmith-trace: 트레이스의 점으로 구분된 순서(dotted order)를 담고 있습니다.
baggage: LangSmith 프로젝트와 기타 선택적 태그 및 메타데이터를 지정합니다.

분산 트레이싱을 선택(opt-in)하려면 클라이언트와 서버 모두 선택해야 합니다.

서버 구성하기

분산 트레이스 컨텍스트를 수락하려면, 그래프가 config에서 트레이스 헤더를 읽고 트레이싱 컨텍스트를 설정해야 합니다. 헤더는 configurable 필드를 통해 langsmith-trace 및 langsmith-project로 전달됩니다.

경고: 분산 트레이싱 헤더(langsmith-trace, baggage)는 신뢰하는 트레이싱 컨텍스트로 소비됩니다. 신뢰하는 내부 서비스가 호출하는 배포에 대해서만 인바운드 트레이스 컨텍스트를 적용하도록 서버를 구성하세요. Agent Server가 신뢰하지 않는 제3자나 공개 인터넷으로부터 직접 요청을 받는 경우에는 이 헤더를 트레이싱 컨텍스트로 전파하지 마세요. 대신 게이트웨이 또는 프록시에서 해당 헤더를 제거하세요. 외부 호출자의 baggage를 신뢰하면 그들이 런이 기록되는 방식에 영향을 줄 수 있습니다.

import contextlib
import langsmith as ls
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에 export하세요:

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

클라이언트에서 연결하기

RemoteGraph — 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,  # 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?")

SDK — 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 trace headers
    ):
        pass
    return chunk

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

더 알아보기

분산 트레이싱의 일반 개념은 Distributed tracing 문서를 참고하세요.
RemoteGraph 사용법 전체 가이드는 RemoteGraph 문서를 확인해 보세요.

Agent Server와 분산 트레이싱