Callbacks

Callbacks sind eine erweiterte Funktion, mit der Sie mithilfe von Python-Code in den Ausführungsprozess eines bestimmten KI-Agenten eingreifen können. Damit können Sie das Verhalten des KI-Agenten an bestimmten, vordefinierten Punkten beobachten, anpassen und sogar steuern.

Es gibt verschiedene Callback-Typen, die Sie verwenden können. Jeder Callback-Typ wird an einem bestimmten Punkt im Gesprächsverlauf 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.

Sandbox- und Netzwerkeinschränkungen

Python-Callback-Code wird in einer Sandbox-Umgebung ausgeführt. Für diese Umgebung gelten die folgenden Einschränkungen:

• Kein Zugriff auf private Netzwerke (Private Network Access, PNA): Auf private IP-Adressen kann nicht direkt zugegriffen werden und private DNS-Domains können nicht aufgelöst werden, auch wenn Service Directory konfiguriert ist. Wenn Sie auf Ressourcen in einem privaten Netzwerk zugreifen müssen, müssen Sie stattdessen einen Tooltyp verwenden, der den Zugriff auf private Netzwerke unterstützt, z. B. OpenAPI- oder MCP-Tools.
• Zugriff auf das öffentliche Internet: Über Callbacks kann nur auf Endpunkte zugegriffen werden, die öffentlich im Internet verfügbar sind.

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 an einem bestimmten Punkt im Gesprächsverlauf 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 innerhalb des Gesprächsbeitrags.
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 vor dem Aufrufen des Agenten aufgerufen.
Zweck	Zum Einrichten von Ressourcen oder Zuständen, die für den Agenten benötigt werden, zum Durchführen von Gültigkeitsprüfungen des Sitzungszustands 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 nach Abschluss des Agenten aufgerufen.
Zweck	Für Bereinigungsaufgaben, zum Validieren nach der Ausführung, zum Ändern des Endzustands oder zum Aktualisieren der Antwort des Agenten.
Argumente	CallbackContext
Zurücksenden	Inhalt(optional): Wenn festgelegt, wird die Ausgabe des Agents durch die bereitgestellte 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	Zur Überprüfung/Änderung der Modellanfrage oder um die Modellnutzung 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-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	Für das Umformatieren von Modellantworten, das Zensieren vertraulicher Informationen, die vom Modell generiert wurden, das Parsen strukturierter Daten aus dem Modell zur Verwendung in Variablen und die Fehlerbehandlung des Modells.
Argumente	CallbackContext, LlmResponse
Zurücksenden	LlmResponse: Wenn festgelegt, wird die Modellantwort durch die bereitgestellte 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 Tool-Aufruf („before_tool_callback“)

X	X
Name	before_tool_callback
Ausführung	Wird vor Toolaufrufen aufgerufen.
Zweck	Zur Prüfung und Änderung von Toolargumenten, für Autorisierungsprüfungen vor der Toolausführung oder die Implementierung von Caching auf Toolebene.
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-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	Für die Überprüfung und Änderung der Toolantwort, bevor sie an das Modell zurückgesendet wird, für die Nachbearbeitung von Toolergebnissen oder zum Speichern bestimmter Teile einer Toolantwort 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 zur Verfügung gestellt wird, damit überschrieben.
ADK-Callback	nach dem 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

Channelspezifische und modalitätsspezifische Aktionen

Bei der Implementierung von Python-Tools oder Callbacks ist es oft am besten, dass alle Modalitäten und Channels dasselbe Verhalten aufweisen. Es gibt jedoch Fälle, in denen ein channelspezifisches oder modalitätsspezifisches Verhalten gewünscht ist. Sie können den Channel oder die Modalität für eine Unterhaltung mit der globalen Variablen _session prüfen.

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 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 im JSON-Format) in die Antwort eines Agenten. Diese Nutzlast ist entscheidend, um die Interaktion des KI-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 im Allgemeinen auf die Ermöglichung umfangreicher, strukturierter Interaktionen ausgerichtet sind:

• Eskalierung/Übergabe an Kundenservicemitarbeiter: 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).
• Rich Content und clientseitige Aktionen: Es können Rich-Widgets und andere Rich-Inhalte direkt in Chats eingebettet werden. Das ist besonders nützlich für benutzerdefinierte Chat-Integrationen.
  • Beispiele hierfür sind die Anzeige von Bild-URLs oder Schnellantwort-Chips und Optionen für einen Kunden, der eine Schnittstelle wie Call Companion verwendet.
• Zusammenstellung von Antworten: 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 vom 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_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 als Struct im Feld payload für RunSession und BidiRunSession eingefügt.

Der Nutzlastwert ist für das LLM nicht sichtbar.