無検閲 LLM API のエージェント：ツール呼び出し、フレームワーク、ガードレール

ツール呼び出しループの解剖学

フレームワークを取り除くと、すべてのエージェントは同じ 4 つのことを行います。会話と tools リストを送信します。モデルは完了した場合はテキスト、または実行を希望する場合は tool_calls で応答します。各呼び出しを実行し、呼び出しの id を引用して呼び出しごとに 1 つの tool メッセージを追加します。その後、より長い会話でモデルに再度質問します。

分岐のシグナルは finish_reason です。これが tool_calls と等しい場合、アシスタントメッセージの content は null に設定され、id、関数名、JSON の arguments string を保持する tool_calls 配列が含まれます。正確な形状は リファレンスページ で確認できます。2 つの注意点があります：引数は解析が必要な文字列であり、呼び出しを含むアシスタントメッセージはツール結果の前に順序を保ったまま履歴に残す必要があります。

ゼロからの完全なループ

以下のスクリプトは1つの関数を定義し、JSONスキーマで説明し、モデルがプレーンテキストを生成するまでループします。APIを指す公式Python SDKを使用します。各パスでアシスタントメッセージ、呼び出しごとのツールメッセージ、ステップカウンターを追加し、暴走を防止します。

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

ループの失敗時の動作に注目してください。エラーは文字列としてツール結果になります。これによりモデルは問題を把握し、無効なチケットIDの修正など、プログラムがクラッシュするのではなく再試行できます。for ... else節は、最終回答なしで上限に達した場合のみ発火し、モデルがループに陥ったことを検出する方法です。

プロンプトを集中させます。ツールをいつ呼び出し、いつ直接応答するかを明記した短いシステムメッセージは、不要な呼び出しを減らし、スキップされた呼び出しごとにラウンドトリップ、トークン、レート制限の 1 スロットを節約します。

ツール実行のガード

モデルの出力は信頼できない入力です。モデルが選択したツール名や引数は間違っている、奇妙である、またはモデルがウェブページやユーザーから読み取ったテキストの影響を受けている可能性があります。すべての呼び出しを、インターネットから到着したリクエストとして扱ってください。

• 名前で Allowlist を設定する。 構築したレジストリで関数を検索します。文字列でコードをインポートまたは評価しないでください。
• 読み取りと書き取りを分離する。 読み取り専用ツールは自由に実行できます。メール送信、金銭支出、データ削除、またはファイル操作を行うものは、明示的な人間の承認が必要です。
• 引数を検証：公開したスキーマに対して必須キー、余分なキーなし、型、長さ制限を確認します。
• 時間と出力を制限する。 タイムアウトを適用し、フィードバックする前に結果を切り捨てて、巨大なツール出力が 100,000 トークンのコンテキストウィンドウを占有しないようにします。
• 反復とコストを制限。 usage ブロックを使用して、タスクあたりのステップ数とセッションあたりのトークン数を制限します。
• 名前、引数、結果を記録してすべての呼び出しをログに記録する ことで、動作を監査できるようにします。

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

ツール結果も信頼できないものとして扱ってください。ツールが指示を含むスクレイピングテキストを返した場合、モデルがそれに従う可能性があります。そのようなコンテンツをツールメッセージで明確な区切り文字で囲み、システムプロンプトでツール出力はデータでありコマンドではないと明記してください。

モデルが使用できるツールスキーマの作成

スキーマはモデルが目にする唯一のドキュメントなので、公開 API と同じ配慮が必要です。タスクに一致する動詞と名詞で関数名を付け（例：ticket_status や lookup_invoice）、run や helper のような汎用的なラベルは避けてください。description は指示として記述します：関数の動作、戻り値、および使用しない場合について。

• パラメータは少なく平坦に保つ。 2 つまたは 3 つの単純なフィールドは、10 個のネストされたオブジェクトよりもはるかに正確に呼び出されます。
• 限定された集合には enums を使用する（例："enum": ["open", "closed"]）。これにより、モデルが値を独自に発明できなくなります。
• 各プロパティを説明：単位や例を含めます（"Ticket id such as T-101" は単なる文字列よりも優れています）。
• コンパクトな結果を返す。 文字列または小さな JSON オブジェクトが最適です。長いダンプはコンテキストを消費するだけで、得られるものはほとんどありません。
• モードフラグ付きの包括的なツール 1 つよりも、複数の狭いツールを優先する。

呼び出しの検証に失敗した場合、例外を発生させるのではなく、検証メッセージをツール結果として返します。モデルは通常、次のターンで呼び出しを修正します（何が間違っていたかを読めるため）。同じ方法で 2 回失敗した場合、ループを書き換えるのではなく、説明を書き換えてください。

LangChain の設定

LangChain の ChatOpenAI クラスは base_url を受け入れるため、カスタムプロバイダーパッケージは不要です。model を uncensored に設定し、環境からキーを渡し、bind_tools でツールをバインドします。langchain-openai と langchain-core をインストールします。

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

invoke の結果は、解析された名前と引数を含む tool_calls リストを持つ AIMessage です。これにより手動の JSON ステップをスキップできます。完全な自律型ループを得るには、同じ llm オブジェクトを LangChain のエージェントコンストラクタに渡します。エンドポイントはツール呼び出しを尊重するだけでよく、それは行われます。エージェントがツールを無視しているように見える場合は、ドキュメント文字列と型ヒントが存在するか確認してください。これらはモデルが目にするスキーマになります。

LlamaIndex の設定

LlamaIndexはチャット完了フォーマットに従うサーバー向けにOpenAILikeを提供します。3つの設定が重要です。api_baseはAPIを指し、is_chat_model=Trueはチャット用ルーティングを選択し、is_function_calling_model=Trueはフレームワークにツール送信を許可します。テキスト分割器とメモリバッファを正しくサイズ設定するため、context_windowを100000に設定します。llama-index-llms-openai-likeをインストールします。

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

ここから、同じ llm をエージェントワーカーまたはクエリエンジンに渡すことができます。取得パイプラインは 1 つのプロンプトに多くのチャンクを詰め込む可能性があることを覚えておいてください。プロンプトと completion で共有される 100,000 トークンのウィンドウでは、取得されるチャンクの数を制限し、max_tokens を控えめに保ってください。

予算内でエージェントを運用する

エージェントは1つのユーザーメッセージを複数のリクエストに変換するため、キャパシティを計画してください。5回のツール呼び出し往復のタスクは6回のモデル呼び出しを要します。10の同時タスクは短期間に60件のリクエストを生成し、1分あたり300件の制限はチャットアプリよりも早く到達します。エラーとレート制限のプラクティブックのトークンバケットスロットルをクライアントの前に配置してください。

各ツール結果が追加されるため、コンテキストはラウンドごとに増大します。使用済みの古いツール出力は要約または削除し、usage.prompt_tokens に注意を払ってください。50,000 に近づいたら履歴を圧縮します。コストについては、6 回の呼び出しで平均 3,000 プロンプトトークンと 200 生成トークンの例示的なタスクを想定します：これは 18,000 入力トークン（100 万トークンあたり $0.25 で約 $0.0045）と 1,200 出力トークン（100 万トークンあたり $1.00 で約 $0.0012）です。これらのトークン数は合計のための仮定であり、測定値ではありません。エンドポイント が初めての場合は、ドキュメント と無料トライアルクレジットから始めてください。

エージェントのテストとデバッグ

エージェントは単一関数のユニットテストでは見逃される方法で失敗するため、会話をテストしてください。すべての実行に対して完全なメッセージリスト（ツール引数と結果を含む）を記録し、ログから失敗を再生します。3つの習慣がすぐに効果をもたらします。

1. モデルのモック。 クライアントをスタブし、スクリプト化された tool_calls メッセージを返させます。これにより、ネットワークとコストなしでレジストリ、検証、ループ終了をテストできます。
2. シードの固定。 ツール駆動型タスクには 0.2 のような低い temperature を使用します。ランダム性の低下により、より再現性のある引数 JSON が得られ、後に書き込み中心のステップではそれを上げることができます。
3. 軌跡のアサート。 固定されたプロンプトのいくつかに対して、最終的なテキストだけでなく、どのツールが呼び出され、どのような順序で呼び出されたかを確認します。

finish_reasonも確認してください。ツール呼び出しの途中でlengthという値が返された場合、引数が切り捨てられたことを意味します。max_tokens（最大32,000）を引き上げると、JSONが完成します。ストリーミングもツールと連携しますが、呼び出しインデックスごとに引数の断片を累積し、ストリームが閉じた後にのみ解析してください。これらのチャンクについての復習には、リファレンスのストリーミングセクションを参照してください。