En este tutorial, lo haremos RizaLa ejecución segura de Python como la piedra angular de un poderoso agente de inteligencia artificial acuático en Google Colab. Comenzando con la administración de claves de la API perfecta, a través de secretos de Colab, variables de entorno o indicaciones ocultas, configuraremos sus credenciales de Riza para habilitar la ejecución de código de sandboxed y listo para auditoría. Integraremos la herramienta Execpython de Riza en un agente de Langchain junto con el modelo generativo de Géminis de Google, definiremos un avanzado CallbackHandler que captura tanto las invocaciones de herramientas como los registros de ejecución de RIZA, y construir utilidades personalizadas para matemáticas complejas y análisis de texto en profundidad.
%pip install --upgrade --quiet langchain-community langchain-google-genai rizaio python-dotenv
import os
from typing import Dict, Any, List
from datetime import datetime
import json
import getpass
from google.colab import userdataInstalaremos y actualizaremos las bibliotecas principales, las extensiones de la comunidad de Langchain, la integración de Google Gemini, el paquete de ejecución seguro de Riza y el soporte Dotenv, en silencio en Colab. Luego importamos utilidades estándar (por ejemplo, OS, DateTime, JSON), anotaciones de escritura, entrada segura a través de GetPass y la API de datos de usuario de Colab para administrar variables de entorno y secretos de los usuarios sin problemas.
def setup_api_keys():
"""Set up API keys using multiple secure methods."""
try:
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")
return True
except:
pass
if os.getenv('GOOGLE_API_KEY') and os.getenv('RIZA_API_KEY'):
print("✅ API keys found in environment")
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
print("✅ API keys set securely via input")
return True
except:
print("❌ Failed to set API keys")
return False
if not setup_api_keys():
print("⚠️ Please set up your API keys using one of these methods:")
print(" 1. Colab Secrets: Go to 🔑 in left panel, add GOOGLE_API_KEY and RIZA_API_KEY")
print(" 2. Environment: Set GOOGLE_API_KEY and RIZA_API_KEY before running")
print(" 3. Manual input: Run the cell and enter keys when prompted")
exit()La celda anterior define una función setup_api_keys () que recupera de forma segura sus claves de la API de Google Gemini y Riza al intentar cargarlas primero de los secretos de Colab, y luego retroceder a las variables de entorno existentes, y finalmente le pide que las ingrese a través de la entrada oculta si es necesario. Si ninguno de estos métodos tiene éxito, imprime instrucciones sobre cómo proporcionar sus claves y sale del cuaderno.
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 BaseCallbackHandlerWe import Riza’s ExecPython tool alongside LangChain’s core components for building a tool‐calling agent, namely the Gemini LLM wrapper (ChatGoogleGenerativeAI), the agent executor and creation functions (AgentExecutor, create_tool_calling_agent), the prompt and message templates, conversation memory buffer, generic Tool wrapper, and the base callback handler for logging and Monitoreo de acciones del agente. Estos bloques de construcción le permiten ensamblar, configurar y rastrear un agente de IA múltiple habilitado para memoria en 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, 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)
})
print(f"🔧 [{timestamp}] Using tool: {action.tool}")
def on_agent_finish(self, finish, **kwargs):
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"✅ [{timestamp}] Agent completed successfully")
def get_execution_summary(self):
return {
"total_actions": len(self.execution_log),
"execution_log": self.execution_log
}
class MathTool:
"""Advanced mathematical operations tool."""
@staticmethod
def complex_calculation(expression: str) -> str:
"""Evaluate complex mathematical expressions safely."""
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:
"""Advanced text analysis tool."""
@staticmethod
def analyze_text(text: str) -> str:
"""Perform comprehensive text analysis."""
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)}"Above Cell reúne tres piezas esenciales: un avanzado CallbackHandler que captura cada invocación de herramientas con un registro de tiempo de tiempo y puede resumir las acciones totales tomadas; una clase de Mathtool que evalúa de manera segura expresiones matemáticas complejas en un entorno restringido para evitar operaciones no deseadas; y una clase Textanalyzer que calcula estadísticas de texto detalladas, como frecuencias de caracteres, recuentos de palabras y longitud promedio de palabras, y devuelve los resultados como JSON formateado.
def validate_api_keys():
"""Validate API keys before creating agents."""
try:
test_llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0
)
test_llm.invoke("test")
print("✅ Gemini API key validated")
test_tool = ExecPython()
print("✅ Riza API key validated")
return True
except Exception as e:
print(f"❌ API key validation failed: {str(e)}")
print("Please check your API keys and try again")
return False
if not validate_api_keys():
exit()
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]
try:
llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0.1,
max_tokens=2048,
top_p=0.8,
top_k=40
)
print("✅ Gemini model initialized successfully")
except Exception as e:
print(f"⚠️ Gemini Pro failed, falling back to Flash: {e}")
llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0.1,
max_tokens=2048
)
En esta celda, primero definimos y ejecutamos Validate_api_Keys () para garantizar que tanto las credenciales de Gemini como Riza funcionen, intentando una llamada de LLM ficticia e instanciando la herramienta Riza Execpython. Salimos del cuaderno si la validación falla. Luego creamos instanciar a Python_Tool para la ejecución de código seguro, envolver nuestros métodos MathTool y TextAnalyzer en objetos de herramienta Langchain y recopilarlos en la lista de herramientas. Finalmente, inicializamos el modelo Gemini con configuraciones personalizadas (temperatura, max_tokens, top_p, top_k), y si la configuración “pro” falla, nos recurrimos con gracia a la variante “flash” más ligera.
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"
)
Esta célula construye el “cerebro” y el flujo de trabajo del agente: define una estructura de chatpromptTtemplate que instruye al sistema en su conjunto de herramientas y estilo de razonamiento, configura una memoria de conversación de ventana deslizante para retener los últimos cinco intercambios e instanciar el avanzado y el control de tiempo real para el registro en tiempo real. Luego crea un agente de reclutamiento de herramientas al unir el Gemini LLM, las herramientas personalizadas y la plantilla de inmediato, y lo envuelve en un Agente Executor que administra la ejecución (hasta diez pasos), aprovecha la memoria para el contexto, transmite la salida detallada y se detiene limpiamente una vez que el agente genera una respuesta final.
def ask_question(question: str) -> Dict[str, Any]:
"""Ask a question to the advanced agent and return detailed results."""
print(f"\n🤖 Processing: {question}")
print("=" * 50)
try:
result = agent_executor.invoke({"input": question})
output = result.get("output", "No output generated")
print("\n📊 Execution Summary:")
summary = callback_handler.get_execution_summary()
print(f"Tools used: {summary['total_actions']}")
return {
"question": question,
"answer": output,
"execution_summary": summary,
"success": True
}
except Exception as e:
print(f"❌ Error: {str(e)}")
return {
"question": question,
"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("🚀 Advanced Gemini Agent with Riza - Ready!")
print("🔐 API keys configured securely")
print("Testing with sample questions...\n")
results = []
for question in test_questions:
result = ask_question(question)
results.append(result)
print("\n" + "="*80 + "\n")
print("📈 FINAL SUMMARY:")
successful = sum(1 for r in results if r["success"])
print(f"Successfully processed: {successful}/{len(results)} questions")Finalmente, definimos una función de ayuda, ask_question (), que envía una consulta de usuario al ejecutor del agente, imprime el encabezado de preguntas, captura la respuesta del agente (o error) y luego genera un breve resumen de ejecución (que muestra cuántas llamadas de herramientas se hicieron). Luego suministra una lista de preguntas de muestra, que cubre los caracteres de conteo, calcula el interés compuesto, analiza la frecuencia de las palabras y la generación de una secuencia de Fibonacci, e itera a través de ellos, invocando al agente en cada uno y recolectando los resultados. Después de ejecutar todas las pruebas, imprime un “resumen final” conciso que indica cuántas consultas se procesaron con éxito, confirmando que su agente Gemini + Riza avanzado está en funcionamiento en Colab.
En conclusión, al centrar la arquitectura en el entorno de ejecución segura de Riza, hemos creado un agente de IA que genera respuestas perspicaces a través de Gemini al tiempo que ejecuta el código de Python arbitrario en un contexto monitoreado totalmente sandboxed. La integración de la herramienta Execpython de Riza garantiza que cada cálculo, desde rutinas numéricas avanzadas hasta análisis de texto dinámicos, se ejecute con rigurosa seguridad y transparencia. Con las llamadas de herramientas de orquestación de Langchain y un contexto de mantenimiento de búfer de memoria, ahora tenemos un marco modular listo para tareas del mundo real, como procesamiento de datos automatizado, creación de prototipos de investigación o demostraciones educativas.
Mira el Computadora portátil. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro 99k+ ml de subreddit y suscribirse a Nuestro boletín.
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.
