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

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

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

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

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

  1. הגדרת Agent Registry.

  2. אם אתם משתמשים בפרויקט אחד, אתרו את מזהה הפרויקט של פרויקט הפיקוח שבו מופעל Agent Registry. אם הגדרתם גבול ניהול לכמה פרויקטים, השתמשו במזהה של פרויקט הניהול כדי לגלות ולרשום את כל הפרויקטים בגבול, או ציינו מזהה של פרויקט חבר כדי לבצע פעולות בפרויקט יחיד.

  3. מתקינים את הגרסה העדכנית של ADK או משדרגים אליה, עם יחסי התלות של A2A, ‏ Agent Identity ו-MCP:

    pip 

    pip install --upgrade "google-adk[a2a,agent-identity,mcp]"

    uv 

    uv add "google-adk[a2a,agent-identity,mcp]"

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

  4. מגדירים את Application Default Credentials (ADC):

    gcloud auth application-default login

לפרטי הכניסה שלכם צריכות להיות הרשאות ניהול הזהויות והרשאות הגישה (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,
)

יצירת מערכת מרובת סוכנים (MAS)

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

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

כדי לאחזר סוכן מרוחק או ערכת כלים של MCP, צריך להפעיל את השיטות הבאות של הלקוח AgentRegistry בקוד הסוכן ב-Python:

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

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

כדי לתכנן ולתזמר סוכנים באמצעות ממשק חזותי עם תכנות מינימלי, מגדירים סוכני משנה רשומים ישירות ב-Agent Studio ב-Gemini Enterprise Agent Platform. מידע נוסף זמין במאמר תכנון נציגים ב-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.')

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

  • ‫AGENT_ID: המזהה של הנציג הרשום.
  • ‫SERVER_ID: המזהה של שרת ה-MCP הרשום.
  • ‫MODEL_ID: מזהה המודל של הסוכן לניהול תהליכים, לדוגמה gemini-1.5-flash.

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

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

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

כדי להשתמש בסוכן שזוהה בכמה סוכנים ראשיים, משתמשים ב-method‏ .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()]
)

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

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

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