Python コードツール

インライン Python コードをエージェントのツールとして提供できます。これにより、エージェントの機能を柔軟に拡張できます。このコードは、エージェントから提供された入力を受け取り、エージェントが会話で使用する結果を返します。カスタム ロジックを実装したり、独自の API やデータベースに接続したり、精度が求められるタスクで決定論的な結果を確保したりできます。

名前と説明

ツールを作成するときは、ツールの名前と呼び出すメイン関数の名前をスネークケースで同じ名前にする必要があります。

Docstring は、Python ツールの定義において重要な部分です。関数の docstring はツールの説明として使用され、ツールを使用するエージェントに提供されます。docstring はプロンプトの拡張機能と考える必要があります。明確で説明的で構造化された docstring は、モデルがツールの機能、使用するタイミング、提供する引数を理解する程度に直接影響します。これが、信頼性が高く正確なツール選択を実現するための鍵となります。

環境

Python ツールのコードでは、コードの作成に役立つ特定のクラスと関数にアクセスできます。詳細については、Python ランタイム リファレンスをご覧ください。

たとえば、context オブジェクトは、現在の会話の状態のスナップショットを提供するグローバルに使用できる変数です。インポートしたり、パラメータとして定義したりする必要はありません。直接アクセスできます。複雑なロジックを実行するための貴重な情報が含まれています。

コンテキスト オブジェクトで使用可能なキーの内訳は次のとおりです。

  • function_call_id: 実行中の特定のツール呼び出しの一意の ID。
  • user_content: ユーザーからの最新のメッセージ(テキストとロールを含む)を含む辞書。これは最もよく使用される属性の 1 つです。
  • state: セッションの状態を表す辞書。これを使用して、会話の複数のターンにわたって保持する必要がある変数(ユーザー プロファイル、ショッピング カートの内容など)を保存して取得できます。
  • events: ここまでの会話履歴のすべてのイベントのリスト。これにより、より複雑なコンテキスト認識を備えたツールを構築できます。
  • session_id: 会話セッション全体の固有識別子。
  • invocation_id: 会話の現在のターンの固有識別子。
  • agent_name: ツールを実行しているエージェントの名前。

他のツールを呼び出す Python ツール

Python コードツールを定義するときに、エージェント アプリケーションで定義された他のツールを明示的に呼び出すことができます。たとえば、crm_service_get_cart_information という OpenAPI ツールがある場合は、次のコードでそのツールを呼び出すことができます。

# Deterministically call another tool from this tool.
# This syntax for OpenAPI spec tool is:
# tools.<tool_name>_<endpoint_name>({tool_args})
res = tools.crm_service_get_cart_information({})

コードサンプル

以降のセクションでは、サンプルを示します。

最後のユーザー入力を取得する

このサンプルは、ユーザーの最新のメッセージにアクセスするという基本的な機能を示しています。このツールは context.user_content オブジェクトを検査し、最後のユーザー ターンからテキストを抽出します。このパターンは、ユーザーが発言した内容に直接基づいてアクションを実行する必要があるツールに不可欠です。

from typing import Optional

# Docstrings in tools are important because they are directly
# sent to the model as the description for the tool. You should
# think of docstrings as an extension of prompting. Clear and
# descriptive docstrings will yield higher quality tool
# selection from the model.
def get_last_user_utterance() -> Optional[str]:
  """
  Retrieves the most recent message sent by the user from the conversation history.

  Returns:
    The text of the last user message, or None if no user messages are found.
  """
  # The 'context.user_content' contains the last input data
  # provided by the user.
  # We can filter it to find only the text input from the user.
  user_messages = [
    part["text"] for part in context.user_content["parts"]
    if context.user_content["role"] == "user"
  ]

  if user_messages:
    # The most recent message is the first one in the list.
    return user_messages[0]

  return None

変数を取得して更新する

言語モデルはセッションの状態を直接変更できません。これは、状態の変化が制御された予測可能な方法で処理されるようにするための意図的なものです。状態は、Python ツールまたはコールバックの 2 つのメカニズムでのみ変更できます。

このサンプルは、Python ツールが状態を管理する方法を示しています。このツールは、まず context.state から現在の customer_profile を読み取ります。次に、ビジネス ロジック(ポイントの追加)を実行し、更新されたプロファイルを context.state に書き戻します。この新しい状態は、セッションの残りの間、エージェントや他のツールで使用できます。

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

# Using Pydantic defines the expected structure of your state variables. This makes your code more reliable and easier to
# debug.
class CustomerProfile(BaseModel):
  email: Optional[str] = None
  loyalty_points: int = Field(default=0, ge=0) # Must be >= 0
  plan: str = "Standard"

# Docstrings in tools are important because they are directly
# sent to the model as the description for the tool. You should
# think of docstrings as an extension of prompting. Clear and
# descriptive docstrings will yield higher quality tool
# selection from the model.
def update_customer_loyalty_points(points_to_add: int) -> Dict[str, Any]:
  """
  Adds loyalty points to the customer's profile and returns the updated profile.

  Args:
    points_to_add: The number of loyalty points to add to the existing total.

  Returns:
    A dictionary containing the customer's updated profile information.
  """
  # 1. Get the current profile data from the session state.
  # The .get() method safely returns an empty dict if
  # 'customer_profile' doesn't exist.
  current_profile_data = context.state.get("customer_profile", {})

  # 2. Load the data into a Pydantic model for validation and easy access.
  profile = CustomerProfile(**current_profile_data)

  # 3. Perform the business logic.
  # Print statements can be used for debugging and will show
  # up in the tracing details.
  profile.loyalty_points += points_to_add
  print(f"Updated loyalty points to: {profile.loyalty_points}")

  # 4. Save the updated data back into the session state.
  # .model_dump() converts the Pydantic model back to a
  # dictionary.
  context.state["customer_profile"] = profile.model_dump()

  return context.state["customer_profile"]

外部ネットワーク リクエスト

Python ツールは外部ネットワーク リクエストを行うことができます。これは、リアルタイム データの取得や、OpenAPI 仕様のないサードパーティ サービスとの統合に役立ちます。これにより、OpenAPI ベースのツールを使用する代わりに、柔軟な方法を利用できます。このサンプルでは、標準の requests ライブラリ(環境で使用可能)を使用して公開 API を呼び出し、ランダムな事実を取得します。

from typing import Optional

# Docstrings in tools are important because they are directly sent to the model as the
# description for the tool. You should think of docstrings as an extension of prompting.
# Clear and descriptive docstrings will yield higher quality tool selection from the model.

def get_random_fact() -> Optional[str]:
  """
  Fetches a random fact from a public API.

  Returns:
      A string containing the fact on success, or None if an error occurs.
  """
  # The 'ces_requests' library is inspired by 'requests', a  standard and powerful way in Python
  # to make HTTP network calls to any external API, just like you would with `curl` or a web browser.
  url = "https://uselessfacts.jsph.pl/api/v2/facts/random"

  try:
      # This example calls a public API that is completely open and requires no authentication
      # (like an API key). Many other APIs for services like weather or e-commerce require you
      # to send credentials, often as an API key in the request headers or parameters.
      res = ces_requests.get(url=url, json=request_body, headers=headers)

      # Example POST request
      #res = ces_requests.post(url=url, json=request_body, headers=headers)

      # This is a standard practice with 'ces_requests' to check if the call was successful. It will
      # raise an error for statuses like 404 or 500.
      res.raise_for_status()

      fact_data = res.json()
      return fact_data.get("text")

  except:
      return None