파일 기반 지식 소스를 사용할 때는 프로젝트 루트에 knowledge 디렉토리를 만들고, 파일을 해당 디렉토리에 배치한다.
또한, 소스를 생성할 때 knowledge 디렉토리를 기준으로 상대 경로를 사용한다.
다음은 문자열 기반 지식을 사용한 예제다:
from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# 지식 소스 생성content = "Users name is John. He is 30 years old and lives in San Francisco."string_source = StringKnowledgeSource( content=content,)# 결정적 출력을 위해 temperature를 0으로 설정한 LLM 생성llm = LLM(model="gpt-4o-mini", temperature=0)# 지식 저장소를 가진 에이전트 생성agent = Agent( role="About User", goal="You know everything about the user.", backstory="""You are a master at understanding people and their preferences.""", verbose=True, allow_delegation=False, llm=llm,)task = Task( description="Answer the following questions about the user: {question}", expected_output="An answer to the question.", agent=agent,)crew = Crew( agents=[agent], tasks=[task], verbose=True, process=Process.sequential, knowledge_sources=[string_source], # 지식 소스를 추가해 지식을 활성화한다. 소스 목록에 더 많은 소스를 추가할 수도 있다.)result = crew.kickoff(inputs={"question": "What city does John live in and how old is he?"})
다음은 CrewDoclingSource를 사용한 예제다. CrewDoclingSource는 MD, PDF, DOCX, HTML 등 다양한 파일 포맷을 처리할 수 있다.
다음 예제를 실행하려면 docling을 설치해야 한다: uv add docling
from crewai import LLM, Agent, Crew, Process, Taskfrom crewai.knowledge.source.crew_docling_source import CrewDoclingSource# 지식 소스 생성content_source = CrewDoclingSource( file_paths=[ "https://lilianweng.github.io/posts/2024-11-28-reward-hacking", "https://lilianweng.github.io/posts/2024-07-07-hallucination", ],)# 결정적 출력을 위해 temperature를 0으로 설정한 LLM 생성llm = LLM(model="gpt-4o-mini", temperature=0)# 지식 저장소를 가진 에이전트 생성agent = Agent( role="About papers", goal="You know everything about the papers.", backstory="""You are a master at understanding papers and their content.""", verbose=True, allow_delegation=False, llm=llm,)task = Task( description="Answer the following questions about the papers: {question}", expected_output="An answer to the question.", agent=agent,)crew = Crew( agents=[agent], tasks=[task], verbose=True, process=Process.sequential, knowledge_sources=[ content_source ], # 지식 소스를 추가해 지식을 활성화한다. 소스 목록에 더 많은 소스를 추가할 수도 있다.)result = crew.kickoff( inputs={ "question": "What is the reward hacking paper about? Be sure to provide sources." })
from crewai.knowledge.source.text_file_knowledge_source import TextFileKnowledgeSource# 텍스트 파일 지식 소스 생성text_source = TextFileKnowledgeSource( file_paths=["document.txt", "another.txt"])# 에이전트 또는 크루 레벨에 텍스트 파일 소스 추가agent = Agent( ... knowledge_sources=[text_source])crew = Crew( ... knowledge_sources=[text_source])
from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource# PDF 지식 소스 생성pdf_source = PDFKnowledgeSource( file_paths=["document.pdf", "another.pdf"])# 에이전트 또는 크루 레벨에 PDF 지식 소스 추가agent = Agent( ... knowledge_sources=[pdf_source])crew = Crew( ... knowledge_sources=[pdf_source])
from crewai.knowledge.source.csv_knowledge_source import CSVKnowledgeSource# CSV 지식 소스 생성csv_source = CSVKnowledgeSource( file_paths=["data.csv"])# 에이전트 또는 크루에 CSV 지식 소스 추가agent = Agent( ... knowledge_sources=[csv_source])crew = Crew( ... knowledge_sources=[csv_source])
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource# JSON 지식 소스 생성json_source = JSONKnowledgeSource( file_paths=["data.json"])# 에이전트 또는 크루 레벨에 JSON 지식 소스 추가agent = Agent( ... knowledge_sources=[json_source])crew = Crew( ... knowledge_sources=[json_source])
지식 소스는 더 나은 처리를 위해 콘텐츠를 자동으로 청크로 나눈다. 지식 소스에서 청킹 동작을 다음과 같이 설정할 수 있다:
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSourcesource = StringKnowledgeSource( content="여기에 콘텐츠를 입력", chunk_size=4000, # 각 청크의 최대 크기 (기본값: 4000) chunk_overlap=200 # 청크 간 겹치는 부분 (기본값: 200))
지식 저장소를 위한 임베딩 설정도 가능하다. 에이전트와는 다른 임베딩 모델을 지식 저장소에 사용하고 싶을 때 유용하다. embedder 파라미터는 다양한 임베딩 모델 프로바이더를 지원한다:
openai: OpenAI의 임베딩 모델
google: Google의 텍스트 임베딩 모델
azure: Azure OpenAI 임베딩
ollama: Ollama를 사용한 로컬 임베딩
vertexai: Google Cloud VertexAI 임베딩
cohere: Cohere의 임베딩 모델
voyageai: VoyageAI의 임베딩 모델
bedrock: AWS Bedrock 임베딩
huggingface: Hugging Face 모델
watson: IBM Watson 임베딩
Google의 text-embedding-004 모델을 사용해 지식 저장소의 임베딩을 설정하는 예제는 다음과 같다:
from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSourceimport os# GEMINI API 키 가져오기GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")# 지식 소스 생성content = "Users name is John. He is 30 years old and lives in San Francisco."string_source = StringKnowledgeSource( content=content,)# 결정적 출력을 위해 온도를 0으로 설정한 LLM 생성gemini_llm = LLM( model="gemini/gemini-1.5-pro-002", api_key=GEMINI_API_KEY, temperature=0,)# 지식 저장소를 가진 에이전트 생성agent = Agent( role="About User", goal="You know everything about the user.", backstory="""You are a master at understanding people and their preferences.""", verbose=True, allow_delegation=False, llm=gemini_llm, embedder={ "provider": "google", "config": { "model": "models/text-embedding-004", "api_key": GEMINI_API_KEY, } })task = Task( description="Answer the following questions about the user: {question}", expected_output="An answer to the question.", agent=agent,)crew = Crew( agents=[agent], tasks=[task], verbose=True, process=Process.sequential, knowledge_sources=[string_source], embedder={ "provider": "google", "config": { "model": "models/text-embedding-004", "api_key": GEMINI_API_KEY, } })result = crew.kickoff(inputs={"question": "What city does John live in and how old is he?"})
crew.knowledge_sources를 사용해 크루 전체에 지식을 제공할 수 있지만, 개별 에이전트도 knowledge_sources 매개변수를 통해 자신만의 지식 소스를 가질 수 있다.
from crewai import Agent, Task, Crewfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# 제품에 대한 에이전트별 지식 생성product_specs = StringKnowledgeSource( content="""XPS 13 노트북 사양: - 13.4인치 4K 디스플레이 - Intel Core i7 프로세서 - 16GB RAM - 512GB SSD 저장 공간 - 12시간 배터리 지속 시간""", metadata={"category": "product_specs"})# 제품 지식을 가진 지원 에이전트 생성support_agent = Agent( role="기술 지원 전문가", goal="정확한 제품 정보와 지원 제공.", backstory="노트북 제품과 사양에 대한 전문가입니다.", knowledge_sources=[product_specs] # 에이전트별 지식)# 제품 지식이 필요한 태스크 생성support_task = Task( description="고객 질문에 답변하세요: {question}", agent=support_agent)# 크루 생성 및 실행crew = Crew( agents=[support_agent], tasks=[support_task])# 노트북 사양에 대한 답변 얻기result = crew.kickoff( inputs={"question": "XPS 13의 저장 용량은 얼마인가요?"})
from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSourceimport requestsfrom datetime import datetimefrom typing import Dict, Anyfrom pydantic import BaseModel, Fieldclass SpaceNewsKnowledgeSource(BaseKnowledgeSource): """Space News API에서 데이터를 가져오는 지식 소스.""" api_endpoint: str = Field(description="API 엔드포인트 URL") limit: int = Field(default=10, description="가져올 기사 수") def load_content(self) -> Dict[Any, str]: """우주 뉴스 기사를 가져와 포맷팅한다.""" try: response = requests.get( f"{self.api_endpoint}?limit={self.limit}" ) response.raise_for_status() data = response.json() articles = data.get('results', []) formatted_data = self._format_articles(articles) return {self.api_endpoint: formatted_data} except Exception as e: raise ValueError(f"우주 뉴스를 가져오는 데 실패: {str(e)}") def _format_articles(self, articles: list) -> str: """기사를 읽기 쉬운 텍스트로 포맷팅한다.""" formatted = "우주 뉴스 기사:\n\n" for article in articles: formatted += f""" 제목: {article['title']} 게시일: {article['published_at']} 요약: {article['summary']} 뉴스 사이트: {article['news_site']} URL: {article['url']} -------------------""" return formatted def add(self) -> None: """기사를 처리하고 저장한다.""" content = self.load_content() for _, text in content.items(): chunks = self._chunk_text(text) self.chunks.extend(chunks) self._save_documents()# 지식 소스 생성recent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles", limit=10,)# 전문가 에이전트 생성space_analyst = Agent( role="우주 뉴스 분석가", goal="우주 뉴스에 대한 질문에 정확하고 포괄적으로 답변한다", backstory="""우주 탐사, 위성 기술, 우주 산업 동향에 전문 지식을 가진 우주 산업 분석가다. 우주 뉴스에 대한 질문에 답변하고 상세하고 정확한 정보를 제공하는 데 뛰어나다.""", knowledge_sources=[recent_news], llm=LLM(model="gpt-4", temperature=0.0))# 사용자 질문을 처리하는 태스크 생성analysis_task = Task( description="우주 뉴스에 대한 이 질문에 답변한다: {user_question}", expected_output="최신 우주 뉴스 기사를 기반으로 한 상세한 답변", agent=space_analyst)# 크루 생성 및 실행crew = Crew( agents=[space_analyst], tasks=[analysis_task], verbose=True, process=Process.sequential)# 사용 예시result = crew.kickoff( inputs={"user_question": "우주 탐사의 최신 동향은 무엇인가?"})
# 더 많은 기사 가져오기recent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles", limit=20, # 기사 수를 늘림)# 검색 파라미터 추가recent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles?search=NASA", # NASA 관련 뉴스 검색 limit=10,)