Callbacks

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

Es gibt verschiedene Arten von Callbacks, die Sie verwenden können. Jede Art von Callback wird an einem bestimmten Punkt in der Unterhaltungsrunde ausgeführt. Diese Arten 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.

Einschränkungen für Sandbox und Netzwerk

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): Callbacks können nicht direkt auf private IP-Adressen zugreifen oder private DNS-Domains auflösen, 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: Callbacks können nur auf Endpunkte zugreifen, 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 beliebigen Namen in Ihrem Callback-Code definieren.

Jede Art von Callback 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	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 aufgerufen, nachdem der Agent abgeschlossen wurde.
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 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	Zur Überprüfung/Änderung der Modellanfrage oder um die Modellnutzung zu vermeiden.
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	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 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	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 Toolausführung übersprungen und diese Ausgabe wird dem Modell zur Verfügung gestellt.
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]: Toolantwort
Zurücksenden	Dict[str,Any]: Wenn festgelegt, wird die Toolantwort überschrieben, die dem Modell zur Verfügung gestellt 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 Agenteneinstellungen.
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 im JSON-Format) 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 verwendet, um die endgültige Antwort zu generieren. 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 darauf ausgerichtet sind, umfangreiche, strukturierte Interaktionen zu ermöglichen:

• Eskalation/Übergabe an einen menschlichen Agenten: Sie wird häufig verwendet, um eine Interaktion an einen menschlichen Agenten 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, was besonders für benutzerdefinierte Chat-Integrationen nützlich ist.
  • Beispiele sind die Anzeige von Bild-URLs oder Chips für schnelle Antworten und Optionen für einen Kunden, der eine Schnittstelle wie Call Companion verwendet.
• Antwortzusammensetzung: Benutzerdefinierte Nutzlasten können so konfiguriert werden, dass sie auf verschiedene Arten zurückgegeben werden:
  • Geben Sie nur die explizite Nutzlast deterministisch zurück.
  • Geben Sie die Nutzlast zusammen mit einer vom LLM generierten Textantwort zurück.
  • Geben Sie die Nutzlast mit einer statischen Textantwort zurück.

Agenteneinrichtung

Benutzerdefinierte Nutzlasten können nur mithilfe von 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 eingefügt.

Der Nutzlastwert ist für das LLM nicht sichtbar.