State
State
State 는 Agent와 Tool 실행 중 공유 정보를 저장하는 컨테이너예요. 여러 도구 사이에서 데이터를 구조적으로 공유하고, 여러 도구 호출에 걸쳐 결과를 누적하며, 에이전트의 최종 답변과 함께 그 결과를 꺼내 볼 수 있게 해 줘요.
출처: 문서
본문
개요 (Overview)
여러 도구를 쓰는 에이전트를 만들 때, 도구들이 정보를 공유하거나 반복 실행에 걸쳐 결과를 누적해야 하는 경우가 많아요. State 는 모든 도구가 읽고 쓸 수 있는 중앙 저장소를 제공해요.
예를 들어 여러 번 호출되는 검색 도구가 공유 documents 리스트에 결과를 계속 추가하고, 그 리스트를 에이전트의 최종 답변과 함께 반환해 출처 검토(inspection)에 쓸 수 있어요.
State 는 스키마 기반 접근을 사용해요. 다음을 정의하죠:
- 어떤 데이터를 저장할 수 있는지
- 각 데이터의 타입
- 업데이트 시 값을 어떻게 병합(merge)할지
Agent 가 내부적으로 State 객체를 만들고 관리해요. 직접 인스턴스화할 필요는 없어요. 도구 정의(inputs_from_state, outputs_to_state, 또는 state: State 파라미터)를 통해 상호작용하고, 결과는 에이전트의 출력 딕셔너리에서 읽으면 돼요.
지원 타입 (Supported Types)
State 는 표준 Python 타입을 지원해요:
- 기본 타입:
str,int,float,bool,dict - 리스트 타입:
list,list[str],list[int],list[Document] - 유니온 타입:
str | int,str | None - 커스텀 클래스와 데이터 클래스
자동 메시지 처리 (Automatic Message Handling)
State 는 실행 중 전체 대화 이력을 저장하는 messages 필드를 자동으로 포함해요. 스키마에 직접 정의할 필요 없어요. merge_lists 핸들러와 함께 list[ChatMessage] 타입을 사용해서, 매 반복마다 새 메시지가 추가돼요.
State API
| 메서드 | 설명 |
|---|---|
state.get(key, default=None) |
값을 읽어요. 키가 없으면 default 를 반환해요. |
state.set(key, value) |
값을 작성해요. 스키마의 핸들러로 병합돼요. |
state.has(key) |
키가 state에 있으면 True 를 반환해요. |
state.data |
현재 모든 state의 스냅샷을 dict 로 반환해요. |
스키마 정의 (Schema Definition)
스키마는 무엇을 저장할 수 있고 값이 어떻게 업데이트되는지를 정의해요. 각 스키마 항목은 다음으로 구성돼요:
type(필수): 이 필드의 Python 타입 (예:str,int,list)handler(선택):set()호출 시 새 값을 어떻게 병합할지 결정하는 호출 가능 객체
{
"parameter_name": {
"type": SomeType, # Required: expected Python type
"handler": some_func, # Optional: merge function
},
}
핸들러를 지정하지 않으면 State 가 타입에 따라 기본값을 자동으로 할당해요.
예약된 키:
Agent는 일부 state 키를 스스로 관리하는데, 사용자가 제공한state_schema에서 해당 키가 나오면ValueError를 던져요.
- 실행 메타데이터 키:
step_count,token_usage,tool_call_counts,exit_reason— 실행 중Agent가 자동으로 채워요. 도구와 훅이 라이브State에서 읽을 수 있고, 결과 딕셔너리에도 반환돼요.exit_reason은 에이전트가 왜 멈췄는지 알려줘요("text","length","content_filter", 도구 종료 조건을 촉발한 도구 이름,"max_agent_steps", 또는 훅이stop_run으로 설정한 커스텀 이유).- 훅 전용 키:
continue_run(on_exit훅이 Agent를 계속 실행하도록 설정),stop_run(어떤 훅이든 실행을 멈추도록 설정, LLM 호출 전에 읽혀exit_reason으로 쓰임),tools(현재 스텝에서 쓸 수 있는 도구들),hook_context(Agent.run(hook_context={...})로 전달된 요청 범위 리소스),context_tokens(현재 컨텍스트 창의 토큰 근사 개수, LLM 호출 후 갱신됨 — 컨텍스트 컴팩션 촉발용). 실행 메타데이터 키와 달리 이 키들은 결과 딕셔너리에 반환되지 않아요.state 키가 충돌하면 이름을 바꾸세요(예:
my_token_usage).
기본 핸들러 (Default Handlers)
State 는 두 가지 내장 병합 동작을 제공해요(haystack.components.agents.state 에서 가져올 수 있어요):
merge_lists: 기존 리스트에 추가해요(리스트 타입의 기본값)replace_values: 기존 값을 덮어써요(비-리스트 타입의 기본값)
from haystack.components.agents import State
schema = {
"documents": {"type": list}, # uses merge_lists by default
"user_name": {"type": str}, # uses replace_values by default
}
state = State(schema=schema)
state.set("documents", [1, 2])
state.set("documents", [3, 4])
print(state.get("documents")) # [1, 2, 3, 4]
state.set("user_name", "Alice")
state.set("user_name", "Bob")
print(state.get("user_name")) # "Bob"
커스텀 핸들러 (Custom Handlers)
커스텀 핸들러는 기본 merge_lists 나 replace_values 동작이 요구에 맞지 않을 때 유용해요. 핸들러는 현재 state 값과 새 값을 받아 병합된 결과를 반환해요. 아래 예시는 전용 중복 제거(deduplication) 핸들러로, 여러 도구 호출이 겹치는 결과를 반환할 때 state에 중복이 쌓이는 것을 방지해요:
def deduplicate(current_value: list | None, new_value: list) -> list:
"""Append new items, skipping any already in the list."""
existing = set(current_value or [])
return (current_value or []) + [item for item in new_value if item not in existing]
schema = {"doc_ids": {"type": list, "handler": deduplicate}}
state = State(schema=schema)
state.set("doc_ids", ["doc-1", "doc-2"])
state.set("doc_ids", ["doc-2", "doc-3"])
print(state.get("doc_ids")) # ["doc-1", "doc-2", "doc-3"]
단일 set() 호출에 대해서만 핸들러를 오버라이드할 수도 있어요:
from haystack.components.agents import State
def concatenate_strings(current: str | None, new: str) -> str:
return f"{current}-{new}" if current else new
state = State(schema={"user_name": {"type": str}})
state.set("user_name", "Alice")
state.set("user_name", "Bob", handler_override=concatenate_strings)
print(state.get("user_name")) # "Alice-Bob"
State 사용하기 (Using State)
Agent 를 만들 때 state_schema 를 정의하세요. state_schema 에 선언된 state 키는 messages 와 last_message 와 함께 에이전트 결과 딕셔너리의 출력 키로 노출돼요.
도구는 세 가지 메커니즘으로 State와 상호작용해요:
outputs_to_state: 도구 실행 후 도구 결과를 state 키에 작성해요.inputs_from_state: 도구 실행 전에 state 값을 도구 파라미터에 주입해요.- 직접
State주입: 도구 함수 시그니처에state: State파라미터를 추가해요.Agent가State어노테이션을 감지해 라이브State객체를 자동으로 주입하므로, 스키마에 정의된 어떤 키든 읽고 쓸 수 있어요.State객체는 LLM의 파라미터 스키마에 노출되지 않아요.
State에서 읽기: inputs_from_state
inputs_from_state 는 {"state_key": "param_name"} 형식으로 state 키를 함수 파라미터 이름에 매핑해요. 값은 도구 실행 전에 state에서 주입되므로 LLM이 제공할 필요가 없어요. inputs_from_state 로 매핑된 파라미터는 LLM의 파라미터 스키마에서 자동으로 제외돼요. 모델은 이를 보고하거나 제공하지 않아요:
from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
@tool(inputs_from_state={"user_name": "user_context"})
def search_documents(
query: Annotated[str, "The search query"],
user_context: str, # injected from state; excluded from LLM schema
) -> dict:
"""Search documents using query and user context."""
return {"results": [f"Found results for '{query}' (user: {user_context})"]}
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[search_documents],
system_prompt="Use the search_documents tool to find information.",
streaming_callback=print_streaming_chunk,
state_schema={"user_name": {"type": str}},
)
result = agent.run(
messages=[ChatMessage.from_user("Search for Python tutorials")],
user_name="Alice", # state key "user_name" is pre-populated by passing user_name= to agent.run()
)
print(result["last_message"].text)
State에 쓰기: outputs_to_state
outputs_to_state 파라미터는 도구 출력 키를 state 키에 매핑해요. 각 항목은 두 개의 선택 필드를 지원해요:
{
"state_key": {
"source": "tool_output_key", # which key to read from the tool's return dict; omit to store the entire dict
"handler": some_func, # override the schema's merge handler for this mapping only
},
}
from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
@tool(
outputs_to_state={
"documents": {"source": "documents"},
"result_count": {"source": "count"},
"last_query": {"source": "query"},
},
)
def retrieve_documents(
query: Annotated[str, "The search query"],
) -> dict:
"""Retrieve relevant documents."""
return {
"documents": [
{"title": "Doc 1", "content": "Content about Python"},
{"title": "Doc 2", "content": "More about Python"},
],
"count": 2,
"query": query,
}
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[retrieve_documents],
system_prompt="Use the retrieve_documents tool to find information.",
streaming_callback=print_streaming_chunk,
state_schema={
"documents": {"type": list},
"result_count": {"type": int},
"last_query": {"type": str},
},
)
result = agent.run(messages=[ChatMessage.from_user("Find information about Python")])
print(f"Documents: {result['documents']}")
print(f"Result count: {result['result_count']}")
print(f"Last query: {result['last_query']}")
source 를 생략하면 전체 도구 결과 딕셔너리가 state 키에 저장돼요:
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
@tool(outputs_to_state={"user_info": {}})
def get_user_info() -> dict:
"""Get user information."""
return {"name": "Alice", "email": "[email protected]", "role": "admin"}
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[get_user_info],
system_prompt="Use the get_user_info tool to look up user details.",
streaming_callback=print_streaming_chunk,
state_schema={"user_info": {"type": dict}},
)
result = agent.run(messages=[ChatMessage.from_user("What are the user's details?")])
print(result["last_message"].text)
print(f"User info: {result['user_info']}")
입력과 출력 결합 (Combining Inputs and Outputs)
도구는 State에서 읽고 쓸 수 있어서, 반복 실행에 걸친 도구 체이닝을 지원해요. 이 예시는 앞 섹션의 retrieve_documents 위에 구축해요:
@tool(
inputs_from_state={"documents": "documents"},
outputs_to_state={
"final_docs": {"source": "processed_docs"},
"final_count": {"source": "processed_count"},
},
)
def process_documents(
max_results: Annotated[int, "Maximum number of documents to return"],
documents: list = None, # injected from state; LLM does not provide this
) -> dict:
"""Process retrieved documents and return a filtered subset."""
processed = (documents or [])[:max_results]
return {"processed_docs": processed, "processed_count": len(processed)}
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[retrieve_documents, process_documents], # chained through state
system_prompt="Use the available tools to retrieve and process documents.",
streaming_callback=print_streaming_chunk,
state_schema={
"documents": {"type": list},
"result_count": {"type": int},
"last_query": {"type": str},
"final_docs": {"type": list},
"final_count": {"type": int},
},
)
result = agent.run(
messages=[ChatMessage.from_user("Find and process 3 documents about Python")],
)
print(f"Processed {result['final_count']} documents")
도구에 State 직접 주입하기 (Injecting State Directly into Tools)
inputs_from_state 와 outputs_to_state 의 대안으로, 도구는 state: State 파라미터를 선언해 호출 시점에 라이브 State 객체를 받을 수 있어요. 이렇게 하면 매핑을 미리 선언하지 않고도 여러 state 키를 읽고 쓸 수 있어요. Agent 가 State 어노테이션을 감지해 객체를 자동으로 주입하고, LLM에 노출되는 스키마에서는 제외돼요. State 와 State | None 어노테이션을 모두 지원해요.
함수 기반 도구는 state 파라미터를 추가하고 @tool 데코레이터를 사용해요:
from typing import Annotated
from haystack.components.agents import Agent, State
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage, Document
from haystack.tools import tool
@tool
def retrieve_and_store(
query: Annotated[str, "The search query"],
state: State,
) -> str:
"""Retrieve documents and store them directly in state."""
documents = [Document(content=f"Result for '{query}'")]
state.set("documents", documents)
user_name = state.get("user_name", "unknown")
return f"Retrieved {len(documents)} document(s) for {user_name}"
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[retrieve_and_store],
system_prompt="Use the retrieve_and_store tool to find documents.",
streaming_callback=print_streaming_chunk,
state_schema={"documents": {"type": list[Document]}, "user_name": {"type": str}},
)
result = agent.run(
messages=[ChatMessage.from_user("Find documents about Python")],
user_name="Alice",
)
print(result["last_message"].text)
print(result["documents"])
컴포넌트 기반 도구는 run 메서드에 State 입력 소켓을 선언하고 ComponentTool 로 감싸요:
from haystack import component
from haystack.components.agents import Agent, State
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.dataclasses import ChatMessage, Document
from haystack.tools import ComponentTool
@component
class DocumentRetriever:
"""Retrieve documents and store them in state."""
@component.output_types(reply=str)
def run(self, query: str, state: State) -> dict[str, str]:
"""
Retrieve documents based on query and store them in state.
:param query: The search query
"""
documents = [Document(content=f"Result for '{query}'")]
state.set("documents", documents)
return {"reply": f"Retrieved {len(documents)} document(s)"}
retriever_tool = ComponentTool(
component=DocumentRetriever(),
name="retrieve",
description="Retrieve documents based on a search query",
)
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[retriever_tool],
system_prompt="Use the retrieve tool to find documents.",
streaming_callback=print_streaming_chunk,
state_schema={"documents": {"type": list[Document]}},
)
result = agent.run(messages=[ChatMessage.from_user("Find documents about Python")])
print(result["last_message"].text)
print(result["documents"])