from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# Create a knowledge sourcecontent ="Users name is John. He is 30 years old and lives in San Francisco."string_source = StringKnowledgeSource( content=content,)# Create an LLM with a temperature of 0 to ensure deterministic outputsllm = LLM(model="gpt-4o-mini", temperature=0)# Create an agent with the knowledge storeagent = 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],# Enable knowledge by adding the sources here. You can also add more sources to the sources list.)result = crew.kickoff(inputs={"question":"What city does John live in and how old is he?"})
from crewai import LLM, Agent, Crew, Process, Taskfrom crewai.knowledge.source.crew_docling_source import CrewDoclingSource# Create a knowledge sourcecontent_source = CrewDoclingSource( file_paths=["https://lilianweng.github.io/posts/2024-11-28-reward-hacking","https://lilianweng.github.io/posts/2024-07-07-hallucination",],)# Create an LLM with a temperature of 0 to ensure deterministic outputsllm = LLM(model="gpt-4o-mini", temperature=0)# Create an agent with the knowledge storeagent = 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],# Enable knowledge by adding the sources here. You can also add more sources to the sources list.)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# Create a text file knowledge sourcetext_source = TextFileKnowledgeSource( file_paths=["document.txt","another.txt"])# Create crew with text file source on agents or crew levelagent = Agent(... knowledge_sources=[text_source])crew = Crew(... knowledge_sources=[text_source])
from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource# Create a PDF knowledge sourcepdf_source = PDFKnowledgeSource( file_paths=["document.pdf","another.pdf"])# Create crew with PDF knowledge source on agents or crew levelagent = Agent(... knowledge_sources=[pdf_source])crew = Crew(... knowledge_sources=[pdf_source])
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSourcesource = StringKnowledgeSource( content="Your content here", chunk_size=4000,# Maximum size of each chunk (default: 4000) chunk_overlap=200# Overlap between chunks (default: 200))
这是一个使用 Google 的 `text-embedding-004` 模型为知识存储配置嵌入器的示例
from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSourceimport os# Get the GEMINI API keyGEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")# Create a knowledge sourcecontent ="Users name is John. He is 30 years old and lives in San Francisco."string_source = StringKnowledgeSource( content=content,)# Create an LLM with a temperature of 0 to ensure deterministic outputsgemini_llm = LLM( model="gemini/gemini-1.5-pro-002", api_key=GEMINI_API_KEY, temperature=0,)# Create an agent with the knowledge storeagent = 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?"})
# Original task prompttask_prompt ="Answer the following questions about the user's favorite movies: What movie did John watch last week? Format your answer in JSON."# Behind the scenes, this might be rewritten as:rewritten_query ="What movies did John watch last week?"
from crewai import Agent, Task, Crewfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# Create agent-specific knowledge about a productproduct_specs = StringKnowledgeSource( content="""The XPS 13 laptop features:-13.4-inch 4K display- Intel Core i7 processor- 16GB RAM- 512GB SSD storage-12-hour battery life""", metadata={"category":"product_specs"})# Create a support agent with product knowledgesupport_agent = Agent( role="Technical Support Specialist", goal="Provide accurate product information and support.", backstory="You are an expert on our laptop products and specifications.", knowledge_sources=[product_specs]# Agent-specific knowledge)# Create a task that requires product knowledgesupport_task = Task( description="Answer this customer question: {question}", agent=support_agent)# Create and run the crewcrew = Crew( agents=[support_agent], tasks=[support_task])# Get answer about the laptop's specificationsresult = crew.kickoff( inputs={"question":"What is the storage capacity of the XPS 13?"})# Resetting the agent specific knowledge via crew objectcrew.reset_memories(command_type ='agent_knowledge')# Resetting the agent specific knowledge via CLIcrewai reset-memories --agent-knowledge crewai reset-memories -akn
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, FieldclassSpaceNewsKnowledgeSource(BaseKnowledgeSource):"""Knowledge source that fetches data from Space News API.""" api_endpoint:str= Field(description="API endpoint URL") limit:int= Field(default=10, description="Number of articles to fetch")defload_content(self)-> Dict[Any,str]:"""Fetch and format space news articles."""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.validate_content(articles)return{self.api_endpoint: formatted_data}except Exception as e:raise ValueError(f"Failed to fetch space news: {str(e)}")defvalidate_content(self, articles:list)->str:"""Format articles into readable text.""" formatted ="Space News Articles:\n\n"for article in articles: formatted +=f""" Title:{article['title']} Published:{article['published_at']} Summary:{article['summary']} News Site:{article['news_site']} URL:{article['url']}-------------------"""return formatteddefadd(self)->None:"""Process and store the articles.""" content = self.load_content()for _, text in content.items(): chunks = self._chunk_text(text) self.chunks.extend(chunks) self._save_documents()# Create knowledge sourcerecent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles", limit=10,)# Create specialized agentspace_analyst = Agent( role="Space News Analyst", goal="Answer questions about space news accurately and comprehensively", backstory="""You are a space industry analyst with expertise in space exploration, satellite technology,and space industry trends. You excel at answering questions about space news and providing detailed, accurate information.""", knowledge_sources=[recent_news], llm=LLM(model="gpt-4", temperature=0.0))# Create task that handles user questionsanalysis_task = Task( description="Answer this question about space news: {user_question}", expected_output="A detailed answer based on the recent space news articles", agent=space_analyst)# Create and run the crewcrew = Crew( agents=[space_analyst], tasks=[analysis_task], verbose=True, process=Process.sequential)# Example usageresult = crew.kickoff( inputs={"user_question":"What are the latest developments in space exploration?"})
# Fetch more articlesrecent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles", limit=20,# Increase the number of articles)# Add search parametersrecent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles?search=NASA",# Search for NASA news limit=10,)