環境構築と準備

公式ドキュメントに基づくツール開発の基本的な流れ：

• ツール作成：name、description、args_schema、return_directのパラメータを定義
• 開発環境の初期化

基本ライブラリのインポート

import asyncio
from typing import Annotated, List, Union

from langchain.agents import initialize_agent, AgentType
from langchain_core.messages import HumanMessage

from utils import *
from langchain_core.tools import tool, BaseTool, ToolException, Tool

1. デコレーターによるツール作成

1.1 基本的なデコレーターツール

@tool
def compute_product_1(x: int, y: int) -> int:
    """Two numbers multiplication operation."""
    # デスクリプションが提供されない場合、docstringが必要
    # ないとエラー：ValueError: Function must have a docstring if description not provided.
    return x * y


print(compute_product_1.name)
print(compute_product_1.description)
print(compute_product_1.args)
print(compute_product_1.invoke({"x": 5, "y": 2}))
print("---------------基本デコレーターツールの実行---------------")

# 出力結果：
# compute_product_1
# Two numbers multiplication operation.
# {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}
# 10

1.2 注釈付きパラメータの使用

@tool
def multiply_with_largest_value_2(
        base_num: Annotated[int, "ベースとなる数値"],
        num_list: Annotated[List[int], "数値リストから最大値を選択して計算"],
) -> int:
    """Multiply base number by the largest value in the list."""
    return base_num * max(num_list)


print(multiply_with_largest_value_2.args_schema.model_json_schema())
print(multiply_with_largest_value_2.invoke({"base_num": 3, "num_list": [2, 7, 4, 8, 1]}))
print("---------------注釈付きパラメータの実装---------------")

# 出力：
# {'description': 'Multiply base number by the largest value in the list.', 'properties': {'base_num': {'description': 'ベースとなる数値', 'title': 'Base Num', 'type': 'integer'}, 'num_list': {'description': '数値リストから最大値を選択して計算', 'items': {'type': 'integer'}, 'title': 'Num List', 'type': 'array'}}, 'required': ['base_num', 'num_list'], 'title': 'multiply_with_largest_value_2', 'type': 'object'}
# 24

1.3 カスタムJSONスキーマの実装

# v1バージョンでは互換性問題が発生する可能性あり
from pydantic import BaseModel, Field


class ProductCalculatorSchema_3(BaseModel):
    first_val: int = Field(description="最初の数値")
    second_val: int = Field(description="二番目の数値")


@tool("product-calculator-tool", args_schema=ProductCalculatorSchema_3, return_direct=True)
def calculate_product_3(first_val: int, second_val: int) -> int:
    """Calculate product of two numbers."""
    return first_val * second_val


print(calculate_product_3.name)
print(calculate_product_3.description)
print(calculate_product_3.args)
print(calculate_product_3.return_direct)
print("---------------カスタムJSONスキーマの利用---------------")

# 出力：
# product-calculator-tool
# Calculate product of two numbers.
# {'first_val': {'description': '最初の数値', 'title': 'First Val', 'type': 'integer'}, 'second_val': {'description': '二番目の数値', 'title': 'Second Val', 'type': 'integer'}}
# True

2. 構造化ツールの作成

from langchain_core.tools import StructuredTool


class ProductSchema_4(BaseModel):
    first_num: int = Field(description="first number")
    second_num: int = Field(description="second number")


def compute_multiplication_4(first_num: int, second_num: int) -> int:
    """Multiply two numbers together."""
    return first_num * second_num


structured_calculator_4 = StructuredTool.from_function(
    func=compute_multiplication_4,
    name="MultiplicationCalculator",
    description="perform multiplication operations",
    args_schema=ProductSchema_4,
    return_direct=True,
)
print(structured_calculator_4.invoke({"first_num": 4, "second_num": 5}))
print(structured_calculator_4.name)
print(structured_calculator_4.description)
print(structured_calculator_4.args)
print("---------------構造化ツールの実装---------------")

# 出力：
# 20
# MultiplicationCalculator
# perform multiplication operations
# {'first_num': {'description': 'first number', 'title': 'First Num', 'type': 'integer'}, 'second_num': {'description': 'second number', 'title': 'Second Num', 'type': 'integer'}}

3. チェーンからツールを作成

from langchain_core.language_models import GenericFakeChatModel
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from utils import *

# チェーンをツールとして変換
greeting_prompt_5 = ChatPromptTemplate.from_messages(
    [("human", "こんにちは、私の名前は {user_name}")]
)

llm_5 = get_llm('ollama')
response_chain_5 = greeting_prompt_5 | llm_5 | StrOutputParser()

converted_tool_5 = response_chain_5.as_tool(name="GreetingResponder", description="ユーザー名を受け取り、中国風の挨拶を返す")
print(converted_tool_5.invoke({"user_name": "田中"}))
print(converted_tool_5.args)
print("---------------チェーンベースのツール作成---------------")

# 出力：
# こんにちは！私は田中です、何かお手伝いできますか？😊
# {'user_name': {'title': 'User Name', 'type': 'string'}}

4. カスタムツールの実装（同期・非同期対応）

from typing import Optional, Type

from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)


class MathOperationSchema_6(BaseModel):
    operand_a: int = Field(description="first number")
    operand_b: int = Field(description="second number")


class CustomMathTool_6(BaseTool):
    name: str = "CustomMathCalculator"
    description: str = "数学演算が必要な場合に使用するツール"
    args_schema: Type[BaseModel] = MathOperationSchema_6
    return_direct: bool = True

    def _run(self, operand_a: int, operand_b: int, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        """Execute the mathematical operation."""
        print("同期ツール実行...")
        return operand_a * operand_b

    async def _arun(self, operand_a: int, operand_b: int, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
        """Asynchronous execution of the tool."""
        print("非同期ツール実行...")
        return self._run(operand_a, operand_b, run_manager=run_manager.get_sync())


math_operation_6 = CustomMathTool_6()
print(math_operation_6.name)
print(math_operation_6.description)
print(math_operation_6.args)
print(math_operation_6.return_direct)

print(math_operation_6.invoke({"operand_a": 3, "operand_b": 4}))

# 非同期実行
async def execute_async_6():
    result = await math_operation_6.ainvoke({"operand_a": 3, "operand_b": 5})
    return result

asyncio.run(execute_async_6())

# 出力：
# CustomMathCalculator
# 数学演算が必要な場合に使用するツール
# {'operand_a': {'description': 'first number', 'title': 'Operand A', 'type': 'integer'}, 'operand_b': {'description': 'second number', 'title': 'Operand B', 'type': 'integer'}}
# True
# 同期ツール実行...
# 12
# 非同期ツール実行...
# 同期ツール実行...
# 15

5. 非同期ツール操作

from langchain_core.tools import StructuredTool


def calculate_product_7(x: int, y: int) -> int:
    """Calculate product of two numbers."""
    return x * y


product_tool_7 = StructuredTool.from_function(func=calculate_product_7)
print(product_tool_7.invoke({"x": 3, "y": 4}))


# 非同期呼び出し
async def async_execution_7():
    async_result = await product_tool_7.ainvoke({"x": 3, "y": 6})
    print(async_result)

asyncio.run(async_execution_7())

# 出力：
# 12
# 18

6. エラーハンドリングと戻り値型の定義

ツールエラーハンドリング

import random
from typing import List, Tuple

from langchain_core.tools import tool


def fetch_climate_data_8(location: str) -> int:
    """都市の気象情報を取得する"""
    raise ToolException(f"Error: This error is defined via handle_tool_error=True for {location}.")
    
climate_tool_8 = StructuredTool.from_function(
    func=fetch_climate_data_8,
    handle_tool_error=True,
)

# 文字列ベースのエラーメッセージ
climate_tool_8_1 = StructuredTool.from_function(
    func=fetch_climate_data_8,
    handle_tool_error="This is defined through string-based handle_tool_error...",
)

# カスタムエラーハンドリング関数
def custom_error_handler_8(error: ToolException) -> str:
    return f"Custom error message via function: `{error.args[0]}`"
    
climate_tool_8_2 = StructuredTool.from_function(
    func=fetch_climate_data_8,
    handle_tool_error=custom_error_handler_8,
)

# 各種エラーハンドリングの結果
print(climate_tool_8.invoke({"location": "Tokyo"}))
print(climate_tool_8_1.invoke({"location": "Osaka"}))
print(climate_tool_8_2.invoke({"location": "Kyoto"}))

Tuple型の戻り値の実装

import random
from typing import List, Tuple

from langchain_core.tools import tool

# Tuple型の戻り値を持つツール
@tool(response_format="content_and_artifact")
def create_random_numbers_8(min_val: int, max_val: int, count: int) -> Tuple[str, List[int]]:
    """Generate random numbers within specified range."""
    numbers_array = [random.randint(min_val, max_val) for _ in range(count)]
    result_message = f"Successfully created array of {count} random numbers in [{min_val}, {max_val}]."
    return result_message, numbers_array

print(f"Basic invocation: {create_random_numbers_8.invoke({'min_val': 1, 'max_val': 10, 'count': 5})}")

tool_invocation_8 = create_random_numbers_8.invoke(
    {
        "name": "create_random_numbers",
        "args": {"min_val": 1, "max_val": 10, "count": 5},
        "id": "456",
        "type": "tool_call",
    }
)
print(f"Tool call result: {tool_invocation_8}")


# BaseToolによるカスタム実装
class GenerateRandomDecimals_8(BaseTool):
    name: str = "generate_random_decimals"
    description: str = "Generate specified count of random decimals within range."
    response_format: str = "content_and_artifact"
    decimal_places: int = 2

    def _run(self, min_decimal: float, max_decimal: float, count: int) -> Tuple[str, List[float]]:
        range_size = max_decimal - min_decimal
        decimal_array = [
            round(min_decimal + (range_size * random.random()), ndigits=self.decimal_places)
            for _ in range(count)
        ]
        message = f"Created {count} decimals in [{min_decimal}, {max_decimal}], rounded to {self.decimal_places} places."
        return message, decimal_array
        
custom_decimal_generator_8 = GenerateRandomDecimals_8(decimal_places=3)
custom_result_8 = custom_decimal_generator_8.invoke(
    {
        "name": "generate_random_decimals",
        "args": {"min_decimal": 0.5, "max_decimal": 2.5, "count": 4},
        "id": "456",
        "type": "tool_call",
    }
)
print(f"Custom tool result: {custom_result_8}")

7. 組み込みツールの利用

詳細なツール情報：

• 日本語リファレンス：https://www.langchain.com.cn/docs/integrations/tools/
• 英語リファレンス：https://python.langchain.com/docs/integrations/tools/

from utils import *
os.environ['serpapi_api_key'] = 'your_api_key_here'
os.environ["TAVILY_API_KEY"] = 'tvly-your_tavily_key'

# 天気検索機能
weather_search = TavilySearchResults(max_results=2)

# 検索機能
search_wrapper = SerpAPIWrapper()

print(weather_search.invoke("2025年4月5日東京の天気"))
print(search_wrapper.run("Langchainとは何か"))

8. 総合的なツール統合：LLM + Tool + パラメータテンプレート

# ツール名とクラス名の一致が重要
@tool("SpecialOperationA_10")
def operation_a_10(input_x: int, input_y: int) -> int:
    """XとYの特別な計算（A演算）
    Args:
        input_x: First parameter
        input_y: Second parameter
    """
    print(f'A operation executed: [{input_x},{input_y}]')
    return input_x + input_y


@tool("SpecialOperationB_10")
def operation_b_10(input_x: int, input_y: int) -> int:
    """XとYの特別な計算（B演算）
    Args:
        input_x: First parameter
        input_y: Second parameter
    """
    print(f'B operation executed: [{input_x},{input_y}]')
    return input_x * input_y

# ツールリストの作成
tool_collection_10 = [operation_b_10, operation_a_10]

# LLMの初期化
llm_10 = ChatOpenAI(base_url='https://open.bigmodel.cn/api/paas/v4/',
                    api_key='your_api_key',
                    model='glm-4-plus',
                    temperature=0.7,
                    verbose=True)

# ツールバインディング
enhanced_llm_10 = llm_10.bind_tools(tool_collection_10)
request_query = "What is 10 A 5? What is 7 B 8?"

# ツール呼び出しの解析
response_message = enhanced_llm_10.invoke(request_query)
print(f"Detected tool calls: {response_message.tool_calls}")

# Pydanticモデルの定義
class SpecialOperationA_10(BaseModel):
    """Add two integers."""
    input_x: int = Field(..., description="First integer")
    input_y: int = Field(..., description="Second integer")


class SpecialOperationB_10(BaseModel):
    """Multiply two integers."""
    input_x: int = Field(..., description="First integer")
    input_y: int = Field(..., description="Second integer")

# Pydanticツールパーサーの実装
from langchain_core.output_parsers import PydanticToolsParser
processing_chain_10 = (enhanced_llm_10
                      | PydanticToolsParser(tools=[SpecialOperationA_10, SpecialOperationB_10]))

print(f"Pydantic parsing result: {processing_chain_10.invoke(request_query)}")

# 最終メッセージ処理
conversation_messages = [HumanMessage(content=request_query)]
conversation_messages.append(response_message)

for tool_call in response_message.tool_calls:
    selected_operation = {"specialoperationa_10": operation_a_10, "specialoperationb_10": operation_b_10}[
        tool_call["name"].lower()]
    tool_response = selected_operation.invoke(tool_call)
    conversation_messages.append(tool_response)
    print(f"Tool execution result: {tool_response}")

final_response = enhanced_llm_10.invoke(conversation_messages).content
print(f"[{request_query}] Final processed response: {final_response}")