1. Giriş
Bu codelab'de, Google Agent Development Kit'i (ADK) kullanarak gelişmiş bir yapay zeka aracısı oluşturmayı öğreneceksiniz. Temel bir sohbet aracısıyla başlayıp uzmanlık alanlarına yönelik özellikleri kademeli olarak ekleyerek doğal bir evrimsel yol izleyeceğiz.
Oluşturduğumuz ajan, ADK, Gemini ve Vertex AI kullanılarak sıfırdan geliştirilmiş bir yönetici asistanıdır. Takviminizi yönetme, görevleri hatırlatma, araştırma yapma ve notları derleme gibi günlük görevlerde size yardımcı olmak için tasarlanmıştır.
Bu laboratuvarın sonunda, tam olarak çalışan bir aracıya ve bunu kendi ihtiyaçlarınıza göre genişletmek için gereken bilgiye sahip olacaksınız.
Ön koşullar
- Python programlama diliyle ilgili temel bilgiler
- Bulut kaynaklarını yönetmek için Google Cloud Console hakkında temel bilgiler
Neler öğreneceksiniz?
- Yapay zeka aracıları için Google Cloud altyapısı sağlama.
- Vertex AI Memory Bank'i kullanarak kalıcı uzun süreli bellek uygulama
- Uzmanlaşmış alt aracılardan oluşan bir hiyerarşi oluşturma.
- Harici veritabanlarını ve Google Workspace ekosistemini entegre etme
İhtiyacınız olanlar
Bu atölye çalışması, gerekli tüm bağımlılıklar (gcloud CLI, kod düzenleyici, Go, Gemini CLI) önceden yüklenmiş olarak gelen Google Cloud Shell'de tamamen yapılabilir.
Alternatif olarak, kendi makinenizde çalışmayı tercih ederseniz aşağıdakilere ihtiyacınız olacaktır:
- Python (3.12 veya sonraki sürümler)
- Kod düzenleyici veya IDE (ör. VS Code ya da
vim). - Python ve
gcloudkomutlarını yürütmek için bir terminal. - Önerilen: Gemini CLI veya Antigravity gibi bir kodlama aracısı
Önemli Teknolojiler
Kullanacağımız teknolojiler hakkında daha fazla bilgiyi burada bulabilirsiniz:
2. Ortam Kurulumu
Aşağıdaki seçeneklerden birini belirleyin: Bu codelab'i kendi makinenizde çalıştırmak istiyorsanız Kendi hızınızda ortam kurulumu'nu, bu codelab'i tamamen bulutta çalıştırmak istiyorsanız Cloud Shell'i başlat'ı seçin.
Yönlendirmesiz ortam kurulumu
- Google Cloud Console'da oturum açın ve yeni bir proje oluşturun veya mevcut bir projeyi yeniden kullanın. Gmail veya Google Workspace hesabınız yoksa hesap oluşturmanız gerekir.
- Proje adı, bu projenin katılımcıları için görünen addır. Google API'leri tarafından kullanılmayan bir karakter dizesidir. Bu bilgiyi istediğiniz zaman güncelleyebilirsiniz.
- Proje kimliği, tüm Google Cloud projelerinde benzersizdir ve sabittir (ayarlandıktan sonra değiştirilemez). Cloud Console, benzersiz bir dizeyi otomatik olarak oluşturur. Genellikle bu dizenin ne olduğuyla ilgilenmezsiniz. Çoğu codelab'de proje kimliğinize (genellikle
PROJECT_IDolarak tanımlanır) başvurmanız gerekir. Oluşturulan kimliği beğenmezseniz başka bir rastgele kimlik oluşturabilirsiniz. Dilerseniz kendi adınızı deneyerek kullanılabilir olup olmadığını kontrol edebilirsiniz. Bu adım tamamlandıktan sonra değiştirilemez ve proje süresince geçerli kalır. - Bazı API'lerin kullandığı üçüncü bir değer olan Proje Numarası da vardır. Bu üç değer hakkında daha fazla bilgiyi belgelerde bulabilirsiniz.
- Ardından, Cloud kaynaklarını/API'lerini kullanmak için Cloud Console'da faturalandırmayı etkinleştirmeniz gerekir. Bu codelab'i tamamlamak neredeyse hiç maliyetli değildir. Bu eğitimin ötesinde faturalandırılmayı önlemek için kaynakları kapatmak üzere oluşturduğunuz kaynakları veya projeyi silebilirsiniz. Yeni Google Cloud kullanıcıları 300 ABD doları değerinde ücretsiz deneme programından yararlanabilir.
Cloud Shell'i başlatma
Google Cloud, dizüstü bilgisayarınızdan uzaktan çalıştırılabilir. Ancak bu codelab'de, Cloud'da çalışan bir komut satırı ortamı olan Google Cloud Shell'i kullanacaksınız.
Google Cloud Console'da sağ üstteki araç çubuğunda Cloud Shell simgesini tıklayın:
Ortamın temel hazırlığı ve bağlanması yalnızca birkaç dakikanızı alır. İşlem tamamlandığında aşağıdakine benzer bir sonuç görürsünüz:
Bu sanal makine, ihtiyaç duyacağınız tüm geliştirme araçlarını içerir. 5 GB boyutunda kalıcı bir ana dizin sunar ve Google Cloud üzerinde çalışır. Bu sayede ağ performansı ve kimlik doğrulama önemli ölçüde güçlenir. Bu codelab'deki tüm çalışmalarınızı tarayıcıda yapabilirsiniz. Herhangi bir şey yüklemeniz gerekmez.
3. Proje Ayarları
Kod yazmadan önce Google Cloud'da gerekli altyapıyı ve izinleri sağlamamız gerekir.
Ortam değişkenlerini ayarlama
Terminali açın ve aşağıdaki ortam değişkenlerini ayarlayın:
export PROJECT_ID=`gcloud config get project`
export LOCATION=us-central1
Gerekli API'leri etkinleştirme
Aracınızın çeşitli Google Cloud hizmetlerine erişmesi gerekir. Bu API'leri etkinleştirmek için aşağıdaki kodu çalıştırın:
gcloud services enable \
aiplatform.googleapis.com \
calendar-json.googleapis.com \
sqladmin.googleapis.com
Uygulama Varsayılan Kimlik Bilgileri ile kimlik doğrulama
Ortamınızdaki Google Cloud hizmetleriyle iletişim kurmak için Uygulama Varsayılan Kimlik Bilgileri (ADC) ile kimlik doğrulaması yapmamız gerekir.
Uygulama varsayılan kimlik bilgilerinizin etkin ve güncel olduğundan emin olmak için aşağıdaki komutu çalıştırın:
gcloud auth application-default login
4. Temel aracı oluşturma
Şimdi, proje kaynak kodunu depolayacağımız dizini başlatmamız gerekiyor:
# setup project directory
mkdir -p adk_ea_codelab && cd adk_ea_codelab
# prepare virtual environment
uv init
# install dependencies
uv add google-adk google-api-python-client tzlocal python-dotenv
uv add cloud-sql-python-connector[pg8000] sqlalchemy
Öncelikle aracının kimliğini ve temel sohbet özelliklerini oluştururuz. ADK'da, Agent sınıfı, temsilci karakterini ve talimatlarını tanımlar.
Bu aşamada, temsilci adı belirlemeyi düşünebilirsiniz. Aida veya Sharon gibi uygun adlar vermeyi tercih ediyorum. Bu şekilde, asistanlara biraz "kişilik" kazandırıldığını düşünüyorum. Ancak asistanı yaptığı işe göre de adlandırabilirsiniz (ör. "executive_assistant", "travel_agent" veya "code_executor").
Bir standart aracı oluşturmak için adk create komutunu çalıştırın:
# replace with your desired agent name
uv run adk create executive_assistant
Lütfen model olarak gemini-2.5-flash'ı, arka uç olarak da Vertex AI'ı seçin. Önerilen proje kimliğinin bu laboratuvar için oluşturduğunuz kimlik olduğunu iki kez kontrol edin ve onaylamak için Enter tuşuna basın. Google Cloud bölgesi için varsayılanı (us-central1) kabul edebilirsiniz. Terminaliniz aşağıdaki gibi görünür:
daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$ uv run adk create executive_assistant Choose a model for the root agent: 1. gemini-2.5-flash 2. Other models (fill later) Choose model (1, 2): 1 1. Google AI 2. Vertex AI Choose a backend (1, 2): 2 You need an existing Google Cloud account and project, check out this link for details: https://google.github.io/adk-docs/get-started/quickstart/#gemini---google-cloud-vertex-ai Enter Google Cloud project ID [your-project-id]: Enter Google Cloud region [us-central1]: Agent created in /home/daniela_petruzalek/adk_ea_codelab/executive_assistant: - .env - __init__.py - agent.py daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$
İşlem tamamlandığında önceki komut, temel aracı tanımını içeren bir agent.py dosyası da dahil olmak üzere birkaç dosyanın bulunduğu, aracı adını (ör. executive_assistant) taşıyan bir klasör oluşturur:
from google.adk.agents.llm_agent import Agent
root_agent = Agent(
model='gemini-2.5-flash',
name='root_agent',
description='A helpful assistant for user questions.',
instruction='Answer user questions to the best of your knowledge',
)
Bu aracıyla etkileşim kurmak istiyorsanız komut satırında uv run adk web komutunu çalıştırıp tarayıcınızda geliştirme kullanıcı arayüzünü açarak bunu yapabilirsiniz. Şuna benzer bir şey görürsünüz:
$ uv run adk web ... INFO: Started server process [1244] INFO: Waiting for application startup. +-----------------------------------------------------------------------------+ | ADK Web Server started | | | | For local testing, access at http://127.0.0.1:8000. | +-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Bu aracı oldukça basit olsa da kurulumun düzgün çalıştığından emin olmak için aracı düzenlemeye başlamadan önce bu işlemi en az bir kez yapmanız faydalıdır. Aşağıdaki ekran görüntüsünde, geliştirme kullanıcı arayüzü kullanılarak yapılan basit bir etkileşim gösterilmektedir:
Şimdi de yönetici asistanı kişiliğimizle ajan tanımını değiştirelim. Aşağıdaki kodu kopyalayın ve agent.py içeriğini bu kodla değiştirin. Temsilci adını ve kişiliğini tercihlerinize göre uyarlayın.
from google.adk.agents.llm_agent import Agent
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant',
instruction='''
You are an elite, warm, and highly efficient AI partner.
Your primary goal is to help the user manage their tasks, schedule, and research.
Always be direct, concise, and high-signal.
''',
)
Ad özelliğinin, aracının dahili adını tanımladığını, talimatlarda ise son kullanıcıyla etkileşimlerde aracının kişiliğinin bir parçası olarak daha kolay anlaşılır bir ad verebileceğinizi unutmayın. Dahili ad, çoğunlukla transfer_to_agent aracı kullanılarak çok agent'lı sistemlerde gözlemlenebilirlik ve devir teslim için kullanılır. Bu araçla kendiniz ilgilenmezsiniz. ADK, bir veya daha fazla alt aracı beyan ettiğinizde aracı otomatik olarak kaydeder.
Yeni oluşturduğumuz aracı çalıştırmak için adk web'i kullanın:
uv run adk web
Tarayıcıda ADK kullanıcı arayüzünü açın ve yeni asistanınıza merhaba deyin.
5. Vertex AI Memory Bank ile kalıcı bellek ekleme
Gerçek bir asistan, sorunsuz ve kişiselleştirilmiş bir deneyim sunmak için tercihleri ve geçmiş etkileşimleri hatırlamalıdır. Bu adımda, kullanıcı sohbetlerine dayalı olarak uzun süreli anıları dinamik olarak oluşturan bir Vertex AI özelliği olan Vertex AI Agent Engine Memory Bank'ı entegre edeceğiz.
Bellek Bankası, temsilcinizin birden fazla oturumda erişilebilen kişiselleştirilmiş bilgiler oluşturmasına olanak tanıyarak oturumlar arası süreklilik sağlar. Bu sistem, oturumdaki mesajların kronolojik sırasını yönetir ve mevcut bağlam için en alakalı anıları temsilciye sağlamak üzere benzerlik arama alımını kullanabilir.
Memory Hizmeti'ni başlatma
ADK, uzun süreli anıları depolamak ve almak için Vertex AI'ı kullanır. Projenizde bir "Memory Engine" başlatmanız gerekir. Bu, esasen bir Memory Bank olarak çalışacak şekilde yapılandırılmış bir Reasoning Engine örneğidir.
Aşağıdaki komut dosyasını setup_memory.py olarak oluşturun:
setup_memory.py
import vertexai
import os
PROJECT_ID=os.getenv("PROJECT_ID")
LOCATION=os.getenv("LOCATION")
client = vertexai.Client(project=PROJECT_ID, location=LOCATION)
# Create Reasoning Engine for Memory Bank
agent_engine = client.agent_engines.create()
# You will need this resource name to give it to ADK
print(agent_engine.api_resource.name)
Şimdi setup_memory.py komutunu çalıştırarak bellek bankası için akıl yürütme motorunu sağlayın:
uv run python setup_memory.py
Elde ettiğiniz çıkış şuna benzer şekilde görünecektir:
$ uv run python setup.py projects/1234567890/locations/us-central1/reasoningEngines/1234567890
Motor kaynağı adını bir ortam değişkenine kaydedin:
export ENGINE_ID="<insert the resource name above>"
Şimdi kodu, kalıcı belleği kullanacak şekilde güncellememiz gerekiyor. agent.py içeriğini aşağıdakiyle değiştirin:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with Persistent Memory',
instruction='''
You are an elite AI partner with long-term memory.
Use load_memory to find context about the user when needed.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
after_agent_callback=auto_save_session_to_memory_callback,
)
PreloadMemoryTool, geçmiş sohbetlerdeki alakalı bağlamı her isteğe otomatik olarak ekler (benzerlik aramasıyla alma yöntemini kullanarak). load_memory_tool ise gerektiğinde modelin Memory Bank'a açıkça bilgi sorgusu göndermesine olanak tanır. Bu kombinasyon, temsilcinize derin ve kalıcı bir bağlam sağlar.
Şimdi, hafıza desteğiyle temsilcinizi başlatmak için adk web'i çalıştırırken memory_service_uri iletmeniz gerekir:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Aracıya kendinizle ilgili birkaç bilgi verip farklı bir oturumda bu bilgiler hakkında soru sormayı deneyin. Örneğin, adınızı söyleyin:
Aracının bulut konsoluna kaydettiği anıları inceleyebilirsiniz. "Agent Engine" ürün sayfasına gidin (arama çubuğunu kullanın).
Ardından aracı motorunuzun adını tıklayın (doğru bölgeyi seçtiğinizden emin olun):
Ardından anılar sekmesine gidin:
Eklenen bazı anıları görmeniz gerekir.
6. Web araştırması özellikleri ekleme
Temsilcimiz, yüksek kaliteli bilgiler sağlamak için tek bir arama sorgusunun ötesine geçen ayrıntılı araştırmalar yapmalıdır. Araştırmayı uzman bir alt aracıya devrederek araştırmacı arka planda karmaşık veri toplama işlemini yaparken ana karakterin yanıt verme hızını koruruz.
Bu adımda, "araştırma derinliği" elde etmek için LoopAgent'ı uyguluyoruz. Bu sayede, ajan tam bir resim elde edene kadar yinelemeli olarak arama yapabiliyor, bulguları değerlendirebiliyor ve sorgularını iyileştirebiliyor. Ayrıca, tüm bulgular için satır içi alıntılar zorunlu tutarak teknik titizliği de uygularız. Böylece her iddianın bir kaynak bağlantısıyla desteklenmesini sağlarız.
Araştırma Uzmanı'nı oluşturun (research.py)
Burada, Google Arama aracıyla donatılmış temel bir Agent tanımlayıp bunu bir LoopAgent'a sarmalıyoruz. max_iterations parametresi, bir sınırlayıcı görevi görerek anlayışında boşluklar kalırsa aracının aramayı en fazla 3 kez yinelemesini sağlar.
research.py
from google.adk.agents.llm_agent import Agent
from google.adk.agents.loop_agent import LoopAgent
from google.adk.tools.google_search_tool import GoogleSearchTool
from google.adk.tools.tool_context import ToolContext
def exit_loop(tool_context: ToolContext):
"""Call this function ONLY when no further research is needed, signaling the iterative process should end."""
print(f" [Tool Call] exit_loop triggered by {tool_context.agent_name}")
tool_context.actions.escalate = True
# Return empty dict as tools should typically return JSON-serializable output
return {}
# --- RESEARCH LOGIC ---
_research_worker = Agent(
model='gemini-2.5-flash',
name='research_worker',
description='Worker agent that performs a single research step.',
instruction='''
Use google_search to find facts and synthesize them for the user.
Critically evaluate your findings. If the data is incomplete or you need more context, prepare to search again in the next iteration.
You must include the links you found as references in your response, formatting them like citations in a research paper (e.g., [1], [2]).
Use the exit_loop tool to terminate the research early if no further research is needed.
If you need to ask the user for clarifications, call the exit_loop function early to interrupt the research cycle.
''',
tools=[GoogleSearchTool(bypass_multi_tools_limit=True), exit_loop],
)
# The LoopAgent iterates the worker up to 3 times for deeper research
research_agent = LoopAgent(
name='research_specialist',
description='Deep web research specialist.',
sub_agents=[_research_worker],
max_iterations=3,
)
Kök Aracıyı (agent.py) güncelleme
research_agent'ı içe aktarın ve Sharon'a araç olarak ekleyin:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our new sub agent
from .research import research_agent
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with persistent memory and research capabilities',
instruction='''
You are an elite AI partner with long-term memory.
1. Use load_memory to recall facts.
2. Delegate research tasks to the research_specialist.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
sub_agents=[research_agent],
after_agent_callback=auto_save_session_to_memory_callback,
)
Araştırma aracısını test etmek için adk web'i tekrar başlatın.
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Örneğin, "İyi bir teknoloji blogu nasıl yazılır?" gibi basit bir araştırma görevi verin.
Temsilcinin, yeni bir oturum olmasına rağmen adımı hatırladığını fark etmiş olabilirsiniz. Lütfen "transfer_to_agent" adlı araç çağrısına da dikkat edin. Bu araç, görevi yeni araştırma temsilcimize aktarır.
Şimdi de görev yönetimine geçelim.
7. Cloud SQL ile görev yönetimi ekleme
Aracı uzun süreli hafızaya sahip olsa da yapılacaklar listesi gibi ayrıntılı ve yapılandırılmış veriler için uygun değildir. Görevler için geleneksel bir ilişkisel veritabanı kullanırız. SQLAlchemy ve Google Cloud SQL (PostgreSQL) veritabanı kullanacağız. Kodu yazmadan önce altyapıyı sağlamamız gerekir.
Altyapıyı sağlama
Veritabanınızı oluşturmak için aşağıdaki komutları çalıştırın. Not: Örnek oluşturma işlemi yaklaşık 5-10 dakika sürer. Bu işlem arka planda çalışırken bir sonraki adıma geçebilirsiniz.
# 1. Define instance variables
export INSTANCE_NAME="assistant-db"
export USER_EMAIL=$(gcloud config get-value account)
# 2. Create the Cloud SQL instance
gcloud sql instances create $INSTANCE_NAME \
--database-version=POSTGRES_18 \
--tier=db-f1-micro \
--region=us-central1 \
--edition=ENTERPRISE
# 3. Create the database for our tasks
gcloud sql databases create tasks --instance=$INSTANCE_NAME
Veritabanı örneği sağlama işlemi birkaç dakika sürer. Bu sırada bir fincan kahve veya çay içebilir ya da işlemin tamamlanmasını beklerken kodu güncelleyebilirsiniz. Ancak erişim denetimini tamamlamak için geri dönmeyi unutmayın.
Erişim denetimini yapılandırma
Şimdi kullanıcı hesabınızı veritabanına erişebilecek şekilde yapılandırmamız gerekiyor. Terminalde aşağıdaki komutları çalıştırın:
# change this to your favorite password
export DB_PASS="correct-horse-battery-staple"
# Create a regular database user
gcloud sql users create assistant_user \
--instance=$INSTANCE_NAME \
--password=$DB_PASS
Ortam yapılandırmasını güncelleme
ADK, yapılandırmayı çalışma zamanında bir .env dosyasından yükler. Temsilcinizin ortamını veritabanı bağlantı ayrıntılarıyla güncelleyin.
# Retrieve the unique connection name
export DB_CONN=$(gcloud sql instances describe $INSTANCE_NAME --format='value(connectionName)')
# Append configuration to your .env file
cat <<EOF >> executive_assistant/.env
DB_CONNECTION_NAME=$DB_CONN
DB_USER=assistant_user
DB_PASSWORD=$DB_PASS
DB_NAME=tasks
EOF
Şimdi kod değişikliklerini yapmaya devam edelim.
Yapılacaklar listesi uzmanını oluşturma (todo.py)
Araştırma aracısına benzer şekilde, yapılacaklar listesi uzmanımızı kendi dosyasında oluşturalım. Oluşturma todo.py:
todo.py
import os
import uuid
import sqlalchemy
from datetime import datetime
from typing import Optional, List
from sqlalchemy import (
Column,
String,
DateTime,
Enum,
select,
delete,
update,
)
from sqlalchemy.orm import declarative_base, Session
from google.cloud.sql.connector import Connector
from google.adk.agents.llm_agent import Agent
# --- DATABASE LOGIC ---
Base = declarative_base()
connector = Connector()
def getconn():
db_connection_name = os.environ.get("DB_CONNECTION_NAME")
db_user = os.environ.get("DB_USER")
db_password = os.environ.get("DB_PASSWORD")
db_name = os.environ.get("DB_NAME", "tasks")
return connector.connect(
db_connection_name,
"pg8000",
user=db_user,
password=db_password,
db=db_name,
)
engine = sqlalchemy.create_engine(
"postgresql+pg8000://",
creator=getconn,
)
class Todo(Base):
__tablename__ = "todos"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
title = Column(String(255), nullable=False)
priority = Column(
Enum("high", "medium", "low", name="priority_levels"), nullable=False, default="medium"
)
due_date = Column(DateTime, nullable=True)
status = Column(Enum("pending", "done", name="status_levels"), default="pending")
created_at = Column(DateTime, default=datetime.utcnow)
def init_db():
"""Builds the table if it's missing."""
Base.metadata.create_all(bind=engine)
def add_todo(
title: str, priority: str = "medium", due_date: Optional[str] = None
) -> dict:
"""
Adds a new task to the list.
Args:
title (str): The description of the task.
priority (str): The urgency level. Must be one of: 'high', 'medium', 'low'.
due_date (str, optional): The due date in ISO format (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS).
Returns:
dict: A dictionary containing the new task's ID and a status message.
"""
init_db()
with Session(engine) as session:
due = datetime.fromisoformat(due_date) if due_date else None
item = Todo(
title=title,
priority=priority.lower(),
due_date=due,
)
session.add(item)
session.commit()
return {"id": item.id, "status": f"Task added ✅"}
def list_todos(status: str = "pending") -> list:
"""
Lists tasks from the database, optionally filtering by status.
Args:
status (str, optional): The status to filter by. 'pending', 'done', or 'all'.
"""
init_db()
with Session(engine) as session:
query = select(Todo)
s_lower = status.lower()
if s_lower != "all":
query = query.where(Todo.status == s_lower)
query = query.order_by(Todo.priority, Todo.created_at)
results = session.execute(query).scalars().all()
return [
{
"id": t.id,
"task": t.title,
"priority": t.priority,
"status": t.status,
}
for t in results
]
def complete_todo(task_id: str) -> str:
"""Marks a specific task as 'done'."""
init_db()
with Session(engine) as session:
session.execute(update(Todo).where(Todo.id == task_id).values(status="done"))
session.commit()
return f"Task {task_id} marked as done."
def delete_todo(task_id: str) -> str:
"""Permanently removes a task from the database."""
init_db()
with Session(engine) as session:
session.execute(delete(Todo).where(Todo.id == task_id))
session.commit()
return f"Task {task_id} deleted."
# --- TODO SPECIALIST AGENT ---
todo_agent = Agent(
model='gemini-2.5-flash',
name='todo_specialist',
description='A specialist agent that manages a structured SQL task list.',
instruction='''
You manage the user's task list using a PostgreSQL database.
- Use add_todo when the user wants to remember something. If no priority is mentioned, mark it as 'medium'.
- Use list_todos to show tasks.
- Use complete_todo to mark a task as finished.
- Use delete_todo to remove a task entirely.
When marking a task as complete or deleting it, if the user doesn't provide the ID,
use list_todos first to find the correct ID for the task they described.
''',
tools=[add_todo, list_todos, complete_todo, delete_todo],
)
Yukarıdaki kod iki temel görevden sorumludur: Cloud SQL veritabanına bağlanma ve ekleme, kaldırma ve tamamlandı olarak işaretleme dahil olmak üzere tüm yaygın yapılacaklar listesi işlemleri için bir araç listesi sağlama.
Bu mantık, yapılacaklar listesi aracısına özgü olduğundan ve yönetici asistanı (kök aracı) açısından bu ayrıntılı yönetimle ilgilenmediğimizden bu aracıyı alt aracı yerine "AgentTool" olarak paketleyeceğiz.
AgentTool veya alt aracı kullanma konusunda karar verirken bağlam paylaşımının gerekli olup olmadığını göz önünde bulundurun:
- Temsilcinizin kök temsilciyle bağlam paylaşması gerekmediğinde AgentTool kullanma
- Temsilcinizin bağlamı kök temsilciyle paylaşmasını istediğinizde alt temsilci kullanma
Araştırma aracısı söz konusu olduğunda bağlam paylaşmak faydalı olabilir ancak basit bir yapılacaklar listesi aracısı için bu pek de yararlı değildir.
AgentTool işlevini agent.py içinde uygulayalım.
Kök Aracıyı (agent.py) güncelleme
Şimdi todo_agent'ı ana dosyanıza aktarın ve araç olarak ekleyin:
agent.py
import os
from datetime import datetime
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, and research.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts or past context about the user.
2. Research: Delegate complex web-based investigations to the research_specialist.
3. Tasks: Delegate all to-do list management (adding, listing, or completing tasks) to the todo_specialist.
Always be direct and professional. If a task is successful, provide a brief confirmation.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent) # Exposes the Todo Specialist as a tool
],
sub_agents=[research_agent], # Exposes the Research Specialist for direct handover
after_agent_callback=auto_save_session_to_memory_callback,
)
Yeni özelliği test etmek için adk web komutunu tekrar çalıştırın:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Ayrıca yapılacaklar listesi oluşturmayı deneyin:
8. Takvim yönetimi ekleme
Son olarak, temsilcinin randevuları yönetebilmesi için Google Takvim ile entegrasyon sağlayacağız. Bu codelab'de, doğru şekilde yapılmadığı takdirde tehlikeli olabilecek bir işlem olan kişisel takviminize erişim izni vermek yerine, temsilcinin yönetebileceği bağımsız bir takvim oluşturacağız.
İlk olarak, aracının kimliği olarak hareket edecek özel bir hizmet hesabı oluşturacağız. Ardından, hizmet hesabını kullanarak temsilcinin takvimini programatik olarak oluştururuz.
Hizmet hesabını sağlama
Terminalinizi açın ve kimliği oluşturmak için aşağıdaki komutları çalıştırın. Ardından, kişisel hesabınıza kimliğe bürünme izni verin:
export SA_NAME="ea-agent"
export SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
# Create the service account
gcloud iam service-accounts create $SA_NAME \
--display-name="Executive Assistant Agent"
# Allow your local user to impersonate it
gcloud iam service-accounts add-iam-policy-binding $SA_EMAIL \
--member="user:$(gcloud config get-value account)" \
--role="roles/iam.serviceAccountTokenCreator"
# Save it to the agent's environment
echo "SERVICE_ACCOUNT_EMAIL=$SA_EMAIL" >> executive_assistant/.env
Takvimi programatik olarak oluşturma
Hizmet Hesabı'na takvim oluşturmasını söyleyen bir komut dosyası yazalım. Projenizin kök dizininde (setup_memory.py ile birlikte) setup_calendar.py adlı yeni bir dosya oluşturun:
setup_calendar.py
import os
import google.auth
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
from dotenv import load_dotenv
load_dotenv('executive_assistant/.env')
SA_EMAIL = os.environ.get("SERVICE_ACCOUNT_EMAIL")
def setup_sa_calendar():
print(f"Authenticating to impersonate {SA_EMAIL}...")
# 1. Base credentials
creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
creds.refresh(Request())
# 2. Impersonate the Service Account
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=SA_EMAIL,
target_scopes=["https://www.googleapis.com/auth/calendar"],
)
service = build("calendar", "v3", credentials=impersonated)
# 3. Create the calendar
print("Creating independent Service Account calendar...")
calendar = service.calendars().insert(body={
"summary": "AI Assistant (SA Owned)",
"description": "An independent calendar managed purely by the AI."
}).execute()
calendar_id = calendar['id']
# 4. Save the ID
with open("executive_assistant/.env", "a") as f:
f.write(f"\nCALENDAR_ID={calendar_id}\n")
print(f"Setup complete! CALENDAR_ID {calendar_id} added to .env")
if __name__ == "__main__":
setup_sa_calendar()
Komut dosyasını terminalinizden çalıştırın:
uv run python setup_calendar.py
Takvim Uzmanı'nı (calendar.py) oluşturun.
Şimdi de takvim uzmanına odaklanalım. Bu temsilciyi, takvim araçlarının tam paketiyle donatacağız: listeleme, oluşturma, güncelleme, silme ve hatta doğal dili anlayan bir "hızlı ekleme" özelliği.
Aşağıdaki kodu calendar.py dosyasına kopyalayın.
calendar.py
import os
from datetime import datetime, timedelta, timezone
import google.auth
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.adk.agents.llm_agent import Agent
def _get_calendar_service():
"""Build the Google Calendar API service using Service Account Impersonation."""
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
target_principal = os.environ.get("SERVICE_ACCOUNT_EMAIL")
if not target_principal:
raise ValueError("SERVICE_ACCOUNT_EMAIL environment variable is missing.")
base_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
creds, _ = google.auth.default(scopes=base_scopes)
creds.refresh(Request())
target_scopes = ["https://www.googleapis.com/auth/calendar"]
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=target_principal,
target_scopes=target_scopes,
)
return build("calendar", "v3", credentials=impersonated)
def _format_event(event: dict) -> dict:
"""Format a raw Calendar API event into a clean dict for the LLM."""
start = event.get("start", {})
end = event.get("end", {})
return {
"id": event.get("id"),
"title": event.get("summary", "(No title)"),
"start": start.get("dateTime", start.get("date")),
"end": end.get("dateTime", end.get("date")),
"location": event.get("location", ""),
"description": event.get("description", ""),
"attendees": [
{"email": a["email"], "status": a.get("responseStatus", "unknown")}
for a in event.get("attendees", [])
],
"link": event.get("htmlLink", ""),
"conference_link": (
event.get("conferenceData", {}).get("entryPoints", [{}])[0].get("uri", "")
if event.get("conferenceData")
else ""
),
"status": event.get("status", ""),
}
def list_events(days_ahead: int = 7) -> dict:
"""List upcoming calendar events."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
now = datetime.now(timezone.utc).isoformat()
end = (datetime.now(timezone.utc) + timedelta(days=days_ahead)).isoformat()
events_result = service.events().list(
calendarId=calendar_id, timeMin=now, timeMax=end,
maxResults=50, singleEvents=True, orderBy="startTime"
).execute()
events = events_result.get("items", [])
if not events:
return {"status": "success", "count": 0, "events": []}
return {"status": "success", "count": len(events), "events": [_format_event(e) for e in events]}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def create_event(title: str, start_time: str, end_time: str, description: str = "", location: str = "", attendees: str = "", add_google_meet: bool = False) -> dict:
"""Create a new calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event_body = {
"summary": title,
"start": {"dateTime": start_time},
"end": {"dateTime": end_time},
}
if description: event_body["description"] = description
if location: event_body["location"] = location
if attendees:
email_list = [e.strip() for e in attendees.split(",") if e.strip()]
event_body["attendees"] = [{"email": e} for e in email_list]
conference_version = 0
if add_google_meet:
event_body["conferenceData"] = {
"createRequest": {"requestId": f"event-{datetime.now().strftime('%Y%m%d%H%M%S')}", "conferenceSolutionKey": {"type": "hangoutsMeet"}}
}
conference_version = 1
event = service.events().insert(calendarId=calendar_id, body=event_body, conferenceDataVersion=conference_version).execute()
return {"status": "success", "message": f"Event created ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def update_event(event_id: str, title: str = "", start_time: str = "", end_time: str = "", description: str = "") -> dict:
"""Update an existing calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
patch_body = {}
if title: patch_body["summary"] = title
if start_time: patch_body["start"] = {"dateTime": start_time}
if end_time: patch_body["end"] = {"dateTime": end_time}
if description: patch_body["description"] = description
if not patch_body: return {"status": "error", "message": "No fields to update."}
event = service.events().patch(calendarId=calendar_id, eventId=event_id, body=patch_body).execute()
return {"status": "success", "message": "Event updated ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def delete_event(event_id: str) -> dict:
"""Delete a calendar event by its ID."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
service.events().delete(calendarId=calendar_id, eventId=event_id).execute()
return {"status": "success", "message": f"Event '{event_id}' deleted ✅"}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def quick_add_event(text: str) -> dict:
"""Create an event using natural language (e.g. 'Lunch with Sarah next Monday noon')."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event = service.events().quickAdd(calendarId=calendar_id, text=text).execute()
return {"status": "success", "message": "Event created from text ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
calendar_agent = Agent(
model='gemini-2.5-flash',
name='calendar_specialist',
description='Manages the user schedule and calendar events.',
instruction='''
You manage the user's Google Calendar.
- Use list_events to check the schedule.
- Use quick_add_event for simple, conversational scheduling requests (e.g., "Lunch tomorrow at noon").
- Use create_event for complex meetings that require attendees, specific durations, or Google Meet links.
- Use update_event to change details of an existing event.
- Use delete_event to cancel or remove an event.
CRITICAL: For update_event and delete_event, you must provide the exact `event_id`.
If the user does not provide the ID, you MUST call list_events first to find the correct `event_id` before attempting the update or deletion.
Always use the current date/time context provided by the root agent to resolve relative dates like "tomorrow".
''',
tools=[list_events, create_event, update_event, delete_event, quick_add_event],
)
Kök aracıyı (agent.py) sonlandırma
agent.py dosyanızı aşağıdaki kodla güncelleyin:
agent.py
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import all our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
from .calendar import calendar_agent
import tzlocal
# Automatically detect the local system timezone
TIMEZONE = tzlocal.get_localzone_name()
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Callback to inject the current time into the prompt
async def setup_agent_context(callback_context, **kwargs):
now = datetime.now(ZoneInfo(TIMEZONE))
callback_context.state["current_time"] = now.strftime("%A, %Y-%m-%d %I:%M %p")
callback_context.state["timezone"] = TIMEZONE
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, research, and schedule.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts.
2. Research: Delegate complex web investigations to the research_specialist.
3. Tasks: Delegate all to-do list management to the todo_specialist.
4. Scheduling: Delegate all calendar queries to the calendar_specialist.
## 🕒 Current State
- Time: {current_time?}
- Timezone: {timezone?}
Always be direct and professional.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent),
AgentTool(calendar_agent)
],
sub_agents=[research_agent],
before_agent_callback=[setup_agent_context],
after_agent_callback=[auto_save_session_to_memory_callback],
)
Takvim aracının yanı sıra yeni bir arama öncesi işlevi de eklediğimizi belirtmek isteriz: setup_agent_context. Bu işlev, ajana geçerli tarih, saat ve saat dilimi hakkında bilgi vererek takvimi daha verimli kullanmasını sağlar. Bu özellik, kısa süreli kalıcılık için tasarlanmış farklı bir aracı belleği türü olan oturum durumu değişkenlerini ayarlayarak çalışır.
Tüm aracıyı test etmek için adk web'i son bir kez daha çalıştırın.
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Oturum durumunu, geliştirici kullanıcı arayüzündeki durum sekmesinde inceleyebilirsiniz:
Artık takvim etkinliklerini ve yapılacaklar listelerini takip edebilen, araştırma yapabilen ve uzun süreli hafızaya sahip bir yardımcınız var.
Laboratuvar çalışmasından sonra temizleme
9. Sonuç
Tebrikler! 5 evrimsel aşamada çok işlevli bir yapay zeka yönetici asistanını başarıyla tasarladınız.
İşlediğimiz konular
- Yapay zeka aracıları için altyapı sağlama.
- ADK yerleşiklerini kullanarak kalıcı bellek ve özel alt aracılar uygulama.
- Harici veritabanlarını ve üretkenlik API'lerini entegre etme
Sonraki Adımlar
Bu platformdaki diğer codelab'leri inceleyerek veya kendi başınıza yönetici asistanında iyileştirmeler yaparak öğrenme yolculuğunuza devam edebilirsiniz.
İyileştirme fikirlerine ihtiyacınız varsa şunları deneyebilirsiniz:
- Uzun görüşmelerde performansı optimize etmek için etkinlik sıkıştırmayı uygulayın.
- Temsilcinin sizin için not almasına ve bunları dosya olarak kaydetmesine olanak tanımak için bir yapay nesne hizmeti ekleyin.
- Google Cloud Run'ı kullanarak aracınızı arka uç hizmeti olarak dağıtın.
Testi tamamladıktan sonra, faturalandırma hesabınıza beklenmedik ücretler yansımaması için ortamı temizlemeyi unutmayın.
Keyifli kodlamalar!