פתרון נקודות קצה ויצירת כלי תזמור

הערכה לפיתוח סוכנים (ADK) מספקת לקוח AgentRegistry ייעודי שמאפשר לכם לגלות, לחפש ולהתחבר לסוכני AI ולשרתים של Model Context Protocol‏ (MCP) שמופיעים בקטלוג של Agent Registry.

במקום להגדיר כתובות URL של נקודות קצה באפליקציה, אפשר להשתמש ב-ADK כדי לפתור את נקודות הקצה האלה בזמן הריצה.

Agent Registry מספק את נקודות הקצה הבסיסיות, אבל בפריסות בסביבת הייצור, בדרך כלל ההפעלות האלה מנותבות דרך Agent Gateway. ה-Agent Gateway עוזר לכם לאכוף מדיניות אבטחה, לבצע גישור בין פרוטוקולים ולהחיל סינון תוכן על הכלים שאתם מגלים.

במסמך הזה מוסבר איך לאחזר סוכנים מרוחקים וערכות כלים של MCP מ-Agent Registry ולכלול אותם בסוכן מתזמן ראשי.

לפני שמתחילים

לפני שמשלבים את ה-ADK עם Agent Registry, צריך לבצע את הפעולות הבאות:

  1. מגדירים את Agent Registry בפרויקט.

  2. מתקינים את הגרסה האחרונה של ADK או משדרגים אליה עם יחסי התלות הנדרשים של A2A:‏

    צִפצוּף 

    pip install --upgrade "google-adk[a2a]"

    uv 

    uv add "google-adk[a2a]"

    צריך לשדרג לפחות ל-google-adk>=1.29.0.

  3. מגדירים Application Default Credentials ‏ (ADC):

    gcloud auth application-default login

לפרטי הכניסה של ADC צריכות להיות הרשאות IAM שנדרשות לשירותים הבסיסיים שאיתם הסוכנים או הכלים מקיימים אינטראקציה. אפשר גם להשתמש בכותרות בהתאמה אישית עבור ערכות כלים חיצוניות. מידע נוסף זמין במאמר אימות לכלים ולמשאבים.

הגדרה של משתני סביבה

כדי לפעול לפי המדריך הזה, צריך להגדיר את משתני הסביבה הבאים:

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

מחליפים את מה שכתוב בשדות הבאים:

  • PROJECT_ID: מזהה הפרויקט.
  • ‫LOCATION: האזור או המיקום של המאגר, למשל us-central1.

אתחול של לקוח המאגר

כדי ליצור אינטראקציה עם המאגר באופן פרוגרמטי, צריך לאתחל את AgentRegistry הלקוח עם הפרויקט והמיקום:

import os
from google.adk.integrations.agent_registry import AgentRegistry

project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")

if not project_id:
    raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")

# Initialize the client
registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

הרכבת מערכת מרובת סוכנים

ה-ADK מבצע הפשטה של מנגנוני החיבור הבסיסיים, ומאפשר לכם לעצב אפליקציות שניתנות להרחבה על ידי שילוב של כמה סוכנים ייעודיים בהיררכיות גמישות.

אפשר להשתמש בלקוח של המאגר כדי לאחזר משאבים ספציפיים ולהעביר אותם ישירות להגדרה של סוכן LlmAgent חדש. האורקסטרטור יכול להפעיל סוכנים מרוחקים כסוכני משנה ולהריץ כלי MCP כאילו היו פונקציות Python מקומיות.

אפשר להשתמש בשיטות הבאות:

  • כדי לאחזר סוכן מרוחק: משתמשים ב-get_remote_a2a_agent()
  • כדי לאחזר ערכת כלים של MCP: משתמשים ב-get_mcp_toolset()

בדוגמה הבאה נדגים איך ליצור מערכת מרובת סוכנים על ידי בניית סוכן לניהול תהליכים שמשתמש בסוכן נסיעות רשום ובשרת MCP רשום של Compute Engine. בדוגמה הזו, האימות מתבצע באמצעות הזהות של הסוכן עצמו, אבל אפשר להשתמש בשיטות אחרות, כמו מפתחות API ו-OAuth. מידע נוסף זמין במאמר אימות לכלים ולמשאבים.

כדי לתכנן ולתזמן סוכנים באמצעות ממשק ויזואלי עם תכנות מינימלי, צריך להגדיר סוכני משנה רשומים ישירות באזור העריכה של Agent Studio. מידע נוסף זמין במאמר תכנון סוכנים ב-Agent Studio.

import httpx
import google.auth
from google.auth.transport.requests import Request
from google.adk.agents.llm_agent import LlmAgent

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Connect to a remote A2A agent using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = "agents/AGENT_ID"

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))
my_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Retrieve an MCP toolset using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "mcpServers/SERVER_ID"
# Full format: f"projects/{project_id}/locations/{location}/mcpServers/SERVER_ID"
mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

# Compose the orchestrator agent
main_agent = LlmAgent(
    model="MODEL_ID", # Replace with a model such as gemini-1.5-flash
    name="travel_orchestrator",
    instruction="""You are a travel coordinator. You can use your
                   subagents to book travel and your tools to query
                   historical travel data.""",
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

# You can now run your orchestrator agent
# response = await main_agent.run('Book a flight to Paris and check my past trips.')

שיטות מומלצות לשימוש חוזר בסוכנים

כדי לצמצם את זמן האחזור ברשת, כדאי לאחזר את הסוכנים ואת ערכות הכלים מהמאגר פעם אחת בהפעלת האפליקציה, במקום לקרוא ל-get_remote_a2a_agent() בכל הפעלה.

לסוכן יכול להיות רק סוכן הורה אחד בכל זמן נתון. אם מנסים להקצות את אותו מופע של סוכן שאוחזר לכמה כלי תזמור, יכול להיות ש-ADK יציג שגיאה שמציינת שלסוכן כבר יש הורה.

כדי לעשות שימוש חוזר בסוכן שזוהה בכמה סוכנים ראשיים, משתמשים בשיטה .clone() כדי ליצור מופע חדש של אובייקט הסוכן.

בדוגמה הבאה אפשר לראות איך מאחזרים סוכן פעם אחת ומשכפלים אותו לשימוש במארגני תנועה שונים:

import httpx
import google.auth
from google.auth.transport.requests import Request
from google.adk.agents.llm_agent import LlmAgent

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))

# Fetch the remote agent once during startup
# Use the resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = f"projects/PROJECT_ID/locations/LOCATION/agents/AGENT_ID"
base_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Use .clone() to assign the agent to different parent orchestrators
flight_orchestrator = LlmAgent(
    model="gemini-1.5-flash",
    name="flight_orchestrator",
    sub_agents=[base_remote_agent.clone()]
)

hotel_orchestrator = LlmAgent(
    model="gemini-1.5-flash",
    name="hotel_orchestrator",
    sub_agents=[base_remote_agent.clone()]
)

המאמרים הבאים