site iconKoei's Digital Garden

Archive

Function Tool Definition

8 min read

Function Tool Definition

参照元: LiveKit Agents Documentation ロードマップ: 学習ロードマップ

What（何についてか）

本ノートは、LiveKit の Tool definition & use 配下に分離された Function tool definition ページを対象に、Python 実装観点でツール定義の実務ポイントを整理する。主題は @function_tool による定義、RunContext の使い方、割り込み制御、動的ツール更新、raw schema、Toolset、構成変更追跡、エラーハンドリングである。

ドキュメント構成変更により、旧 tools/ ページにまとまっていた内容は tools/definition/, tools/mcp/, tools/forwarding/ に分割された。したがって本ノートは tools/definition/ のみを対象にする。

Why（なぜ必要か）

Function Tool は、LLM の推論結果を「実処理」に接続する最小インターフェースであり、引数定義・返却形式・割り込み挙動の設計がユーザー体験と運用安定性を直接決定する。特に音声エージェントでは、再生同期や中断時の整合性管理が必要になり、通常の関数呼び出し設計より厳密な制御が要求される。

また、マルチエージェント運用では handoff 後もセッション文脈が連続するため、動的な tools / instructions 変更履歴の追跡（AgentConfigUpdate）がデバッグ性と再現性に寄与する。

How（どう動くか）

1. 基本定義

@function_tool で関数を公開すると、LLM が tool call 可能になる。デフォルトでは関数名が tool 名、docstring が説明になる。

from typing import Any
from livekit.agents import Agent, RunContext, function_tool
 
 
class MyAgent(Agent):
    @function_tool()
    async def lookup_weather(
        self,
        context: RunContext,
        location: str,
    ) -> dict[str, Any]:
        """Look up weather information for a given location."""
        return {"weather": "sunny", "temperature_f": 70}

2. Tool ID と更新時の重複解決

Python では各 tool が安定 id を持つ。id は関数名または name= で上書きした値になる。update_tools() 時は同一 ID が dedup され、後勝ちで残る。

@function_tool()
async def lookup_weather(context: RunContext, location: str) -> str:
    return "sunny"
 
# lookup_weather.id == "lookup_weather"
 
@function_tool(name="get_weather")
async def my_func(context: RunContext, location: str) -> str:
    return "sunny"
 
# my_func.id == "get_weather"

3. 返り値と handoff

返り値は文字列化されて LLM に渡る。None を返すと「tool 実行のみ」で会話出力を抑制できる。handoff では (Agent, result) あるいは Agent を返せる。

@function_tool()
async def my_tool(context: RunContext):
    return SomeAgent(), "Transferring the user to SomeAgent"

4. speech in tool calls（再生同期）

tool 内で session.generate_reply() を使えるが、待機は speech handle を直接 await せず context.wait_for_playout() を使う。これは tool 実行と再生制御を同一文脈で同期するため。

@function_tool()
async def process_order(self, context: RunContext, order_id: str):
    self.session.generate_reply(
        instructions=f"Processing order {order_id}. This may take a moment."
    )
    await context.wait_for_playout()
 
    result = await process_order_internal(order_id)
 
    self.session.generate_reply(
        instructions=f"Order {order_id} has been processed successfully."
    )
    await context.wait_for_playout()
    return result

5. 割り込み制御

長時間 tool はユーザー発話で中断されうる。中断可能にする場合はバックグラウンド task と wait_if_not_interrupted を使う。取り消せない副作用を伴う処理では run_ctx.disallow_interruptions() を先に呼ぶ。

wait_for_result = asyncio.ensure_future(self._a_long_running_task(query))
await run_ctx.speech_handle.wait_if_not_interrupted([wait_for_result])
 
if run_ctx.speech_handle.interrupted:
    wait_for_result.cancel()
    return None

6. 動的追加・共有・プログラム生成

agent.update_tools() は差分追加ではなく全置換である。既存を残すなら agent.tools を取り込んで再構成する。複数 agent で共有する tool はクラス外定義が有効。

また、function_tool をデコレータではなく関数として使うことで、closure を使った用途別 tool を生成できる。

class Assistant(Agent):
    def _set_profile_field_func_for(self, field: str):
        async def set_value(context: RunContext, value: str):
            return f"field {field} was set to {value}"
        return set_value
 
    def __init__(self):
        super().__init__(
            tools=[
                function_tool(
                    self._set_profile_field_func_for("phone"),
                    name="set_phone_number",
                    description="Call this function when user has provided their phone number.",
                ),
                function_tool(
                    self._set_profile_field_func_for("email"),
                    name="set_email",
                    description="Call this function when user has provided their email.",
                ),
            ],
        )

7. raw schema

Python シグネチャに寄せられない既存定義や外部由来仕様は raw_schema で受ける。handler は raw_arguments dict を受け取り、schema 側キーを参照して処理する。

raw_schema = {
    "type": "function",
    "name": "get_weather",
    "description": "Get weather for a given location.",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City and country e.g. New York"
            }
        },
        "required": ["location"],
        "additionalProperties": False
    }
}
 
@function_tool(raw_schema=raw_schema)
async def get_weather(raw_arguments: dict[str, object], context: RunContext):
    location = raw_arguments["location"]
    return f"The weather of {location} is ..."

8. Toolset / AgentConfigUpdate / ToolError

Toolset は関連 tool 群を 1 ID で束ねる管理単位で、LLM 送信時にフラット化される。名前重複は ValueError。

update_tools() と update_instructions() の変更は会話履歴に AgentConfigUpdate として記録される。handoff 自体が分離境界ではなく、同一セッション文脈上での役割交代として扱われるため、構成変更履歴は依然有効である。

業務エラーは ToolError を投げる。

from livekit.agents.llm import ToolError
 
@function_tool()
async def lookup_weather(context: RunContext, location: str):
    if location == "mars":
        raise ToolError(
            "This location is coming soon. Please join our mailing list to stay updated."
        )
    return {"weather": "sunny", "temperature_f": 70}
graph TD
    A[User utterance] --> B[LLM decides tool call]
    B --> C[Function Tool]
    C --> D{Interrupted?}
    D -- Yes --> E[Cancel/cleanup and return None]
    D -- No --> F[Return result or ToolError]
    F --> G[LLM follow-up response]
    G --> H[Optional handoff]

sequenceDiagram
    participant U as User
    participant L as LLM
    participant T as Tool
    participant S as Session

    U->>L: Request
    L->>T: tool_call(args)
    T->>S: generate_reply("processing...")
    T->>T: wait_for_playout()
    T->>T: external API / DB op
    T-->>L: result or ToolError
    L-->>U: final response

Key Concepts

用語説明
function_tool関数を LLM 呼び出し可能な tool として公開するデコレータ/ファクトリ。
RunContexttool 実行中の session, function_call, speech_handle, userdata などへアクセスするコンテキスト。
tool idPython での安定識別子。update_tools() の dedup と変更追跡に使用。
update_tools利用可能 tool を実行時に全置換する API。
raw_schema関数シグネチャではなく JSON schema で tool 定義する方式。
Toolset関連 tool 群の管理単位。追加・削除を束で扱える。
AgentConfigUpdatetools/instructions の変更履歴を会話文脈に保持するレコード。
ToolErrortool 実行時の業務エラーを LLM へ伝えるための例外。

一言まとめ

Function Tool Definition は、単なる関数公開機能ではなく、音声再生同期・割り込み・動的構成変更・監査可能性まで含めて、LiveKit エージェントの実運用品質を決定する中核レイヤーである。

Graph View

Backlinks