[langgraph] 그래프에 런타임 구성(configuration)을 하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/configuration/

에이전트를 호출하는 시점에 설정할 필요가 있다. 예를 들어, 어떤 LLM을 사용할지 구성하는 경우가 해당된다. 아래에서는 이를 수행하는 예제를 설명한다.

준비

우선, 필요한 패키지를 설치하자.

pip install langgraph langchain_openai

그래프 정의

우선, 간단한 그래프를 만들자.

import operator
from typing import Annotated, Sequence

from langchain_core.messages import BaseMessage
from langchain_anthropic import ChatAnthropic
from langgraph.graph import END, StateGraph, START
from typing_extensions import TypedDict

model = ChatAnthropic(model_name="claude-2.1")


class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]


def _call_model(state):
    state["messages"]
    response = model.invoke(state["messages"])
    return {"messages": [response]}


builder = StateGraph(AgentState)
builder.add_node("model", _call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

그래프 구성

이제 사용자가 여러 LLM 중에서 선택할 수 있도록 이 예제를 확장한다고 가정해 보자. 쉽게 이를 구성할 수 있다. 아래와 같이 구성(config)을 전달함으로써 가능하다. 이 구성은 입력에 포함되지 않는 정보(따라서 상태의 일부로 추적하지 않으려는 정보)를 포함하도록 설계되었다.

from langchain_openai import ChatOpenAI
from typing import Optional
from langchain_core.runnables.config import RunnableConfig

openai_model = ChatOpenAI()

models = {
    "anthropic": model,
    "openai": openai_model,
}


def _call_model(state: AgentState, config: RunnableConfig):
    model_name = config["configurable"].get("model", "anthropic")
    model = models[model_name]
    response = model.invoke(state["messages"])
    return {"messages": [response]}


builder = StateGraph(AgentState)
builder.add_node("model", _call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

구성 없이 호출하면, 정의한 기본값(Anthropic)을 사용하게 된다.

result = graph.invoke({"messages": [HumanMessage(content="hi")]})
print(result)

{'messages': [HumanMessage(content='hi', additional_kwargs={}, response_metadata={}),
  AIMessage(content='Hello!', additional_kwargs={}, response_metadata={'id': 'msg_01WFXkfgK8AvSckLvYYrHshi', 'model': 'claude-2.1', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 10, 'output_tokens': 6}}, id='run-ece54b16-f8fc-4201-8405-b97122edf8d8-0', usage_metadata={'input_tokens': 10, 'output_tokens': 6, 'total_tokens': 16})]}

구성을 전달하면 다른 모델을 사용하도록 호출할 수 있다.

config = {"configurable": {"model": "openai"}}
result = graph.invoke({"messages": [HumanMessage(content="hi")]}, config=config)
print(result)

{'messages': [HumanMessage(content='hi', additional_kwargs={}, response_metadata={}),
  AIMessage(content='Hello! How can I assist you today?', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 9, 'prompt_tokens': 8, 'total_tokens': 17, 'completion_tokens_details': {'reasoning_tokens': 0}}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-f8331964-d811-4b44-afb8-56c30ade7c15-0', usage_metadata={'input_tokens': 8, 'output_tokens': 9, 'total_tokens': 17})]}

그래프를 더 많은 구성을 받을 수 있도록 조정할 수도 있다. 예를 들어, 시스템 메시지를 추가하는 것이다.

from langchain_core.messages import SystemMessage


class ConfigSchema(TypedDict):
    model: Optional[str]
    system_message: Optional[str]


def _call_model(state: AgentState, config: RunnableConfig):
    model_name = config["configurable"].get("model", "anthropic")
    model = models[model_name]
    messages = state["messages"]
    if "system_message" in config["configurable"]:
        messages = [
            SystemMessage(content=config["configurable"]["system_message"])
        ] + messages
    response = model.invoke(messages)
    return {"messages": [response]}


workflow = StateGraph(AgentState, ConfigSchema)
workflow.add_node("model", _call_model)
workflow.add_edge(START, "model")
workflow.add_edge("model", END)

graph = workflow.compile()

result = graph.invoke({"messages": [HumanMessage(content="hi")]})
print(result)

{'messages': [HumanMessage(content='hi', additional_kwargs={}, response_metadata={}),
  AIMessage(content='Hello!', additional_kwargs={}, response_metadata={'id': 'msg_01VgCANVHr14PsHJSXyKkLVh', 'model': 'claude-2.1', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 10, 'output_tokens': 6}}, id='run-f8c5f18c-be58-4e44-9a4e-d43692d7eed1-0', usage_metadata={'input_tokens': 10, 'output_tokens': 6, 'total_tokens': 16})]}

config = {"configurable": {"system_message": "respond in italian"}}
result = graph.invoke({"messages": [HumanMessage(content="hi")]}, config=config)
print(result)

{'messages': [HumanMessage(content='hi', additional_kwargs={}, response_metadata={}),
  AIMessage(content='Ciao!', additional_kwargs={}, response_metadata={'id': 'msg_011YuCYQk1Rzc8PEhVCpQGr6', 'model': 'claude-2.1', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 14, 'output_tokens': 7}}, id='run-a583341e-5868-4e8c-a536-881338f21252-0', usage_metadata={'input_tokens': 14, 'output_tokens': 7, 'total_tokens': 21})]}

LangGraph 참고 자료

• Controllability
  • 병렬 실행 브랜치 생성하는 방법
  • map-reduce 브랜치를 병렬 실행하는 방법
  • 그래프 재귀 한도를 제한하는 방법
• Persistence
  • 그래프에 스레드 수준 영속성 추가하는 방법
  • 서브그래프에 스레드 수준 영속성을 추가하는 방법
  • 그래프에 여러 스레드 간 영속성을 추가하는 방법
  • Postgres 영속성 체크포인터 사용하는 방법
• Memory
  • 대화 이력을 관리하는 방법
  • 메시지 삭제하는 방법
  • 대화 이력을 요약하는 방법
• Human-in-the-loop
  • 중단 지점 추가하는 방법
  • 동적 중단 지점을 추가하는 방법
  • 그래프 상태를 수정하는 방법
  • 사용자 입력을 대기하는 방법
  • 그래프의 과거 상태를 보고 수정하는 방법
  • 도구 호출을 리뷰하는 방법
• Streaming
  • 그래프의 전체 상태를 스트리밍하는 방법
  • 그래프의 상태 업데이트를 스트리밍하는 방법
  • 그래프에서 LLM 토큰을 스트리밍하는 방법
  • LLM 토큰을 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 커스텀 데이터를 스트리밍하는 방법
  • 동시에 여러 스트리밍 모드를 구성하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 최종 노드에서 스트리밍하는 방법
  • 서브그래프에서 스트리밍하는 방법
  • 지원하지 않는 모델에서 스트리밍을 비활성화하는 방법
• Tool calling
  • ToolNode를 사용하여 도구를 호출하는 방법
  • 도구 호출 시 에러 처리하는 방법
  • 도구에 런타임 값을 전달하는 방법
  • 도구에 구성 전달하는 방법
  • 많은 수의 도구를 처리하는 방법
• Subgraphs
  • 서브 그래프를 추가하고 사용하는 방법
  • 서브 그래프에서 상태를 조회하고 업데이트하는 방법
  • 서브 그래프의 입/출력값을 변환하는 방법
• State Management
  • 상태에 pydantic 모델을 사용하는 방법
  • 그래프의 입출력 스키마를 정의하는 방법
  • 노드 간 비공개 상태 전달하는 방법
• Other
  • 그래프를 비동기로 실행하는 방법
  • 그래프를 시각화하는 방법
  • 그래프에 런타임 구성을 하는 방법
  • 노드 재시도 정책 추가하는 방법
  • ReAct 스타일 에이전트로 구조화된 출력을 리턴하는 방법
  • LangSmith에서 그래프 실행 시 실행 ID, 태그와 메타데이터를 전달하는 방법
  • 재귀 횟수에 도달하기 전에 상태를 리턴하는 방법
  • AutoGen, CrewAI 외 다른 프레임워크와 LangGraph 통합하는 방법
• Prebuilt ReAct Agent
  • ReAct 에이전트를 사용하는 방법
  • ReAct 에이전트에 메모리 추가하는 방법
  • ReAct 에이전트에 사용자 정의 시스템 프롬프트 추가하는 방법
  • ReAct 에이전트에 사람이 개입하는 프로세스를 추가하는 방법
  • 처음부터 ReAct 에이전트를 만드는 방법