Callbacks

Callbacks sind eine erweiterte Funktion, die einen leistungsstarken Mechanismus bietet, um mit Python-Code in den Ausführungsprozess eines bestimmten Agenten einzugreifen. Sie ermöglichen es Ihnen, das Verhalten des Agents an bestimmten, vordefinierten Punkten zu beobachten, anzupassen und sogar zu steuern.

Es gibt verschiedene Rückruftypen, die Sie verwenden können. Jeder Rückruftyp wird an einem bestimmten Punkt in der Unterhaltungsrunde ausgeführt. Diese Typen werden in den folgenden Abschnitten beschrieben.

Python-Laufzeit und -Klassen

In Ihrem Python-Callback-Code haben Sie Zugriff auf bestimmte Klassen und Funktionen, die Ihnen beim Schreiben des Codes helfen. Weitere Informationen finden Sie in der Python-Laufzeitreferenz.

Callback-Typen

Je nach Art des Callbacks muss Ihre primäre Callback-Funktion einen bestimmten Namen haben. So können Sie Hilfsfunktionen mit beliebigem Namen in Ihrem Callback-Code definieren.

Jeder Callback-Typ wird zu einem bestimmten Zeitpunkt in der Unterhaltungsrunde ausgeführt:

Callback-Ablauf

Wenn Sie mehrere Callbacks eines bestimmten Typs definieren, werden sie in der Reihenfolge ausgeführt, in der Sie sie definieren.

In den folgenden Abschnitten werden die einzelnen Callback-Typen beschrieben. Für jeden Typ werden die folgenden Informationen angegeben:

X X
Name Erforderlicher Name der Callback-Funktion
Ausführung Ausführungspunkt innerhalb der Unterhaltungsrunde.
Zweck Nützliche Szenarien für die Verwendung des Callbacks.
Argumente Eingabeargumente für die Funktion.
Zurücksenden Rückgabewert für die Funktion.
ADK-Rückruf Link zur entsprechenden ADK-Callback-Dokumentation.

Vor dem Start des Agenten (before_agent_callback)

X X
Name before_agent_callback
Ausführung Wird aufgerufen, bevor der Agent aufgerufen wird.
Zweck Nützlich, um Ressourcen oder den für den Agenten erforderlichen Status einzurichten, Validierungsprüfungen für den Sitzungsstatus durchzuführen oder den Aufruf des Agenten zu vermeiden.
Argumente CallbackContext
Zurücksenden Inhalt(optional): Wenn festgelegt, wird der Agent nicht aufgerufen und die bereitgestellte Antwort wird verwendet.
ADK-Rückruf before_agent_callback

Codebeispiel:

import random

def before_agent_callback(
  callback_context: CallbackContext
) -> Optional[Content]:
  username = callback_context.variables.get("username", None)
  if not username:
    # default user
    final_name = "Default Name"
  else:
    # add a random integer to the username
    final_name = f"{username} {random.randint(1,10)}"
  # update the username variable
  callback_context.variables["username"] = final_name

Nachdem der Agent fertig ist (after_agent_callback)

X X
Name after_agent_callback
Ausführung Wird nach Abschluss des Agenten aufgerufen.
Zweck Nützlich für Bereinigungsaufgaben, zum Validieren nach der Ausführung, zum Ändern des Endzustands oder zum Aktualisieren der Agent-Antwort.
Argumente CallbackContext
Zurücksenden Inhalt(optional): Wenn festgelegt, wird die Ausgabe des Agents durch die angegebene Ausgabe ersetzt.
ADK-Rückruf after_agent_callback

Codebeispiel:

def after_agent_callback(
  callback_context: CallbackContext
) -> Optional[Content]:
  if callback_context.agent_name == "Routing Agent":
    counter = callback_context.variables.get("counter", 0)
    counter += 1
    # increment the invoked counter for this agent
    callback_context.variables["counter"] = int(counter)

Vor dem LLM-Aufruf (before_model_callback)

X X
Name before_model_callback
Ausführung Wird vor der Modellanfrage aufgerufen.
Zweck Nützlich für die Überprüfung/Änderung der Modellanfrage oder um die Verwendung des Modells zu vermeiden.
Argumente CallbackContext, LlmRequest
Zurücksenden LlmResponse: Wenn diese Option festgelegt ist, wird der Modellaufruf übersprungen und die Antwort wird so verwendet, als käme sie vom Modell.
ADK-Rückruf before_model_callback

Codebeispiel:

def before_model_callback(
  callback_context: CallbackContext,
  llm_request: LlmRequest
) -> Optional[LlmResponse]:
  """
  This callback executes *before* a request is sent to the LLM.

  By returning an `LlmResponse` object, we are intercepting the call to the
  LLM. The LLM will *not* be called, and the framework will instead use the
  `LlmResponse` we provide as if it came from the model.

  This is the core mechanism for implementing input guardrails, prompt
  validation, or serving responses from a cache. Here, we force the agent to
  call a function instead of thinking with the LLM.
  """
  # Modify the shared session state.
  callback_context.variables['foo'] = 'baz'

  # Skip the LLM call and return a custom response telling the agent to
  # execute a specific function.
  return LlmResponse(
    content=Content(parts=[Part(
      function_call=FunctionCall(
        name="function_name", args={"arg_name": "arg_value"}))],
      role="model"))

Nach dem LLM-Aufruf (after_model_callback)

X X
Name after_model_callback
Ausführung Wird aufgerufen, nachdem eine Modellantwort empfangen wurde.
Zweck Nützlich zum Umformatieren von Modellantworten, zum Zensieren sensibler Informationen, die vom Modell generiert wurden, zum Parsen strukturierter Daten aus dem Modell zur Verwendung in Variablen und zur Fehlerbehandlung des Modells.
Argumente CallbackContext, LlmResponse
Zurücksenden LlmResponse: Wenn dieser Wert festgelegt ist, wird die Modellantwort durch die angegebene Antwort ersetzt.
ADK-Rückruf after_model_callback

Codebeispiel:

def after_model_callback(
  callback_context: CallbackContext,
  llm_response: LlmResponse
) -> Optional[LlmResponse]:
  """
  This callback executes *after* a response has been received from the LLM,
  but before the agent processes it.

  The `llm_response` parameter contains the actual data from the LLM.
  By returning `None`, we are approving this response and allowing the agent
  to use it as-is.

  If we returned a new `LlmResponse` object, it would *replace* the original,
  which is useful for redacting sensitive information, enforcing output
  formatting, or adding disclaimers.
  """
  # Returning None allows the LLM's actual response to be used.
  return None

Vor dem Tool-Aufruf („before_tool_callback“)

X X
Name before_tool_callback
Ausführung Wird vor Tool-Aufrufen aufgerufen.
Zweck Nützlich für die Überprüfung und Änderung von Tool-Argumenten, Autorisierungsprüfungen vor der Tool-Ausführung oder die Implementierung von Caching auf Tool-Ebene.
Argumente Tool, Dict[str,Any]: Tool-Eingaben, CallbackContext
Zurücksenden Dict[str,Any] : Wenn festgelegt, wird die Tool-Ausführung übersprungen und diese Ausgabe wird dem Modell bereitgestellt.
ADK-Rückruf vor dem Tool-Callback

Codebeispiel:

def before_tool_callback(
  tool: Tool,
  input: dict[str, Any],
  callback_context: CallbackContext
) -> Optional[dict[str, Any]]:
  """
  This callback executes *before* a specific tool is called by the agent.

  Here, we modify the input arguments intended for the tool and then return
  a dictionary. By returning a dictionary instead of `None`, we are
  overriding the default behavior. The actual tool function will *not* be
  executed. Instead, the dictionary we return will be treated as the
  llm.tool's result and passed back to the LLM for the next step.

  This is ideal for validating tool inputs, applying policies, or returning
  mocked/cached data for testing.
  """
  # Modify the shared session state.
  callback_context.variables['foo'] = 'baz'

  # Modify the arguments for the tool call in-place.
  input['input_arg'] = 'updated_val1'
  input['additional_arg'] = 'updated_val2'

  # Override the tool call and return a mocked result.
  return {"result": "ok"}

Nach dem Toolaufruf (after_tool_callback)

X X
Name after_tool_callback
Ausführung Wird nach Abschluss des Tools aufgerufen.
Zweck Nützlich für die Überprüfung und Änderung von Tool-Antworten, bevor sie an das Modell zurückgesendet werden, für die Nachbearbeitung von Tool-Ergebnissen oder zum Speichern bestimmter Teile einer Tool-Antwort in Variablen.
Argumente Tool, Dict[str,Any]: Tool-Eingaben, CallbackContext, Dict[str,Any]: Tool-Antwort
Zurücksenden Dict[str,Any]: Wenn festgelegt, wird die Tool-Antwort, die dem Modell bereitgestellt wird, damit überschrieben.
ADK-Rückruf nach dem Tool-Rückruf

Codebeispiel:

# Previous tool was named `get_user_info`
# Previous tool returned the payload:
# {"username": "Patrick", "fave_food": ["pizza"]}

def after_tool_callback(
  tool: Tool,
  input: dict[str, Any],
  callback_context: CallbackContext,
  tool_response: dict
) -> Optional[dict]:

  if tool.name == "get_user_info":
    tool_response["username"] = "Gary"
    tool_response["pet"] = "dog"

    # Override tool response
    return tool_response

Rückruf erstellen

So erstellen Sie einen Rückruf:

  1. Öffnen Sie die Agent-Einstellungen.
  2. Klicken Sie auf Code hinzufügen.
  3. Wählen Sie einen Rückruftyp aus.
  4. Python-Code bereitstellen
  5. Klicken Sie auf Speichern.

Benutzerdefinierte Nutzlasten (custom_payloads)

Benutzerdefinierte Nutzlasten ermöglichen die Einbeziehung zusätzlicher, nicht textbasierter, strukturierter Daten (in der Regel als JSON formatiert) in die Antwort eines Agenten. Diese Nutzlast ist entscheidend, um die Interaktion des Agents mit externen Systemen oder Clientanwendungen zu steuern oder zu erweitern.

Der Nutzlastwert ist für das Large Language Model (LLM) nicht sichtbar. Er wird nur zum Generieren der endgültigen Antwort verwendet. Benutzerdefinierte Nutzlasten werden mithilfe von Callbacks generiert und festgelegt, insbesondere mit before_model_callback oder after_model_callback.

Die benutzerdefinierte Nutzlast kann für verschiedene Zwecke verwendet werden, die in der Regel auf die Ermöglichung umfangreicher, strukturierter Interaktionen ausgerichtet sind:

  • Eskalierung/Übergabe an Kundenservicemitarbeiter: Diese Funktion wird häufig verwendet, um eine Interaktion an einen menschlichen Kundenservicemitarbeiter zu übergeben. Dazu werden Weiterleitungsanweisungen bereitgestellt, z. B. die spezifische Warteschlange, an die weitergeleitet werden soll.
  • Rich Content und clientseitige Aktionen: Es unterstützt das Einbetten von Rich-Widgets und anderen Rich-Content-Elementen direkt in Chat-Umgebungen, was besonders für benutzerdefinierte Chat-Integrationen nützlich ist.
    • Beispiele hierfür sind die Anzeige von Bild-URLs oder Schnellantwort-Chips und Optionen für einen Kunden über eine Schnittstelle wie Call Companion.
  • Antwortzusammensetzung: Benutzerdefinierte Nutzlasten können so konfiguriert werden, dass sie auf verschiedene Arten zurückgegeben werden:
    • Gibt nur die explizite Nutzlast deterministisch zurück.
    • Gibt die Nutzlast zusammen mit einer von einem LLM generierten Textantwort zurück.
    • Nutzlast mit einer statischen Textantwort zurückgeben

Agent-Einrichtung

Benutzerdefinierte Nutzlasten können nur mit Callbacks generiert und festgelegt werden. Die Nutzlast wird als Blob mit einem mime_type von application/json festgelegt.

Part.from_json(data=payload_string)

Beispiel für after_model_callback

Dies ist ein Beispiel für einen „after_model_callback“, der die Modellantwort zusammen mit einer zusätzlichen benutzerdefinierten Nutzlastantwort zurückgibt.

import json

def after_model_callback(callback_context: CallbackContext, llm_response: LlmResponse) -> Optional[LlmResponse]:
 """
 Adds a custom payload to every model response which is a text
 """
 if (llm_response.content.parts[0].text is not None):
   # construct payload
   payload_dict = { "custom_payload_key": "custom_payload_value"}
   payload_json_string = json.dumps(payload_dict)

   new_parts = []
   # Keep the origial agent response part, as model only sees text in the historical context.
   new_parts.append(Part(text=llm_response.content.parts[0].text))

   # Append custom payload
   new_parts.append(Part.from_json(data=payload_string))

   return LlmResponse(content=Content(parts=new_parts))

Beispiel für before_model_callback

Dies ist ein Beispiel für „before_model_callback“, das nach dem Auslösen eines bestimmten Tools eine zusätzliche benutzerdefinierte Nutzlast zurückgibt.

import json

def has_escalate(llm_request: LlmRequest) -> bool:
  for content in llm_request.contents:
    for part in content.parts:
      if part.function_call and part.function_call.name == 'escalate':
        return True
  return False

def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]:
  # checks if `escalate` tool is being called
  if not has_escalate(llm_request):
    return None
  payload_dict = { "escalate": "user ask for escalation"}
  payload_json_string = json.dumps(payload_dict)

  return LlmResponse(content=Content(parts=[Part(text="ESCALATE!!!"), Part.from_json(data=payload_json_string)]))

Nutzlast in der Antwort zur Laufzeit prüfen

Die Nutzlast wird als Struct im Feld payload für RunSession und BidiRunSession eingefügt.

Der Nutzlastwert ist für das LLM nicht sichtbar.