Agent
Agent
Agent 컴포넌트는 도구를 사용하는 에이전트예요. 채팅 기반 LLM과 도구를 번갈아 사용하면서 복잡한 질의를 반복적으로 해결하죠. 외부 도구를 실행하고, 여러 번의 LLM 호출에 걸쳐 상태를 유지하며, 설정 가능한 exit_conditions에 따라 실행을 멈출 수 있어요.
파이프라인에서 가장 흔한 위치: ChatPromptBuilder 뒤나 사용자 입력 직후
필수 init 변수: chat_generator — 도구를 지원하는 Chat Generator 인스턴스
필수 run 변수: messages — ChatMessage 목록
출력 변수: messages(도구·모델 응답이 담긴 채팅 기록), last_message(실행의 마지막 ChatMessage), step_count·token_usage·tool_call_counts·exit_reason(실행 메타데이터), 그리고 state_schema에 정의한 키마다 출력 하나씩
API reference: Agents
GitHub link: https://github.com/deepset-ai/haystack/blob/main/haystack/components/agents/agent.py
Package name: haystack-ai
출처: 문서
본문
Overview
Agent 컴포넌트는 루프 기반 시스템이에요. 채팅 기반 대형 언어 모델(LLM)과 외부 도구를 조합해 복잡한 사용자 질의를 해결하죠. 도구를 호출하고, 상태를 갱신하고, 프롬프트를 생성하는 과정을 설정된 exit_conditions 중 하나가 충족될 때까지 반복합니다.
할 수 있는 일은 이렇게 정리돼요:
- 사용자 입력에 따라 도구를 동적으로 선택할 수 있고,
- 스키마를 이용해 실행 시간 상태를 유지하고 검증할 수 있으며,
- LLM의 토큰 단위 출력을 스트리밍할 수 있어요.
Agent는 다음을 담은 딕셔너리를 반환합니다:
messages: 전체 대화 기록last_message: 에이전트가 낸 마지막ChatMessagestep_count: 에이전트가 실행한 스텝 수token_usage: 실행 중 모든 LLM 호출에 걸쳐 합산한 토큰 사용량tool_call_counts: 도구 이름별 호출 횟수exit_reason: 에이전트가 왜 멈췄는지 — 출력을 다운스트림으로 라우팅할 때 유용해요state_schema에 따른 추가 동적 키들
Run Metadata
step_count, token_usage, tool_call_counts, exit_reason 출력은 실행 중 자동으로 채워져요. 이들은 내부적으로 에이전트의 state_schema에 추가되기 때문에, inputs_from_state로 등록한 도구나 hooks가 실행 중인 State에서 읽을 수 있죠. 이 값은 출력 전용이라 run()이나 run_async()의 입력으로는 넘길 수 없고, 여러분이 직접 state_schema의 키로 쓰면 ValueError가 발생해요. 자세한 내용은 State를 참고하세요.
response = agent.run(messages=[ChatMessage.from_user("What is 7 * (4 + 2)?")])
print(response["step_count"]) # 2
print(response["token_usage"]) # {"prompt_tokens": 512, "completion_tokens": 86, ...}
print(response["tool_call_counts"]) # {"calculator": 1}
print(response["exit_reason"]) # "text"
Exit reason
exit_reason 출력은 에이전트가 왜 멈췄는지 알려줘요. 덕분에 ConditionalRouter 같은 걸로 출력을 다운스트림에 쉽게 라우팅할 수 있죠. 값은 다음 중 하나입니다:
"text": 모델이 도구 호출 없이 완전한 답변을 반환했어요."length": 모델이 출력 토큰 한도에 도달했어요.last_message에 부분 응답이 들어 있을 수 있어요."content_filter": 콘텐츠 필터가 모델 응답을 멈췄어요.last_message에 부분 응답이 들어 있을 수 있어요.- 도구 종료 조건을 충족한 도구의 이름. 이 경우
last_message는 그 도구의 결과 —text가 빈 도구-결과ChatMessage— 이므로,exit_reason이 그 결과를 어떻게 소비할지 알려줘요. "max_agent_steps": 에이전트가 종료 조건을 만나기 전에max_agent_steps에 도달했어요.- 훅이
stop_run상태 키를 통해 설정한 사용자 지정 이유. 예를 들어 TokenBudgetHook의"token_budget_exceeded"같은 값이죠.
Agent는 기본적으로 "length"와 "content_filter"에서 멈춰서, 변경되지 않은 요청을 반복 제출하지 않아요. 이 이유 때문에 복구가 불가능해지진 않아요: on_exit 훅이 이유를 확인하고 메시지를 다시 쓴 뒤 continue_run을 True로 설정할 수 있거든요. 또는 결과를 다운스트림으로 라우팅해 다른 생성 설정으로 재시도하거나 사람의 검토를 요청할 수도 있어요.
exit_reason은 실행 중인 State에서 확인 가능하므로, after_run hook이 실행이 어떻게 끝났는지에 따라 반응할 수 있어요. 예를 들어 스텝 예산이 에이전트가 끝나기도 전에 소진되면 대체 답변을 붙이는 방식이죠:
from haystack.components.agents.state import State
from haystack.dataclasses import ChatMessage
from haystack.hooks import hook
@hook
def fallback_on_max_steps(state: State) -> None:
if state.get("exit_reason") == "max_agent_steps":
state.set(
"messages",
[ChatMessage.from_assistant("Sorry, I ran out of steps before finishing.")],
)
예를 들어 이 훅은 출력 한도 종료 후 짧은 이어말하기 하나를 요청해요. 에이전트의 기존 max_agent_steps 설정은 여전히 실행을 제한합니다.
@hook
def recover_from_output_limit(state: State) -> None:
recovery_prompt = "Continue with a shorter answer."
recovery_attempted = any(
message.text == recovery_prompt for message in state.get("messages", [])
)
if state.get("exit_reason") == "length" and not recovery_attempted:
state.set(
"messages",
[ChatMessage.from_user(recovery_prompt)],
)
state.set("continue_run", True)
Parameters
chat_generator는 유일한 필수 파라미터로, 도구를 지원하는 Chat Generator 인스턴스예요. 나머지 파라미터는 모두 선택입니다.
tools: 에이전트가 호출할 수 있는 도구·도구셋 인스턴스 목록. 지원 타입: Tool, ComponentTool, PipelineTool, AgentTool, MCPTool, Toolset, MCPToolset, SearchableToolset. 도구 이름은 유일해야 해요. 중복 이름은 매 에이전트 스텝 시작 시, 채팅 생성기가 호출되기 전에 감지됩니다.system_prompt: 매 실행의 시스템 메시지로 쓰는 평문 문자열 또는 Jinja2 템플릿. 템플릿에 Jinja2 변수가 있으면 그 변수들이run()의 추가 입력이 돼요.user_prompt: 매 실행 시 사용자 제공 메시지 뒤에 붙는 Jinja2 템플릿. 템플릿 변수는run()의 추가 입력이 됩니다. 어떤 변수가 제공돼야 하는지 강제하려면required_variables를 쓰세요.exit_conditions: 에이전트를 멈추게 하는 조건 목록. LLM이 도구 호출 없이 답할 때 멈추려면"text"를, 특정 도구가 실행된 뒤 멈추려면 그 도구 이름을 쓰세요. 기본값은["text"]예요. 종료 조건은 초기화 때 검증되는 게 아니라 실행 시간에 평가되므로, 나중에 로드되는 도구 — 예를 들어run(tools=...)으로 전달하거나 SearchableToolset이 발견한 도구 — 를 조건이 가리킬 수 있어요.state_schema: 에이전트의 실행 시간 상태를 정의해요 — 키 이름을 타입 설정으로 매핑한 dict예요(예:{"docs": {"type": list[Document]}}). 도구는inputs_from_state와outputs_to_state로 상태 키를 읽고 쓸 수 있어요. 전체 내용은 State를 참고하세요.streaming_callback: 스트리밍되는 토큰마다 호출되는 콜백. 콘솔 출력에는 내장print_streaming_chunk를 쓰세요.max_agent_steps: 에이전트가 멈추기 전에 수행할 최대 LLM+도구 호출 반복 횟수. 기본값은100.raise_on_tool_invocation_failure:True면 도구 호출 실패 시 예외를 던져요.False(기본값)면 에러를 메시지로 LLM에 돌려줘 LLM이 복구할 수 있게 합니다.hooks: 훅 포인트("before_run","before_llm","before_tool","after_tool","on_exit","after_run")를 그 시점에 실행할 훅 목록으로 매핑한 dict. 훅은 실행 중인State를 받아, 그걸 변형해 실행에 영향을 줍니다 — 예를 들어 런타임 컨텍스트를 만들거나 도구 호출에 사람 확인을 요구하는 식이죠. Hooks와 Human in the Loop를 참고하세요.tool_concurrency_limit: 동시에 실행할 최대 도구 호출 수. 기본값은4. 병렬 도구 실행을 끄려면1로 설정하세요.tool_streaming_callback_passthrough:True면 스트리밍 콜백을 이를 받아들이는 도구에도 전달해요.
Runtime overrides
run()은 단일 호출에서 init 시점 설정을 덮어쓰는 파라미터도 받습니다:
tools:Tool/Toolset객체 목록, 또는 에이전트가 설정한 도구 중 일부를 선택하는 도구 이름 문자열 목록을 넘길 수 있어요.generation_kwargs: 채팅 생성기로 전달되는 추가 키워드 인자(예:{"temperature": 0.2}). 채팅 생성기 초기화에 설정된generation_kwargs와 키별로 병합돼요: 여기 넘긴 키가 우선하고, 초기화에서만 설정된 키는 유지됩니다.hook_context: hooks가state.data["hook_context"]를 통해 접근하는 요청 범위 리소스 dict — 예를 들어 사용자 ID나 WebSocket 연결이요.
info 전체 파라미터 참조는 Agents API Documentation을 확인하세요.
Cloning and modifying an agent
Agent 속성은 초기화 후 재할당하도록 설계되지 않았어요. 일부 파라미터는 init 시점에 처리되므로, 빌드된 에이전트에 속성을 설정해도 확실하게 적용되지 않죠. 기존 에이전트의 수정 버전을 얻는 권장 방법은 clone()입니다.
clone()은 같은 설정으로, 선택적으로 일부 init 파라미터를 교체한 새 에이전트를 반환해요. 직접 만들지 않은 에이전트 — 예를 들어 Agent Pack의 팩토리 함수가 반환한 것 — 의 변형을 만들 때 유용하죠.
variant = agent.clone(system_prompt="Answer in German.", max_agent_steps=20)
오버라이드는 원래 값을 교체합니다. 목록이나 dict를 덮어쓰는 대신 확장하려면 기존 값을 풀어서 항목을 추가하세요:
extended = agent.clone(
tools=[*agent.tools, my_new_tool],
state_schema={**agent.state_schema, "notes": {"type": str}},
hooks={**agent.hooks, "before_llm": [my_hook]},
)
Usage
On its own
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
from haystack.components.agents import Agent
from typing import Annotated
@tool(outputs_to_state={"calc_result": {"source": "result"}})
def calculator(
expression: Annotated[str, "Math expression to evaluate, e.g. '7 * (4 + 2)'"],
) -> dict:
"""Evaluate basic math expressions."""
try:
result = eval(expression, {"__builtins__": {}})
return {"result": result}
except Exception as e:
return {"error": str(e)}
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[calculator],
system_prompt="You are a helpful assistant. Always use the calculator tool to evaluate math expressions.",
state_schema={"calc_result": {"type": int}},
)
response = agent.run(messages=[ChatMessage.from_user("What is 7 * (4 + 2)?")])
print(response["last_message"].text)
print("Calc Result:", response.get("calc_result"))
In a pipeline
아래 예시 파이프라인은 OpenAIChatGenerator, LinkContentFetcher, 그리고 사용자 정의 데이터베이스 도구로 데이터베이스 어시스턴트를 만듭니다. 주어진 URL을 읽고 페이지 내용을 처리한 뒤, AI용 프롬프트를 만듭니다. 어시스턴트는 이 정보를 이용해 페이지에서 인물의 이름과 직함을 데이터베이스에 기록해요.
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders.chat_prompt_builder import ChatPromptBuilder
from haystack.components.converters.html import HTMLToDocument
from haystack.components.fetchers.link_content import LinkContentFetcher
from haystack import Document, Pipeline
from haystack.dataclasses import ChatMessage
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.tools import tool
from typing import Annotated, Optional
document_store = InMemoryDocumentStore() # create a document store or an SQL database
@tool
def add_database_tool(
name: Annotated[str, "First name of the person"],
surname: Annotated[str, "Last name of the person"],
job_title: Annotated[Optional[str], "Job title or role of the person"] = None,
other: Annotated[Optional[str], "Any other relevant information"] = None,
) -> str:
"""Add a person to the database with information about them."""
document_store.write_documents(
[
Document(
content=name + " " + surname + " " + (job_title or ""),
meta={"other": other},
),
],
)
# Returning a confirmation lets the agent know the tool call succeeded
return f"Successfully added {name} {surname} to the database."
database_assistant = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[add_database_tool],
system_prompt="""
You are a database assistant.
Your task is to extract the names of people mentioned in the given context and add them to a knowledge base,
along with additional relevant information about them that can be extracted from the context.
Do not use your own knowledge, stay grounded to the given context.
Do not ask the user for confirmation.
Instead, automatically update the knowledge base and return a brief summary of the people added,
including the information stored for each.
""",
)
# ... (파이프라인 빌드 계속)
In YAML
아래 예시 파이프라인은 웹페이지를 가져와 HTML을 텍스트로 변환하고, 페이지 내용과 사용자 질의를 합친 채팅 프롬프트를 만듭니다. Agent가 제공된 내용을 바탕으로 질문에 답하고, 필요하면 웹 검색 도구로 추가 정보를 찾을 수 있어요.
components:
agent:
init_parameters:
chat_generator:
init_parameters:
api_base_url: null
api_key:
env_vars:
- OPENAI_API_KEY
strict: true
type: env_var
generation_kwargs: {}
http_client_kwargs: null
max_retries: null
model: gpt-5.4-nano
organization: null
streaming_callback: null
timeout: null
tools: null
tools_strict: false
type: haystack.components.generators.chat.openai.OpenAIChatGenerator
exit_conditions:
- text
hooks: null
max_agent_steps: 5
raise_on_tool_invocation_failure: false
required_variables: null
state_schema: {}
streaming_callback: null
system_prompt: You are a helpful assistant. Use the web search tool to find
information when needed.
tool_concurrency_limit: 4
tool_streaming_callback_passthrough: false
tools:
- data:
component:
init_parameters:
allowed_domains: null
api_key:
env_vars:
- SERPERDEV_API_KEY
strict: true
type: env_var
exclude_subdomains: false
search_params: {}
top_k: 3
type: haystack_integrations.components.websearch.serperdev.websearch.SerperDevWebSearch
description: Search the web for current information on any topic
inputs_from_state: null
name: web_search
outputs_to_state: null
outputs_to_string: null
parameters: null
type: haystack.tools.component_tool.ComponentTool
user_prompt: null
type: haystack.components.agents.agent.Agent
converter:
init_parameters:
extraction_kwargs: {}
store_full_path: false
type: haystack.components.converters.html.HTMLToDocument
fetcher:
init_parameters:
client_kwargs:
follow_redirects: true
timeout: 3
# ... (생략)
Streaming
생성되는 대로 출력을 스트리밍할 수 있어요. streaming_callback에 콜백을 넘기면 됩니다. 내장 print_streaming_chunk를 쓰면 텍스트 토큰과 도구 이벤트(도구 호출·도구 결과)를 출력할 수 있어요.
from haystack.components.generators.utils import print_streaming_chunk
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5.4-nano"),
tools=[...],
system_prompt="...",
streaming_callback=print_streaming_chunk,
)
StreamingChunk가 어떻게 동작하는지, 사용자 정의 콜백을 어떻게 작성하는지는 Streaming Support 문서를 참고하세요.
기본적으로 print_streaming_chunk를 우선 쓰는 걸 권장해요. 특정 전송(예: SSE/WebSocket)이나 사용자 정의 UI 포맷이 필요할 때만 직접 콜백을 작성하세요.
Multimodal Inputs
에이전트는 gpt-5(OpenAI)이나 gemini-2.5-flash(Google) 같은 비전 지원 모델과 함께 쓰면 멀티모달 입력을 지원해요. ChatMessage의 content_parts에 ImageContent 객체를 넣어 텍스트와 함께 이미지를 전달하세요:
from haystack.dataclasses import ChatMessage, ImageContent
image = ImageContent.from_url("https://example.com/chart.png")
result = agent.run(
messages=[
ChatMessage.from_user(content_parts=["What does this chart show?", image]),
],
)
도구가 직접 ImageContent를 반환할 수도 있어요. 그러면 에이전트가 루프 중에 이미지를 동적으로 가져와 추론할 수 있죠. 두 가지가 필요합니다: outputs_to_string={"raw_result": True}로 설정해 Agent의 도구 실행이 문자열 변환을 건너뛰게 하고, list[ImageContent]를 반환하면 돼요(도구 결과 타입은 str | Sequence[TextContent | ImageContent]).
표준 Chat Completions API는 도구 결과의 이미지를 지원하지 않아요 — 대신 OpenAIResponsesChatGenerator(OpenAI의 Responses API)를 쓰세요:
from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIResponsesChatGenerator
from haystack.dataclasses import ChatMessage, ImageContent
from haystack.tools import tool
@tool(outputs_to_string={"raw_result": True})
def fetch_image(
url: Annotated[str, "URL of the image to fetch and analyze"],
) -> list[ImageContent]:
"""Fetch an image from a URL so the agent can analyze its contents."""
return [ImageContent.from_url(url)]
agent = Agent(
chat_generator=OpenAIResponsesChatGenerator(model="gpt-5"),
tools=[fetch_image],
system_prompt="You are a helpful assistant that can fetch and analyze images from URLs.",
)
result = agent.run(
messages=[
ChatMessage.from_user(
"Fetch the image at https://picsum.photos/seed/haystack/640/480 and describe what you see.",
),
],
)
print(result["last_message"].text)
ImageContent는 URL, 로컬 파일 경로, 또는 PDFToImageContent 컨버터를 통해 PDF 페이지로부터 만들 수 있어요.
In a pipeline
Agent가 파이프라인 안에 있을 때는 ChatPromptBuilder를 문자열 템플릿 형식과 함께 쓰고, | templatize_part 필터로 이미지를 구조화된 콘텐츠 파트로 전달하세요:
from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ImageContent
template = """{% message role="user" %}{{ question }}{{ image | templatize_part }}{% endmessage %}"""
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-5"),
system_prompt="You are a helpful assistant that can analyze images.",
)
prompt_builder = ChatPromptBuilder(
template=template,
required_variables=["question", "image"],
)
pipeline = Pipeline()
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("agent", agent)
pipeline.connect("prompt_builder.prompt", "agent.messages")
# Download or provide your own chart image as "chart.png"
image = ImageContent.from_file_path("chart.png")
result = pipeline.run(
{
"prompt_builder": {"question": "What does this chart show?", "image": image},
},
)
print(result["agent"]["last_message"].text)
tip 완전한 멀티모달 에이전트 예시는 이 쿡북들을 참고하세요:
- Multimodal Agents — 에이전트의 이미지 입력과 도구 사용
- Gemma Chat RAG — RAG 파이프라인의 비전 모델
Multi-Agent Systems
Agent를 도구로 감싸면 멀티 에이전트 시스템을 만들 수 있어요. 전문 에이전트가 집중된 하위 작업을 맡고, 코디네이터 에이전트가 계획을 세우고 위임하는 구조죠.
가장 간단한 방법은 AgentTool이에요. 에이전트를 감싸 작업을 단일 사용자 메시지로 위임하고 최종 답변만 반환하죠.
전체 가이드는 Multi-Agent Systems에서 확인하세요.
MCP Integration
Agent는 MCP와 두 방향으로 작동해요:
- MCP 도구 소비하기:
tools목록에MCPTool이나MCPToolset인스턴스를 넘겨 모든 MCP 호환 서버(파일시스템, 브라우저, 데이터베이스 등)의 도구를 호출할 수 있어요. MCPTool과 MCPToolset을 참고하세요. - MCP 서버로 노출하기: Hayhooks로 에이전트를 배포해 MCP 서버로 노출하면, Claude Desktop이나 Cursor 같은 MCP 호환 클라이언트에서 호출할 수 있어요.
더 알아보기 (Learn more)
📖 관련 문서:
- State — 도구 간 공유 데이터 관리
- Hooks — 실행 루프의 정의된 시점에 사용자 정의 로직 실행
- Human in the Loop — 도구 호출을 가로채 사람 검토에 넘기기
- Tool Result Offloading — 큰 도구 결과를 컨텍스트 윈도우 밖에 두기
📚 튜토리얼:
🧑🍳 쿡북: