멀티 에이전트 시스템 구축
멀티 에이전트 시스템 구축 (Multi-Agent System)
각자 다른 역할을 맡은 전문화된 에이전트들이 협력하는 멀티 에이전트 시스템을 만들어 볼게요. 리서치 에이전트(정보 조사)와 라이터 에이전트(저장)를 각각 만들고, 이 둘을 메인 에이전트가 조율하도록 묶는 구조예요.
출처: 공식문서
개요
이 튜토리얼은 Haystack 3.1 이상이 필요해요. 두 개의 서브 에이전트와 각자의 도구를 만들고, AgentTool로 감싸서 메인 에이전트에 넘겨 최종 멀티 에이전트 시스템을 구성해요.
환경 준비
pip install haystack-ai duckduckgo-api-haystack
pip install openai "datasets>=2.6.1"
API 키를 입력받아 저장해요. Notion 연동을 쓰려면 NOTION_API_KEY도 필요해요.
from getpass import getpass
import os
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API key:")
if not os.environ.get("NOTION_API_KEY"):
os.environ["NOTION_API_KEY"] = getpass("Enter your NOTION API key:")
리서치 에이전트용 도구 만들기
리서치 에이전트는 정보를 모으는 역할이에요. 여기서는 웹 전체 검색과 위키피디아 검색 두 도구를 DuckduckgoApiWebSearch 컴포넌트로 만들어요. allowed_domain을 주면 해당 도메인으로 검색을 제한할 수 있어요.
DuckduckgoApiWebSearch는 문서 리스트를 반환하는데, 에이전트는 도구 결과가 문자열 하나일 때 가장 잘 동작해요. 그래서 문서 리스트를 문자열로 바꾸는 doc_to_string 함수를 outputs_to_string 핸들러로 넘겨줘요.
from haystack.tools import ComponentTool
from duckduckgo_api_haystack import DuckduckgoApiWebSearch
def doc_to_string(documents) -> str:
"""
Handles the tool output before conversion to ChatMessage.
"""
result_str = ""
for document in documents:
result_str += f"File Content for {document.meta['link']}\n\n {document.content}"
if len(result_str) > 150_000: # trim if the content is too large
result_str = result_str[:150_000] + "...(large file can't be fully displayed)"
return result_str
web_search = ComponentTool(
component=DuckduckgoApiWebSearch(top_k=5, backend="lite"),
name="web_search",
description="Search the web",
outputs_to_string={"source": "documents", "handler": doc_to_string},
)
wiki_search = ComponentTool(
component=DuckduckgoApiWebSearch(top_k=5, backend="lite", allowed_domain="https://en.wikipedia.org"),
name="wiki_search",
description="Search Wikipedia",
outputs_to_string={"source": "documents", "handler": doc_to_string},
)
리서치 에이전트 초기화
web_search와 wiki_search 도구를 넘겨 리서치 에이전트를 만들어요. OpenAIChatGenerator나 함수 호출을 지원하는 다른 챗 생성기를 쓰면 돼요. 실시간 진행 상황을 보려면 print_streaming_chunk로 스트리밍을 켜요.
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
from haystack.components.generators.chat import OpenAIChatGenerator
research_agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
system_prompt="""
You are a research agent that can find information on web or specifically on wikipedia.
Use wiki_search tool if you need facts and use web_search tool for latest news on topics.
Use one tool at a time. Try different queries if you need more information.
Only use the retrieved context, do not use your own knowledge.
Summarize the all retrieved information before returning response to the user.
""",
tools=[web_search, wiki_search],
streaming_callback=print_streaming_chunk,
)
result = research_agent.run(
messages=[ChatMessage.from_user("Can you tell me about Florence Nightingale's contributions to nursery?")]
)
print("Final Answer:", result["last_message"].text)
라이터 에이전트용 도구 만들기
라이터 에이전트는 내용을 저장하는 역할이에요. 여기서는 Notion과 문서 스토어 두 곳에 저장하는 도구를 만들어요.
Notion 작성 도구
커스텀 컴포넌트로 Notion 작업공간에 새 페이지를 만드는 NotionPageCreator를 만들고, ComponentTool로 감싸요.
from haystack import component
from typing import Optional
from haystack.utils import Secret
import requests
@component
class NotionPageCreator:
"""
Create a page in Notion using provided title and content.
"""
def __init__(
self,
page_id: str,
notion_version: str = "2022-06-28",
api_key: Secret = Secret.from_env_var("NOTION_API_KEY"), # to use the environment variable NOTION_API_KEY
):
"""
Initialize with the target Notion database ID and API version.
"""
self.api_key = api_key
self.notion_version = notion_version
self.page_id = page_id
@component.output_types(success=bool, status_code=int, error=Optional[str])
def run(self, title: str, content: str):
"""
:param title: The title of the Notion page.
:param content: The content of the Notion page.
"""
headers = {
"Authorization": f"Bearer {self.api_key.resolve_value()}",
"Content-Type": "application/json",
"Notion-Version": self.notion_version,
}
payload = {
"parent": {"page_id": self.page_id},
"properties": {"title": [{"text": {"content": title}}]},
"children": [
{
"object": "block",
"type": "paragraph",
"paragraph": {"rich_text": [{"type": "text", "text": {"content": content}}]},
}
],
}
response = requests.post("https://api.notion.com/v1/pages", headers=headers, json=payload)
if response.status_code == 200 or response.status_code == 201:
return {"success": True, "status_code": response.status_code}
else:
return {"success": False, "status_code": response.status_code, "error": response.text}
from haystack.tools import ComponentTool
notion_writer = ComponentTool(
component=NotionPageCreator(page_id="<your_page_id>"),
name="notion_writer",
description="Use this tool to write/save content to Notion.",
)
notion_writer.parameters # see how parameters are automatically generated by the ComponentTool
커스텀 컴포넌트를 ComponentTool 도구로 만들 때는 입력 파라미터를 잘 정의해야 해요. properties 사전을 넘기거나, run 메서드의 docstring에 파라미터 어노테이션을 쓰는 방식으로 정의할 수 있어요. 도구의 description 설정에도 같은 방식이 적용돼요.
문서 스토어 작성 도구
DocumentAdapter 커스텀 컴포넌트와 DocumentWriter로 파이프라인을 만들고, PipelineTool로 감싸 문서 스토어에 저장하는 도구를 만들어요.
from haystack import Pipeline, component, Document
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.tools import PipelineTool
from typing import List
@component
class DocumentAdapter:
@component.output_types(documents=List[Document])
def run(self, content: str, title: str):
return {"documents": [Document(content=content, meta={"title": title})]}
document_store = InMemoryDocumentStore()
doc_store_writer_pipeline = Pipeline()
doc_store_writer_pipeline.add_component("adapter", DocumentAdapter())
doc_store_writer_pipeline.add_component("writer", DocumentWriter(document_store=document_store))
doc_store_writer_pipeline.connect("adapter", "writer")
doc_store_writer = PipelineTool(
pipeline=doc_store_writer_pipeline,
name="doc_store_writer",
description="Use this tool to write/save content to document store",
parameters={
"type": "object",
"properties": {
"title": {"type": "string", "description": "The title of the Document"},
"content": {"type": "string", "description": "The content of the Document"},
},
"required": ["title", "content"],
},
)
doc_store_writer.parameters
PipelineTool은 파이프라인 전체를 선언적 구성 방식으로 도구화할 때 좋아요. 반면 파이프라인을 @tool 데코레이터 함수로 감싸면 LLM이 보는 입력 파라미터, 출력 포맷, 오류 처리를 세밀하게 제어할 수 있어요.
라이터 에이전트 초기화
notion_writer와 doc_store_writer 도구를 넘겨 라이터 에이전트를 만들어요. 이 에이전트는 응답이 아니라 행동이 목적이므로, exit_conditions를 도구 이름으로 설정해 해당 도구를 한 번 호출하면 멈추게 해요.
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
from haystack.components.generators.chat import OpenAIChatGenerator
writer_agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
system_prompt="""
You are a writer agent that saves given information to different locations.
Do not change the provided content before saving.
Infer the title from the text if not provided.
When you need to save provided information to Notion, use notion_writer tool.
When you need to save provided information to document store, use doc_store_writer tool
If no location is mentioned, use notion_writer tool to save the information.
""",
tools=[doc_store_writer, notion_writer],
streaming_callback=print_streaming_chunk,
exit_conditions=["notion_writer", "doc_store_writer"],
)
멀티 에이전트 시스템 만들기
두 서브 에이전트를 AgentTool로 감싼 뒤, 메인 에이전트의 도구로 넘겨요. AgentTool은 위임에 맞는 기본 동작을 제공해요.
- 작업은 단일 사용자 메시지로 전달돼서, 메인 에이전트는 서브 에이전트 내부를 몰라도 돼요.
- 서브 에이전트의 최종 답변만 메인 에이전트로 돌아가요. 중간 단계(도구 호출, 도구 결과, 초안)는 메인 에이전트 컨텍스트에 들어가지 않아요.
- 서브 에이전트가
max_agent_steps때문에 멈췄다면 결과에 경고가 붙어서, 절단된 답을 완전한 답으로 오인하지 않게 해요.
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
from haystack.tools import AgentTool
research_tool = AgentTool(
agent=research_agent,
name="research_tool",
description="Use this tool to find information on web or specifically on wikipedia",
)
writer_tool = AgentTool(
agent=writer_agent, name="writer_tool", description="Use this tool to write content into document store or Notion"
)
main_agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
system_prompt="""
You are an assistant that has access to several tools.
Understand the user query and use relevant tool to answer the query.
You can use `research_tool` to make research on web and wikipedia and `writer_tool` to save information into the document store or Notion.
""",
streaming_callback=print_streaming_chunk,
tools=[research_tool, writer_tool],
)
이제 메인 에이전트에 연구와 저장을 함께 시킬 수 있어요. 예를 들어 실크로드의 역사를 연구하고, RAG 파이프라인이 어떻게 동작하는지 요약해서 Notion에 저장하라고 지시할 수 있어요.
result = main_agent.run(
messages=[
ChatMessage.from_user(
"""
Can you research the history of the Silk Road?
"""
)
]
)
result["last_message"].text