«`html
Develop a Multi-Tool AI Agent with Secure Python Execution using Riza and Gemini
In this tutorial, we’ll harness Riza’s secure Python execution as the cornerstone of a powerful, tool-augmented AI agent in Google Colab. We will begin with seamless API key management through Colab secrets, environment variables, or hidden prompts to configure your Riza credentials for sandboxed, audit-ready code execution. We will integrate Riza’s ExecPython tool into a LangChain agent alongside Google’s Gemini generative model, define an AdvancedCallbackHandler to capture both tool invocations and Riza execution logs, and build custom utilities for complex math and in-depth text analysis.
Target Audience Analysis
The target audience for this tutorial includes:
- Data scientists and AI developers looking to enhance their projects with secure execution environments.
- Business managers interested in leveraging AI for data processing and analysis.
- Educators and researchers seeking to prototype advanced AI applications.
Common pain points include:
- Concerns about security and data integrity when executing code in cloud environments.
- Challenges in integrating multiple AI tools and frameworks effectively.
- Need for clear documentation and examples to facilitate learning and implementation.
Goals of the audience include:
- Building robust AI applications that can handle complex tasks.
- Ensuring compliance with security standards in AI deployments.
- Gaining practical knowledge of integrating AI tools for enhanced functionality.
Interests may revolve around:
- Latest advancements in AI technology.
- Best practices for secure coding and execution.
- Real-world applications of AI in business contexts.
Preferred communication styles include clear, concise instructions with practical examples and visual aids.
Setting Up Your Environment
We will install and upgrade the core libraries, LangChain Community extensions, Google Gemini integration, Riza’s secure execution package, and dotenv support in Google Colab:
%pip install --upgrade --quiet langchain-community langchain-google-genai rizaio python-dotenvNext, we import standard utilities:
import os
from typing import Dict, Any, List
from datetime import datetime
import json
import getpass
from google.colab import userdataAPI Key Management
We define a setup_api_keys() function to securely retrieve your Google Gemini and Riza API keys:
def setup_api_keys():
try:
os.environ['GOOGLE_API_KEY'] = userdata.get('GOOGLE_API_KEY')
os.environ['RIZA_API_KEY'] = userdata.get('RIZA_API_KEY')
return True
except:
pass
if os.getenv('GOOGLE_API_KEY') and os.getenv('RIZA_API_KEY'):
return True
try:
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
return True
except:
return FalseThe function attempts to load keys from Colab secrets, checks for environment variables, and prompts for manual input if necessary.
Integrating Riza and Gemini
We import Riza’s ExecPython tool alongside LangChain’s core components:
from langchain_community.tools.riza.command import ExecPython
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from langchain.memory import ConversationBufferWindowMemory
from langchain.tools import Tool
from langchain.callbacks.base import BaseCallbackHandlerCallback Handler and Utility Classes
We create an AdvancedCallbackHandler for detailed logging:
class AdvancedCallbackHandler(BaseCallbackHandler):
def __init__(self):
self.execution_log = []
self.start_time = None
self.token_count = 0
def on_agent_action(self, action, **kwargs):
timestamp = datetime.now().strftime("%H:%M:%S")
self.execution_log.append({
"timestamp": timestamp,
"action": action.tool,
"input": str(action.tool_input)[:100] + "..." if len(str(action.tool_input)) > 100 else str(action.tool_input)
})
def on_agent_finish(self, finish, **kwargs):
timestamp = datetime.now().strftime("%H:%M:%S")
def get_execution_summary(self):
return {
"total_actions": len(self.execution_log),
"execution_log": self.execution_log
}We also define MathTool and TextAnalyzer classes for advanced operations:
class MathTool:
@staticmethod
def complex_calculation(expression: str) -> str:
try:
import math
import numpy as np
safe_dict = {
"__builtins__": {},
"abs": abs, "round": round, "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
}
result = eval(expression, safe_dict)
return f"Result: {result}"
except Exception as e:
return f"Math Error: {str(e)}"
class TextAnalyzer:
@staticmethod
def analyze_text(text: str) -> str:
try:
char_freq = {}
for char in text.lower():
if char.isalpha():
char_freq[char] = char_freq.get(char, 0) + 1
words = text.split()
word_count = len(words)
avg_word_length = sum(len(word) for word in words) / max(word_count, 1)
specific_chars = {}
for char in set(text.lower()):
if char.isalpha():
specific_chars[char] = text.lower().count(char)
analysis = {
"total_characters": len(text),
"total_words": word_count,
"average_word_length": round(avg_word_length, 2),
"character_frequencies": dict(sorted(char_freq.items(), key=lambda x: x[1], reverse=True)[:10]),
"specific_character_counts": specific_chars
}
return json.dumps(analysis, indent=2)
except Exception as e:
return f"Analysis Error: {str(e)}"Validating API Keys
We validate API keys before creating agents:
def validate_api_keys():
try:
test_llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
test_llm.invoke("test")
test_tool = ExecPython()
return True
except Exception as e:
return FalseCreating Tools and Initializing the Agent
We instantiate the tools and initialize the Gemini model:
python_tool = ExecPython()
math_tool = Tool(
name="advanced_math",
description="Perform complex mathematical calculations and evaluations",
func=MathTool.complex_calculation
)
text_analyzer_tool = Tool(
name="text_analyzer",
description="Analyze text for character frequencies, word statistics, and specific character counts",
func=TextAnalyzer.analyze_text
)
tools = [python_tool, math_tool, text_analyzer_tool]
llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0.1,
max_tokens=2048,
top_p=0.8,
top_k=40
)Prompt Template and Memory Management
We define a structured prompt template and set up memory management:
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}"),
])
memory = ConversationBufferWindowMemory(k=5, return_messages=True, memory_key="chat_history")
callback_handler = AdvancedCallbackHandler()
agent = create_tool_calling_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
memory=memory,
callbacks=[callback_handler],
max_iterations=10,
early_stopping_method="generate"
)Asking Questions to the Agent
We define a function to ask questions and process results:
def ask_question(question: str) -> Dict[str, Any]:
try:
result = agent_executor.invoke({"input": question})
output = result.get("output", "No output generated")
return {
"question": question,
"answer": output,
"execution_summary": callback_handler.get_execution_summary(),
"success": True
}
except Exception as e:
return {
"question": question,
"error": str(e),
"success": False
}We test the agent with a series of sample questions:
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?"
]
results = []
for question in test_questions:
result = ask_question(question)
results.append(result)Conclusion
By centering the architecture on Riza’s secure execution environment, we’ve created an AI agent that generates insightful responses via Gemini while also running arbitrary Python code in a fully sandboxed, monitored context. The integration of Riza’s ExecPython tool ensures that every computation, from advanced numerical routines to dynamic text analyses, is executed with rigorous security and transparency. With LangChain orchestrating tool calls and a memory buffer maintaining context, we now have a modular framework ready for real-world tasks such as automated data processing, research prototyping, or educational demos.
Check out the Notebook. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 99k+ ML SubReddit and Subscribe to our Newsletter.
«`