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 Agenten an bestimmten, vordefinierten Punkten zu beobachten, anzupassen und sogar zu steuern.

Es gibt verschiedene Callback-Typen, die Sie verwenden können. Jeder Callback-Typ 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 Referenz zur Python-Laufzeit.

Callback-Typen

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

Jeder Callback-Typ wird an einem bestimmten Punkt in der Unterhaltungsrunde ausgeführt:

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

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

X	X
Name	Erforderlicher Name der Callback-Funktion
Ausführung	Ausführungspunkt in 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-Callback	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 zum Einrichten von Ressourcen oder Status, die für den Agenten erforderlich sind, zum Ausführen von Validierungsprüfungen für den Sitzungsstatus oder zum Vermeiden des Aufrufs des Agenten.
Argumente	CallbackContext
Zurücksenden	Inhalt(optional): Wenn festgelegt, wird der Agent nicht aufgerufen und die angegebene Antwort wird verwendet.
ADK-Callback	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

Nach Abschluss des Agenten (after_agent_callback)

X	X
Name	after_agent_callback
Ausführung	Wird aufgerufen, nachdem der Agent abgeschlossen wurde.
Zweck	Nützlich für Bereinigungsaufgaben, zum Validieren nach der Ausführung, zum Ändern des Endstatus oder zum Aktualisieren der Antwort des Agenten.
Argumente	CallbackContext
Zurücksenden	Inhalt(optional): Wenn festgelegt, wird die Ausgabe des Agenten durch die angegebene Ausgabe ersetzt.
ADK-Callback	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 zum Überprüfen/Ändern der Modellanfrage oder zum Vermeiden der Modellnutzung.
Argumente	CallbackContext, LlmRequest
Zurücksenden	LlmResponse: Wenn festgelegt, wird der Modellaufruf übersprungen und die Antwort wird so verwendet, als käme sie vom Modell.
ADK-Callback	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 vertraulicher 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 festgelegt, wird die Modellantwort durch die angegebene Antwort ersetzt.
ADK-Callback	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 Toolaufruf (before_tool_callback)

X	X
Name	before_tool_callback
Ausführung	Wird vor Toolaufrufen aufgerufen.
Zweck	Nützlich zum Überprüfen und Ändern von Tool-Argumenten, zum Ausführen von Autorisierungsprüfungen vor der Toolausführung oder zum Implementieren von Caching auf Tool-Ebene.
Argumente	Tool, Dict[str,Any]: Tool-Eingaben, CallbackContext
Zurücksenden	Dict[str,Any] : Wenn festgelegt, wird die Toolausführung übersprungen und diese Ausgabe wird dem Modell bereitgestellt.
ADK-Callback	before_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 zum Überprüfen und Ändern der Toolantwort, bevor sie an das Modell zurückgesendet wird, zur Nachbearbeitung von Toolergebnissen oder zum Speichern bestimmter Teile einer Toolantwort in Variablen.
Argumente	Tool, Dict[str,Any]: Tool-Eingaben, CallbackContext, Dict[str,Any]: Toolantwort
Zurücksenden	Dict[str,Any]: Wenn festgelegt, wird die Toolantwort überschrieben, die dem Modell bereitgestellt wird.
ADK-Callback	after_tool_callback

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

Callback erstellen

So erstellen Sie einen Callback:

1. Öffnen Sie die Agent-Einstellungen.
2. Klicken Sie auf Code hinzufügen.
3. Wählen Sie einen Callback-Typ aus.
4. Geben Sie Python-Code an.
5. Klicken Sie auf Speichern.

Benutzerdefinierte Nutzlasten (custom_payloads)

Benutzerdefinierte Nutzlasten ermöglichen das Einbeziehen zusätzlicher, nicht textueller, strukturierter Daten (in der Regel als JSON formatiert) in die Antwort eines Agenten. Diese Nutzlast ist entscheidend, um die Interaktion des Agenten 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 mit Callbacks generiert und festgelegt, insbesondere mit before_model_callback oder after_model_callback.

Die benutzerdefinierte Nutzlast kann für verschiedene Zwecke verwendet werden, die im Allgemeinen darauf ausgerichtet sind, umfangreiche, strukturierte Interaktionen zu ermöglichen:

• Eskalation/Übergabe des Agenten: Sie wird häufig verwendet, um eine Interaktion an einen menschlichen Kundenservicemitarbeiter zu übergeben, indem Routinganweisungen bereitgestellt werden (z. B. die spezifische Warteschlange, an die weitergeleitet werden soll).
• Umfangreiche Inhalte und clientseitige Aktionen: Sie unterstützt das Einbetten umfangreicher Widgets und anderer umfangreicher Inhalte direkt in Chat-Interaktionen. Dies ist besonders nützlich für benutzerdefinierte Chatintegrationen.
  • Beispiele sind das Anzeigen von Bild-URLs oder Chips für Schnellantworten und Optionen für einen Kunden, der eine Schnittstelle wie den Anruf-Assistenten verwendet.
• Antwortzusammensetzung: Benutzerdefinierte Nutzlasten können so konfiguriert werden, dass sie auf verschiedene Arten zurückgegeben werden:
  • Nur die explizite Nutzlast deterministisch zurückgeben.
  • Die Nutzlast zusammen mit einer vom LLM generierten Textantwort zurückgeben.
  • Die 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 after_model_callback, das 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_json_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 überprüfen

Die Nutzlast wird sowohl für RunSession als auch für BidiRunSessionals Struct im Feld payload ausgefüllt.

Der Nutzlastwert ist für das LLM nicht sichtbar.