런타임에 그래프 재구축하기

런타임에 그래프 재구축하기

ServerRuntime을 사용해 각 실행마다 다른 구성으로 그래프를 재구축할 수 있어요.

새 실행을 위해 다른 구성으로 그래프를 재구축해야 할 수 있습니다. 예를 들어 사용자의 자격증명에 따라 다른 도구를 로드하고 싶을 수 있어요. 이 가이드는 ServerRuntime을 사용해 이를 수행하는 방법을 보여드립니다.

대부분의 경우 사용자 정의는 전체 그래프 구조를 동적으로 변경하기보다 개별 노드 내에서 config를 조건적으로 처리하는 것이 가장 좋습니다. 이렇게 하면 테스트와 관리가 더 쉬워져요.

출처: 문서

본문

사전 요구사항 (Prerequisites)

  • 먼저 배포용 앱을 설정하는 하우투 가이드를 확인하세요.
  • ServerRuntimelanggraph-api >= 0.7.31langgraph-sdk >= 0.3.5가 필요합니다. 그 이전에는 그래프 팩토리가 단일 config: RunnableConfig 인자만 허용했습니다.

그래프 정의하기 (Define graphs)

LLM을 호출하고 사용자에게 응답을 반환하는 간단한 그래프가 있는 앱이 있다고 가정해 보겠습니다. 앱 파일 디렉터리는 다음과 같습니다:

my-app/
|-- langgraph.json
|-- my_project/
|   |-- __init__.py
|   |-- agents.py     # code for your graph
|-- pyproject.toml

여기서 그래프는 agents.py에 정의되어 있습니다.

재구축 없음 (No rebuild)

Agent Server를 배포하는 가장 일반적인 방법은 파일 최상위에 정의된 컴파일된 그래프 인스턴스를 참조하는 것입니다. 예시는 아래와 같습니다:

# my_project/agents.py
from langgraph.graph import StateGraph, MessagesState, START

async def model(state: MessagesState):
    return {"messages": [{"role": "assistant", "content": "Hi, there!"}]}

graph_workflow = StateGraph(MessagesState)
graph_workflow.add_node("model", model)
graph_workflow.add_edge(START, "model")
agent = graph_workflow.compile()

서버가 그래프를 인식하도록 하려면 LangGraph API 구성(langgraph.json)에서 CompiledStateGraph 인스턴스를 담고 있는 변수에 대한 경로를 지정해야 합니다. 예:

{
    "$schema": "https://langgra.ph/schema.json",
    "dependencies": ["."],
    "graphs": {
        "chat_agent": "my_project.agents:agent",
    }
}

재구축 (Rebuild)

각 새 실행에서 그래프를 재구축하려면 그래프를 반환(또는 yield)하는 팩토리 함수(factory function) 를 제공하세요. 팩토리는 선택적으로 ServerRuntime 매개변수 또는 RunnableConfig를 받을 수 있습니다. 서버는 함수의 타입 어노테이션을 검사해 어떤 인자를 주입할지 결정하므로 올바른 타입 힌트를 포함해야 합니다. 서버의 큐 워커는 실행을 처리해야 할 때마다 팩토리 함수를 호출합니다. 이 함수는 상태를 업데이트하거나 상태를 읽거나 어시스턴트 스키마를 가져오는 일부 다른 엔드포인트에서도 호출됩니다. ServerRuntime은 어떤 컨텍스트가 호출을 트리거했는지 알려줍니다.

ServerRuntime베타 상태이며 향후 릴리스에서 변경될 수 있습니다.

단순 팩토리 (Simple factory)

가장 단순한 형태는 컴파일된 그래프를 반환하는 평범한 async def입니다:

from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langchain_core.runnables import RunnableConfig
from langgraph_sdk.runtime import ServerRuntime

from my_agent.utils.state import AgentState

model = ChatOpenAI(model="gpt-5.5")


def make_graph_for_user(user_id: str):
    """Build a graph customized per user."""
    graph_workflow = StateGraph(AgentState)

    async def call_model(state):
        return {"messages": [await model.ainvoke(state["messages"])]}

    graph_workflow.add_node("agent", call_model)
    graph_workflow.add_edge(START, "agent")
    return graph_workflow.compile()


async def make_graph(config: RunnableConfig, runtime: ServerRuntime):
    user = runtime.ensure_user()
    return make_graph_for_user(user.identity)
컨텍스트 매니저 팩토리 (Context manager factory)

리소스(데이터베이스 연결, MCP 도구 로드 등)를 설정하고 해제해야 하는 경우 비동기 컨텍스트 매니저를 사용하세요. runtime.execution_runtime을 사용해 그래프가 실제 실행을 위해 호출되는지 아니면 인트로스펙션(스키마, 시각화)만을 위해 호출되는지 확인합니다:

import contextlib

from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langchain_core.runnables import RunnableConfig
from langgraph_sdk.runtime import ServerRuntime

from my_agent.utils.state import AgentState

model = ChatOpenAI(model="gpt-5.5")


def make_agent_graph(tools: list):
    """Make a simple LLM agent."""
    graph_workflow = StateGraph(AgentState)
    bound = model.bind_tools(tools)

    async def call_model(state):
        return {"messages": [await bound.ainvoke(state["messages"])]}

    graph_workflow.add_node("agent", call_model)
    graph_workflow.add_edge(START, "agent")
    return graph_workflow.compile()


@contextlib.asynccontextmanager
async def make_graph(runtime: ServerRuntime):
    if ert := runtime.execution_runtime:
        # Only set up expensive resources during actual execution.
        # Introspection calls (get_schema, get_graph, ...) skip this.
        mcp_tools = await connect_mcp(ert.ensure_user())  # your setup logic
        yield make_agent_graph(tools=mcp_tools)
        await disconnect_mcp()  # your teardown logic
    else:
        # For schema/state reads, return a graph with the same
        # topology but no expensive resource setup.
        yield make_agent_graph(tools=[])

마지막으로, langgraph.json에 팩토리 경로를 지정합니다:

{
    "$schema": "https://langgra.ph/schema.json",
    "dependencies": ["."],
    "graphs": {
        "chat_agent": "my_project.agents:make_graph",
    }
}

ServerRuntime 참조 (ServerRuntime reference)

팩토리 함수는 다음 속성을 가진 ServerRuntime 인스턴스를 받습니다:

속성 (Attribute) 타입 (Type) 설명 (Description)
access_context str 팩토리가 호출된 이유: "threads.create_run", "threads.update", "threads.read", 또는 "assistants.read".
user BaseUser | None 인증된 사용자, 또는 커스텀 인증이 구성되지 않은 경우 None.
store BaseStore 지속성 및 메모리용 스토어 인스턴스.

메서드:

메서드 (Method) 설명 (Description)
ensure_user() 인증된 사용자를 반환. 사용자가 제공되지 않으면 PermissionError를 발생시킴.
execution_runtime access_context"threads.create_run"일 때 실행 런타임을 반환하고, 그 외에는 None을 반환. 이 속성을 사용해 비용이 큰 리소스를 실행 중에만 조건적으로 설정.

접근 컨텍스트 (Access contexts)

서버는 실행 실행뿐만 아니라 여러 컨텍스트에서 팩토리를 호출합니다. 모든 컨텍스트에서 반환된 그래프는 동일한 토폴로지(노드, 엣지, 상태 스키마)를 가져야 합니다. 쓰기 컨텍스트(threads.create_run, threads.update)에서 토폴로지가 일치하지 않으면 잘못된 상태 업데이트가 발생할 수 있습니다. 읽기 컨텍스트(threads.read, assistants.read)에서 불일치는 보고된 보류 작업, 스키마, 시각화에 영향을 주지만 데이터가 손상되지는 않습니다. execution_runtime을 사용해 그래프 구조를 변경하지 않고 비용이 큰 리소스를 조건적으로 설정하세요.

컨텍스트 (Context) 설명 (Description)
threads.create_run 전체 그래프 실행. execution_runtime 사용 가능.
threads.update aupdate_state를 통한 상태 업데이트. 노드 함수를 실행하지 않지만 보류 작업을 변경할 수 있음.
threads.read aget_state / aget_state_history를 통한 상태 읽기.
assistants.read 시각화, MCP, A2A 등을 위한 스키마 및 그래프 인트로스펙션.

그래프별 추적 사용자 지정하기 (Customize tracing per graph)

팩토리 함수를 사용해 특정 그래프의 추적을 사용자 지정하거나 비활성화할 수 있습니다. 예시는 조건부 추적: 배포된 에이전트에서 추적 사용자 지정을 참고하세요.

LangGraph API 구성 파일에 대한 자세한 내용도 확인하세요.

더 알아보기 (Learn more)