도구

pip install 'crewai[tools]'

import os
from crewai import Agent, Task, Crew
# crewAI 도구 임포트
from crewai_tools import (
    DirectoryReadTool,
    FileReadTool,
    SerperDevTool,
    WebsiteSearchTool
)

# API 키 설정
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API 키
os.environ["OPENAI_API_KEY"] = "Your Key"

# 도구 인스턴스 생성
docs_tool = DirectoryReadTool(directory='./blog-posts')
file_tool = FileReadTool()
search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

# 에이전트 생성
researcher = Agent(
    role='시장 조사 분석가',
    goal='AI 산업의 최신 시장 분석 제공',
    backstory='시장 트렌드를 꿰뚫어 보는 전문 분석가',
    tools=[search_tool, web_rag_tool],
    verbose=True
)

writer = Agent(
    role='콘텐츠 작가',
    goal='AI 산업에 대한 흥미로운 블로그 글 작성',
    backstory='기술에 열정을 가진 숙련된 작가',
    tools=[docs_tool, file_tool],
    verbose=True
)

# 작업 정의
research = Task(
    description='AI 산업의 최신 트렌드를 조사하고 요약 제공',
    expected_output='AI 산업의 상위 3개 트렌드 개발 동향과 그 중요성에 대한 독창적인 관점을 담은 요약',
    agent=researcher
)

write = Task(
    description='조사 분석가의 요약을 바탕으로 AI 산업에 대한 흥미로운 블로그 글 작성. 최신 블로그 글에서 영감을 얻는다.',
    expected_output='복잡한 전문 용어를 피하고, 흥미롭고 정보가 풍부하며 접근 가능한 콘텐츠로 구성된 4단락의 마크다운 형식 블로그 글',
    agent=writer,
    output_file='blog-posts/new_post.md'  # 최종 블로그 글 저장 위치
)

# 플래닝 기능을 활성화한 크루 구성
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    verbose=True,
    planning=True,  # 플래닝 기능 활성화
)

# 작업 실행
crew.kickoff()

from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    """MyCustomTool의 입력 스키마."""
    argument: str = Field(..., description="인자의 설명.")

class MyCustomTool(BaseTool):
    name: str = "내 도구의 이름"
    description: str = "이 도구가 수행하는 작업. 효과적인 활용을 위해 중요하다."
    args_schema: Type[BaseModel] = MyToolInput

    def _run(self, argument: str) -> str:
        # 도구의 로직을 여기에 작성
        return "도구의 결과"

from crewai.tools import tool
@tool("내 도구 이름")
def my_tool(question: str) -> str:
    """이 도구가 어떤 용도로 사용되는지 명확히 설명한다. 에이전트가 이를 사용하려면 이 정보가 필요하다."""
    # 함수 로직 작성
    return "커스텀 도구의 결과"

from crewai.tools.structured_tool import CrewStructuredTool
from pydantic import BaseModel

# Pydantic을 사용해 도구의 입력 스키마 정의
class APICallInput(BaseModel):
    endpoint: str
    parameters: dict

# API 호출을 실행하는 래퍼 함수
def tool_wrapper(*args, **kwargs):
    # 여기서는 일반적으로 매개변수를 사용해 API를 호출
    # 예시를 위해 플레이스홀더 문자열 반환
    return f"Call the API at {kwargs['endpoint']} with parameters {kwargs['parameters']}"

# 구조화된 도구 생성 및 반환
def create_structured_tool():
    return CrewStructuredTool.from_function(
        name='Wrapper API',
        description="구조화된 입력으로 API 호출을 감싸는 도구.",
        args_schema=APICallInput,
        func=tool_wrapper,
    )

# 예시 사용법
structured_tool = create_structured_tool()

# 구조화된 입력으로 도구 실행
result = structured_tool._run(**{
    "endpoint": "https://example.com/api",
    "parameters": {"key1": "value1", "key2": "value2"}
})
print(result)  # 출력: Call the API at https://example.com/api with parameters {'key1': 'value1', 'key2': 'value2'}

from crewai.tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """두 숫자를 곱해야 할 때 유용한 도구이다."""
    return first_number * second_number

def cache_func(args, result):
    # 이 경우, 결과가 2의 배수일 때만 캐시한다
    cache = result % 2 == 0
    return cache

multiplication_tool.cache_function = cache_func

writer1 = Agent(
        role="Writer",
        goal="어린이를 위한 수학 교재를 작성한다.",
        backstory="글쓰기 전문가이며 아이들을 가르치는 것을 좋아하지만, 수학에 대한 지식은 전무하다.",
        tools=[multiplication_tool],
        allow_delegation=False,
    )
    #...