A Coding Implementation to Build a Multi-Agent Research and Content Pipeline with CrewAI and Gemini

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!")

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 datetime

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() 

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)

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}")

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}") 

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 """)