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)
