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"""
strive:
genai.configure(api_key=self.api_key)
mannequin = genai.GenerativeModel('gemini-1.5-flash')
response = mannequin.generate_content("Hey, it is a take a look at.")
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)}")
elevate
def setup_tools(self):
"""Initialize obtainable instruments"""
self.file_tool = FileReadTool()
print("🛠️ Instruments initialized efficiently!")
def setup_agents(self):
"""Create specialised brokers optimized for Colab"""
self.researcher = Agent(
function="Senior Analysis Analyst",
purpose="Conduct complete analysis and supply detailed insights",
backstory="""You're an knowledgeable analysis analyst with intensive expertise in
gathering, analyzing, and synthesizing info. 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(
function="Information Evaluation Skilled",
purpose="Analyze info and supply statistical insights",
backstory="""You're a expert knowledge analyst who excels at deciphering
complicated info, 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(
function="Content material Technique Skilled",
purpose="Remodel analysis into partaking, accessible content material",
backstory="""You're a artistic content material strategist who excels at
reworking 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(
function="High quality Assurance Specialist",
purpose="Guarantee high-quality, correct, and coherent deliverables",
backstory="""You're a meticulous high quality assurance knowledgeable 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 setting"""
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 = Job(
description=f"""
Analysis the subject: {matter}
Present a complete evaluation together with:
1. Key ideas and definitions
2. Present developments and developments
3. Most important 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 = Job(
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 suggestions",
context=[research_task]
)
content_task = Job(
description=f"""
Create partaking content material about: {matter}
Primarily based on analysis and evaluation, create:
1. Government abstract (2-3 paragraphs)
2. Key takeaways (5-7 bullet factors)
3. Actionable suggestions
4. Future implications
Make it accessible and interesting 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 = Job(
description=f"""
Overview and enhance all content material for: {matter}
Guarantee:
1. Accuracy and consistency
2. Clear construction and circulate
3. Completeness of data
4. Readability and engagement
Present the ultimate polished model.
""",
agent=self.qa_agent,
expected_output="Closing 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 quicker execution"""
quick_research = Job(
description=f"""
Present a fast however thorough evaluation of: {matter}
Embrace:
1. Transient overview and key factors
2. Most important 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 = Job(
description=f"""
Create a abstract report for: {matter}
Format:
1. Government 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 = Job(
description=f"""
Carry out deep evaluation of: {matter}
Give attention to:
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 venture optimized for Colab"""
print(f"n🚀 Beginning Colab AI Agent Venture")
print(f"📋 Subject: {matter}")
print(f"🔧 Job Sort: {task_type}")
print("=" * 60)
start_time = time.time()
strive:
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✅ Venture 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❌ Venture execution failed: {str(e)}")
print("💡 Attempt utilizing 'fast' process kind for quicker 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)
strive:
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 obtainable")
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" Job 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, function, purpose, backstory, max_iter=2):
"""Create a customized agent"""
return Agent(
function=function,
purpose=purpose,
backstory=backstory,
llm=self.llm,
instruments=[self.file_tool],
verbose=True,
allow_delegation=False,
max_iter=max_iter,
reminiscence=True
)
We architect the guts of the workflow: a ColabGeminiAgentSystem class that wires Gemini into LangChain, defines a file-reading instrument, and spawns 4 specialised brokers, analysis, knowledge, content material, and QA, every able to collaborate on duties.
print("🔧 Initializing Colab AI Agent System...")
strive:
agent_system = ColabGeminiAgentSystem(GEMINI_API_KEY)
print("✅ System prepared to be used!")
besides Exception as e:
print(f"❌ System initialization failed: {e}")
print("Please examine your API key and take a look at once more.")
We instantiate the agent system with our API key, waiting for successful 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 reveal 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 Outcome:")
print(result1)
print("n2. Deep Evaluation Instance:")
topic2 = "Sustainable Vitality Options"
result2 = agent_system.execute_colab_project(topic2, task_type="evaluation")
if result2:
print(f"n📋 Deep Evaluation Outcome:")
print(result2)
