In this tutorial, we set up an end-to-end AI agent system powered by CrewAI and Google’s Gemini models. We start by installing all required packages, configuring the Gemini key securely, and then building a suite of specialized agents, including research, data analysis, content creation, and quality assurance, each optimized for rapid, sequential collaboration. With clear utility classes and interactive commands, we streamline everything from quick one-off analyses to comprehensive multi-agent research projects right inside the notebook.
import subprocess
import sys
import os
def install_packages():
"""Install required packages in Colab"""
packages = [
"crewai",
"crewai-tools",
"google-generativeai",
"python-dotenv",
"langchain-google-genai"
]
for package in packages:
try:
print(f"
Installing {package}...")
subprocess.check_call([sys.executable, "-m", "pip", "install", package, "-q"])
print(f"
{package} installed successfully!")
except Exception as e:
print(f"
Failed to install {package}: {e}")
print("
Setting up Google Colab environment...")
install_packages()
print(" All packages installed!")Check out the full Codes
We kick things off by auto-installing CrewAI, Gemini client libraries, and other helpers, ensuring that every dependency is ready within the Colab runtime. As the loop runs, we see each package installed quietly and verify its success before proceeding.
import warnings
warnings.filterwarnings('ignore')
from crewai import Agent, Task, Crew, Process
from crewai_tools import FileReadTool
from langchain_google_genai import ChatGoogleGenerativeAI
import google.generativeai as genai
from google.colab import userdata
import time
import json
from datetime import datetimeCheck out the full Codes
We silence warnings for a cleaner log, import the CrewAI core classes and Gemini wrappers, and pull in utility modules such as time and datetime; this provides us with all the building blocks we’ll utilize throughout the notebook.
def setup_api_key():
"""Setup Gemini API key in Colab"""
try:
api_key = userdata.get('GEMINI_API_KEY')
print(" API key loaded from Colab secrets!")
return api_key
except:
print("
Gemini API key not found in Colab secrets.")
print("Please follow these steps:")
print("1. Go to https://makersuite.google.com/app/apikey")
print("2. Create a free API key")
print("3. In Colab, go to (Secrets) in the left sidebar")
print("4. Add a new secret named 'GEMINI_API_KEY' with your API key")
print("5. Enable notebook access for the secret")
print("6. Re-run this cell")
from getpass import getpass
api_key = getpass("Or enter your Gemini API key here (it will be hidden): ")
return api_key
GEMINI_API_KEY = setup_api_key()
Check out the full Codes
We retrieve our Gemini API key from Colab Secrets, or, if it’s missing, we prompt ourselves to securely paste it. A quick test call confirms the key works, ensuring our LLM is authenticated before any real tasks begin.
class ColabGeminiAgentSystem:
def __init__(self, api_key):
"""Initialize the Colab-optimized Gemini agent system"""
self.api_key = api_key
self.setup_gemini()
self.setup_tools()
self.setup_agents()
self.results_history = []
def setup_gemini(self):
"""Configure Gemini API for Colab"""
try:
genai.configure(api_key=self.api_key)
model = genai.GenerativeModel('gemini-1.5-flash')
response = model.generate_content("Hello, this is a test.")
print(" Gemini API connection successful!")
self.llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
google_api_key=self.api_key,
temperature=0.7,
convert_system_message_to_human=True
)
except Exception as e:
print(f" Gemini API setup failed: {str(e)}")
raise
def setup_tools(self):
"""Initialize available tools"""
self.file_tool = FileReadTool()
print("
Tools initialized successfully!")
def setup_agents(self):
"""Create specialized agents optimized for Colab"""
self.researcher = Agent(
role="Senior Research Analyst",
goal="Conduct comprehensive research and provide detailed insights",
backstory="""You are an expert research analyst with extensive experience in
gathering, analyzing, and synthesizing information. You excel at identifying
key trends, patterns, and providing actionable insights.""",
llm=self.llm,
tools=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
memory=True
)
self.data_analyst = Agent(
role="Data Analysis Expert",
goal="Analyze information and provide statistical insights",
backstory="""You are a skilled data analyst who excels at interpreting
complex information, identifying patterns, and creating actionable
recommendations based on data-driven insights.""",
llm=self.llm,
tools=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
memory=True
)
self.content_creator = Agent(
role="Content Strategy Expert",
goal="Transform research into engaging, accessible content",
backstory="""You are a creative content strategist who excels at
transforming complex research and analysis into clear, engaging
content that resonates with target audiences.""",
llm=self.llm,
tools=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
memory=True
)
self.qa_agent = Agent(
role="Quality Assurance Specialist",
goal="Ensure high-quality, accurate, and coherent deliverables",
backstory="""You are a meticulous quality assurance expert who ensures
all deliverables meet high standards of accuracy, clarity, and coherence.""",
llm=self.llm,
tools=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=1,
memory=True
)
print("
All agents initialized successfully!")
def create_colab_tasks(self, topic, task_type="comprehensive"):
"""Create optimized tasks for Colab environment"""
if task_type == "comprehensive":
return self._create_comprehensive_tasks(topic)
elif task_type == "quick":
return self._create_quick_tasks(topic)
elif task_type == "analysis":
return self._create_analysis_tasks(topic)
else:
return self._create_comprehensive_tasks(topic)
def _create_comprehensive_tasks(self, topic):
"""Create comprehensive research tasks"""
research_task = Task(
description=f"""
Research the topic: {topic}
Provide a comprehensive analysis including:
1. Key concepts and definitions
2. Current trends and developments
3. Main challenges and opportunities
4. Future outlook and implications
Format your response in clear sections with bullet points.
""",
agent=self.researcher,
expected_output="Structured research report with clear sections and key insights"
)
analysis_task = Task(
description=f"""
Analyze the research findings for: {topic}
Provide:
1. Key insights and patterns
2. Statistical observations (if applicable)
3. Comparative analysis
4. Actionable recommendations
5. Risk assessment
Present findings in a clear, analytical format.
""",
agent=self.data_analyst,
expected_output="Analytical report with insights and recommendations",
context=[research_task]
)
content_task = Task(
description=f"""
Create engaging content about: {topic}
Based on research and analysis, create:
1. Executive summary (2-3 paragraphs)
2. Key takeaways (5-7 bullet points)
3. Actionable recommendations
4. Future implications
Make it accessible and engaging for a general audience.
""",
agent=self.content_creator,
expected_output="Engaging, well-structured content for general audience",
context=[research_task, analysis_task]
)
qa_task = Task(
description=f"""
Review and improve all content for: {topic}
Ensure:
1. Accuracy and consistency
2. Clear structure and flow
3. Completeness of information
4. Readability and engagement
Provide the final polished version.
""",
agent=self.qa_agent,
expected_output="Final polished content with quality improvements",
context=[research_task, analysis_task, content_task]
)
return [research_task, analysis_task, content_task, qa_task]
def _create_quick_tasks(self, topic):
"""Create quick analysis tasks for faster execution"""
quick_research = Task(
description=f"""
Provide a quick but thorough analysis of: {topic}
Include:
1. Brief overview and key points
2. Main benefits and challenges
3. Current status and trends
4. Quick recommendations
Keep it concise but informative.
""",
agent=self.researcher,
expected_output="Concise analysis with key insights"
)
quick_content = Task(
description=f"""
Create a summary report for: {topic}
Format:
1. Executive summary
2. Key findings (3-5 points)
3. Recommendations (3-5 points)
4. Next steps
Make it actionable and clear.
""",
agent=self.content_creator,
expected_output="Clear summary report with actionable insights",
context=[quick_research]
)
return [quick_research, quick_content]
def _create_analysis_tasks(self, topic):
"""Create analysis-focused tasks"""
deep_analysis = Task(
description=f"""
Perform deep analysis of: {topic}
Focus on:
1. Detailed examination of key components
2. Pros and cons analysis
3. Comparative evaluation
4. Strategic implications
5. Data-driven conclusions
Provide thorough analytical insights.
""",
agent=self.data_analyst,
expected_output="Deep analytical report with detailed insights"
)
return [deep_analysis]
def execute_colab_project(self, topic, task_type="comprehensive", save_results=True):
"""Execute project optimized for Colab"""
print(f"n Starting Colab AI Agent Project")
print(f"
Topic: {topic}")
print(f"
Task Type: {task_type}")
print("=" * 60)
start_time = time.time()
try:
tasks = self.create_colab_tasks(topic, task_type)
if task_type == "quick":
agents = [self.researcher, self.content_creator]
elif task_type == "analysis":
agents = [self.data_analyst]
else:
agents = [self.researcher, self.data_analyst, self.content_creator, self.qa_agent]
crew = Crew(
agents=agents,
tasks=tasks,
process=Process.sequential,
verbose=1,
memory=True,
max_rpm=20
)
result = crew.kickoff()
execution_time = time.time() - start_time
print(f"n Project completed in {execution_time:.2f} seconds!")
print("=" * 60)
if save_results:
self._save_results(topic, task_type, result, execution_time)
return result
except Exception as e:
print(f"n Project execution failed: {str(e)}")
print("
Try using 'quick' task type for faster execution")
return None
def _save_results(self, topic, task_type, result, execution_time):
"""Save results to history"""
result_entry = {
'timestamp': datetime.now().isoformat(),
'topic': topic,
'task_type': task_type,
'execution_time': execution_time,
'result': str(result)
}
self.results_history.append(result_entry)
try:
with open('colab_agent_results.json', 'w') as f:
json.dump(self.results_history, f, indent=2)
print("
Results saved to colab_agent_results.json")
except Exception as e:
print(f"
Could not save results: {e}")
def show_results_history(self):
"""Display results history"""
if not self.results_history:
print("
No results history available")
return
print("n
Results History:")
print("=" * 50)
for i, entry in enumerate(self.results_history, 1):
print(f"n{i}. Topic: {entry['topic']}")
print(f" Task Type: {entry['task_type']}")
print(f" Execution Time: {entry['execution_time']:.2f}s")
print(f" Timestamp: {entry['timestamp']}")
print("-" * 30)
def create_custom_agent(self, role, goal, backstory, max_iter=2):
"""Create a custom agent"""
return Agent(
role=role,
goal=goal,
backstory=backstory,
llm=self.llm,
tools=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=max_iter,
memory=True
)
We architect the heart of the workflow: a ColabGeminiAgentSystem class that wires Gemini into LangChain, defines a file-reading tool, and spawns four specialized agents, research, data, content, and QA, each ready to collaborate on tasks.
print(" Initializing Colab AI Agent System...")
try:
agent_system = ColabGeminiAgentSystem(GEMINI_API_KEY)
print(" System ready for use!")
except Exception as e:
print(f" System initialization failed: {e}")
print("Please check your API key and try again.")
We instantiate the agent system with our API key, watching for a success message that tells us the model handshake and agent initialization all land smoothly, our framework is officially alive.
def run_quick_examples():
"""Run quick examples to demonstrate the system"""
print("n
Quick Start Examples")
print("=" * 40)
print("n1. Quick Analysis Example:")
topic1 = "Machine Learning in Business"
result1 = agent_system.execute_colab_project(topic1, task_type="quick")
if result1:
print(f"n Quick Analysis Result:")
print(result1)
print("n2. Deep Analysis Example:")
topic2 = "Sustainable Energy Solutions"
result2 = agent_system.execute_colab_project(topic2, task_type="analysis")
if result2:
print(f"n Deep Analysis Result:")
print(result2)Check out the full Codes
We demonstrate the workflow with two lightning-round projects: a “quick” analysis of machine-learning trends and a deeper dive into sustainable energy, printing each result so we can see the agents in action right away.
def interactive_agent_system():
"""Interactive interface for the agent system"""
print("n
Interactive AI Agent System")
print("=" * 40)
print("Available commands:")
print("1. 'research [topic]' - Comprehensive research")
print("2. 'quick [topic]' - Quick analysis")
print("3. 'analyze [topic]' - Deep analysis")
print("4. 'history' - Show results history")
print("5. 'help' - Show this help")
print("6. 'exit' - Exit the system")
print("=" * 40)
while True:
try:
command = input("n Enter command: ").strip().lower()
if command == 'exit':
print("
Goodbye!")
break
elif command == 'help':
print("nAvailable commands:")
print("- research [topic] - Comprehensive research")
print("- quick [topic] - Quick analysis")
print("- analyze [topic] - Deep analysis")
print("- history - Show results history")
print("- exit - Exit the system")
elif command == 'history':
agent_system.show_results_history()
elif command.startswith('research '):
topic = command[9:]
agent_system.execute_colab_project(topic, task_type="comprehensive")
elif command.startswith('quick '):
topic = command[6:]
agent_system.execute_colab_project(topic, task_type="quick")
elif command.startswith('analyze '):
topic = command[8:]
agent_system.execute_colab_project(topic, task_type="analysis")
else:
print(" Unknown command. Type 'help' for available commands.")
except KeyboardInterrupt:
print("n System interrupted. Goodbye!")
break
except Exception as e:
print(f" Error: {e}")Check out the full Codes
We build a mini command-line loop that lets us type “research,” “quick,” or “analyze” followed by a topic to spin up new projects on demand; this turns the notebook into an interactive sandbox without requiring extra coding.
class ColabUtils:
"""Utility functions for Colab"""
@staticmethod
def download_results():
"""Download results file"""
try:
from google.colab import files
files.download('colab_agent_results.json')
print("
Results file downloaded!")
except Exception as e:
print(f" Download failed: {e}")
@staticmethod
def display_formatted_result(result):
"""Display result in formatted way"""
from IPython.display import display, Markdown
if isinstance(result, str):
display(Markdown(f"### AI Agent Resultnn{result}"))
else:
display(Markdown(f"### AI Agent Resultnn{str(result)}"))
@staticmethod
def save_to_drive():
"""Save results to Google Drive"""
try:
from google.colab import drive
drive.mount('/content/drive')
import shutil
shutil.copy('colab_agent_results.json', '/content/drive/MyDrive/agent_results.json')
print(" Results saved to Google Drive!")
except Exception as e:
print(f" Drive save failed: {e}")
Check out the full Codes
We add finishing touches, helpers to download results, pretty-print Markdown summaries, and push our JSON history to Google Drive, so we can share or archive our findings with a single call.
def run_demo():
"""Run a comprehensive demo"""
print("n
AI Agent System Demo")
print("=" * 50)
demo_topics = [
("Artificial Intelligence Ethics", "quick"),
("Climate Change Solutions", "analysis"),
("Future of Work", "comprehensive")
]
for topic, task_type in demo_topics:
print(f"n Demo: {topic} ({task_type})")
result = agent_system.execute_colab_project(topic, task_type)
if result:
ColabUtils.display_formatted_result(result)
time.sleep(2)
print("""
Google Colab AI Agent System Ready!
How to use:
1. **Quick Start**:
```python
result = agent_system.execute_colab_project("Your Topic", task_type="quick")
```
2. **Comprehensive Analysis**:
```python
result = agent_system.execute_colab_project("Your Topic", task_type="comprehensive")
```
3. **Deep Analysis**:
```python
result = agent_system.execute_colab_project("Your Topic", task_type="analysis")
```
4. **Interactive Mode**:
```python
interactive_agent_system()
```
5. **Run Demo**:
```python
run_demo()
```
6. **View History**:
```python
agent_system.show_results_history()
```
7. **Download Results**:
```python
ColabUtils.download_results()
```
**Example Usage**:
```python
# Quick analysis
result = agent_system.execute_colab_project("Machine Learning Trends", "quick")
print(result)
# Show formatted result
ColabUtils.display_formatted_result(result)
```
**Tips for Colab**:
- Use "quick" task type for faster execution
- Results are automatically saved
- Use ColabUtils for better formatting
- Download results before closing the session
**Troubleshooting**:
- If you get API errors, check your rate limits
- For memory issues, restart runtime
- Use quick tasks for better performance
""")
We script a showcase that cycles through three topics and task types, displaying formatted outputs between short pauses; this proves the system scales from rapid briefs to full, multi-agent studies.
In conclusion, we have a fully operational, reusable framework that lets us spin up research pipelines, generate polished outputs, and store our results with just a few commands. We can now run quick tests, deep dives, or interactive sessions on any topic, download the findings, and even mount them to Google Drive.
Check out the full Codes. All credit for this research goes to the researchers of this project. Ready to connect with 1 Million+ AI Devs/Engineers/Researchers? See how NVIDIA, LG AI Research, and top AI companies leverage MarkTechPost to reach their target audience [Learn More]
The post A Coding Implementation to Build a Multi-Agent Research and Content Pipeline with CrewAI and Gemini appeared first on MarkTechPost.
