CrewAI is a framework for orchestrating role-playing, autonomous AI agents. It enables multiple agents with specialized roles to collaborate on complex tasks through structured workflows and task delegation.
Agents are specialized team members with specific roles, goals, and backstories:
from crewai import Agent, LLMimport os# Create a researcher agentresearcher = Agent( role='Senior Researcher', goal='Discover groundbreaking technologies', backstory=( 'A curious mind fascinated by cutting-edge innovation ' 'and the potential to change the world, you know ' 'everything about tech.' ), verbose=True, llm=LLM( model="nebius/Qwen/Qwen3-235B-A22B", api_key=os.getenv("NEBIUS_API_KEY") ),)
Tasks define specific work to be done, with expected outputs and assigned agents:
from crewai import Taskresearch_task = Task( description='Identify the next big trend in AI', expected_output='5 paragraphs on the next big AI trend', agent=researcher, # Assign to specific agent)
Combine CrewAI with RAG for knowledge-grounded agents:
1
Create Vector Store Tool
from langchain_qdrant import QdrantVectorStorefrom langchain_community.embeddings import HuggingFaceEmbeddingsfrom crewai import Toolimport os# Setup Qdrantembeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2")vector_store = QdrantVectorStore( url=os.getenv("QDRANT_URL"), api_key=os.getenv("QDRANT_API_KEY"), collection_name="documents", embedding=embeddings)# Create search tooldef search_knowledge_base(query: str) -> str: """Search the knowledge base for relevant information.""" docs = vector_store.similarity_search(query, k=5) return "\n\n".join([doc.page_content for doc in docs])qdrant_tool = Tool( name="Knowledge Base Search", description="Search internal documents for information", func=search_knowledge_base)
2
Create Web Search Tool
from exa_py import Exaexa_client = Exa(api_key=os.getenv("EXA_API_KEY"))def web_search(query: str) -> str: """Search the web for current information.""" results = exa_client.search_and_contents( query, num_results=3, text=True ) return "\n\n".join([r.text for r in results.results])web_tool = Tool( name="Web Search", description="Search the internet for current information", func=web_search)
3
Create Agents with Tools
researcher = Agent( role='Research Agent', goal='Find relevant information from multiple sources', tools=[qdrant_tool, web_tool], # Multiple tools llm=llm, verbose=True,)analyst = Agent( role='Analysis Agent', goal='Synthesize findings into insights', tools=[qdrant_tool], # Can access knowledge base llm=llm, verbose=True,)
4
Define Task Flow
research_task = Task( description='Research {query} using knowledge base and web search', expected_output='Comprehensive research findings', agent=researcher,)analysis_task = Task( description='Analyze research and provide insights on {query}', expected_output='Executive summary with key insights', agent=analyst,)crew = Crew( agents=[researcher, analyst], tasks=[research_task, analysis_task], process=Process.sequential,)# Run with input variablesresult = crew.kickoff(inputs={"query": "AI agent frameworks"})
# .env fileNEBIUS_API_KEY=your_nebius_api_keyOPENAI_API_KEY=your_openai_api_key # If using OpenAI# For RAG applicationsQDRANT_URL=your_qdrant_urlQDRANT_API_KEY=your_qdrant_api_keyEXA_API_KEY=your_exa_api_key # For web search
# ✓ Good: Specific role and expertiseresearcher = Agent( role='Senior AI Research Analyst', goal='Analyze latest AI research papers and identify trends', backstory='PhD in AI with 10 years analyzing ML research')# ✗ Bad: Vague roleagent = Agent( role='Helper', goal='Help with stuff', backstory='Knows things')
Define Clear Expected Outputs
Be specific about what each task should produce:
task = Task( description='Research AI agent frameworks', expected_output=( '1. Executive summary (2-3 paragraphs)\n' '2. Comparison table of top 5 frameworks\n' '3. Pros and cons for each\n' '4. Recommendation with justification' ), agent=researcher,)
Use Task Context for Dependencies
When tasks depend on previous outputs:
research_task = Task(description='Research topic', ...)analysis_task = Task(description='Analyze findings', ...)writing_task = Task( description='Write article based on research and analysis', context=[research_task, analysis_task], # Access previous outputs agent=writer,)