Skip to content

Instantly share code, notes, and snippets.

@PonDad
Last active September 7, 2023 03:54
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save PonDad/5df3ac736daab1c8d79c311291908a15 to your computer and use it in GitHub Desktop.
Save PonDad/5df3ac736daab1c8d79c311291908a15 to your computer and use it in GitHub Desktop.

ツール(Tools)

ツールは、エージェントが現実世界とやり取りするための手段です。

カスタムツールの定義

独自のエージェントを構築する際には、そのエージェントが利用可能なツールのリストを提供する必要があります。ツールは、次のような要素で構成されます。

  • 名前(str):必須で、提供されるツールのセット内で一意である必要があります。
  • 説明(str):オプションですが、ツールの使用方法を決定するためにエージェントによって利用されることがおすすめされます。
  • 直接返す(return_direct)(bool):デフォルトはFalseです。
  • 引数スキーマ(args_schema)(Pydantic BaseModel):オプションですが、予想されるパラメータの詳細情報(例:few-shot examples)や検証の提供に使用できます。

ツールを定義する主な方法は2つあり、以下の例で両方をカバーします。

# Import things that are needed generically
from langchain import LLMMathChain, SerpAPIWrapper
from langchain.agents import AgentType, initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import BaseTool, StructuredTool, Tool, tool

llm = ChatOpenAI(temperature=0)

まったく新しいツール - 文字列の入力と出力

最も単純なツールは、単一のクエリ文字列を受け入れ、文字列の出力を返します。ツール関数に複数の引数が必要な場合、以下の「StructuredTool」セクションに進むことができます。

これを行う方法は2つあります。Toolデータクラスを使用するか、BaseToolクラスをサブクラス化するかです。

ツールデータクラス

「ツール」データクラスは、単一の文字列を受け取り、文字列を返す関数をラップします。

# Load the tool configs that are needed.
search = SerpAPIWrapper()
llm_math_chain = LLMMathChain(llm=llm, verbose=True)
tools = [
    Tool.from_function(
        func=search.run,
        name="Search",
        description="useful for when you need to answer questions about current events"
        # coroutine= ... <- you can specify an async method if desired as well
    ),
]

さらに、入力に関する詳細情報を提供するために独自のargs_schemaを定義することもできます。

from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    question: str = Field()


tools.append(
    Tool.from_function(
        func=llm_math_chain.run,
        name="Calculator",
        description="useful for when you need to answer questions about math",
        args_schema=CalculatorInput
        # coroutine= ... <- you can specify an async method if desired as well
    )
)

# Construct the agent. We will use the default agent type here.
# See documentation for a full list of options.
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

agent.run(
    "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"
)

BaseTool クラスをサブクラス化

BaseTool クラスを直接サブクラス化することもできます。これは、インスタンス変数に対するさらなる制御が必要な場合や、ネストされたチェーンや他のツールにコールバックを伝えたい場合に便利です。

from typing import Optional, Type

from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)


class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"

    def _run(
        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return search.run(query)

    async def _arun(
        self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("custom_search does not support async")


class CustomCalculatorTool(BaseTool):
    name = "Calculator"
    description = "useful for when you need to answer questions about math"
    args_schema: Type[BaseModel] = CalculatorInput

    def _run(
        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return llm_math_chain.run(query)

    async def _arun(
        self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("Calculator does not support async")

tools = [CustomSearchTool(), CustomCalculatorTool()]
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

agent.run(
    "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"
)

ツールデコレーター(tool decorator) を使用する

カスタムツールを定義しやすくするために、@tool デコレータが提供されています。このデコレータを使用すると、簡単な関数から迅速にツールを作成できます。デコレータは、デフォルトで関数名をツール名として使用しますが、最初の引数として文字列を渡すことでこれを上書きできます。さらに、デコレータはツールの説明として関数のドキュメンテーション文字列を使用します。

from langchain.tools import tool


@tool
def search_api(query: str) -> str:
    """Searches the API for the query."""
    return f"Results for query {query}"


search_api

@tool("search", return_direct=True)
def search_api(query: str) -> str:
    """Searches the API for the query."""
    return "Results"

class SearchInput(BaseModel):
    query: str = Field(description="should be a search query")


@tool("search", return_direct=True, args_schema=SearchInput)
def search_api(query: str) -> str:
    """Searches the API for the query."""
    return "Results"

カスタムな構造化ツール

もし関数がより構造化された引数を必要とする場合、StructuredToolクラスを直接使用するか、またはBaseToolクラスをサブクラス化することができます。

StructuredToolデータクラス 特定の関数から構造化ツールを動的に生成するには、最速の方法はStructuredTool.from_function()を使用することです。

import requests
from langchain.tools import StructuredTool


def post_message(url: str, body: dict, parameters: Optional[dict] = None) -> str:
    """Sends a POST request to the given url with the given body and parameters."""
    result = requests.post(url, json=body, params=parameters)
    return f"Status: {result.status_code} - {result.text}"


tool = StructuredTool.from_function(post_message)

BaseToolをサブクラス化

BaseToolは、_runメソッドのシグネチャからスキーマを自動的に推論します。

from typing import Optional, Type

from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)


class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"

    def _run(
        self,
        query: str,
        engine: str = "google",
        gl: str = "us",
        hl: str = "en",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool."""
        search_wrapper = SerpAPIWrapper(params={"engine": engine, "gl": gl, "hl": hl})
        return search_wrapper.run(query)

    async def _arun(
        self,
        query: str,
        engine: str = "google",
        gl: str = "us",
        hl: str = "en",
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("custom_search does not support async")


# You can provide a custom args schema to add descriptions or custom validation


class SearchSchema(BaseModel):
    query: str = Field(description="should be a search query")
    engine: str = Field(description="should be a search engine")
    gl: str = Field(description="should be a country code")
    hl: str = Field(description="should be a language code")


class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"
    args_schema: Type[SearchSchema] = SearchSchema

    def _run(
        self,
        query: str,
        engine: str = "google",
        gl: str = "us",
        hl: str = "en",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool."""
        search_wrapper = SerpAPIWrapper(params={"engine": engine, "gl": gl, "hl": hl})
        return search_wrapper.run(query)

    async def _arun(
        self,
        query: str,
        engine: str = "google",
        gl: str = "us",
        hl: str = "en",
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("custom_search does not support async")

デコレーターの使用

ツールデコレーターは、シグネチャに複数の引数がある場合、自動的に構造化ツールを作成します。

import requests
from langchain.tools import tool


@tool
def post_message(url: str, body: dict, parameters: Optional[dict] = None) -> str:
    """Sends a POST request to the given url with the given body and parameters."""
    result = requests.post(url, json=body, params=parameters)
    return f"Status: {result.status_code} - {result.text}"

既存のツールを変更する

次に、既存のツールをロードして直接変更する方法を示します。以下の例では、非常にシンプルなことを行い、検索ツールの名前を「Google検索」に変更します。

from langchain.agents import load_tools

tools = load_tools(["serpapi", "llm-math"], llm=llm)

tools[0].name = "Google Search"

agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

agent.run(
    "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"
)

ツール間の優先順位の定義

カスタムツールを作成した場合、通常のツールよりもカスタムツールをエージェントにより頻繁に使用させたいことがあるかもしれません。

たとえば、データベースから音楽情報を取得するカスタムツールを作成したとしましょう。ユーザーが曲に関する情報を求める場合、通常の検索ツールよりもカスタムツールを優先して使用させたいかもしれません。しかし、エージェントは通常の検索ツールを優先するかもしれません。

これは、説明に「これは音楽に関する質問の場合に通常の検索よりもこれを使用してください。たとえば 'who is the singer of yesterday?' や 'what is the most popular song in 2022?' など」といった文を追加することで実現できます。

以下に例を示します。

# Import things that are needed generically
from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.llms import OpenAI
from langchain import LLMMathChain, SerpAPIWrapper

search = SerpAPIWrapper()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="useful for when you need to answer questions about current events",
    ),
    Tool(
        name="Music Search",
        func=lambda x: "'All I Want For Christmas Is You' by Mariah Carey.",  # Mock Function
        description="A Music search engine. Use this more than the normal search if the question is about Music, like 'who is the singer of yesterday?' or 'what is the most popular song in 2022?'",
    ),
]

agent = initialize_agent(
    tools,
    OpenAI(temperature=0),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)

ツールを使用して直接出力を返す

しばしば、ツールを呼び出した場合にそのツールの出力をユーザーに直接返すことが望ましいことがあります。LangChainでは、ツールのreturn_directフラグをTrueに設定することで、これを簡単に行うことができます。

llm_math_chain = LLMMathChain(llm=llm)
tools = [
    Tool(
        name="Calculator",
        func=llm_math_chain.run,
        description="useful for when you need to answer questions about math",
        return_direct=True,
    )
]

llm = OpenAI(temperature=0)
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

agent.run("whats 2**.12")

ツールのエラー処理

ツールがエラーに遭遇し、例外がキャッチされない場合、エージェントは実行を停止します。エージェントの実行を継続させたい場合、ToolExceptionを発生させ、handle_tool_errorを適切に設定します。

ToolExceptionがスローされると、エージェントは作業を停止せず、ツールのhandle_tool_error変数に従って例外を処理し、処理結果が観測としてエージェントに返され、赤で表示されます。

handle_tool_errorをTrueに設定、統一された文字列値に設定、または関数として設定できます。関数として設定する場合、関数はToolExceptionをパラメーターとして受け取り、str値を返す必要があります。

ただし、ToolExceptionをスローしただけでは効果がありません。まず、ツールのhandle_tool_errorを設定する必要があります。デフォルト値はFalseです。

from langchain.tools.base import ToolException

from langchain import SerpAPIWrapper
from langchain.agents import AgentType, initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool

from langchain.chat_models import ChatOpenAI


def _handle_error(error: ToolException) -> str:
    return (
        "The following errors occurred during tool execution:"
        + error.args[0]
        + "Please try another tool."
    )


def search_tool1(s: str):
    raise ToolException("The search tool1 is not available.")


def search_tool2(s: str):
    raise ToolException("The search tool2 is not available.")


search_tool3 = SerpAPIWrapper()

description = "useful for when you need to answer questions about current events.You should give priority to using it."
tools = [
    Tool.from_function(
        func=search_tool1,
        name="Search_tool1",
        description=description,
        handle_tool_error=True,
    ),
    Tool.from_function(
        func=search_tool2,
        name="Search_tool2",
        description=description,
        handle_tool_error=_handle_error,
    ),
    Tool.from_function(
        func=search_tool3.run,
        name="Search_tool3",
        description="useful for when you need to answer questions about current events",
    ),
]

agent = initialize_agent(
    tools,
    ChatOpenAI(temperature=0),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)


agent.run("Who is Leo DiCaprio's girlfriend?")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment