Mengelola sesi dengan Agent Development Kit

Halaman ini menjelaskan cara menghubungkan agen Agent Development Kit (ADK) dengan Sesi Vertex AI Agent Engine dan menggunakan sesi terkelola di lingkungan lokal dan produksi.

Sebelum memulai

Pastikan lingkungan Anda disiapkan dengan mengikuti langkah-langkah Mendapatkan peran yang diperlukan dan Autentikasi di Menyiapkan lingkungan Anda.

Membuat instance Vertex AI Agent Engine

Untuk mengakses Sesi Vertex AI Agent Engine, Anda harus membuat instance Vertex AI Agent Engine terlebih dahulu. Anda tidak perlu men-deploy kode apa pun untuk mulai menggunakan Sesi. Tanpa deployment kode, pembuatan instance Vertex AI Agent Engine hanya memerlukan waktu beberapa detik.

import vertexai
from vertexai import agent_engines

# Create an agent engine instance
agent_engine = agent_engines.create()

Mengembangkan agen ADK

Untuk membuat agen ADK, ikuti petunjuk di Agent Development Kit, atau gunakan kode berikut untuk membuat agen yang menyapa pengguna dengan sapaan tetap:

from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

Menyiapkan runner ADK

ADK Runtime mengatur eksekusi agen, alat, dan callback Anda, serta mengatur panggilan untuk membaca dan menulis sesi. Lakukan inisialisasi Runner dengan VertexAiSessionService, yang terhubung dengan Sesi Vertex AI Agent Engine.

from google.adk.sessions import VertexAiSessionService

app_name="AGENT_ENGINE_ID"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
       "PROJECT_ID", "LOCATION")
runner = adk.Runner(
    agent=root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Ganti kode berikut:

• PROJECT_ID: Project ID Anda.

• LOCATION: Region Anda.

• AGENT_ENGINE_ID: ID resource instance Vertex AI Agent Engine.

  • Untuk agen yang di-deploy, ID resource dicantumkan sebagai variabel lingkungan GOOGLE_CLOUD_AGENT_ENGINE_ID

  • Untuk agen lokal, Anda dapat mengambil ID resource menggunakan agent_engine.name.split("/")[-1].

• USER_ID: ID unik yang tidak boleh kosong untuk pengguna, dengan panjang maksimum 128 karakter.

Berinteraksi dengan agen Anda

Setelah menentukan agen dan menyiapkan Sesi Vertex AI Agent Engine, Anda dapat berinteraksi dengan agen untuk memeriksa apakah histori dan status sesi tetap ada.

agent_engine_id="AGENT_ENGINE_ID"

adk web --session_db_url=agentengine://${agent_engine_id}

# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

# Create a session with state
session = await session_service.create_session(
      app_name=app_name,
      user_id=user_id,
      state={'key': 'value'})

print(session.state['key'])
# value

from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
  invocation_id="invocation_id",
  author="system", # Or 'agent', 'tool' etc.
  actions=actions_with_update,
  timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
      app_name=app_name,
      user_id=user_id,
      session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

Men-deploy agen Anda ke Vertex AI Agent Engine

Setelah menguji agen secara lokal, Anda dapat men-deploy agen ke produksi dengan memperbarui instance Vertex AI Agent Engine dengan parameter:

agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)

Ganti kode berikut:

• AGENT: Aplikasi yang menerapkan metode query / stream_query (misalnya, AdkApp untuk agen ADK). Untuk mengetahui informasi selengkapnya, lihat Pertimbangan deployment.

Pembersihan

Untuk membersihkan semua resource yang digunakan dalam project ini, Anda dapat menghapus instance Vertex AI Agent Engine beserta resource turunannya:

agent_engine.delete(force=True)

Langkah berikutnya

• Mengelola sesi menggunakan panggilan API.