יצירת סוכן בהתאמה אישית

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

כך יוצרים תבנית בהתאמה אישית ליצירת מופעים של סוכנים שאפשר לפרוס ב-Agent Platform:

  1. דוגמה בסיסית
  2. אופציונלי: הצגת התשובות באופן שוטף
  3. אופציונלי: רישום שיטות בהתאמה אישית
  4. אופציונלי: הוספת הערות מסוג Type
  5. אופציונלי: שליחת נתוני מעקב אל Cloud Trace
  6. אופציונלי: עבודה עם משתני סביבה
  7. אופציונלי: שילוב עם Secret Manager
  8. אופציונלי: טיפול בפרטי כניסה
  9. אופציונלי: טיפול בשגיאות

דוגמה בסיסית

כדוגמה בסיסית, המחלקה הבאה ב-Python היא תבנית ליצירת מופעים של סוכנים שאפשר לפרוס ב-Agent Platform (אפשר לתת למשתנה CLASS_NAME ערך כמו MyAgent):

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

שיקולים לגבי הפריסה

כשכותבים את המחלקה ב-Python, חשוב להשתמש בשלוש השיטות הבאות:

  1. __init__():
    • השיטה הזו מתאימה רק לפרמטרים של הגדרת הנציג. לדוגמה, אתם יכולים להשתמש בשיטה הזו כדי לאסוף את הפרמטרים של המודל ומאפייני הבטיחות כארגומנטים של קלט מהמשתמשים. אפשר גם להשתמש בשיטה הזו כדי לאסוף פרמטרים כמו מזהה הפרויקט, האזור, פרטי הכניסה לאפליקציה ומפתחות ה-API.
    • ה-constructor מחזיר אובייקט שחייב להיות "pickle-able" כדי שאפשר יהיה לפרוס אותו ב-Agent Runtime. לכן, צריך להפעיל לקוחות שירות וליצור חיבורים למסדי נתונים ב-method‏ .set_up במקום ב-method‏ __init__.
    • השיטה הזו היא אופציונלית. אם לא מציינים פונקציה, Agent Runtime משתמש ב-constructor של Python שמוגדר כברירת מחדל למחלקה.
  2. set_up():
    • אתם חייבים להשתמש ב-method הזה כדי להגדיר את הלוגיקה של אתחול הנציג. לדוגמה, אתם יכולים להשתמש בשיטה הזו כדי ליצור חיבורים למסדי נתונים או לשירותים תלויים, לייבא חבילות תלויות או לחשב מראש נתונים שמשמשים להצגת שאילתות.
    • השיטה הזו היא אופציונלית. אם לא מציינים את זה, Agent Runtime מניח שהנציג לא צריך להפעיל שיטה של .set_up לפני שהוא מטפל בשאילתות של המשתמשים.
  3. query() / stream_query():
    • משתמשים ב-query() כדי להחזיר את התשובה המלאה כתוצאה אחת.
    • כדי להחזיר את התשובה בחלקים כשהיא הופכת לזמינה, צריך להשתמש ב-stream_query(), וכך לאפשר חוויית סטרימינג. השיטה stream_query צריכה להחזיר אובייקט שניתן לחזור עליו (לדוגמה, גנרטור) כדי לאפשר סטרימינג.
    • אפשר להטמיע את שתי השיטות אם רוצים לתמוך גם באינטראקציות של תשובה אחת וגם באינטראקציות בסטרימינג עם הנציג.
    • צריך לתת למתודה הזו מחרוזת תיעוד ברורה שמגדירה מה היא עושה, מתעדת את המאפיינים שלה ומספקת הערות לגבי סוגי הקלט שלה. מומלץ להימנע מארגומנטים משתנים בשיטות query ו-stream_query.

יצירת מופע של הסוכן באופן מקומי

אתם יכולים ליצור מכונה מקומית של הנציג באמצעות הקוד הבא:

agent = CLASS_NAME(
    model=model,  # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

בדיקת השיטה query

כדי לבדוק את הסוכן, אפשר לשלוח שאילתות למכונה המקומית:

response = agent.query(
    input="What is the exchange rate from US dollars to Swedish currency?"
)

print(response)

התגובה היא מילון שדומה לזה:

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

שליחת שאילתות באופן אסינכרוני

כדי להגיב לשאילתות באופן אסינכרוני, אתם יכולים להגדיר שיטה (כמו async_query) שמחזירה Python Coroutine. לדוגמה, התבנית הבאה מרחיבה את הדוגמה הבסיסית כדי להגיב באופן אסינכרוני, ואפשר לפרוס אותה ב-Agent Platform:

class AsyncAgent(CLASS_NAME):

    async def async_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.ainvoke(**kwargs):
            yield dumpd(chunk)

agent = AsyncAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

בדיקת השיטה async_query

כדי לבדוק את הנציג באופן מקומי, אפשר להפעיל את ה-method‏ async_query. לדוגמה:

response = await agent.async_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
)
print(response)

התגובה היא מילון שדומה לזה:

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

תשובות בסטרימינג

כדי להעביר תשובות לשאילתות בסטרימינג, אתם יכולים להגדיר שיטה בשם stream_query שמניבה תשובות. לדוגמה, התבנית הבאה מרחיבה את הדוגמה הבסיסית כדי להעביר תשובות בסטרימינג, ואפשר לפרוס אותה ב-Agent Platform:

from typing import Iterable

class StreamingAgent(CLASS_NAME):

    def stream_query(self, **kwargs) -> Iterable:
        from langchain.load.dump import dumpd

        for chunk in self.graph.stream(**kwargs):
            yield dumpd(chunk)

agent = StreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

הנה כמה דברים חשובים שכדאי לזכור כשמשתמשים ב-API לסטרימינג:

  • זמן קצוב מקסימלי לתפוגה: הזמן הקצוב המקסימלי לתפוגה של תשובות בסטרימינג הוא 15 דקות. אם לנציג נדרש זמן עיבוד ארוך יותר, כדאי לפצל את המשימה לחלקים קטנים יותר.
  • מודלים ושרשראות בסטרימינג: ממשק ה-Runnable של LangChain תומך בסטרימינג, כך שאפשר להעביר תשובות בסטרימינג לא רק מסוכנים, אלא גם ממודלים ומשרשראות.
  • תאימות ל-LangChain: שימו לב ששיטות אסינכרוניות כמו השיטה astream_event של LangChain לא נתמכות כרגע.
  • הגבלת קצב יצירת התוכן: אם אתם נתקלים בבעיות של לחץ נגדי (מצב שבו המפיק יוצר נתונים מהר יותר מהקצב שבו הצרכן יכול לעבד אותם), כדאי להגביל את קצב יצירת התוכן. כך אפשר למנוע גלישת מאגר נתונים (buffer overflow) וליהנות מחוויית סטרימינג חלקה.

בדיקת השיטה stream_query

כדי לבדוק את שאילתת הסטרימינג באופן מקומי, אפשר להפעיל את ה-method‏ stream_query ולעבור על התוצאות. הנה דוגמה:

import pprint

for chunk in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

הקוד הזה מדפיס כל חלק בתשובה כשהוא נוצר. הפלט עשוי להיראות כך:

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

בדוגמה הזו, כל קטע מכיל מידע שונה על התשובה, כמו הפעולות שהנציג ביצע, ההודעות שהוחלפו והפלט הסופי.

הצגת תשובות באופן אסינכרוני

כדי להעביר תשובות בסטרימינג באופן אסינכרוני, אתם יכולים להגדיר שיטה (כמו async_stream_query) שמחזירה גנרטור אסינכרוני. לדוגמה, התבנית הבאה מרחיבה את הדוגמה הבסיסית כדי להעביר תשובות בסטרימינג באופן אסינכרוני, ואפשר לפרוס אותה ב-Agent Platform:

class AsyncStreamingAgent(CLASS_NAME):

    async def async_stream_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.astream(**kwargs):
            yield dumpd(chunk)

agent = AsyncStreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

בדיקת השיטה async_stream_query

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

import pprint

async for chunk in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

הקוד הזה מדפיס כל חלק בתשובה כשהוא נוצר. הפלט עשוי להיראות כך:

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

רישום שיטות בהתאמה אישית

כברירת מחדל, ה-methods‏ query ו-stream_query רשומות כפעולות בנציג שנפרס. אפשר לשנות את התנהגות ברירת המחדל ולהגדיר את קבוצת הפעולות שצריך לרשום באמצעות השיטה register_operations. אפשר לרשום פעולות במצב הפעלה רגיל (מיוצג על ידי מחרוזת ריקה "") או במצב הפעלה בסטרימינג ("stream").

כדי לרשום כמה פעולות, אפשר להגדיר שיטה בשם register_operations שמפרטת את השיטות שיהיו זמינות למשתמשים כשהנציג ייפרס. בקוד לדוגמה הבא, השיטה register_operations תגרום לסוכן שנפרס לרשום את query ו-get_state כפעולות שפועלות באופן סינכרוני, ואת stream_query ו-get_state_history כפעולות שמעבירות את התשובות בסטרימינג:

from typing import Iterable

class CustomAgent(StreamingAgent):

    def get_state(self) -> dict: # new synchronous method
        return self.graph.get_state(**kwargs)._asdict()

    def get_state_history(self) -> Iterable: # new streaming operation
        for state_snapshot in self.graph.get_state_history(**kwargs):
            yield state_snapshot._asdict()

    def register_operations(self):
        return {
            # The list of synchronous operations to be registered
            "": ["query", "get_state"],
            # The list of streaming operations to be registered
            "stream": ["stream_query", "get_state_history"],
        }

כדי לבדוק את השיטות בהתאמה אישית, אתם יכולים להפעיל אותן ישירות במכונה המקומית של הסוכן, בדומה לבדיקה של השיטות query ו-stream_query.

הוספת הערות לגבי סוגים

אתם יכולים להשתמש בהערות מסוגים שונים כדי לציין את סוגי הקלט והפלט הצפויים של השיטות של הנציג. כשפורסים את הנציג, רק סוגים שניתן להמיר ל-JSON נתמכים בקלט ובפלט של הפעולות שהנציג תומך בהן. אפשר להוסיף הערות לסכימות של הקלט והפלט באמצעות מודלים של TypedDict או Pydantic.

בדוגמה הבאה, אנחנו מוסיפים הערה לקלט בתור TypedDict, וממירים את הפלט הגולמי מ-.get_state (שהוא NamedTuple) למילון שניתן לסדרתיות באמצעות השיטה ._asdict():

from typing import Any, Dict, TypedDict

# schemas.py
class RunnableConfig(TypedDict, total=False):
    metadata: Dict[str, Any]
    configurable: Dict[str, Any]

# agents.py
class AnnotatedAgent(CLASS_NAME):

    def get_state(self, config: RunnableConfig) -> dict:
        return self.graph.get_state(config=config)._asdict()

    def register_operations(self):
        return {"": ["query", "get_state"]}

שליחת נתוני מעקב אל Cloud Trace

כדי לשלוח נתוני מעקב ל-Cloud Trace באמצעות ספריות אינסטרומנטציה שתומכות ב-OpenTelemetry, אפשר לייבא ולאתחל אותן ב-method‏ .set_up. במסגרות נפוצות של סוכנים, יכול להיות שתוכלו להשתמש בשילוב של Google CloudOpen Telemetry עם מסגרת אינסטרומנטציה כמו OpenInference או OpenLLMetry.

לדוגמה, התבנית הבאה היא שינוי של הדוגמה הבסיסית לייצוא של נתוני מעקב ל-Cloud Trace:

OpenInference

קודם מתקינים את החבילה הנדרשת באמצעות pip על ידי הפעלת הפקודה הבאה:

pip install openinference-instrumentation-langchain==0.1.34

בשלב הבא, מייבאים ומפעילים את ה-instrumentor:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from openinference.instrumentation.langchain import LangChainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangChainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

OpenLLMetry

קודם מתקינים את החבילה הנדרשת באמצעות pip על ידי הפעלת הפקודה הבאה:

pip install opentelemetry-instrumentation-langchain==0.38.10

בשלב הבא, מייבאים ומפעילים את ה-instrumentor:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from opentelemetry.instrumentation.langchain import LangchainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangchainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

עבודה עם משתני סביבה

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

שילוב עם Secret Manager

כדי לשלב עם Secret Manager:

  1. מתקינים את ספריית הלקוח באמצעות הפקודה הבאה:

    pip install google-cloud-secret-manager

  2. פועלים לפי ההוראות במאמר הקצאת תפקידים לסוכן שנפרס כדי להקצות לחשבון השירות את התפקיד "מנהל סודות עם גישה לסודות" (roles/secretmanager.secretAccessor) באמצעות מסוף Google Cloud .

  3. מייבאים ומפעילים את הלקוח ב-method‏ .set_up ומקבלים את הסוד המתאים לפי הצורך. לדוגמה, התבנית הבאה היא שינוי של הדוגמה הבסיסית, כדי להשתמש במפתח API של ChatAnthropic שמאוחסן ב-Secret Manager:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.secret_id = secret_id # <- new

    def set_up(self):
        from google.cloud import secretmanager
        from langchain_anthropic import ChatAnthropic
        from langgraph.prebuilt import create_react_agent

        # Get the API Key from Secret Manager here.
        self.secret_manager_client = secretmanager.SecretManagerServiceClient()
        secret_version = self.secret_manager_client.access_secret_version(request={
            "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
        })
        # Use the API Key from Secret Manager here.
        model = ChatAnthropic(
            model_name=self.model_name,
            model_kwargs={"api_key": secret_version.payload.data.decode()},  # <- new
        )
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

טיפול בפרטי כניסה

כשפורסים את הסוכן, יכול להיות שהוא יצטרך לטפל בסוגים שונים של פרטי כניסה:

  1. Application Default Credentials‏ (ADC), שנוצרים בדרך כלל מחשבונות שירות.
  2. OAuth, שנובע בדרך כלל מחשבונות משתמשים, וגם
  3. ספקי זהויות לפרטי כניסה מחשבונות חיצוניים (איחוד שירותי אימות הזהות של עומסי עבודה).

Application Default Credentials

import google.auth

credentials, project = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

אפשר להשתמש בו בקוד באופן הבא:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str = "meta/llama3-405b-instruct-maas",
        tools: Sequence[Callable],
        location: str,
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.endpoint = f"https://{location}-aiplatform.googleapis.com"
        self.base_url = f'{self.endpoint}/v1beta1/projects/{project}/locations/{location}/endpoints/openapi'

    def query(self, **kwargs):
        import google.auth
        from langchain_openai import ChatOpenAI
        from langgraph.prebuilt import create_react_agent

        # Note: the credential lives for 1 hour by default.
        # After expiration, it must be refreshed.
        creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
        creds.refresh(google.auth.transport.requests.Request())

        model = ChatOpenAI(
            model=self.model_name,
            base_url=self.base_url,
            api_key=creds.token,  # Use the token from the credentials here.
        )
        graph = create_react_agent(model, tools=self.tools)
        return graph.invoke(**kwargs)

לפרטים נוספים, קראו את המאמר סקירה כללית על Application Default Credentials.

OAuth

בדרך כלל מקבלים את פרטי הכניסה של המשתמש באמצעות OAuth 2.0.

אם יש לכם אסימון גישה (למשל מ-oauthlib), אתם יכולים ליצור מכונה של google.oauth2.credentials.Credentials. בנוסף, אם אתם מקבלים טוקן רענון, אתם יכולים גם לציין את טוקן הרענון ואת ה-URI של הטוקן כדי לאפשר רענון אוטומטי של פרטי הכניסה:

credentials = google.oauth2.credentials.Credentials(
    token="ACCESS_TOKEN",
    refresh_token="REFRESH_TOKEN",  # Optional
    token_uri="TOKEN_URI",          # E.g. "https://oauth2.googleapis.com/token"
    client_id="CLIENT_ID",          # Optional
    client_secret="CLIENT_SECRET"   # Optional
)

כאן, הערכים TOKEN_URI,‏ CLIENT_ID ו-CLIENT_SECRET מבוססים על יצירת פרטי כניסה של לקוח OAuth.

אם אין לכם טוקן גישה, אתם יכולים להשתמש ב-google_auth_oauthlib.flow כדי לבצע את תהליך הענקת ההרשאה של OAuth 2.0 ולקבל מופע google.oauth2.credentials.Credentials תואם:

from google.cloud import secretmanager
from google_auth_oauthlib.flow import InstalledAppFlow
import json

# Get the client config from Secret Manager here.
secret_manager_client = secretmanager.SecretManagerServiceClient()
secret_version = client.access_secret_version(request={
    "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
})
client_config = json.loads(secret_version.payload.data.decode())

# Create flow instance to manage the OAuth 2.0 Authorization Grant Flow steps.
flow = InstalledAppFlow.from_client_config(
    client_config,
    scopes=['https://www.googleapis.com/auth/cloud-platform'],
    state="OAUTH_FLOW_STATE"  # from flow.authorization_url(...)
)

# You can get the credentials from the flow object.
credentials: google.oauth2.credentials.Credentials = flow.credentials

# After obtaining the credentials, you can then authorize API requests on behalf
# of the given user or service account. For example, to authorize API requests
# to vertexai services, you'll specify it in vertexai.init(credentials=)
import vertexai

vertexai.init(
    project="PROJECT_ID",
    location="LOCATION",
    credentials=credentials, # specify the credentials here
)

פרטים נוספים זמינים במסמכי התיעוד של המודול google_auth_oauthlib.flow.

ספק הזהות

אם אתם רוצים לאמת משתמשים באמצעות אימייל/סיסמה, מספר טלפון, ספקים חברתיים כמו Google, ‏ Facebook או GitHub, או מנגנון אימות בהתאמה אישית, אתם יכולים להשתמש ב-Identity Platform או באימות ב-Firebase, או בכל ספק זהויות שתומך ב-OpenID Connect ‏ (OIDC).

פרטים נוספים זמינים במאמר איך ניגשים למשאבים מספק זהויות מסוג OIDC.

טיפול בשגיאות

כדי לוודא שהשגיאות ב-API יוחזרו בפורמט JSON מובנה, מומלץ להטמיע טיפול בשגיאות בקוד של הנציג באמצעות בלוק try...except, שאפשר להפוך אותו ל-decorator.

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

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

בדוגמה הבאה אפשר לראות איך מטמיעים טיפול בשגיאות בנציג:

from functools import wraps
import json

def error_wrapper(func):
    @wraps(func)  # Preserve original function metadata
    def wrapper(*args, **kwargs):
        try:
            # Execute the original function with its arguments
            return func(*args, **kwargs)
        except Exception as err:
            error_code = getattr(err, 'code')
            error_message = getattr(err, 'error')

            # Construct the error response dictionary
            error_response = {
                "error": {
                    "code": error_code,
                    "message": f"'{func.__name__}': {error_message}"
                }
            }
            # Return the Python dictionary directly.
            return error_response

    return wrapper

# Example exception
class SessionNotFoundError(Exception):
    def __init__(self, session_id, message="Session not found"):
        self.code = 404
        self.error = f"{message}: {session_id}"
        super().__init__(self.error)

# Example Agent Class
class MyAgent:
    @error_wrapper
    def get_session(self, session_id: str):
        # Simulate the condition where the session isn't found
        raise SessionNotFoundError(session_id=session_id)


# Example Usage: Session Not Found
agent = MyAgent()
error_result = agent.get_session(session_id="nonexistent_session_123")
print(json.dumps(error_result, indent=2))

הקוד שלמעלה יוביל לפלט הבא: json { "error": { "code": 404, "message": "Invocation error in 'get_session': Session not found: nonexistent_session_123" } }

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

מדריך

חמש דרכים לפרוס סוכן ב-Agent Platform Runtime בהתאם לצורכי הפיתוח שלכם.

מדריך

שימוש בסוכן עם Agent Platform Runtime.

מדריך

ליצור ולפרוס סוכן בסיסי ולהשתמש בשירות ההערכה של AI גנרטיבי כדי להעריך את הסוכן

פתרון בעיות

איך פותרים שגיאות נפוצות כשיוצרים סוכנים בהתאמה אישית.

משאב

מקורות מידע ותמיכה ל-Google Agent Platform.