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"""
attempt:
genai.configure(api_key=self.api_key)
mannequin = genai.GenerativeModel('gemini-1.5-flash')
response = mannequin.generate_content("Hey, it is a check.")
print("✅ Gemini API connection profitable!")
self.llm = ChatGoogleGenerativeAI(
mannequin="gemini-1.5-flash",
google_api_key=self.api_key,
temperature=0.7,
convert_system_message_to_human=True
)
besides Exception as e:
print(f"❌ Gemini API setup failed: {str(e)}")
increase
def setup_tools(self):
"""Initialize accessible instruments"""
self.file_tool = FileReadTool()
print("🛠️ Instruments initialized efficiently!")
def setup_agents(self):
"""Create specialised brokers optimized for Colab"""
self.researcher = Agent(
position="Senior Analysis Analyst",
objective="Conduct complete analysis and supply detailed insights",
backstory="""You're an professional analysis analyst with in depth expertise in
gathering, analyzing, and synthesizing data. You excel at figuring out
key developments, patterns, and offering actionable insights.""",
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
reminiscence=True
)
self.data_analyst = Agent(
position="Information Evaluation Skilled",
objective="Analyze data and supply statistical insights",
backstory="""You're a expert knowledge analyst who excels at deciphering
complicated data, figuring out patterns, and creating actionable
suggestions primarily based on data-driven insights.""",
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
reminiscence=True
)
self.content_creator = Agent(
position="Content material Technique Skilled",
objective="Rework analysis into partaking, accessible content material",
backstory="""You're a artistic content material strategist who excels at
remodeling complicated analysis and evaluation into clear, partaking
content material that resonates with goal audiences.""",
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=2,
reminiscence=True
)
self.qa_agent = Agent(
position="High quality Assurance Specialist",
objective="Guarantee high-quality, correct, and coherent deliverables",
backstory="""You're a meticulous high quality assurance professional who ensures
all deliverables meet excessive requirements of accuracy, readability, and coherence.""",
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=1,
reminiscence=True
)
print("🤖 All brokers initialized efficiently!")
def create_colab_tasks(self, matter, task_type="complete"):
"""Create optimized duties for Colab atmosphere"""
if task_type == "complete":
return self._create_comprehensive_tasks(matter)
elif task_type == "fast":
return self._create_quick_tasks(matter)
elif task_type == "evaluation":
return self._create_analysis_tasks(matter)
else:
return self._create_comprehensive_tasks(matter)
def _create_comprehensive_tasks(self, matter):
"""Create complete analysis duties"""
research_task = Activity(
description=f"""
Analysis the subject: {matter}
Present a complete evaluation together with:
1. Key ideas and definitions
2. Present developments and developments
3. Predominant challenges and alternatives
4. Future outlook and implications
Format your response in clear sections with bullet factors.
""",
agent=self.researcher,
expected_output="Structured analysis report with clear sections and key insights"
)
analysis_task = Activity(
description=f"""
Analyze the analysis findings for: {matter}
Present:
1. Key insights and patterns
2. Statistical observations (if relevant)
3. Comparative evaluation
4. Actionable suggestions
5. Threat evaluation
Current findings in a transparent, analytical format.
""",
agent=self.data_analyst,
expected_output="Analytical report with insights and proposals",
context=[research_task]
)
content_task = Activity(
description=f"""
Create partaking content material about: {matter}
Primarily based on analysis and evaluation, create:
1. Govt abstract (2-3 paragraphs)
2. Key takeaways (5-7 bullet factors)
3. Actionable suggestions
4. Future implications
Make it accessible and fascinating for a common viewers.
""",
agent=self.content_creator,
expected_output="Participating, well-structured content material for common viewers",
context=[research_task, analysis_task]
)
qa_task = Activity(
description=f"""
Evaluation and enhance all content material for: {matter}
Guarantee:
1. Accuracy and consistency
2. Clear construction and movement
3. Completeness of data
4. Readability and engagement
Present the ultimate polished model.
""",
agent=self.qa_agent,
expected_output="Remaining polished content material with high quality enhancements",
context=[research_task, analysis_task, content_task]
)
return [research_task, analysis_task, content_task, qa_task]
def _create_quick_tasks(self, matter):
"""Create fast evaluation duties for sooner execution"""
quick_research = Activity(
description=f"""
Present a fast however thorough evaluation of: {matter}
Embody:
1. Temporary overview and key factors
2. Predominant advantages and challenges
3. Present standing and developments
4. Fast suggestions
Preserve it concise however informative.
""",
agent=self.researcher,
expected_output="Concise evaluation with key insights"
)
quick_content = Activity(
description=f"""
Create a abstract report for: {matter}
Format:
1. Govt abstract
2. Key findings (3-5 factors)
3. Suggestions (3-5 factors)
4. Subsequent steps
Make it actionable and clear.
""",
agent=self.content_creator,
expected_output="Clear abstract report with actionable insights",
context=[quick_research]
)
return [quick_research, quick_content]
def _create_analysis_tasks(self, matter):
"""Create analysis-focused duties"""
deep_analysis = Activity(
description=f"""
Carry out deep evaluation of: {matter}
Deal with:
1. Detailed examination of key elements
2. Execs and cons evaluation
3. Comparative analysis
4. Strategic implications
5. Information-driven conclusions
Present thorough analytical insights.
""",
agent=self.data_analyst,
expected_output="Deep analytical report with detailed insights"
)
return [deep_analysis]
def execute_colab_project(self, matter, task_type="complete", save_results=True):
"""Execute challenge optimized for Colab"""
print(f"n🚀 Beginning Colab AI Agent Mission")
print(f"📋 Subject: {matter}")
print(f"🔧 Activity Sort: {task_type}")
print("=" * 60)
start_time = time.time()
attempt:
duties = self.create_colab_tasks(matter, task_type)
if task_type == "fast":
brokers = [self.researcher, self.content_creator]
elif task_type == "evaluation":
brokers = [self.data_analyst]
else:
brokers = [self.researcher, self.data_analyst, self.content_creator, self.qa_agent]
crew = Crew(
brokers=brokers,
duties=duties,
course of=Course of.sequential,
verbose=1,
reminiscence=True,
max_rpm=20
)
consequence = crew.kickoff()
execution_time = time.time() - start_time
print(f"n✅ Mission accomplished in {execution_time:.2f} seconds!")
print("=" * 60)
if save_results:
self._save_results(matter, task_type, consequence, execution_time)
return consequence
besides Exception as e:
print(f"n❌ Mission execution failed: {str(e)}")
print("💡 Strive utilizing 'fast' activity kind for sooner execution")
return None
def _save_results(self, matter, task_type, consequence, execution_time):
"""Save outcomes to historical past"""
result_entry = {
'timestamp': datetime.now().isoformat(),
'matter': matter,
'task_type': task_type,
'execution_time': execution_time,
'consequence': str(consequence)
}
self.results_history.append(result_entry)
attempt:
with open('colab_agent_results.json', 'w') as f:
json.dump(self.results_history, f, indent=2)
print("💾 Outcomes saved to colab_agent_results.json")
besides Exception as e:
print(f"⚠️ Couldn't save outcomes: {e}")
def show_results_history(self):
"""Show outcomes historical past"""
if not self.results_history:
print("📭 No outcomes historical past accessible")
return
print("n📊 Outcomes Historical past:")
print("=" * 50)
for i, entry in enumerate(self.results_history, 1):
print(f"n{i}. Subject: {entry['topic']}")
print(f" Activity Sort: {entry['task_type']}")
print(f" Execution Time: {entry['execution_time']:.2f}s")
print(f" Timestamp: {entry['timestamp']}")
print("-" * 30)
def create_custom_agent(self, position, objective, backstory, max_iter=2):
"""Create a customized agent"""
return Agent(
position=position,
objective=objective,
backstory=backstory,
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=max_iter,
reminiscence=True
)
We architect the center of the workflow: a ColabGeminiAgentSystem class that wires Gemini into LangChain, defines a file-reading device, and spawns 4 specialised brokers, analysis, knowledge, content material, and QA, every able to collaborate on duties.
print("🔧 Initializing Colab AI Agent System...")
attempt:
agent_system = ColabGeminiAgentSystem(GEMINI_API_KEY)
print("✅ System prepared to be used!")
besides Exception as e:
print(f"❌ System initialization failed: {e}")
print("Please test your API key and take a look at once more.")
We instantiate the agent system with our API key, anticipating a hit message that tells us the mannequin handshake and agent initialization all land easily, our framework is formally alive.
def run_quick_examples():
"""Run fast examples to show the system"""
print("n🎯 Fast Begin Examples")
print("=" * 40)
print("n1. Fast Evaluation Instance:")
topic1 = "Machine Studying in Enterprise"
result1 = agent_system.execute_colab_project(topic1, task_type="fast")
if result1:
print(f"n📋 Fast Evaluation Consequence:")
print(result1)
print("n2. Deep Evaluation Instance:")
topic2 = "Sustainable Power Options"
result2 = agent_system.execute_colab_project(topic2, task_type="evaluation")
if result2:
print(f"n📋 Deep Evaluation Consequence:")
print(result2)