חוזה זמן ריצה של Gemini Enterprise Agent Platform

‫Agent Runtime מתוכנן כך שהוא לא תלוי ב-framework של האפליקציה. אם אתם בוחרים לפרוס את הנציג באמצעות קונטיינר בהתאמה אישית או קובץ Docker, הקונטיינר צריך לעמוד בדרישות של חוזה זמן הריצה כדי לטפל בשאילתות בהצלחה.

מידע נוסף על שיטות פריסה זמין במאמר פריסת סוכן.

מגבלות ודרישות

כדי לפרוס קונטיינר בהתאמה אישית ב-Agent Runtime, הקונטיינר צריך להאזין לבקשות HTTP ב-0.0.0.0 ביציאה 8080.

נקודות קצה (אופציונלי)

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

נקודות הקצה האלה הן אופציונליות ברמת ה-API, אבל הטמעה שלהן מאפשרת את תכונות השילוב הבאות:

• תמיכה ב-Python SDK: הטמעה של /api/reasoning_engine ו-/api/stream_reasoning_engine מאפשרת להשתמש בנציג שנפרס באמצעות Gemini Enterprise Agent Platform Python SDK.
• תמיכה ב-Playground: צריך להטמיע את /api/stream_reasoning_engine אם רוצים לקיים אינטראקציה עם הנציג דרךGoogle Cloud console playground. מידע נוסף זמין במאמר תמיכה ב-Playground.

כדי להשתמש בתכונות האלה, צריך להטמיע את נקודות הקצה הבאות:

• ‫/api/reasoning_engine: משמשת להצגת שאילתות שנשלחות ל-API בארכיטקטורת REST של reasoningEngines/query או לשיטות השאילתות הסינכרוניות והאסינכרוניות של Python SDK.
• ‫/api/stream_reasoning_engine: משמש להצגת שאילתות שנשלחות ל-API בארכיטקטורת REST של reasoningEngines/streamQuery או ל-methods של שאילתות סטרימינג ב-Python SDK.

שיטות של מחלקות ומצבי הפעלה

כשפורסים קונטיינר בהתאמה אישית, צריך להצהיר על השיטות של המחלקות הנתמכות ברשימה classMethods של מפרט הפריסה. השיטות האלה תואמות לפעולות שהגדרתם כשפיתחתם את הסוכן (ראו רישום שיטות בהתאמה אישית ושליחת שאילתות לסוכן באמצעות פעולות נתמכות). לכל שיטה מוצהרת יש name ו-api_mode שקובעים איך היא מנותבת:

מצב API	סוג הביצוע	נקודת קצה לניתוב
‫"" (מחרוזת ריקה) או "async"	Unary (בקשה-תגובה)	/api/reasoning_engine
"stream" או "async_stream"	סטרימינג	/api/stream_reasoning_engine

כשלקוח מפעיל שיטה, שירות Agent Runtime שולח בקשת POST לנקודת הקצה התואמת לניתוב בקונטיינר. גוף ה-JSON של הבקשה מכיל את השדה class_method (שתואם לשם השיטה) ואת השדה input.

השיטות הנדרשות לשילוב

כדי להשתמש ב-Python SDK או במגרש המשחקים של מסוף Google Cloud , הקונטיינר צריך להטמיע את השיטות הספציפיות שהשילובים האלה מצפים להן:

• שאילתה רגילה של SDK: נדרשות ההרשאות query (מצב "" או "async") ו-stream_query (מצב "stream" או "async_stream").
• ‫Playground: נדרש stream_query (מצב "stream" או "async_stream").

שילוב ADK

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

אפשר להשתמש בתבנית AdkApp כהטמעה לדוגמה. למידע נוסף, קראו את מאמרי העזרה של AdkApp ואת קוד המקור של AdkApp.

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

מפרט ל-API

גם /api/reasoning_engine וגם /api/stream_reasoning_engine מקבלים בקשות HTTP POST עם גוף JSON שמכיל את השדות הבאים:

• ‫class_method (מחרוזת): שם השיטה של הסוכן הבסיסי שצריך להפעיל (לדוגמה, query או stream_query).
• ‫input (אובייקט JSON): הארגומנטים שיועברו ל-method של המחלקה שצוינה.

/api/reasoning_engine (אונרי)

• שיטת הבקשה: POST
• גוף הבקשה: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
• ‫Response: אובייקט JSON שמכיל את הפלט של הפעלת הנציג. json { "output": "The capital of France is Paris." }

‫/api/stream_reasoning_engine (סטרימינג)

• שיטת הבקשה: POST
• גוף הבקשה: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
• ‫Response: סטרימינג של JSON מופרד בשורות (ndjson), שבו כל שורה היא קטע מהתשובה בקידוד JSON. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

דוגמה לשרת API ‏ (Python)

הדוגמה הבאה היא שרת FastAPI ב-Python שמטמיע את חוזה זמן הריצה של Agent Platform. השרת הזה כולל סוכן לדוגמה (SimpleAgent) ומטפל בניתוב ובקידוד. אתם יכולים להחליף את SimpleAgent בהטמעה של הסוכן שלכם.

כדי להריץ את הדוגמה הזו, צריך לוודא שהתקנתם את fastapi, uvicorn ו-pydantic.

import inspect
import json
import logging
import os
import uvicorn
from fastapi import FastAPI, encoders, responses
from pydantic import BaseModel

app = FastAPI()

# Define the request body structure
class QueryRequest(BaseModel):
    input: dict | None = None
    class_method: str

# Example Agent implementation
class SimpleAgent:
    def query(self, message: str) -> str:
        return f"Echo: {message}"

    async def stream_query(self, message: str):
        words = message.split()
        for word in words:
            yield {"output": word + " "}

agent = SimpleAgent()

def _encode_chunk_to_json(chunk):
    """Encodes a chunk to a JSON string with a newline."""
    try:
        json_chunk = encoders.jsonable_encoder(chunk)
        return json.dumps(json_chunk) + "\n"
    except Exception:
        logging.exception("Failed to encode chunk")
        return None

async def json_generator(output):
    async for chunk in output:
        encoded_chunk = _encode_chunk_to_json(chunk)
        if encoded_chunk is None:
            break
        yield encoded_chunk

async def _invoke_callable_or_raise(invocation_callable, invocation_payload):
    if inspect.iscoroutinefunction(invocation_callable):
        return await invocation_callable(**invocation_payload)
    else:
        return invocation_callable(**invocation_payload)

@app.post("/api/reasoning_engine")
async def query_endpoint(request: QueryRequest) -> responses.JSONResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.JSONResponse(
            status_code=400,
            content={"error": f"Method {request.class_method} not found on agent"}
        )

    output = await _invoke_callable_or_raise(method, request.input or {})

    try:
        json_serialized_content = encoders.jsonable_encoder({"output": output})
    except ValueError as encoding_error:
        logging.exception("Failed to JSON-encode response: %s", encoding_error)
        raise encoding_error
    return responses.JSONResponse(content=json_serialized_content)

@app.post("/api/stream_reasoning_engine")
async def stream_query_endpoint(request: QueryRequest) -> responses.StreamingResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.StreamingResponse(
            content=iter([json.dumps({"error": f"Method {request.class_method} not found"})]),
            status_code=400,
            media_type="application/json"
        )

    output = await _invoke_callable_or_raise(method, request.input or {})
    return responses.StreamingResponse(
        content=json_generator(output),
        media_type="application/json",
    )

if __name__ == "__main__":
    # The container must listen on 0.0.0.0 and port 8080
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))