On this tutorial, we’ll harness Riza’s safe Python execution because the cornerstone of a robust, tool-augmented AI agent in Google Colab. Starting with seamless API key administration, via Colab secrets and techniques, surroundings variables, or hidden prompts, we’ll configure your Riza credentials to allow sandboxed, audit-ready code execution. We’ll combine Riza’s ExecPython software right into a LangChain agent alongside Google’s Gemini generative mannequin, outline an AdvancedCallbackHandler that captures each software invocations and Riza execution logs, and construct customized utilities for complicated math and in-depth textual content evaluation.
%pip set up --upgrade --quiet langchain-community langchain-google-genai rizaio python-dotenv
import os
from typing import Dict, Any, Record
from datetime import datetime
import json
import getpass
from google.colab import userdataWe are going to set up and improve the core libraries, LangChain Neighborhood extensions, Google Gemini integration, Riza’s safe execution package deal, and dotenv help, quietly in Colab. We then import normal utilities (e.g., os, datetime, json), typing annotations, safe enter through getpass, and Colab’s person information API to handle surroundings variables and person secrets and techniques seamlessly.
def setup_api_keys():
"""Arrange API keys utilizing a number of safe strategies."""
strive:
os.environ['GOOGLE_API_KEY'] = userdata.get('GOOGLE_API_KEY')
os.environ['RIZA_API_KEY'] = userdata.get('RIZA_API_KEY')
print("✅ API keys loaded from Colab secrets and techniques")
return True
besides:
move
if os.getenv('GOOGLE_API_KEY') and os.getenv('RIZA_API_KEY'):
print("✅ API keys present in surroundings")
return True
strive:
if not os.getenv('GOOGLE_API_KEY'):
google_key = getpass.getpass("🔑 Enter your Google Gemini API key: ")
os.environ['GOOGLE_API_KEY'] = google_key
if not os.getenv('RIZA_API_KEY'):
riza_key = getpass.getpass("🔑 Enter your Riza API key: ")
os.environ['RIZA_API_KEY'] = riza_key
print("✅ API keys set securely through enter")
return True
besides:
print("❌ Did not set API keys")
return False
if not setup_api_keys():
print("⚠️ Please arrange your API keys utilizing considered one of these strategies:")
print(" 1. Colab Secrets and techniques: Go to 🔑 in left panel, add GOOGLE_API_KEY and RIZA_API_KEY")
print(" 2. Atmosphere: Set GOOGLE_API_KEY and RIZA_API_KEY earlier than working")
print(" 3. Guide enter: Run the cell and enter keys when prompted")
exit()The above cell defines a setup_api_keys() operate that securely retrieves your Google Gemini and Riza API keys by first making an attempt to load them from Colab secrets and techniques, then falling again to current surroundings variables, and at last prompting you to enter them through hidden enter if wanted. If none of those strategies succeed, it prints directions on easy methods to present your keys and exits the pocket book.
from langchain_community.instruments.riza.command import ExecPython
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.brokers import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from langchain.reminiscence import ConversationBufferWindowMemory
from langchain.instruments import Instrument
from langchain.callbacks.base import BaseCallbackHandlerWe import Riza’s ExecPython software alongside LangChain’s core elements for constructing a software‐calling agent, particularly the Gemini LLM wrapper (ChatGoogleGenerativeAI), the agent executor and creation capabilities (AgentExecutor, create_tool_calling_agent), the immediate and message templates, dialog reminiscence buffer, generic Instrument wrapper, and the bottom callback handler for logging and monitoring agent actions. These constructing blocks allow you to assemble, configure, and monitor a memory-enabled, multi-tool AI agent in Colab.
class AdvancedCallbackHandler(BaseCallbackHandler):
"""Enhanced callback handler for detailed logging and metrics."""
def __init__(self):
self.execution_log = []
self.start_time = None
self.token_count = 0
def on_agent_action(self, motion, **kwargs):
timestamp = datetime.now().strftime("%H:%M:%S")
self.execution_log.append({
"timestamp": timestamp,
"motion": motion.software,
"enter": str(motion.tool_input)[:100] + "..." if len(str(motion.tool_input)) > 100 else str(motion.tool_input)
})
print(f"🔧 [{timestamp}] Utilizing software: {motion.software}")
def on_agent_finish(self, end, **kwargs):
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"✅ [{timestamp}] Agent accomplished efficiently")
def get_execution_summary(self):
return {
"total_actions": len(self.execution_log),
"execution_log": self.execution_log
}
class MathTool:
"""Superior mathematical operations software."""
@staticmethod
def complex_calculation(expression: str) -> str:
"""Consider complicated mathematical expressions safely."""
strive:
import math
import numpy as np
safe_dict = {
"__builtins__": {},
"abs": abs, "spherical": spherical, "min": min, "max": max,
"sum": sum, "len": len, "pow": pow,
"math": math, "np": np,
"sin": math.sin, "cos": math.cos, "tan": math.tan,
"log": math.log, "sqrt": math.sqrt, "pi": math.pi, "e": math.e
}
consequence = eval(expression, safe_dict)
return f"Consequence: {consequence}"
besides Exception as e:
return f"Math Error: {str(e)}"
class TextAnalyzer:
"""Superior textual content evaluation software."""
@staticmethod
def analyze_text(textual content: str) -> str:
"""Carry out complete textual content evaluation."""
strive:
char_freq = {}
for char in textual content.decrease():
if char.isalpha():
char_freq[char] = char_freq.get(char, 0) + 1
phrases = textual content.cut up()
word_count = len(phrases)
avg_word_length = sum(len(phrase) for phrase in phrases) / max(word_count, 1)
specific_chars = {}
for char in set(textual content.decrease()):
if char.isalpha():
specific_chars[char] = textual content.decrease().depend(char)
evaluation = {
"total_characters": len(textual content),
"total_words": word_count,
"average_word_length": spherical(avg_word_length, 2),
"character_frequencies": dict(sorted(char_freq.objects(), key=lambda x: x[1], reverse=True)[:10]),
"specific_character_counts": specific_chars
}
return json.dumps(evaluation, indent=2)
besides Exception as e:
return f"Evaluation Error: {str(e)}"Above cell brings collectively three important items: an AdvancedCallbackHandler that captures each software invocation with a timestamped log and might summarize the entire actions taken; a MathTool class that safely evaluates complicated mathematical expressions in a restricted surroundings to forestall undesirable operations; and a TextAnalyzer class that computes detailed textual content statistics, resembling character frequencies, phrase counts, and common phrase size, and returns the outcomes as formatted JSON.
def validate_api_keys():
"""Validate API keys earlier than creating brokers."""
strive:
test_llm = ChatGoogleGenerativeAI(
mannequin="gemini-1.5-flash",
temperature=0
)
test_llm.invoke("take a look at")
print("✅ Gemini API key validated")
test_tool = ExecPython()
print("✅ Riza API key validated")
return True
besides Exception as e:
print(f"❌ API key validation failed: {str(e)}")
print("Please test your API keys and check out once more")
return False
if not validate_api_keys():
exit()
python_tool = ExecPython()
math_tool = Instrument(
identify="advanced_math",
description="Carry out complicated mathematical calculations and evaluations",
func=MathTool.complex_calculation
)
text_analyzer_tool = Instrument(
identify="text_analyzer",
description="Analyze textual content for character frequencies, phrase statistics, and particular character counts",
func=TextAnalyzer.analyze_text
)
instruments = [python_tool, math_tool, text_analyzer_tool]
strive:
llm = ChatGoogleGenerativeAI(
mannequin="gemini-1.5-flash",
temperature=0.1,
max_tokens=2048,
top_p=0.8,
top_k=40
)
print("✅ Gemini mannequin initialized efficiently")
besides Exception as e:
print(f"⚠️ Gemini Professional failed, falling again to Flash: {e}")
llm = ChatGoogleGenerativeAI(
mannequin="gemini-1.5-flash",
temperature=0.1,
max_tokens=2048
)
On this cell, we first outline and run validate_api_keys() to make sure that each the Gemini and Riza credentials work, making an attempt a dummy LLM name and instantiating the Riza ExecPython software. We exit the pocket book if validation fails. We then instantiate python_tool for safe code execution, wrap our MathTool and TextAnalyzer strategies into LangChain Instrument objects, and gather them into the instruments checklist. Lastly, we initialize the Gemini mannequin with customized settings (temperature, max_tokens, top_p, top_k), and if the “Professional” configuration fails, we gracefully fall again to the lighter “Flash” variant.
prompt_template = ChatPromptTemplate.from_messages([
("system", """You are an advanced AI assistant with access to powerful tools.
Key capabilities:
- Python code execution for complex computations
- Advanced mathematical operations
- Text analysis and character counting
- Problem decomposition and step-by-step reasoning
Instructions:
1. Always break down complex problems into smaller steps
2. Use the most appropriate tool for each task
3. Verify your results when possible
4. Provide clear explanations of your reasoning
5. For text analysis questions (like counting characters), use the text_analyzer tool first, then verify with Python if needed
Be precise, thorough, and helpful."""),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
reminiscence = ConversationBufferWindowMemory(
okay=5,
return_messages=True,
memory_key="chat_history"
)
callback_handler = AdvancedCallbackHandler()
agent = create_tool_calling_agent(llm, instruments, prompt_template)
agent_executor = AgentExecutor(
agent=agent,
instruments=instruments,
verbose=True,
reminiscence=reminiscence,
callbacks=[callback_handler],
max_iterations=10,
early_stopping_method="generate"
)
This cell constructs the agent’s “mind” and workflow: it defines a structured ChatPromptTemplate that instructs the system on its toolset and reasoning fashion, units up a sliding-window dialog reminiscence to retain the final 5 exchanges, and instantiates the AdvancedCallbackHandler for real-time logging. It then creates a software‐calling agent by binding the Gemini LLM, customized instruments, and immediate template, and wraps it in an AgentExecutor that manages execution (as much as ten steps), leverages reminiscence for context, streams verbose output, and halts cleanly as soon as the agent generates a closing response.
def ask_question(query: str) -> Dict[str, Any]:
"""Ask a query to the superior agent and return detailed outcomes."""
print(f"n🤖 Processing: {query}")
print("=" * 50)
strive:
consequence = agent_executor.invoke({"enter": query})
output = consequence.get("output", "No output generated")
print("n📊 Execution Abstract:")
abstract = callback_handler.get_execution_summary()
print(f"Instruments used: {abstract['total_actions']}")
return {
"query": query,
"reply": output,
"execution_summary": abstract,
"success": True
}
besides Exception as e:
print(f"❌ Error: {str(e)}")
return {
"query": query,
"error": str(e),
"success": False
}
test_questions = [
"How many r's are in strawberry?",
"Calculate the compound interest on $1000 at 5% for 3 years",
"Analyze the word frequency in the sentence: 'The quick brown fox jumps over the lazy dog'",
"What's the fibonacci sequence up to the 10th number?"
]
print("🚀 Superior Gemini Agent with Riza - Prepared!")
print("🔐 API keys configured securely")
print("Testing with pattern questions...n")
outcomes = []
for query in test_questions:
consequence = ask_question(query)
outcomes.append(consequence)
print("n" + "="*80 + "n")
print("📈 FINAL SUMMARY:")
profitable = sum(1 for r in outcomes if r["success"])
print(f"Efficiently processed: {profitable}/{len(outcomes)} questions")Lastly, we outline a helper operate, ask_question(), that sends a person question to the agent executor, prints the query header, captures the agent’s response (or error), after which outputs a quick execution abstract (displaying what number of software calls have been made). It then provides a listing of pattern questions, protecting counting characters, computing compound curiosity, analyzing phrase frequency, and producing a Fibonacci sequence, and iterates via them, invoking the agent on every and amassing the outcomes. After working all checks, it prints a concise “FINAL SUMMARY” indicating what number of queries have been processed efficiently, confirming that your Superior Gemini + Riza agent is up and working in Colab.
In conclusion, by centering the structure on Riza’s safe execution surroundings, we’ve created an AI agent that generates insightful responses through Gemini whereas additionally working arbitrary Python code in a completely sandboxed, monitored context. The mixing of Riza’s ExecPython software ensures that each computation, from superior numerical routines to dynamic textual content analyses, is executed with rigorous safety and transparency. With LangChain orchestrating software calls and a reminiscence buffer sustaining context, we now have a modular framework prepared for real-world duties resembling automated information processing, analysis prototyping, or instructional demos.
Take a look at the Pocket book. All credit score for this analysis goes to the researchers of this venture. Additionally, be happy to observe us on Twitter and don’t neglect to affix our 99k+ ML SubReddit and Subscribe to our Publication.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its recognition amongst audiences.
