Hooks
Hooks (훅)
Hook을 사용하면 Agent의 실행 루프에서 정의된 지점(실행의 시작과 끝, 각 LLM 호출 전, tool 실행 전후, 종료 시점)에 커스텀 로직을 실행할 수 있어요.
출처: 문서
본문
hooks를 Agent에 딕셔너리로 전달해요 — hook 포인트 를 그 지점에서 Agent가 실행할 hook 리스트에 매핑하는 형태예요. 각 hook은 라이브 State를 받고, 그것을 제자리에서 변경해 실행에 영향을 줘요. 특정 hook 포인트의 hook은 리스트 순서대로 실행되며, 같은 hook을 여러 hook 포인트에 등록할 수 있어요.
이를 통해 런타임 시스템 컨텍스트 구축, 첫 LLM 호출 전 메모리 검색, tool 호출 감사·차단, Agent가 끝내기 전에 특정 조건이 성립하도록 요구하기 같은 패턴을 만들 수 있어요.
Hook points
before_run: 실행당 한 번, 상태가 초기화되고 첫 chat-generator 호출 전에 실행돼요.before_llm처럼 매 스텝 재실행하지 않고 초기 메시지를 다시 쓰거나 state를 시드하는 데 사용하세요 — 예를 들어 사용자 쿼리를 작업 브리프로 바꾸기.before_llm: 매 chat-generator 호출 전에 실행돼요.before_tool: 모델이 tool 호출을 요청한 뒤, tool 실행 전에 실행돼요. 이 hook들이 실행된 후 Agent는state.data["messages"]에서 현재 마지막 메시지를 다시 읽어요. 그 메시지에 tool 호출이 있으면 그 호출이 실행되고, 없으면 그 스텝에서는 tool이 실행되지 않으며 tool 기반 종료 조건도 트리거되지 않고,max_agent_steps에 도달하지 않았다면 Agent는 다음 LLM 호출로 돌아가요.after_tool: tool이 실행된 뒤, 그 결과 메시지가state.data["messages"]에 들어간 다음, 종료 조건 확인과 다음 LLM 호출 전에 실행돼요. 새로 생성된 tool 결과 메시지를 다시 쓰는 데 사용하세요 — 예를 들어 결과를 오프로드하거나, 교정하거나, 자르거나, 요약하기. 일반 텍스트 종료 스텝에서는 실행되지 않아요.before_toolhook이 대기 중인 tool 호출을 제거했을 때도 여전히 실행되지만, 그 스텝에서 tool이 실행되지 않았으므로 마지막 메시지가 새 tool 결과라고 가정하지 마세요.on_exit: Agent가 종료 조건에서 멈추려 할 때 실행돼요.on_exithook은continue_run제어 플래그를 설정해(state.set("continue_run", True)) Agent가 계속 실행되게 할 수 있어요. 보통 모델에게 무엇을 할지 알려주는 메시지와 함께요.on_exithook은 Agent가 종료 조건에서 멈출 때 실행되지만,max_agent_steps에 도달해 멈출 때는 실행되지 않아요 — 실행이 어떻게 끝나든 실행되어야 하는 로직에는after_run을 사용하세요.after_run: 실행당 한 번, 스텝 루프가 끝난 뒤 Agent가 반환 값을 만들기 전에 실행돼요 — 실행이 종료 조건에서 멈췄든max_agent_steps에 도달했든 관계없이요(on_exit와 다름). 마지막 메시지 추가 같은 상태 변경은 반환되는messages/last_message와state_schema출력에 반영돼요. 여기서continue_run을 설정해도 효과가 없어요.
알 수 없는 hook 포인트에 hook을 등록하면 생성 시점에 ValueError가 발생해요. hook 클래스는 지원하는 hook 포인트를 나열하는 allowed_hook_points 속성을 선언할 수 있어요. Agent가 이를 검증하고, hook이 속하지 않는 곳에 등록되면 빠르게 실패해요.
State keys for hooks
Agent는 hook이 상호작용하는 몇 가지 state 키를 관리해요. 실행 메타데이터 키(step_count, token_usage, tool_call_counts)처럼 예약되어 있어서, 자신의 state_schema에서 그것들을 사용하면 ValueError가 발생해요. 전체 목록은 State를 보세요:
continue_run:on_exithook이 Agent를 계속 실행시키기 위해 설정해요.stop_run: 어떤 hook이든 실행을 멈추기 위해 설정하며, 각 LLM 호출 전에 읽히고exit_reason으로 사용돼요.tools: 현재 스텝에서 사용 가능한 tool — hook이 검사할 수 있게.hook_context:Agent.run(hook_context={...})/run_async(hook_context={...})로 전달되는 요청 범위 리소스. Hook은state.data["hook_context"]또는state.data.get("hook_context")로 읽어요 — 사용자 ID, WebSocket, 데이터베이스 클라이언트 같은 요청별 리소스에 사용하세요. 여기서State.get은 값을 깊은 복사해 반환하므로 피하세요. WebSocket이나 DB 클라이언트 같은 이 딕셔너리에 저장된 리소스에선 자주 실패해요.context_tokens: 현재 컨텍스트 창에 있는 토큰의 근사 개수. 각 LLM 호출 후 그 답변의 프롬프트+완료 토큰으로 갱신돼요(state.get("context_tokens")로 읽음). 실행 동안 누적되는token_usage와 달리 매 호출마다 교체돼요. 사용량을 보고하는 첫 답변까지는0이며, 최신 호출 이후 추가된 메시지는 세지 않아요.before_llmhook이 이 값을 읽어 임계값을 넘으면 컨텍스트 압축을 트리거할 수 있어요.
Hook은 자동 추적되는 실행 메타데이터(step_count, token_usage, tool_call_counts)도 읽을 수 있어요.
Creating hooks
With the @hook decorator
@hook 데코레이터는 단일 State 인자를 받는 함수를 hook으로 감싸요. 일반 함수는 hook의 동기 경로, 코루틴 함수는 비동기 경로가 돼요. 단일 hook에 두 경로를 모두 주려면 function과 async_function을 모두 가진 FunctionHook을 직접 구성하세요.
아래 예시는 before_llm, before_tool, on_exit 각각에 hook을 등록해 hook이 할 수 있는 일을 보여줘요:
from datetime import datetime, timezone
from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.agents.state import State, replace_values
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.hooks import hook
from haystack.tools import tool
@tool
def search(query: Annotated[str, "The search query"]) -> str:
"""Search the web."""
# Placeholder: would call a real search API
return "Fusion startups reported net-energy-gain milestones this year."
@hook
def build_context(state: State) -> None:
# before_llm: build run-time system context once, before the first model call.
if state.get("step_count") == 0:
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
system = ChatMessage.from_system(
f"You are a research assistant. The current time is {now}.",
)
state.set(
"messages",
[system, *state.data["messages"]],
handler_override=replace_values,
)
@hook
def audit_tool_calls(state: State) -> None:
# before_tool: see which tools the model is about to run.
pending = state.data["messages"][-1].tool_calls
print(f"about to run: {[tc.tool_name for tc in pending]}")
@hook
def require_search(state: State) -> None:
# on_exit: keep going until the agent has actually searched.
if state.get("tool_call_counts", {}).get("search", 0) == 0:
state.set("messages", [ChatMessage.from_system("Search before answering.")])
state.set("continue_run", True)
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[search],
hooks={
"before_llm": [build_context],
"before_tool": [audit_tool_calls],
"on_exit": [require_search],
},
)
result = agent.run(
messages=[
ChatMessage.from_user("What are the latest developments in fusion energy?"),
],
)
print(result["last_message"].text)
Class-based hooks
hook은 run(state) 메서드를 가진 어떤 객체든 가능하며, 진짜 비동기 동작을 위해 run_async(state)를 추가로 정의할 수도 있어요. 클래스 기반 hook은 선택적 라이프사이클 메서드인 warm_up/warm_up_async와 close/close_async를 구현할 수도 있어요. Agent는 자신의 warm_up/close에서 그것들을 호출하므로, hook은 클라이언트를 열거나 자격 증명을 읽는 것을 warm-up까지 미루고 close에서 해제할 수 있어요. warm-up이 모든 Agent 실행 전에 실행되므로 hook이 비싼 초기화를 반복해서는 안 돼요:if self._client is not None: return처럼 작업이 이미 끝났으면 일찍 반환하세요.
클래스 기반 hook이 직렬화 가능해야 하면(그것을 사용하는 Agent를 직렬화할 수 있도록) to_dict/from_dict를 구현하세요: 직렬화 가능한 생성자 인자를 hook에 저장하고 그 값들로 런타임 클라이언트를 다시 만드세요.
아래 예시는 자신의 LLM으로 Agent의 답을 채점하고, 끝내기 전에 약한 답을 개선하라고 요청하는 on_exit hook이에요:
from typing import Any
from haystack.components.agents import Agent
from haystack.components.agents.state import State
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.core.serialization import default_from_dict, default_to_dict
from haystack.dataclasses import ChatMessage
class GradeFinalAnswer:
"""Grade the Agent's answer with an LLM and ask it to improve a weak answer before finishing."""
def __init__(self, model: str = "gpt-5.4-nano"):
self.model = model
self._judge = OpenAIChatGenerator(model=self.model)
def warm_up(self) -> None:
# The Agent calls this before every run, but OpenAIChatGenerator.warm_up
# creates its client only on the first call, so repeating it is safe and cheap.
self._judge.warm_up()
def close(self) -> None:
# Release the judge's client during the Agent's close.
self._judge.close()
def run(self, state: State) -> None:
answer = state.data["messages"][-1].text or ""
verdict = (
self._judge.run(
messages=[
ChatMessage.from_user(
f"Reply with only PASS or FAIL. Is this answer complete?\n\n{answer}",
),
],
)["replies"][0].text
or ""
)
if "FAIL" in verdict.upper():
state.set(
"messages",
[
ChatMessage.from_user(
"Your answer was incomplete. Please improve it.",
),
],
)
state.set("continue_run", True)
def to_dict(self) -> dict[str, Any]:
return default_to_dict(self, model=self.model)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "GradeFinalAnswer":
return default_from_dict(cls, data)
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
hooks={"on_exit": [GradeFinalAnswer()]},
)
result = agent.run(messages=[ChatMessage.... ])
Ready-made hooks
Haystack는 haystack.hooks의 각 하위 모듈에 여러 준비된 hook을 제공해요:
CompactionHook(haystack.hooks.compaction에서): 대화가 모델 컨텍스트 창의 구성된 비율에 도달하면 Agent의 대화를 줄이는before_llmhook.Compactor가 대화를 어떻게 줄일지 결정해요. Context Compaction 참조.ConfirmationHook(haystack.hooks.human_in_the_loop에서): 대기 중인 tool 호출에 Human-in-the-Loop 확인 전략을 적용하는before_toolhook — 실행 전에 사람이 모델이 요청한 tool 호출을 확인, 수정, 거부할 수 있어요. Human in the Loop 참조.ToolResultOffloadHook(haystack.hooks.tool_result_offloading에서): tool 결과를ToolResultStore(예:FileSystemToolResultStore)로 오프로드하고 대화에서는 컴팩트 포인터로 교체해, 다음 LLM 호출이 전체 결과 대신 참조를 보게 하는after_toolhook. Per-tool 정책(AlwaysOffload,NeverOffload,OffloadOverChars)이 어떤 결과를 오프로드할지 제어해요. Tool Result Offloading 참조.TokenBudgetHook(haystack.hooks.budget에서): 누적 토큰 사용량이 구성된 예산에 도달하면 실행을 종료하고,"token_budget_exceeded"를exit_reason으로 보고하는before_llmhook. Token Budget 참조.