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

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

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

1. דוגמה בסיסית
2. אופציונלי: הצגת התשובות באופן שוטף
3. אופציונלי: רישום של שיטות מותאמות אישית
4. אופציונלי: הוספת הערות לגבי סוגים
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.
   • הבנאי מחזיר אובייקט שצריך להיות ניתן ל-pickle כדי שאפשר יהיה לפרוס אותו ב-Agent Runtime. לכן, צריך לאתחל לקוחות שירותים וליצור חיבורים למסדי נתונים בשיטה .set_up במקום בשיטה __init__.
   • השיטה הזו היא אופציונלית. אם לא מציינים, Agent Runtime משתמש בבונה ברירת המחדל של Python עבור המחלקה.
2. set_up():
   • חובה להשתמש בשיטה הזו כדי להגדיר את הלוגיקה של אתחול הסוכן. לדוגמה, אתם יכולים להשתמש בשיטה הזו כדי ליצור חיבורים למסדי נתונים או לשירותים תלויים, לייבא חבילות תלויות או לבצע חישוב מראש של נתונים שמשמשים להצגת שאילתות.
   • השיטה הזו היא אופציונלית. אם לא מציינים, 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

אפשר לבדוק את הנציג באופן מקומי באמצעות קריאה לשיטה 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()

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

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

בדיקת השיטה stream_query

כדי לבדוק את שאילתת הסטרימינג באופן מקומי, מפעילים את השיטה 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

בדומה לקוד לבדיקת שאילתות סטרימינג, אפשר לבדוק את הסוכן באופן מקומי על ידי קריאה לשיטה 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 רשומות כפעולות בסוכן שנפרס. אפשר לשנות את התנהגות ברירת המחדל ולהגדיר את קבוצת הפעולות שיירשמו באמצעות ה-method‏ register_operations. אפשר לרשום פעולות במצבי ביצוע רגילים (מיוצגים על ידי מחרוזת ריקה "") או במצבי ביצוע של סטרימינג ("stream").

כדי לרשום כמה פעולות, אפשר להגדיר method בשם register_operations שמפרטת את ה-methods שיהיו זמינות למשתמשים כשהסוכן יופעל. בדוגמת הקוד הבאה, המתודה 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", "get_state"]} בשיטה register_operations בלי לציין רשימה תואמת של שיטות ל-"stream".

כדי לבדוק את השיטות המותאמות אישית, קוראים להן ישירות במופע המקומי של הנציג, בדומה לבדיקה של השיטות 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, אפשר לייבא אותן ולהפעיל אותן בשיטה .set_up. במקרה של מסגרות נפוצות של סוכנים, יכול להיות שאפשר להשתמש בשילוב של Open Telemetry Google Cloud בשילוב עם מסגרת מכשור כמו OpenInference או OpenLLMetry.

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

OpenInference

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

pip install openinference-instrumentation-langchain==0.1.34

בשלב הבא, מייבאים ומפעילים את הכלי למעקב אחר ביצועים:

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

בשלב הבא, מייבאים ומפעילים את הכלי למעקב אחר ביצועים:

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. פועלים לפי ההוראות במאמר הקצאת תפקידים לסוכן שנפרס כדי להקצות לחשבון השירות את התפקיד Secret Manager Secret Accessor (roles/secretmanager.secretAccessor) דרך מסוף Google Cloud .

3. מייבאים ומאתחלים את הלקוח בשיטה .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. פרטי כניסה שמוגדרים כברירת מחדל באפליקציה (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, שאפשר להפוך אותו למעטפת.

פלטפורמת הסוכנים יכולה לטפל בקודים שונים של סטטוסים באופן פנימי, אבל ב-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 גנרטיבי כדי להעריך את הסוכן

פתרון בעיות

פתרון בעיות ביצירת סוכנים

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

Resource

קבלת תמיכה

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