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

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

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

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

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

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

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

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

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

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

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

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

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

מצב API	סוג הביצוע	נקודת קצה לניתוב
‫"" (מחרוזת ריקה) או "async"	אונארי (בקשה-תגובה)	/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").
• סביבת משחקים: נדרש stream_query (מצב "stream" או "async_stream").

שילוב של ADK

אם אתם פורסים סוכן שנוצר באמצעות ערכת פיתוח סוכנים (ADK), אתם יכולים ליצור קובץ proxy משלכם בשפת התכנות ובמסגרת השרת שתבחרו. כדי לתמוך בכל התכונות של 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 (string): שם השיטה של הסוכן הבסיסי שיופעל (לדוגמה, query או stream_query).
• ‫input (אובייקט JSON): הארגומנטים שיועברו לשיטת המחלקה שצוינה.

‫/api/reasoning_engine (Unary)

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

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

• Request method: ‏ 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)))