워크플로와 에이전트 (Workflows and Agents)

LangGraph로 애플리케이션을 만들 때 자주 쓰는 워크플로와 에이전트 패턴들을 정리한 가이드예요.

워크플로(workflow) 는 코드의 흐름이 미리 정해져 있어요. 정해진 순서대로 동작하도록 설계됩니다.
에이전트(agent) 는 동적이에요. 스스로 처리 과정과 도구 사용을 정합니다.

LangGraph는 에이전트와 워크플로를 만들 때 쓸모 있는 기능을 여럿 제공하는데, 대표적으로 영속성(persistence), 스트리밍(streaming), 그리고 디버깅 지원과 배포(deployment)를 꼽을 수 있어요.

이런 워크플로 패턴을 LangSmith에서 추적하고 비교해 볼 수도 있어요. tracing quickstart를 따라가면 각 단계를 통해 데이터가 어떻게 흐르는지 볼 수 있어요. 또 LangSmith Engine을 함께 세팅하는 것도 추천해요. 이건 트레이스를 모니터링하고, 문제를 감지하고, 수정안을 제안해 줍니다.

Setup

워크플로나 에이전트를 만들려면 structured output(구조화된 출력)과 tool calling(도구 호출)을 지원하는 챗 모델이라면 어떤 것이든 쓸 수 있어요. 아래 예시는 Anthropic 모델을 사용합니다.

의존성을 설치합니다:

pip install langchain_core langchain-anthropic langgraph

LLM을 초기화합니다:

import os
import getpass

from langchain_anthropic import ChatAnthropic

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

llm = ChatAnthropic(model="claude-sonnet-4-6")

LLM과 증강 (LLMs and augmentations)

워크플로와 에이전트 시스템은 LLM과, 여기에 더하는 여러 가지 증강(augmentation) 을 바탕으로 만들어져요. LLM을 내 필요에 맞게 조정하는 방법으로는 도구 호출(Tool calling), 구조화된 출력(structured outputs), 단기 기억(short term memory) 같은 것들이 있어요.

structured output을 쓰는 방법을 코드로 먼저 볼게요. 출력 형태를 스키마로 정의하고, with_structured_output로 그 스키마를 LLM에 얹으면 모델이 정해진 형태의 인스턴스를 돌려줍니다. 도구는 bind_tools로 얹어요.

# Schema for structured output
from pydantic import BaseModel, Field


class SearchQuery(BaseModel):
    search_query: str | None = Field(
        default=None, description="Query that is optimized web search."
    )
    justification: str | None = Field(
        default=None, description="Why this query is relevant to the user's request."
    )


# Augment the LLM with schema for structured output
structured_llm = llm.with_structured_output(SearchQuery)

# Invoke the augmented LLM
output = structured_llm.invoke("How does Calcium CT score relate to high cholesterol?")
print(output)  # The model returns an instance of SearchQuery.

# Define a tool
def multiply(a: int, b: int) -> int:
    return a * b

# Augment the LLM with tools
llm_with_tools = llm.bind_tools([multiply])

# Invoke the LLM with input that triggers the tool call
msg = llm_with_tools.invoke("What is 2 times 3?")
print(msg.tool_calls)  # The model returns a request to call the tool.

프롬프트 체이닝 (Prompt chaining)

프롬프트 체이닝은 각 LLM 호출이 이전 호출의 출력을 입력으로 받아 처리하는 방식이에요. 작고, 검증 가능한 단계들로 쪼갤 수 있는 명확한 작업을 수행할 때 자주 쓰여요. 예를 들면:

문서를 다른 언어로 번역하기
생성된 콘텐츠의 일관성을 검증하기

아래 예시는 농담을 세 단계에 걸쳐 다듬는 워크플로예요. 먼저 주제로 초기 농담을 만들고, 그 농담에 반전(punchline)이 있는지 확인한 다음, 조건에 따라 농담을 개선하거나 그대로 끝냅니다.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display


# Graph state
class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str


# Nodes
def generate_joke(state: State):
    """First LLM call to generate initial joke"""

    msg = llm.invoke(f"Write a short joke about {state['topic']}")
    return {"joke": msg.content}


def check_punchline(state: State):
    """Gate function to check if the joke has a punchline"""

    # Simple check - does the joke contain "?" or "!"
    if "?" in state["joke"] or "!" in state["joke"]:
        return "Pass"
    return "Fail"


def improve_joke(state: State):
    """Second LLM call to improve the joke"""

    msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}")
    return {"improved_joke": msg.content}


def polish_joke(state: State):
    """Third LLM call for final polish"""
    msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}")
    return {"final_joke": msg.content}


# Build workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)

# Add edges to connect nodes
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)

# Compile
chain = workflow.compile()

# Show workflow
display(Image(chain.get_graph().draw_mermaid_png()))

# Invoke
state = chain.invoke({"topic": "cats"})
print("Initial joke:")
print(state["joke"])
print("\n--- --- ---\n")
if "improved_joke" in state:
    print("Improved joke:")
    print(state["improved_joke"])
    print("\n--- --- ---\n")

    print("Final joke:")
    print(state["final_joke"])
else:
    print("Final joke:")
    print(state["joke"])

핵심은 check_punchline이라는 게이트 함수가 있어서, 반전이 없으면 improve_joke로 보내 다듬고 반전이 있으면 바로 끝낸다는 점이에요. 이런 결정을 조건부 엣지(add_conditional_edges)로 표현합니다.

병렬화 (Parallelization)

병렬화에서는 LLM이 하나의 작업을 동시에 처리해요. 여러 독립적인 하위 작업을 동시에 실행하거나, 같은 작업을 여러 번 실행해서 서로 다른 출력을 확인하는 방식이에요. 병렬화는 주로 두 가지 목적으로 써요:

하위 작업을 쪼개 병렬로 실행해 속도를 높이기
작업을 여러 번 실행해 서로 다른 출력을 확인함으로써 신뢰도(confidence) 높이기

예시로는:

한쪽 하위 작업은 문서에서 키워드를 처리하고, 다른 하위 작업은 형식 오류가 있는지 검사하기
기준을 달리해 문서의 정확도를 여러 번 평가하기. 예를 들어 인용 수, 사용된 출처 수, 출처의 품질처럼 서로 다른 기준으로 채점하는 식이에요.

아래 코드는 같은 주제로 농담, 이야기, 시를 각각 병렬로 만들고, aggregator 노드가 세 결과를 하나로 합칩니다. 세 call_llm 노드가 모두 START에서 바로 갈라져 나오는 게 병렬 실행의 핵심이에요.

# Graph state
class State(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str


# Nodes
def call_llm_1(state: State):
    """First LLM call to generate initial joke"""

    msg = llm.invoke(f"Write a joke about {state['topic']}")
    return {"joke": msg.content}


def call_llm_2(state: State):
    """Second LLM call to generate story"""

    msg = llm.invoke(f"Write a story about {state['topic']}")
    return {"story": msg.content}


def call_llm_3(state: State):
    """Third LLM call to generate poem"""

    msg = llm.invoke(f"Write a poem about {state['topic']}")
    return {"poem": msg.content}


def aggregator(state: State):
    """Combine the joke, story and poem into a single output"""

    combined = f"Here's a story, joke, and poem about {state['topic']}!\n\n"
    combined += f"STORY:\n{state['story']}\n\n"
    combined += f"JOKE:\n{state['joke']}\n\n"
    combined += f"POEM:\n{state['poem']}"
    return {"combined_output": combined}


# Build workflow
parallel_builder = StateGraph(State)

# Add nodes
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)

# Add edges to connect nodes
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
parallel_workflow = parallel_builder.compile()

# Show workflow
display(Image(parallel_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = parallel_workflow.invoke({"topic": "cats"})
print(state["combined_output"])

라우팅 (Routing)

라우팅 워크플로는 입력을 처리한 다음, 그 입력을 상황에 맞는 전용 작업으로 보내요. 이렇게 하면 복잡한 작업에 대해 특화된 흐름을 정의할 수 있어요. 예를 들어 제품 관련 질문에 답하는 워크플로가 있다면, 먼저 질문 유형을 파악한 뒤 요청을 가격, 환불, 반품 같은 각각의 프로세스로 보낼 수 있어요.

아래 예시를 볼게요. llm_call_router가 structured output으로 입력을 poem/story/joke 중 하나로 분류하고, route_decision 함수가 그 결과에 따라 다음에 방문할 노드를 골라요.

from typing_extensions import Literal
from langchain.messages import HumanMessage, SystemMessage


# Schema for structured output to use as routing logic
class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(
        None, description="The next step in the routing process"
    )


# Augment the LLM with schema for structured output
router = llm.with_structured_output(Route)


# State
class State(TypedDict):
    input: str
    decision: str
    output: str


# Nodes
def llm_call_1(state: State):
    """Write a story"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_2(state: State):
    """Write a joke"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_3(state: State):
    """Write a poem"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_router(state: State):
    """Route the input to the appropriate node"""

    # Run the augmented LLM with structured output to serve as routing logic
    decision = router.invoke(
        [
            SystemMessage(
                content="Route the input to story, joke, or poem based on the user's request."
            ),
            HumanMessage(content=state["input"]),
        ]
    )

    return {"decision": decision.step}


# Conditional edge function to route to the appropriate node
def route_decision(state: State):
    # Return the node name you want to visit next
    if state["decision"] == "story":
        return "llm_call_1"
    elif state["decision"] == "joke":
        return "llm_call_2"
    elif state["decision"] == "poem":
        return "llm_call_3"


# Build workflow
router_builder = StateGraph(State)

# Add nodes
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)

# Add edges to connect nodes
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
    "llm_call_router",
    route_decision,
    {  # Name returned by route_decision : Name of next node to visit
        "llm_call_1": "llm_call_1",
        "llm_call_2": "llm_call_2",
        "llm_call_3": "llm_call_3",
    },
)
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)

# Compile workflow
router_workflow = router_builder.compile()

# Show the workflow
display(Image(router_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = router_workflow.invoke({"input": "Write me a joke about cats"})
print(state["output"])

오케스트레이터-워커 (Orchestrator-worker)

오케스트레이터-워커 구성에서 오케스트레이터는:

작업을 하위 작업(subtask)으로 쪼개고
하위 작업을 워커(worker)에게 위임하며
워커의 출력을 합쳐 최종 결과를 만듭니다

오케스트레이터-워커 워크플로는 더 유연해서, 병렬화처럼 하위 작업을 미리 정해 둘 수 없는 경우에 자주 쓰여요. 코드를 작성하거나 여러 파일에 걸쳐 내용을 갱신해야 하는 워크플로가 대표적이에요. 예를 들어 여러 Python 라이브러리의 설치 안내를, 그 수를 알 수 없는 여러 문서에 걸쳐 갱신해야 하는 워크플로가 이 패턴을 쓸 수 있어요.

계획(planning)을 위한 structured output 스키마부터 정의합니다:

from typing import Annotated, List
import operator


# Schema for structured output to use in planning
class Section(BaseModel):
    name: str = Field(
        description="Name for this section of the report.",
    )
    description: str = Field(
        description="Brief overview of the main topics and concepts to be covered in this section.",
    )


class Sections(BaseModel):
    sections: List[Section] = Field(
        description="Sections of the report.",
    )


# Augment the LLM with schema for structured output
planner = llm.with_structured_output(Sections)

LangGraph에서 워커 만들기 (Creating workers in LangGraph)

오케스트레이터-워커 워크플로는 흔해서, LangGraph가 이를 위한 지원을 내장하고 있어요. Send API를 쓰면 워커 노드를 동적으로 만들고 각 워커에 특정 입력을 보낼 수 있어요. 각 워커는 자신만의 상태를 갖고, 모든 워커의 출력은 오케스트레이터 그래프가 접근할 수 있는 공유 상태 키(shared state key) 에 기록됩니다. 그래서 오케스트레이터가 모든 워커 출력에 접근해 최종 결과로 합성할 수 있어요. 아래 예시는 섹션 목록을 순회하면서 Send API로 각 워커에 섹션 하나씩을 보냅니다.

from langgraph.types import Send


# Graph state
class State(TypedDict):
    topic: str  # Report topic
    sections: list[Section]  # List of report sections
    completed_sections: Annotated[
        list, operator.add
    ]  # All workers write to this key in parallel
    final_report: str  # Final report


# Worker state
class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]


# Nodes
def orchestrator(state: State):
    """Orchestrator that generates a plan for the report"""

    # Generate queries
    report_sections = planner.invoke(
        [
            SystemMessage(content="Generate a plan for the report."),
            HumanMessage(content=f"Here is the report topic: {state['topic']}"),
        ]
    )

    return {"sections": report_sections.sections}


def llm_call(state: WorkerState):
    """Worker writes a section of the report"""

    # Generate section
    section = llm.invoke(
        [
            SystemMessage(
                content="Write a report section following the provided name and description. Include no preamble for each section. Use markdown formatting."
            ),
            HumanMessage(
                content=f"Here is the section name: {state['section'].name} and description: {state['section'].description}"
            ),
        ]
    )

    # Write the updated section to completed sections
    return {"completed_sections": [section.content]}


def synthesizer(state: State):
    """Synthesize full report from sections"""

    # List of completed sections
    completed_sections = state["completed_sections"]

    # Format completed section to str to use as context for final sections
    completed_report_sections = "\n\n---\n\n".join(completed_sections)

    return {"final_report": completed_report_sections}


# Conditional edge function to create llm_call workers that each write a section of the report
def assign_workers(state: State):
    """Assign a worker to each section in the plan"""

    # Kick off section writing in parallel via Send() API
    return [Send("llm_call", {"section": s}) for s in state["sections"]]


# Build workflow
orchestrator_worker_builder = StateGraph(State)

# Add the nodes
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)

# Add edges to connect nodes
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
    "orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)

# Compile the workflow
orchestrator_worker = orchestrator_worker_builder.compile()

# Show the workflow
display(Image(orchestrator_worker.get_graph().draw_mermaid_png()))

# Invoke
state = orchestrator_worker.invoke({"topic": "Create a report on LLM scaling laws"})

from IPython.display import Markdown
Markdown(state["final_report"])

눈여겨볼 부분은 assign_workers 함수예요. 이 함수가 Send를 반환해서 워커를 병렬로 생성하고, completed_sections가 Annotated[list, operator.add]로 선언되어 여러 워커가 동시에 같은 키에 결과를 쌓을 수 있다는 점이에요.

평가자-최적화자 (Evaluator-optimizer)

평가자-최적화자 워크플로에서는 LLM 호출 하나가 응답을 만들고, 다른 호출이 그 응답을 평가해요. 평가자나 인간의 개입(human-in-the-loop)이 응답을 다듬어야 한다고 판단하면 피드백을 주고 응답을 다시 만듭니다. 이 루프는 받아들일 만한 응답이 만들어질 때까지 계속돼요. 평가자-최적화자 워크플로는 작업에 특정한 성공 기준이 있는데, 그 기준을 맞추려면 반복이 필요한 경우에 자주 쓰여요. 예를 들어 두 언어 사이의 텍스트 번역이 항상 완벽하게 맞아떨어지지는 않는데요, 두 언어에서 뜻이 같은 번역을 만들려면 몇 번의 반복이 필요할 수 있어요.

아래 예시는 농담이 재미있는지 평가자(평가기)가 판단하고, 재미없으면 피드백을 붙여 생성기로 돌려보냅니다. route_joke가 "funny"면 끝내고, "not funny"면 생성기로 다시 보내요.

# Graph state
class State(TypedDict):
    joke: str
    topic: str
    feedback: str
    funny_or_not: str


# Schema for structured output to use in evaluation
class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(
        description="Decide if the joke is funny or not.",
    )
    feedback: str = Field(
        description="If the joke is not funny, provide feedback on how to improve it.",
    )


# Augment the LLM with schema for structured output
evaluator = llm.with_structured_output(Feedback)


# Nodes
def llm_call_generator(state: State):
    """LLM generates a joke"""

    if state.get("feedback"):
        msg = llm.invoke(
            f"Write a joke about {state['topic']} but take into account the feedback: {state['feedback']}"
        )
    else:
        msg = llm.invoke(f"Write a joke about {state['topic']}")
    return {"joke": msg.content}


def llm_call_evaluator(state: State):
    """LLM evaluates the joke"""

    grade = evaluator.invoke(f"Grade the joke {state['joke']}")
    return {"funny_or_not": grade.grade, "feedback": grade.feedback}


# Conditional edge function to route back to joke generator or end based upon feedback from the evaluator
def route_joke(state: State):
    """Route back to joke generator or end based upon feedback from the evaluator"""

    if state["funny_or_not"] == "funny":
        return "Accepted"
    elif state["funny_or_not"] == "not funny":
        return "Rejected + Feedback"


# Build workflow
optimizer_builder = StateGraph(State)

# Add the nodes
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)

# Add edges to connect nodes
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
    "llm_call_evaluator",
    route_joke,
    {  # Name returned by route_joke : Name of next node to visit
        "Accepted": END,
        "Rejected + Feedback": "llm_call_generator",
    },
)

# Compile the workflow
optimizer_workflow = optimizer_builder.compile()

# Show the workflow
display(Image(optimizer_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = optimizer_workflow.invoke({"topic": "Cats"})
print(state["joke"])

에이전트 (Agents)

에이전트는 보통 도구(tools)를 사용해 동작을 수행하는 LLM으로 구현돼요. 에이전트는 연속적인 피드백 루프로 작동하며, 문제와 해결책을 예측하기 어려운 상황에 쓰입니다. 에이전트는 워크플로보다 자율성이 높아서, 어떤 도구를 쓸지, 문제를 어떻게 풀지 스스로 결정할 수 있어요. 물론 사용 가능한 도구 세트와 에이전트가 어떻게 행동해야 하는지에 대한 지침은 여전히 정의할 수 있어요.

에이전트를 시작하려면 quickstart를 보거나, LangChain에서 에이전트가 어떻게 동작하는지를 더 읽어보세요.

도구 사용하기 (Using tools)

먼저 @tool 데코레이터로 도구를 정의하고, bind_tools로 LLM에 얹어요.

from langchain.tools import tool


# Define tools
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`.

    Args:
        a: First int
        b: Second int
    """
    return a / b


# Augment the LLM with tools
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)

이제 에이전트 루프를 만듭니다. llm_call은 도구를 호출할지 말지 결정하고, should_continue가 마지막 메시지에 tool call이 있는지 확인해 tool_node로 보낼지 그대로 끝낼지 정해요. tool_node는 호출된 도구를 실제로 실행하고 결과 메시지를 상태에 추가합니다.

from langgraph.graph import MessagesState
from langchain.messages import SystemMessage, HumanMessage, ToolMessage


# Nodes
def llm_call(state: MessagesState):
    """LLM decides whether to call a tool or not"""

    return {
        "messages": [
            llm_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ]
    }


def tool_node(state: MessagesState):
    """Performs the tool call"""

    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}


# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""

    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END


# Build workflow
agent_builder = StateGraph(MessagesState)

# Add nodes
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# Add edges to connect nodes
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# Compile the agent
agent = agent_builder.compile()

# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

여기서 tool_node에서 llm_call로 다시 엣지가 이어지는 게 핵심이에요. 이 순환 엣지 덕분에 에이전트가 "도구 호출 → 결과 확인 → 다시 도구 호출" 루프를 계속 돌다가, LLM이 더 이상 도구를 호출하지 않으면 END로 빠져나옵니다.

ToolNode

ToolNode는 LangGraph 워크플로에서 도구를 실행하는 사전 제작(prebuilt) 노드예요. 도구의 병렬 실행, 오류 처리, 상태 주입(state injection)을 자동으로 처리해 줍니다. 그래프가 도구를 실행하는 방식을 세밀하게 제어해야 할 때 ToolNode를 쓰세요. 많은 LangGraph 에이전트 패턴에서 도구 실행을 담당하는 핵심 구성 요소예요.

from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import MessagesState, StateGraph

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

builder = StateGraph(MessagesState)
builder.add_node("tools", ToolNode([search, calculator]))
# ... add other nodes and edges
graph = builder.compile()

도구에서 그래프 상태와 컨텍스트 접근하기 (Access graph state and context from tools)

ToolNode가 실행하는 도구는 첫 번째 인자로 모델이 생성한 인자를 받아요. 모델이 만들지 않은 그래프 쪽 데이터(그래프 상태나 실행 컨텍스트)를 읽으려면 다음 중 하나를 쓰세요:

Python에서는 주입된 ToolRuntime 인자에서 상태와 실행 범위 컨텍스트를 읽어요.
JavaScript에서는 도구의 두 번째 인자(타입은 ToolRuntime)에서 상태와 실행 범위 컨텍스트를 읽어요.

도구는 ToolNode에 전달된 상태 값에만 접근할 수 있어요. ToolNode를 StateGraph의 노드로 직접 추가하면, 그 입력이 현재 그래프 상태가 됩니다. 다른 노드에서 ToolNode를 수동으로 invoke한다면, 도구가 커스텀 상태 필드가 필요할 때 전체 상태(full state) 를 넘겨야 해요. 예를 들어 tool_node.invoke(state) 또는 toolNode.invoke(state, config)처럼 호출하면 전체 상태가 노출되지만, {"messages": state["messages"]} 또는 { messages: state.messages }처럼 일부만 넘기면 messages만 노출돼요.

from dataclasses import dataclass

from langchain.messages import AIMessage
from langchain.tools import ToolRuntime, tool
from langgraph.graph import MessagesState, START, StateGraph
from langgraph.prebuilt import ToolNode


class State(MessagesState):
    user_id: str


@dataclass
class Context:
    organization_id: str


@tool
def get_user_info(runtime: ToolRuntime[Context, State]) -> str:
    """Look up user information."""
    # Read the current graph state passed to the ToolNode.
    user_id = runtime.state["user_id"]

    # Read explicit per-run values that are not part of graph state.
    organization_id = runtime.context.organization_id

    return f"User {user_id} in organization {organization_id}"


builder = StateGraph(State, context_schema=Context)
builder.add_node("tools", ToolNode([get_user_info]))
builder.add_edge(START, "tools")
graph = builder.compile()

result = graph.invoke(
    {
        "messages": [
            AIMessage(
                content="",
                tool_calls=[
                    {
                        "name": "get_user_info",
                        "args": {},
                        "id": "call_user_info",
                    }
                ],
            )
        ],
        "user_id": "user_123",
    },
    context=Context(organization_id="org_456"),
)

위 예시에서는 State에 그래프 상태인 user_id를, Context에는 상태에 속하지 않는 실행별 값인 organization_id를 담아요. StateGraph(State, context_schema=Context)로 컨텍스트 스키마를 지정하고, invoke 시점에 context=...로 값을 넘겨요. 그러면 도구 안에서 runtime.state["user_id"]와 runtime.context.organization_id로 각각 읽을 수 있어요.