개요
AgentState는 LangGraph V1.0에서 새롭게 추가된 핵심 클래스로, 에이전트 그래프의 상태 관리를 위한 표준화된 스키마를 제공합니다. create_agent() 함수에 기본으로 설정되어 있어 대부분의 에이전트 개발 시 추가적인 설정 없이 바로 사용할 수 있습니다.
구현체:
from langchain.agents import AgentState from typing import TypedDict, Required, NotRequired, Annotated, Generic, TypeVar from langgraph.graph.message import add_messages
ResponseT = TypeVar("ResponseT")
class AgentState(TypedDict, Generic[ResponseT]):
"""State schema for the agent."""
messages: Required[Annotated[list[AnyMessage], add_messages]]
jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]주요 필드 상세 설명
1. messages
에이전트와 사용자 간의 모든 메시지 내역을 관리합니다.
특징:
- Required 타입: 모든 에이전트 상태에 필수적으로 포함되어야 하는 필드
- AnyMessage:
BaseMessage,HumanMessage,AIMessage,ToolMessage등 모든 메시지 타입 지원 - add_messages 리듀서: 새로운 메시지를 기존 메시지 리스트에 자동으로 누적시킵니다
- 중복된 메시지 필터링
- 메시지 순서 유지
- 효율적인 상태 업데이트
사용 예시:
from langchain_core.messages import HumanMessage, AIMessage
state = {
"messages": [
HumanMessage(content="안녕하세요"),
AIMessage(content="안녕하세요! 무엇을 도와드릴까요?"),
]
}2. jump_to
에이전트의 제어 흐름을 동적으로 변경합니다.
특징:
- NotRequired 타입: 선택적 필드로 필요할 때만 사용
- EphemeralValue: 임시 값으로, 그래프 실행 후 자동으로 제거됨
- PrivateStateAttr: 개인 속성으로, 외부 입출력에서 제외됨
- 특정 노드로 직접 이동(점프)하거나 현재 노드를 건너뛸 수 있음
사용 사례:
# 특정 노드로 즉시 이동
def agent_node(state: AgentState):
if should_exit:
return {
"messages": [...],
"jump_to": "end_node"
}
return {"messages": [...]}3. structured_response
에이전트의 최종 응답을 구조화된 형식으로 저장합니다.
특징:
- Generic 타입
ResponseT: 응답의 타입을 유연하게 정의 가능 - OmitFromInput: 그래프 입력에서 제외되는 필드로, 내부 처리용만 해당
- 반환값이나 중간 결과를 저장할 때 유용
사용 예시:
from pydantic import BaseModel
class SearchResult(BaseModel):
title: str
url: str
snippet: str
# AgentState를 SearchResult로 특정
class SearchAgentState(AgentState[SearchResult]):
pass
def search_agent(state: SearchAgentState) -> SearchAgentState:
# 검색 수행
result = SearchResult(
title="예제",
url="https://example.com",
snippet="예제 설명"
)
return {
"messages": [...],
"structured_response": result
}create_agent()와의 관계
create_agent() 함수는 AgentState를 기본 상태 스키마로 사용합니다.
from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")
tools = [...] # 도구 정의
# AgentState가 자동으로 적용됨
agent = create_agent(
model=model,
tools=tools,
# state_schema=AgentState # 기본값이므로 명시하지 않아도 됨
)
# 실행
result = agent.invoke({"messages": [{"role": "user", "content": "..."}]})커스텀 상태 확장
AgentState를 상속하여 추가 필드를 정의할 수 있습니다.
from langchain.agents import AgentState
from typing import Annotated
import operator
class CustomAgentState(AgentState[dict]):
"""커스텀 에이전트 상태"""
# 기존 필드들 (messages, jump_to, structured_response) 상속
# 새로운 필드 추가
search_results: Annotated[list[dict], operator.add]
iteration_count: int
context: str메시지 관리의 우수한 점
add_messages의 이점
- 자동 누적: 새 메시지를 수동으로 추가할 필요 없음
- 중복 제거: 같은 메시지가 중복으로 추가되지 않음
- 상태 동기화: 메시지 순서가 자동으로 유지됨
- 메모리 효율: 불필요한 메시지 복제 방지
# add_messages 동작 예시
from langgraph.graph.message import add_messages
messages1 = [HumanMessage(content="첫 번째")]
messages2 = [AIMessage(content="응답")]
# add_messages는 내부적으로 이렇게 작동
result = add_messages(messages1, messages2)
# 결과: [HumanMessage(...), AIMessage(...)]실제 사용 예시: 다중 에이전트 시스템
from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 상태 정의
class MultiAgentState(AgentState[dict]):
"""다중 에이전트 시스템 상태"""
agent_name: str
research_results: list[str]
# 에이전트 생성
model = ChatOpenAI(model="gpt-4o")
research_agent = create_agent(model, tools=[...])
analysis_agent = create_agent(model, tools=[...])
# 그래프 구성
def research_node(state: MultiAgentState):
result = research_agent.invoke(state)
return {
"messages": result["messages"],
"agent_name": "researcher",
"research_results": [result.get("structured_response", "")]
}
def analysis_node(state: MultiAgentState):
result = analysis_agent.invoke(state)
return {
"messages": result["messages"],
"agent_name": "analyzer"
}
# StateGraph 구성
graph = StateGraph(MultiAgentState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_edge(START, "research")
graph.add_edge("research", "analysis")
graph.add_edge("analysis", END)
compiled_graph = graph.compile()TypedDict와 Generic의 역할
TypedDict
- 상태 필드와 그 타입을 명시적으로 정의
- IDE의 자동완성과 타입 검사 지원
- 런타임에서도 검증 가능
Generic[ResponseT]
- 응답 타입을 유연하게 지정 가능
- 다양한 구조화된 응답 타입 지원
- 타입 안전성 제공
마이그레이션 가이드 (기존 State → AgentState)
이전 방식:
class CustomState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# 추가 필드들...V1.0 권장 방식:
from langchain.agents import AgentState
class CustomState(AgentState):
# 추가 필드들...주의사항
- messages 필드는 필수:
Required타입이므로 항상 포함해야 함 - jump_to와 structured_response는 옵션: 필요할 때만 사용
- 커스텀 필드 리듀서: 추가 필드에는 명시적으로 리듀서(operator.add 등)를 지정해야 함
- EphemeralValue의 자동 제거: jump_to는 한 번 사용 후 자동으로 제거됨
참고자료
- LangGraph 공식 문서: https://langchain-ai.github.io/langgraph/
- LangChain 깃허브: https://github.com/langchain-ai/langchain
- 관련 예제: https://github.com/LangChain-OpenTutorial/LangChain-OpenTutorial