المعرفة في CrewAI هي نظام قوي يتيح لوكلاء الذكاء الاصطناعي الوصول إلى مصادر المعلومات الخارجية واستخدامها أثناء مهامهم.
فكّر فيها كمنح وكلائك مكتبة مرجعية يمكنهم الرجوع إليها أثناء العمل.
لمصادر المعرفة المستندة إلى الملفات، تأكد من وضع ملفاتك في مجلد knowledge في جذر مشروعك.
أيضًا، استخدم المسارات النسبية من مجلد knowledge عند إنشاء المصدر.
يوفر CrewAI تجريدًا لعميل RAG محايد بالنسبة للمزود لمتاجر المتجهات. المزود الافتراضي هو ChromaDB، ويتم دعم Qdrant أيضًا. يمكنك التبديل بين المزودين باستخدام أدوات الإعداد.المدعوم حاليًا:
ChromaDB (افتراضي)
Qdrant
Code
نسخ
اسأل الذكاء الاصطناعي
from crewai.rag.config.utils import set_rag_config, get_rag_client, clear_rag_config# ChromaDB (default)from crewai.rag.chromadb.config import ChromaDBConfigset_rag_config(ChromaDBConfig())chromadb_client = get_rag_client()# Qdrantfrom crewai.rag.qdrant.config import QdrantConfigset_rag_config(QdrantConfig())qdrant_client = get_rag_client()# Example operations (same API for any provider)client = qdrant_client # or chromadb_clientclient.create_collection(collection_name="docs")client.add_documents( collection_name="docs", documents=[{"id": "1", "content": "CrewAI enables collaborative AI agents."}],)results = client.search(collection_name="docs", query="collaborative agents", limit=3)clear_rag_config() # optional reset
عميل RAG هذا منفصل عن التخزين المدمج في المعرفة. استخدمه عندما تحتاج إلى تحكم مباشر في متجر المتجهات أو خطوط أنابيب استرجاع مخصصة.
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)result = crew.kickoff(inputs={"question": "What city does John live in and how old is he?"})
تحتاج إلى تثبيت docling لكي يعمل المثال التالي: uv add docling
Code
نسخ
اسأل الذكاء الاصطناعي
from crewai import LLM, Agent, Crew, Process, Taskfrom crewai.knowledge.source.crew_docling_source import CrewDoclingSource# Create a knowledge source from web contentcontent_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],)result = crew.kickoff( inputs={"question": "What is the reward hacking paper about? Be sure to provide sources."})
فهم مستويات المعرفة: يدعم CrewAI المعرفة على مستوى كل من Agent و Crew. يوضح هذا القسم بالضبط كيف يعمل كل منهما، ومتى يتم تهيئتهما، ويعالج المفاهيم الخاطئة الشائعة حول التبعيات.
from crewai import Agent, Task, Crewfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# Agent-specific knowledgeagent_knowledge = StringKnowledgeSource( content="Agent-specific information that only this agent needs")agent = Agent( role="Specialist", goal="Use specialized knowledge", backstory="Expert with specific knowledge", knowledge_sources=[agent_knowledge], embedder={ # Agent can have its own embedder "provider": "openai", "config": {"model": "text-embedding-3-small"} })task = Task( description="Answer using your specialized knowledge", agent=agent, expected_output="Answer based on agent knowledge")# No crew knowledge neededcrew = Crew(agents=[agent], tasks=[task])result = crew.kickoff() # Works perfectly
# Different knowledge for different agentssales_knowledge = StringKnowledgeSource(content="Sales procedures and pricing")tech_knowledge = StringKnowledgeSource(content="Technical documentation")support_knowledge = StringKnowledgeSource(content="Support procedures")sales_agent = Agent( role="Sales Representative", knowledge_sources=[sales_knowledge], embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}})tech_agent = Agent( role="Technical Expert", knowledge_sources=[tech_knowledge], embedder={"provider": "ollama", "config": {"model": "mxbai-embed-large"}})support_agent = Agent( role="Support Specialist", knowledge_sources=[support_knowledge] # Will use crew embedder as fallback)crew = Crew( agents=[sales_agent, tech_agent, support_agent], tasks=[...], embedder={ # Fallback embedder for agents without their own "provider": "google-generativeai", "config": {"model_name": "gemini-embedding-001"} })# Each agent gets only their specific knowledge# Each can use different embedding providers
على عكس الاسترجاع من قاعدة بيانات متجهات باستخدام أداة، فإن الوكلاء المُحمّلين مسبقًا بالمعرفة لن يحتاجوا إلى شخصية أو مهمة استرجاع.
ما عليك سوى إضافة مصادر المعرفة ذات الصلة التي يحتاجها Agent أو Crew للعمل.يمكن إضافة مصادر المعرفة على مستوى Agent أو Crew.
مصادر المعرفة على مستوى Crew سيستخدمها جميع الوكلاء في الفريق.
مصادر المعرفة على مستوى Agent سيستخدمها الوكيل المحدد المُحمّل بالمعرفة.
results_limit: هو عدد المستندات ذات الصلة المُعادة. القيمة الافتراضية هي 3.
score_threshold: هو الحد الأدنى لدرجة اعتبار المستند ذا صلة. القيمة الافتراضية هي 0.35.
فهم تخزين المعرفة: يخزّن CrewAI مصادر المعرفة تلقائيًا في مجلدات خاصة بالمنصة باستخدام ChromaDB للتخزين المتجهي. فهم هذه المواقع والإعدادات الافتراضية يساعد في النشر في بيئة الإنتاج وتصحيح الأخطاء وإدارة التخزين.
import osfrom crewai import Crew# Set custom storage location for all CrewAI dataos.environ["CREWAI_STORAGE_DIR"] = "./my_project_storage"# All knowledge will now be stored in ./my_project_storage/knowledge/crew = Crew( agents=[...], tasks=[...], knowledge_sources=[...])
import osfrom pathlib import Path# Store knowledge in project directoryproject_root = Path(__file__).parentknowledge_dir = project_root / "knowledge_storage"os.environ["CREWAI_STORAGE_DIR"] = str(knowledge_dir)# Now all knowledge will be stored in your project directory
مزود التضمين الافتراضي: يستخدم CrewAI افتراضيًا تضمينات OpenAI (text-embedding-3-small) لتخزين المعرفة، حتى عند استخدام مزودي LLM مختلفين. يمكنك تخصيص هذا بسهولة ليتوافق مع إعدادك.
from crewai import Agent, Crew, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# When using Claude as your LLM...agent = Agent( role="Researcher", goal="Research topics", backstory="Expert researcher", llm=LLM(provider="anthropic", model="claude-3-sonnet") # Using Claude)# CrewAI will still use OpenAI embeddings by default for knowledge# This ensures consistency but may not match your LLM provider preferenceknowledge_source = StringKnowledgeSource(content="Research data...")crew = Crew( agents=[agent], tasks=[...], knowledge_sources=[knowledge_source] # Default: Uses OpenAI embeddings even with Claude LLM)
agent = Agent( role="Researcher", goal="Research topics", backstory="Expert researcher", knowledge_sources=[knowledge_source], embedder={ "provider": "azure", "config": { "api_key": "your-azure-api-key", "model": "text-embedding-ada-002", # change to the model you are using and is deployed in Azure "api_base": "https://your-azure-endpoint.openai.azure.com/", "api_version": "2024-02-01" } })
ينفذ CrewAI آلية إعادة صياغة استعلام ذكية لتحسين استرجاع المعرفة. عندما يحتاج وكيل إلى البحث في مصادر المعرفة، يتم تحويل موجّه المهمة الخام تلقائيًا إلى استعلام بحث أكثر فعالية.
# 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?"
الاستعلام المُعاد صياغته أكثر تركيزًا على الحاجة الأساسية للمعلومات ويزيل التعليمات غير ذات الصلة حول تنسيق المخرجات.
هذه الآلية تلقائية بالكامل ولا تتطلب أي إعداد من المستخدمين. يُستخدم LLM الخاص بالوكيل لتنفيذ إعادة صياغة الاستعلام، لذا فإن استخدام LLM أكثر قدرة يمكن أن يحسّن جودة الاستعلامات المُعاد صياغتها.
يُصدر CrewAI أحداثًا أثناء عملية استرجاع المعرفة يمكنك الاستماع إليها باستخدام نظام الأحداث. تتيح لك هذه الأحداث مراقبة وتصحيح أخطاء وتحليل كيفية استرجاع المعرفة واستخدامها بواسطة وكلائك.
يتيح لك CrewAI إنشاء مصادر معرفة مخصصة لأي نوع من البيانات عن طريق توسيع فئة BaseKnowledgeSource. لنقم بإنشاء مثال عملي يجلب ويعالج مقالات أخبار الفضاء.
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): """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") def load_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)}") def validate_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 formatted def add(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?"})
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource# Create a test knowledge sourcetest_source = StringKnowledgeSource( content="Test knowledge content for debugging", chunk_size=100, # Small chunks for testing chunk_overlap=20)# Check chunking behaviorprint(f"Original content length: {len(test_source.content)}")print(f"Chunk size: {test_source.chunk_size}")print(f"Chunk overlap: {test_source.chunk_overlap}")# Process and inspect chunkstest_source.add()print(f"Number of chunks created: {len(test_source.chunks)}")for i, chunk in enumerate(test_source.chunks[:3]): # Show first 3 chunks print(f"Chunk {i+1}: {chunk[:50]}...")
# Ensure files are in the correct locationfrom crewai.utilities.constants import KNOWLEDGE_DIRECTORYimport osknowledge_dir = KNOWLEDGE_DIRECTORY # Usually "knowledge"file_path = os.path.join(knowledge_dir, "your_file.pdf")if not os.path.exists(file_path): print(f"File not found: {file_path}") print(f"Current working directory: {os.getcwd()}") print(f"Expected knowledge directory: {os.path.abspath(knowledge_dir)}")
أخطاء “عدم تطابق أبعاد التضمين”:
نسخ
اسأل الذكاء الاصطناعي
# This happens when switching embedding providers# Reset knowledge storage to clear old embeddingscrew.reset_memories(command_type='knowledge')# Or use consistent embedding providerscrew = Crew( agents=[...], tasks=[...], knowledge_sources=[...], embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}})