xDev Asia

複雑なツールを作成します: Web スクレイピング、Google 検索、Python コードサンドボックス、データベースクエリ、REST API 呼び出し元。レジストリツール、エラー処理、および再試行ロジックを管理します。

AI エージェントの構築: ゼロから本番環境まで

パート 2: 関数呼び出しとツールの使用

xdev.asia

はじめに

前のレッスンでは、関数呼び出しの仕組み**を学習しました。この記事では、運用環境でエージェントが必要とする 実際のビルドツール (Web 検索、コード実行サンドボックス、データベースクエリ、REST API 統合) に焦点を当てます。

1. Web検索ツール

1.1 SerpAPI / Tavily の使用

import httpx

def web_search(query: str, num_results: int = 5) -> str:
    """Tìm kiếm web và trả về kết quả top."""
    response = httpx.get("https://api.tavily.com/search", params={
        "api_key": TAVILY_API_KEY,
        "query": query,
        "max_results": num_results,
    })
    results = response.json()["results"]
    return "\n".join([
        f"- [{r['title']}]({r['url']}): {r['content'][:200]}"
        for r in results
    ])

2. コード実行サンドボックス

import subprocess
import tempfile

def execute_python(code: str, timeout: int = 10) -> str:
    """Chạy Python code trong sandbox."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        f.flush()
        try:
            result = subprocess.run(
                ['python', f.name],
                capture_output=True, text=True, timeout=timeout,
                env={"PATH": "/usr/bin"}  # Restricted env
            )
            return result.stdout or result.stderr
        except subprocess.TimeoutExpired:
            return "Error: Code execution timed out"

3. データベースクエリツール

import sqlite3

def query_database(sql: str) -> str:
    """Chạy SQL query (read-only) trên database."""
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries allowed"
    
    conn = sqlite3.connect("app.db")
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        return json.dumps({"columns": columns, "rows": rows[:50]})
    except Exception as e:
        return f"SQL Error: {e}"
    finally:
        conn.close()

4. エラー処理と再試行ロジック

def safe_tool_execute(tool_func, args, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            result = tool_func(**args)
            return {"status": "success", "data": result}
        except Exception as e:
            if attempt == max_retries:
                return {"status": "error", "error": str(e)}
            time.sleep(1)

概要

Web 検索、コード実行、データベースクエリ - 3 つの最も重要なツール
セキュリティ第一: サンドボックスコード実行、読み取り専用 DB、入力検証
本番エージェントにはエラー処理 + 再試行ロジックが必要です
ツールの可観測性: デバッグのためにすべてのツール呼び出しをログに記録します。

演習

Tavily API を使用して実際の web_search ツールを実装する
より安全なサンドボックス実行コードを構築する (Docker を使用)
任意のエンドポイントを呼び出すことができる REST API ツールを作成する
ツールのレート制限を実装します (スパム API 呼び出しを回避します)。

レッスン 5: カスタム ツールの構築 — Web 検索、コード実行、API 統合