내장 도구¶
이러한 내장 도구는 Google 검색 또는 코드 실행기와 같은 즉시 사용 가능한 기능을 제공하여 에이전트에게 일반적인 기능을 제공합니다. 예를 들어 웹에서 정보를 검색해야 하는 에이전트는 추가 설정 없이 google_search 도구를 직접 사용할 수 있습니다.
사용 방법¶
- 가져오기: 도구 모듈에서 원하는 도구를 가져옵니다. Python에서는
agents.tools이고, Go에서는google.golang.org/adk/tool/geminitool이며, Java에서는com.google.adk.tools입니다. - 구성: 필요한 매개변수가 있는 경우 제공하여 도구를 초기화합니다.
- 등록: 초기화된 도구를 에이전트의 도구 목록에 추가합니다.
에이전트에 추가되면 에이전트는 사용자 프롬프트와 지침을 기반으로 도구 사용을 결정할 수 있습니다. 프레임워크는 에이전트가 도구를 호출할 때 도구 실행을 처리합니다. 중요: 이 페이지의 제한 사항 섹션을 확인하세요.
사용 가능한 내장 도구¶
참고: Go는 Google 검색 도구 및 geminitool 패키지를 통한 기타 내장 도구를 지원합니다.
참고: Java는 현재 Google 검색 및 코드 실행 도구만 지원합니다.
Google 검색¶
ADK에서 지원Python v0.1.0Go v0.1.0Java v0.2.0
google_search 도구는 에이전트가 Google 검색을 사용하여 웹 검색을 수행할 수 있도록 합니다. google_search 도구는 Gemini 2 모델과만 호환됩니다. 도구에 대한 자세한 내용은 Google 검색 그라운딩 이해를 참조하세요.
google_search 도구 사용 시 추가 요구 사항
Google 검색으로 그라운딩을 사용하고 응답에서 검색 제안을 받는 경우, 프로덕션 및 애플리케이션에 검색 제안을 표시해야 합니다.
Google 검색으로 그라운딩에 대한 자세한 내용은 Google AI Studio 또는 Vertex AI의 Google 검색으로 그라운딩 설명서를 참조하세요. UI 코드(HTML)는 Gemini 응답에서 renderedContent로 반환되며, 정책에 따라 앱에 HTML을 표시해야 합니다.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
APP_NAME="google_search_agent"
USER_ID="user1234"
SESSION_ID="1234"
root_agent = Agent(
name="basic_search_agent",
model="gemini-2.0-flash",
description="Agent to answer questions using Google Search.",
instruction="I can answer your questions by searching the internet. Just ask me anything!",
# google_search is a pre-built tool which allows the agent to perform Google searches.
tools=[google_search]
)
# Session and Runner
async def setup_session_and_runner():
session_service = InMemorySessionService()
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
return session, runner
# Agent Interaction
async def call_agent_async(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
session, runner = await setup_session_and_runner()
events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
async for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
await call_agent_async("what's the latest ai news?")
// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"context"
"fmt"
"log"
"google.golang.org/adk/agent"
"google.golang.org/adk/agent/llmagent"
"google.golang.org/adk/model/gemini"
"google.golang.org/adk/runner"
"google.golang.org/adk/session"
"google.golang.org/adk/tool"
"google.golang.org/adk/tool/geminitool"
"google.golang.org/genai"
)
func createSearchAgent(ctx context.Context) (agent.Agent, error) {
model, err := gemini.NewModel(ctx, "gemini-2.5-flash", &genai.ClientConfig{})
if err != nil {
return nil, fmt.Errorf("failed to create model: %v", err)
}
return llmagent.New(llmagent.Config{
Name: "basic_search_agent",
Model: model,
Description: "Agent to answer questions using Google Search.",
Instruction: "I can answer your questions by searching the web. Just ask me anything!",
Tools: []tool.Tool{geminitool.GoogleSearch{}},
})
}
const (
userID = "user1234"
appName = "Google Search_agent"
)
func callAgent(ctx context.Context, a agent.Agent, prompt string) error {
sessionService := session.InMemoryService()
session, err := sessionService.Create(ctx, &session.CreateRequest{
AppName: appName,
UserID: userID,
})
if err != nil {
return fmt.Errorf("failed to create the session service: %v", err)
}
config := runner.Config{
AppName: appName,
Agent: a,
SessionService: sessionService,
}
r, err := runner.New(config)
if err != nil {
return fmt.Errorf("failed to create the runner: %v", err)
}
sessionID := session.Session.ID()
userMsg := &genai.Content{
Parts: []*genai.Part{{Text: prompt}},
Role: string(genai.RoleUser),
}
// The r.Run method streams events and errors.
// The loop iterates over the results, handling them as they arrive.
for event, err := range r.Run(ctx, userID, sessionID, userMsg, agent.RunConfig{
StreamingMode: agent.StreamingModeSSE,
}) {
if err != nil {
fmt.Printf("\nAGENT_ERROR: %v\n", err)
} else if event.Partial {
for _, p := range event.LLMResponse.Content.Parts {
fmt.Print(p.Text)
}
}
}
return nil
}
func main() {
agent, err := createSearchAgent(context.Background())
if err != nil {
log.Fatalf("Failed to create agent: %v", err)
}
fmt.Println("Agent created:", agent.Name())
prompt := "what's the latest ai news?"
fmt.Printf("\nPrompt: %s\nResponse: ", prompt)
if err := callAgent(context.Background(), agent, prompt); err != nil {
log.Fatalf("Error calling agent: %v", err)
}
fmt.Println("\n---")
}
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.LlmAgent;
import com.google.adk.runner.Runner;
import com.google.adk.sessions.InMemorySessionService;
import com.google.adk.sessions.Session;
import com.google.adk.tools.GoogleSearchTool;
import com.google.common.collect.ImmutableList;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
public class GoogleSearchAgentApp {
private static final String APP_NAME = "Google Search_agent";
private static final String USER_ID = "user1234";
private static final String SESSION_ID = "1234";
/**
* Calls the agent with the given query and prints the final response.
*
* @param runner The runner to use.
* @param query The query to send to the agent.
*/
public static void callAgent(Runner runner, String query) {
Content content =
Content.fromParts(Part.fromText(query));
InMemorySessionService sessionService = (InMemorySessionService) runner.sessionService();
Session session =
sessionService
.createSession(APP_NAME, USER_ID, /* state= */ null, SESSION_ID)
.blockingGet();
runner
.runAsync(session.userId(), session.id(), content)
.forEach(
event -> {
if (event.finalResponse()
&& event.content().isPresent()
&& event.content().get().parts().isPresent()
&& !event.content().get().parts().get().isEmpty()
&& event.content().get().parts().get().get(0).text().isPresent()) {
String finalResponse = event.content().get().parts().get().get(0).text().get();
System.out.println("Agent Response: " + finalResponse);
}
});
}
public static void main(String[] args) {
// Google Search is a pre-built tool which allows the agent to perform Google searches.
GoogleSearchTool googleSearchTool = new GoogleSearchTool();
BaseAgent rootAgent =
LlmAgent.builder()
.name("basic_search_agent")
.model("gemini-2.0-flash") // Ensure to use a Gemini 2.0 model for Google Search Tool
.description("Agent to answer questions using Google Search.")
.instruction(
"I can answer your questions by searching the internet. Just ask me anything!")
.tools(ImmutableList.of(googleSearchTool))
.build();
// Session and Runner
InMemorySessionService sessionService = new InMemorySessionService();
Runner runner = new Runner(rootAgent, APP_NAME, null, sessionService);
// Agent Interaction
callAgent(runner, "what's the latest ai news?");
}
}
코드 실행¶
ADK에서 지원Python v0.1.0Java v0.2.0
built_in_code_execution 도구는 에이전트가 코드를 실행할 수 있게 하며,
특히 Gemini 2 모델을 사용할 때 그렇습니다. 이를 통해 모델은 계산, 데이터 조작 또는 작은 스크립트 실행과 같은 작업을 수행할 수 있습니다.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types
AGENT_NAME = "calculator_agent"
APP_NAME = "calculator"
USER_ID = "user1234"
SESSION_ID = "session_code_exec_async"
GEMINI_MODEL = "gemini-2.0-flash"
# Agent Definition
code_agent = LlmAgent(
name=AGENT_NAME,
model=GEMINI_MODEL,
code_executor=BuiltInCodeExecutor(),
instruction="""You are a calculator agent.
When given a mathematical expression, write and execute Python code to calculate the result.
Return only the final numerical result as plain text, without markdown or code blocks.
""",
description="Executes Python code to perform calculations.",
)
# Session and Runner
session_service = InMemorySessionService()
session = asyncio.run(session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
))
runner = Runner(agent=code_agent, app_name=APP_NAME,
session_service=session_service)
# Agent Interaction (Async)
async def call_agent_async(query):
content = types.Content(role="user", parts=[types.Part(text=query)])
print(f"\n--- Running Query: {query} ---")
final_response_text = "No final text response captured."
try:
# Use run_async
async for event in runner.run_async(
user_id=USER_ID, session_id=SESSION_ID, new_message=content
):
print(f"Event ID: {event.id}, Author: {event.author}")
# --- Check for specific parts FIRST ---
has_specific_part = False
if event.content and event.content.parts:
for part in event.content.parts: # Iterate through all parts
if part.executable_code:
# Access the actual code string via .code
print(
f" Debug: Agent generated code:\n```python\n{part.executable_code.code}\n```"
)
has_specific_part = True
elif part.code_execution_result:
# Access outcome and output correctly
print(
f" Debug: Code Execution Result: {part.code_execution_result.outcome} - Output:\n{part.code_execution_result.output}"
)
has_specific_part = True
# Also print any text parts found in any event for debugging
elif part.text and not part.text.isspace():
print(f" Text: '{part.text.strip()}'")
# Do not set has_specific_part=True here, as we want the final response logic below
# --- Check for final response AFTER specific parts ---
# Only consider it final if it doesn't have the specific code parts we just handled
if not has_specific_part and event.is_final_response():
if (
event.content
and event.content.parts
and event.content.parts[0].text
):
final_response_text = event.content.parts[0].text.strip()
print(f"==> Final Agent Response: {final_response_text}")
else:
print(
"==> Final Agent Response: [No text content in final event]")
except Exception as e:
print(f"ERROR during agent run: {e}")
print("-" * 30)
# Main async function to run the examples
async def main():
await call_agent_async("Calculate the value of (5 + 7) * 3")
await call_agent_async("What is 10 factorial?")
# Execute the main async function
try:
asyncio.run(main())
except RuntimeError as e:
# Handle specific error when running asyncio.run in an already running loop (like Jupyter/Colab)
if "cannot be called from a running event loop" in str(e):
print("\nRunning in an existing event loop (like Colab/Jupyter).")
print("Please run `await main()` in a notebook cell instead.")
# If in an interactive environment like a notebook, you might need to run:
# await main()
else:
raise e # Re-raise other runtime errors
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.LlmAgent;
import com.google.adk.runner.Runner;
import com.google.adk.sessions.InMemorySessionService;
import com.google.adk.sessions.Session;
import com.google.adk.tools.BuiltInCodeExecutionTool;
import com.google.common.collect.ImmutableList;
import com.google.genai.types.Content;
import com.google.genai.types.Part;
public class CodeExecutionAgentApp {
private static final String AGENT_NAME = "calculator_agent";
private static final String APP_NAME = "calculator";
private static final String USER_ID = "user1234";
private static final String SESSION_ID = "session_code_exec_sync";
private static final String GEMINI_MODEL = "gemini-2.0-flash";
/**
* Calls the agent with a query and prints the interaction events and final response.
*
* @param runner The runner instance for the agent.
* @param query The query to send to the agent.
*/
public static void callAgent(Runner runner, String query) {
Content content =
Content.builder().role("user").parts(ImmutableList.of(Part.fromText(query))).build();
InMemorySessionService sessionService = (InMemorySessionService) runner.sessionService();
Session session =
sessionService
.createSession(APP_NAME, USER_ID, /* state= */ null, SESSION_ID)
.blockingGet();
System.out.println("\n--- Running Query: " + query + " ---");
final String[] finalResponseText = {"No final text response captured."};
try {
runner
.runAsync(session.userId(), session.id(), content)
.forEach(
event -> {
System.out.println("Event ID: " + event.id() + ", Author: " + event.author());
boolean hasSpecificPart = false;
if (event.content().isPresent() && event.content().get().parts().isPresent()) {
for (Part part : event.content().get().parts().get()) {
if (part.executableCode().isPresent()) {
System.out.println(
" Debug: Agent generated code:\n```python\n"
+ part.executableCode().get().code()
+ "\n```");
hasSpecificPart = true;
} else if (part.codeExecutionResult().isPresent()) {
System.out.println(
" Debug: Code Execution Result: "
+ part.codeExecutionResult().get().outcome()
+ " - Output:\n"
+ part.codeExecutionResult().get().output());
hasSpecificPart = true;
} else if (part.text().isPresent() && !part.text().get().trim().isEmpty()) {
System.out.println(" Text: '" + part.text().get().trim() + "'");
}
}
}
if (!hasSpecificPart && event.finalResponse()) {
if (event.content().isPresent()
&& event.content().get().parts().isPresent()
&& !event.content().get().parts().get().isEmpty()
&& event.content().get().parts().get().get(0).text().isPresent()) {
finalResponseText[0] =
event.content().get().parts().get().get(0).text().get().trim();
System.out.println("==> Final Agent Response: " + finalResponseText[0]);
} else {
System.out.println(
"==> Final Agent Response: [No text content in final event]");
}
}
});
} catch (Exception e) {
System.err.println("ERROR during agent run: " + e.getMessage());
e.printStackTrace();
}
System.out.println("------------------------------");
}
public static void main(String[] args) {
BuiltInCodeExecutionTool codeExecutionTool = new BuiltInCodeExecutionTool();
BaseAgent codeAgent =
LlmAgent.builder()
.name(AGENT_NAME)
.model(GEMINI_MODEL)
.tools(ImmutableList.of(codeExecutionTool))
.instruction(
"""
You are a calculator agent.
When given a mathematical expression, write and execute Python code to calculate the result.
Return only the final numerical result as plain text, without markdown or code blocks.
""")
.description("Executes Python code to perform calculations.")
.build();
InMemorySessionService sessionService = new InMemorySessionService();
Runner runner = new Runner(codeAgent, APP_NAME, null, sessionService);
callAgent(runner, "Calculate the value of (5 + 7) * 3");
callAgent(runner, "What is 10 factorial?");
}
}
GKE 코드 실행기¶
ADK에서 지원Python v1.14.0
GKE 코드 실행기(GkeCodeExecutor)는 gVisor를 사용하여 워크로드 격리를 수행하는 GKE(Google Kubernetes Engine) 샌드박스 환경을 활용하여 LLM 생성 코드를 실행하기 위한 안전하고 확장 가능한 방법을 제공합니다. 각 코드 실행 요청에 대해 강화된 Pod 구성으로 임시 샌드박스형 Kubernetes 작업을 동적으로 생성합니다. 보안 및 격리가 중요한 GKE의 프로덕션 환경에 이 실행기를 사용해야 합니다.
작동 방식¶
코드 실행 요청이 이루어지면 GkeCodeExecutor는 다음 단계를 수행합니다.
- ConfigMap 생성: 실행해야 하는 Python 코드를 저장하기 위해 Kubernetes ConfigMap이 생성됩니다.
- 샌드박스형 Pod 생성: 강화된 보안 컨텍스트와 gVisor 런타임이 활성화된 Pod를 생성하는 새 Kubernetes 작업이 생성됩니다. ConfigMap의 코드가 이 Pod에 마운트됩니다.
- 코드 실행: 코드는 기본 노드 및 다른 워크로드와 격리된 샌드박스형 Pod 내에서 실행됩니다.
- 결과 검색: 실행의 표준 출력 및 오류 스트림이 Pod의 로그에서 캡처됩니다.
- 리소스 정리: 실행이 완료되면 작업 및 관련 ConfigMap이 자동으로 삭제되어 아티팩트가 남지 않도록 합니다.
주요 이점¶
- 향상된 보안: 코드는 커널 수준 격리가 있는 gVisor 샌드박스 환경에서 실행됩니다.
- 임시 환경: 각 코드 실행은 자체 임시 Pod에서 실행되어 실행 간 상태 전송을 방지합니다.
- 리소스 제어: 실행 Pod에 대한 CPU 및 메모리 제한을 구성하여 리소스 남용을 방지할 수 있습니다.
- 확장성: GKE가 기본 노드의 예약 및 확장을 처리하므로 많은 수의 코드 실행을 병렬로 실행할 수 있습니다.
시스템 요구 사항¶
GKE 코드 실행기 도구를 사용하여 ADK 프로젝트를 성공적으로 배포하려면 다음 요구 사항을 충족해야 합니다.
- gVisor 지원 노드 풀이 있는 GKE 클러스터.
- 에이전트의 서비스 계정에는 다음을 수행할 수 있는 특정 RBAC 권한이 필요합니다.
- 각 실행 요청에 대한 작업 생성, 감시 및 삭제.
- 작업의 Pod에 코드를 주입하기 위한 ConfigMap 관리.
- 실행 결과를 검색하기 위한 Pod 나열 및 로그 읽기.
- GKE 추가 기능과 함께 클라이언트 라이브러리 설치:
pip install google-adk[gke]
완전하고 즉시 사용 가능한 구성은 deployment_rbac.yaml 샘플을 참조하세요. ADK 워크플로를 GKE에 배포하는 방법에 대한 자세한 내용은 Google Kubernetes Engine(GKE)에 배포를 참조하세요.
from google.adk.agents import LlmAgent
from google.adk.code_executors import GkeCodeExecutor
# 서비스 계정에 필요한 RBAC 권한이 있는 네임스페이스를 대상으로 실행기 초기화.
# 이 예제에서는 사용자 지정 시간 초과 및 리소스 제한도 설정합니다.
gke_executor = GkeCodeExecutor(
namespace="agent-sandbox",
timeout_seconds=600,
cpu_limit="1000m", # 1 CPU 코어
mem_limit="1Gi",
)
# 이제 에이전트는 생성하는 모든 코드에 이 실행기를 사용합니다.
gke_agent = LlmAgent(
name="gke_coding_agent",
model="gemini-2.0-flash",
instruction="당신은 Python 코드를 작성하고 실행하는 유용한 AI 에이전트입니다.",
code_executor=gke_executor,
)
구성 매개변수¶
GkeCodeExecutor는 다음 매개변수로 구성할 수 있습니다.
| 매개변수 | 유형 | 설명 |
|---|---|---|
namespace |
str |
실행 작업이 생성될 Kubernetes 네임스페이스. 기본값은 "default"입니다. |
image |
str |
실행 Pod에 사용할 컨테이너 이미지. 기본값은 "python:3.11-slim"입니다. |
timeout_seconds |
int |
코드 실행 시간 초과(초). 기본값은 300입니다. |
cpu_requested |
str |
실행 Pod에 요청할 CPU 양. 기본값은 "200m"입니다. |
mem_requested |
str |
실행 Pod에 요청할 메모리 양. 기본값은 "256Mi"입니다. |
cpu_limit |
str |
실행 Pod가 사용할 수 있는 최대 CPU 양. 기본값은 "500m"입니다. |
mem_limit |
str |
실행 Pod가 사용할 수 있는 최대 메모리 양. 기본값은 "512Mi"입니다. |
kubeconfig_path |
str |
인증에 사용할 kubeconfig 파일 경로. 클러스터 내 구성 또는 기본 로컬 kubeconfig로 대체됩니다. |
kubeconfig_context |
str |
사용할 kubeconfig 컨텍스트. |
Vertex AI RAG 엔진¶
ADK에서 지원Python v0.1.0Java v0.2.0
vertex_ai_rag_retrieval 도구는 에이전트가 Vertex
AI RAG 엔진을 사용하여 개인 데이터 검색을 수행할 수 있도록 합니다.
Vertex AI RAG 엔진으로 그라운딩을 사용하는 경우 사전에 RAG 코퍼스를 준비해야 합니다. 설정 방법은 RAG ADK 에이전트 샘플 또는 Vertex AI RAG 엔진 페이지를 참조하세요.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
from google.adk.agents import Agent
from google.adk.tools.retrieval.vertex_ai_rag_retrieval import VertexAiRagRetrieval
from vertexai.preview import rag
from dotenv import load_dotenv
from .prompts import return_instructions_root
load_dotenv()
ask_vertex_retrieval = VertexAiRagRetrieval(
name='retrieve_rag_documentation',
description=(
'Use this tool to retrieve documentation and reference materials for the question from the RAG corpus,'
),
rag_resources=[
rag.RagResource(
# please fill in your own rag corpus
# here is a sample rag corpus for testing purpose
# e.g. projects/123/locations/us-central1/ragCorpora/456
rag_corpus=os.environ.get("RAG_CORPUS")
)
],
similarity_top_k=10,
vector_distance_threshold=0.6,
)
root_agent = Agent(
model='gemini-2.0-flash-001',
name='ask_rag_agent',
instruction=return_instructions_root(),
tools=[
ask_vertex_retrieval,
]
)
Vertex AI 검색¶
ADK에서 지원Python v0.1.0
vertex_ai_search_tool은 Google Cloud Vertex AI Search를 사용하여 에이전트가
비공개로 구성된 데이터 저장소(예: 내부 문서, 회사 정책, 지식 기반)를 검색할 수 있도록 합니다. 이 내장 도구는 구성 중에 특정 데이터 저장소 ID를 제공해야 합니다. 도구에 대한 자세한 내용은 Vertex AI 검색 그라운딩 이해를 참조하세요.
# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.tools import VertexAiSearchTool
# Replace with your Vertex AI Search Datastore ID, and respective region (e.g. us-central1 or global).
# Format: projects/<PROJECT_ID>/locations/<REGION>/collections/default_collection/dataStores/<DATASTORE_ID>
DATASTORE_PATH = "DATASTORE_PATH_HERE"
# Constants
APP_NAME_VSEARCH = "vertex_search_app"
USER_ID_VSEARCH = "user_vsearch_1"
SESSION_ID_VSEARCH = "session_vsearch_1"
AGENT_NAME_VSEARCH = "doc_qa_agent"
GEMINI_2_FLASH = "gemini-2.0-flash"
# Tool Instantiation
# You MUST provide your datastore ID here.
vertex_search_tool = VertexAiSearchTool(data_store_id=DATASTORE_PATH)
# Agent Definition
doc_qa_agent = LlmAgent(
name=AGENT_NAME_VSEARCH,
model=GEMINI_2_FLASH, # Requires Gemini model
tools=[vertex_search_tool],
instruction=f"""You are a helpful assistant that answers questions based on information found in the document store: {DATASTORE_PATH}.
Use the search tool to find relevant information before answering.
If the answer isn't in the documents, say that you couldn't find the information.
""",
description="Answers questions using a specific Vertex AI Search datastore.",
)
# Session and Runner Setup
session_service_vsearch = InMemorySessionService()
runner_vsearch = Runner(
agent=doc_qa_agent, app_name=APP_NAME_VSEARCH, session_service=session_service_vsearch
)
session_vsearch = session_service_vsearch.create_session(
app_name=APP_NAME_VSEARCH, user_id=USER_ID_VSEARCH, session_id=SESSION_ID_VSEARCH
)
# Agent Interaction Function
async def call_vsearch_agent_async(query):
print("\n--- Running Vertex AI Search Agent ---")
print(f"Query: {query}")
if "DATASTORE_PATH_HERE" in DATASTORE_PATH:
print("Skipping execution: Please replace DATASTORE_PATH_HERE with your actual datastore ID.")
print("-" * 30)
return
content = types.Content(role='user', parts=[types.Part(text=query)])
final_response_text = "No response received."
try:
async for event in runner_vsearch.run_async(
user_id=USER_ID_VSEARCH, session_id=SESSION_ID_VSEARCH, new_message=content
):
# Like Google Search, results are often embedded in the model's response.
if event.is_final_response() and event.content and event.content.parts:
final_response_text = event.content.parts[0].text.strip()
print(f"Agent Response: {final_response_text}")
# You can inspect event.grounding_metadata for source citations
if event.grounding_metadata:
print(f" (Grounding metadata found with {len(event.grounding_metadata.grounding_attributions)} attributions)")
except Exception as e:
print(f"An error occurred: {e}")
print("Ensure your datastore ID is correct and the service account has permissions.")
print("-" * 30)
# --- Run Example ---
async def run_vsearch_example():
# Replace with a question relevant to YOUR datastore content
await call_vsearch_agent_async("Summarize the main points about the Q2 strategy document.")
await call_vsearch_agent_async("What safety procedures are mentioned for lab X?")
# Execute the example
# await run_vsearch_example()
# Running locally due to potential colab asyncio issues with multiple awaits
try:
asyncio.run(run_vsearch_example())
except RuntimeError as e:
if "cannot be called from a running event loop" in str(e):
print("Skipping execution in running event loop (like Colab/Jupyter). Run locally.")
else:
raise e
BigQuery¶
ADK에서 지원Python v1.1.0
다음은 BigQuery와의 통합을 제공하기 위한 도구 모음입니다.
list_dataset_ids: GCP 프로젝트에 있는 BigQuery 데이터세트 ID를 가져옵니다.get_dataset_info: BigQuery 데이터세트에 대한 메타데이터를 가져옵니다.list_table_ids: BigQuery 데이터세트에 있는 테이블 ID를 가져옵니다.get_table_info: BigQuery 테이블에 대한 메타데이터를 가져옵니다.execute_sql: BigQuery에서 SQL 쿼리를 실행하고 결과를 가져옵니다.forecast:AI.FORECAST함수를 사용하여 BigQuery AI 시계열 예측을 실행합니다.ask_data_insights: 자연어를 사용하여 BigQuery 테이블의 데이터에 대한 질문에 답변합니다.
이들은 BigQueryToolset 도구 세트에 패키지되어 있습니다.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.bigquery import BigQueryCredentialsConfig
from google.adk.tools.bigquery import BigQueryToolset
from google.adk.tools.bigquery.config import BigQueryToolConfig
from google.adk.tools.bigquery.config import WriteMode
from google.genai import types
import google.auth
# Define constants for this example agent
AGENT_NAME = "bigquery_agent"
APP_NAME = "bigquery_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-2.0-flash"
# Define a tool configuration to block any write operations
tool_config = BigQueryToolConfig(write_mode=WriteMode.BLOCKED)
# Uses externally-managed Application Default Credentials (ADC) by default.
# This decouples authentication from the agent / tool lifecycle.
# https://cloud.google.com/docs/authentication/provide-credentials-adc
credentials_config = BigQueryCredentialsConfig()
# Instantiate a BigQuery toolset
bigquery_toolset = BigQueryToolset(
credentials_config=credentials_config, bigquery_tool_config=tool_config
)
# Agent Definition
bigquery_agent = Agent(
model=GEMINI_MODEL,
name=AGENT_NAME,
description=(
"Agent to answer questions about BigQuery data and models and execute"
" SQL queries."
),
instruction="""\
You are a data science agent with access to several BigQuery tools.
Make use of those tools to answer the user's questions.
""",
tools=[bigquery_toolset],
)
# Session and Runner
session_service = InMemorySessionService()
session = asyncio.run(
session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
)
)
runner = Runner(
agent=bigquery_agent, app_name=APP_NAME, session_service=session_service
)
# Agent Interaction
def call_agent(query):
"""
Helper function to call the agent with a query.
"""
content = types.Content(role="user", parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
print("USER:", query)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("AGENT:", final_response)
call_agent("Are there any ml datasets in bigquery-public-data project?")
call_agent("Tell me more about ml_datasets.")
call_agent("Which all tables does it have?")
call_agent("Tell me more about the census_adult_income table.")
call_agent("How many rows are there per income bracket?")
call_agent(
"What is the statistical correlation between education_num, age, and the income_bracket?"
)
Spanner¶
ADK에서 지원Python v1.11.0
다음은 Spanner와의 통합을 제공하기 위한 도구 모음입니다.
list_table_names: GCP Spanner 데이터베이스에 있는 테이블 이름을 가져옵니다.list_table_indexes: GCP Spanner 데이터베이스에 있는 테이블 인덱스를 가져옵니다.list_table_index_columns: GCP Spanner 데이터베이스에 있는 테이블 인덱스 열을 가져옵니다.list_named_schemas: Spanner 데이터베이스에 대한 명명된 스키마를 가져옵니다.get_table_schema: Spanner 데이터베이스 테이블 스키마 및 메타데이터 정보를 가져옵니다.execute_sql: Spanner 데이터베이스에서 SQL 쿼리를 실행하고 결과를 가져옵니다.similarity_search: 텍스트 쿼리를 사용하여 Spanner에서 유사성 검색을 수행합니다.
이들은 SpannerToolset 도구 세트에 패키지되어 있습니다.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
# from google.adk.sessions import DatabaseSessionService
from google.adk.tools.google_tool import GoogleTool
from google.adk.tools.spanner import query_tool
from google.adk.tools.spanner.settings import SpannerToolSettings
from google.adk.tools.spanner.settings import Capabilities
from google.adk.tools.spanner.spanner_credentials import SpannerCredentialsConfig
from google.adk.tools.spanner.spanner_toolset import SpannerToolset
from google.genai import types
from google.adk.tools.tool_context import ToolContext
import google.auth
from google.auth.credentials import Credentials
# Define constants for this example agent
AGENT_NAME = "spanner_agent"
APP_NAME = "spanner_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-2.5-flash"
# Define Spanner tool config with read capability set to allowed.
tool_settings = SpannerToolSettings(capabilities=[Capabilities.DATA_READ])
# Define a credentials config - in this example we are using application default
# credentials
# https://cloud.google.com/docs/authentication/provide-credentials-adc
application_default_credentials, _ = google.auth.default()
credentials_config = SpannerCredentialsConfig(
credentials=application_default_credentials
)
# Instantiate a Spanner toolset
spanner_toolset = SpannerToolset(
credentials_config=credentials_config, spanner_tool_settings=tool_settings
)
# Optional
# Create a wrapped function tool for the agent on top of the built-in
# `execute_sql` tool in the Spanner toolset.
# For example, this customized tool can perform a dynamically-built query.
def count_rows_tool(
table_name: str,
credentials: Credentials, # GoogleTool handles `credentials`
settings: SpannerToolSettings, # GoogleTool handles `settings`
tool_context: ToolContext, # GoogleTool handles `tool_context`
):
"""Counts the total number of rows for a specified table.
Args:
table_name: The name of the table for which to count rows.
Returns:
The total number of rows in the table.
"""
# Replace the following settings for a specific Spanner database.
PROJECT_ID = "<PROJECT_ID>"
INSTANCE_ID = "<INSTANCE_ID>"
DATABASE_ID = "<DATABASE_ID>"
query = f"""
SELECT count(*) FROM {table_name}
"""
return query_tool.execute_sql(
project_id=PROJECT_ID,
instance_id=INSTANCE_ID,
database_id=DATABASE_ID,
query=query,
credentials=credentials,
settings=settings,
tool_context=tool_context,
)
# Agent Definition
spanner_agent = Agent(
model=GEMINI_MODEL,
name=AGENT_NAME,
description=(
"Agent to answer questions about Spanner database and execute SQL queries."
),
instruction="""\
You are a data assistant agent with access to several Spanner tools.
Make use of those tools to answer the user's questions.
""",
tools=[
spanner_toolset,
# Add customized Spanner tool based on the built-in Spanner toolset.
GoogleTool(
func=count_rows_tool,
credentials_config=credentials_config,
tool_settings=tool_settings,
),
],
)
# Session and Runner
session_service = InMemorySessionService()
# Optionally, Spanner can be used as the Database Session Service for production.
# Note that it's suggested to use a dedicated instance/database for storing sessions.
# session_service_spanner_db_url = "spanner+spanner:///projects/PROJECT_ID/instances/INSTANCE_ID/databases/my-adk-session"
# session_service = DatabaseSessionService(db_url=session_service_spanner_db_url)
session = asyncio.run(
session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
)
)
runner = Runner(
agent=spanner_agent, app_name=APP_NAME, session_service=session_service
)
# Agent Interaction
def call_agent(query):
"""
Helper function to call the agent with a query.
"""
content = types.Content(role="user", parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
print("USER:", query)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("AGENT:", final_response)
# Replace the Spanner database and table names below with your own.
call_agent("List all tables in projects/<PROJECT_ID>/instances/<INSTANCE_ID>/databases/<DATABASE_ID>")
call_agent("Describe the schema of <TABLE_NAME>")
call_agent("List the top 5 rows in <TABLE_NAME>")
Bigtable¶
ADK에서 지원Python v1.12.0
다음은 Bigtable과의 통합을 제공하기 위한 도구 모음입니다.
list_instances: Google Cloud 프로젝트의 Bigtable 인스턴스를 가져옵니다.get_instance_info: Google Cloud 프로젝트의 메타데이터 인스턴스 정보를 가져옵니다.list_tables: GCP Bigtable 인스턴스의 테이블을 가져옵니다.get_table_info: GCP Bigtable의 메타데이터 테이블 정보를 가져옵니다.execute_sql: Bigtable 테이블에서 SQL 쿼리를 실행하고 결과를 가져옵니다.
이들은 BigtableToolset 도구 세트에 패키지되어 있습니다.
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.google_tool import GoogleTool
from google.adk.tools.bigtable import query_tool
from google.adk.tools.bigtable.settings import BigtableToolSettings
from google.adk.tools.bigtable.bigtable_credentials import BigtableCredentialsConfig
from google.adk.tools.bigtable.bigtable_toolset import BigtableToolset
from google.genai import types
from google.adk.tools.tool_context import ToolContext
import google.auth
from google.auth.credentials import Credentials
# Define constants for this example agent
AGENT_NAME = "bigtable_agent"
APP_NAME = "bigtable_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-2.5-flash"
# Define Bigtable tool config with read capability set to allowed.
tool_settings = BigtableToolSettings()
# Define a credentials config - in this example we are using application default
# credentials
# https://cloud.google.com/docs/authentication/provide-credentials-adc
application_default_credentials, _ = google.auth.default()
credentials_config = BigtableCredentialsConfig(
credentials=application_default_credentials
)
# Instantiate a Bigtable toolset
bigtable_toolset = BigtableToolset(
credentials_config=credentials_config, bigtable_tool_settings=tool_settings
)
# Optional
# Create a wrapped function tool for the agent on top of the built-in
# `execute_sql` tool in the bigtable toolset.
# For example, this customized tool can perform a dynamically-built query.
def count_rows_tool(
table_name: str,
credentials: Credentials, # GoogleTool handles `credentials`
settings: BigtableToolSettings, # GoogleTool handles `settings`
tool_context: ToolContext, # GoogleTool handles `tool_context`
):
"""Counts the total number of rows for a specified table.
Args:
table_name: The name of the table for which to count rows.
Returns:
The total number of rows in the table.
"""
# Replace the following settings for a specific bigtable database.
PROJECT_ID = "<PROJECT_ID>"
INSTANCE_ID = "<INSTANCE_ID>"
query = f"""
SELECT count(*) FROM {table_name}
"""
return query_tool.execute_sql(
project_id=PROJECT_ID,
instance_id=INSTANCE_ID,
query=query,
credentials=credentials,
settings=settings,
tool_context=tool_context,
)
# Agent Definition
bigtable_agent = Agent(
model=GEMINI_MODEL,
name=AGENT_NAME,
description=(
"Agent to answer questions about bigtable database and execute SQL queries."
),
instruction="""\
You are a data assistant agent with access to several bigtable tools.
Make use of those tools to answer the user's questions.
""",
tools=[
bigtable_toolset,
# Add customized bigtable tool based on the built-in bigtable toolset.
GoogleTool(
func=count_rows_tool,
credentials_config=credentials_config,
tool_settings=tool_settings,
),
],
)
# Session and Runner
session_service = InMemorySessionService()
session = asyncio.run(
session_service.create_session(
app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
)
)
runner = Runner(
agent=bigtable_agent, app_name=APP_NAME, session_service=session_service
)
# Agent Interaction
def call_agent(query):
"""
Helper function to call the agent with a query.
"""
content = types.Content(role="user", parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
print("USER:", query)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("AGENT:", final_response)
# Replace the bigtable instance and table names below with your own.
call_agent("List all tables in projects/<PROJECT_ID>/instances/<INSTANCE_ID>")
call_agent("List the top 5 rows in <TABLE_NAME>")
다른 도구와 함께 내장 도구 사용하기¶
ADK에서 지원PythonJava
다음 코드 샘플은 여러 내장 도구를 사용하거나 여러 에이전트를 사용하여 다른 도구와 함께 내장 도구를 사용하는 방법을 보여줍니다.
from google.adk.tools.agent_tool import AgentTool
from google.adk.agents import Agent
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
search_agent = Agent(
model='gemini-2.0-flash',
name='SearchAgent',
instruction="""
당신은 Google 검색 전문가입니다.
""",
tools=[google_search],
)
coding_agent = Agent(
model='gemini-2.0-flash',
name='CodeAgent',
instruction="""
당신은 코드 실행 전문가입니다.
""",
code_executor=BuiltInCodeExecutor(),
)
root_agent = Agent(
name="RootAgent",
model="gemini-2.0-flash",
description="루트 에이전트",
tools=[AgentTool(agent=search_agent), AgentTool(agent=coding_agent)],
)
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.LlmAgent;
import com.google.adk.tools.AgentTool;
import com.google.adk.tools.BuiltInCodeExecutionTool;
import com.google.adk.tools.GoogleSearchTool;
import com.google.common.collect.ImmutableList;
public class NestedAgentApp {
private static final String MODEL_ID = "gemini-2.0-flash";
public static void main(String[] args) {
// 검색 에이전트 정의
LlmAgent searchAgent =
LlmAgent.builder()
.model(MODEL_ID)
.name("SearchAgent")
.instruction("당신은 Google 검색 전문가입니다.")
.tools(new GoogleSearchTool()) // GoogleSearchTool 인스턴스화
.build();
// 코딩 에이전트 정의
LlmAgent codingAgent =
LlmAgent.builder()
.model(MODEL_ID)
.name("CodeAgent")
.instruction("당신은 코드 실행 전문가입니다.")
.tools(new BuiltInCodeExecutionTool()) // BuiltInCodeExecutionTool 인스턴스화
.build();
// 루트 에이전트 정의, AgentTool.create()를 사용하여 검색 에이전트와 코딩 에이전트를 래핑
BaseAgent rootAgent =
LlmAgent.builder()
.name("RootAgent")
.model(MODEL_ID)
.description("루트 에이전트")
.tools(
AgentTool.create(searchAgent), // create 메서드 사용
AgentTool.create(codingAgent) // create 메서드 사용
)
.build();
// 참고: 이 샘플은 에이전트 정의만 보여줍니다.
// 이 에이전트들을 실행하려면 Runner와 SessionService와 통합해야 합니다.
// 이전 예제와 유사하게.
System.out.println("에이전트가 성공적으로 정의되었습니다:");
System.out.println(" 루트 에이전트: " + rootAgent.name());
System.out.println(" 검색 에이전트 (중첩됨): " + searchAgent.name());
System.out.println(" 코드 에이전트 (중첩됨): " + codingAgent.name());
}
}
제한 사항¶
Warning
현재 각 루트 에이전트 또는 단일 에이전트당 하나의 내장 도구만 지원됩니다. 동일한 에이전트에서 다른 유형의 도구를 사용할 수 없습니다.
예를 들어, 단일 에이전트 내에서 다른 도구와 함께 내장 도구를 사용하는 다음 접근 방식은 현재 지원되지 않습니다:
ADK Python에는 이 제한을 우회하는 내장 해결 방법이 있습니다.
GoogleSearchTool 및 VertexAiSearchTool의 경우 (bypass_multi_tools_limit=True를 사용하여 활성화), 예:
샘플 에이전트.
Warning
내장 도구는 하위 에이전트 내에서 사용할 수 없습니다. 단,
위에서 언급한 해결 방법 때문에 ADK Python의 GoogleSearchTool 및 VertexAiSearchTool은 예외입니다.
예를 들어, 하위 에이전트 내에서 내장 도구를 사용하는 다음 접근 방식은 현재 지원되지 않습니다:
url_context_agent = Agent(
model='gemini-2.0-flash',
name='UrlContextAgent',
instruction="""
당신은 URL 컨텍스트 전문가입니다.
""",
tools=[url_context],
)
coding_agent = Agent(
model='gemini-2.0-flash',
name='CodeAgent',
instruction="""
당신은 코드 실행 전문가입니다.
""",
code_executor=BuiltInCodeExecutor(),
)
root_agent = Agent(
name="RootAgent",
model="gemini-2.0-flash",
description="루트 에이전트",
sub_agents=[
url_context_agent,
coding_agent
],
)
LlmAgent searchAgent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("SearchAgent")
.instruction("당신은 Google 검색 전문가입니다.")
.tools(new GoogleSearchTool())
.build();
LlmAgent codingAgent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("CodeAgent")
.instruction("당신은 코드 실행 전문가입니다.")
.tools(new BuiltInCodeExecutionTool())
.build();
LlmAgent rootAgent =
LlmAgent.builder()
.name("RootAgent")
.model("gemini-2.0-flash")
.description("루트 에이전트")
.subAgents(searchAgent, codingAgent) // 지원되지 않음, 하위 에이전트가 내장 도구를 사용하기 때문.
.build();