devbox
运维做开发
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【可能是全网最丝滑的LangChain教程】十八、LangChain进阶之Tools_langchain tools

作者：运维做开发 | 2024-07-24 14:25:58

赞

踩

langchain tools

永远不要说再见，因为再见意味着离去，离去意味着遗忘。

01 Tool介绍

在LangChain中，Tools 是一个核心概念，它们允许模型与外部系统进行交互，从而扩展了基础语言模型的功能。Tools 可以被看作是代理(agent)可用的一系列功能，这些功能包括但不限于网络搜索、API调用、数据库查询、文件操作等。

LangChain Tools 的基本结构

  1. 名称 (name)：工具的标识符。
  2. 输入（input）：工具的输入
  3. 描述 (description)：描述工具用途的字符串，这将帮助代理决定何时使用此工具。
  4. 函数 (func)：实际执行工具功能的可调用对象。
  5. 返回直接结果 (return_direct)：如果设置为True，则代理将直接返回工具的结果，而不是将其作为思考过程的一部分。

名称、描述和 JSON 架构可用于提示 LLM，以便它知道如何指定要执行的操作，然后要调用的函数等同于执行该操作。

工具的输入越简单，LLM 就越容易使用它。

02 LangChain中的默认Tools

LangChain内置了很多的预定义工具，我们可以格式化打印看下有多少：

class BaseTool
  class Tool
  class StructuredTool
  class WikipediaQueryRun
  class AINBaseTool
    class AINAppOps
    class AINOwnerOps
    class AINRuleOps
    class AINTransfer
    class AINValueOps
  class AIPluginTool
  class ArxivQueryRun
  class AzureAiServicesDocumentIntelligenceTool
  class AzureAiServicesImageAnalysisTool
  class AzureAiServicesSpeechToTextTool
  class AzureAiServicesTextAnalyticsForHealthTool
  class AzureAiServicesTextToSpeechTool
  class AzureCogsFormRecognizerTool
  class AzureCogsImageAnalysisTool
  class AzureCogsSpeech2TextTool
  class AzureCogsText2SpeechTool
  class AzureCogsTextAnalyticsHealthTool
  class BaseGraphQLTool
  class RequestsGetTool
  class RequestsPostTool
  class RequestsPatchTool
  class RequestsPutTool
  class RequestsDeleteTool
  class QuerySQLDataBaseTool
  class InfoSQLDatabaseTool
  class ListSQLDatabaseTool
  class QuerySQLCheckerTool
  class QuerySparkSQLTool
  class InfoSparkSQLTool
  class ListSparkSQLTool
  class QueryCheckerTool
  class BingSearchRun
  class BingSearchResults
  class BraveSearch
  class BaseBrowserTool
    class ClickTool
    class CurrentWebPageTool
    class ExtractHyperlinksTool
    class ExtractTextTool
    class GetElementsTool
    class NavigateTool
    class NavigateBackTool
  class CogniswitchKnowledgeRequest
  class CogniswitchKnowledgeStatus
  class CogniswitchKnowledgeSourceFile
  class CogniswitchKnowledgeSourceURL
  class ConneryAction
  class CopyFileTool
  class DeleteFileTool
  class FileSearchTool
  class ListDirectoryTool
  class MoveFileTool
  class ReadFileTool
  class WriteFileTool
  class DataheraldTextToSQL
  class DuckDuckGoSearchRun
  class DuckDuckGoSearchResults
  class E2BDataAnalysisTool
  class EdenaiTool
    class EdenAiSpeechToTextTool
    class EdenAiTextToSpeechTool
    class EdenAiExplicitImageTool
    class EdenAiObjectDetectionTool
    class EdenAiParsingIDTool
    class EdenAiParsingInvoiceTool
    class EdenAiTextModerationTool
  class ElevenLabsText2SpeechTool
  class GmailBaseTool
    class GmailCreateDraft
    class GmailGetMessage
    class GmailGetThread
    class GmailSearch
    class GmailSendMessage
  class GoogleCloudTextToSpeechTool
  class GooglePlacesTool
  class GoogleSearchRun
  class GoogleSearchResults
  class GoogleSerperRun
  class GoogleSerperResults
  class HumanInputRun
  class IFTTTWebhook
  class QueryPowerBITool
  class InfoPowerBITool
  class ListPowerBITool
  class JiraAction
  class JsonListKeysTool
  class JsonGetValueTool
  class MerriamWebsterQueryRun
  class MetaphorSearchResults
  class MojeekSearch
  class NasaAction
  class O365BaseTool
    class O365CreateDraftMessage
    class O365SearchEvents
    class O365SearchEmails
    class O365SendEvent
    class O365SendMessage
  class OpenWeatherMapQueryRun
  class PolygonAggregates
  class PolygonFinancials
  class PolygonLastQuote
  class PolygonTickerNews
  class PubmedQueryRun
  class RedditSearchRun
  class SceneXplainTool
  class SearchAPIRun
  class SearchAPIResults
  class SearxSearchRun
  class SearxSearchResults
  class ShellTool
  class SlackBaseTool
    class SlackGetChannel
    class SlackGetMessage
    class SlackScheduleMessage
    class SlackSendMessage
  class SleepTool
  class StackExchangeTool
  class SteamWebAPIQueryRun
  class SteamshipImageGenerationTool
  class VectorStoreQATool
  class VectorStoreQAWithSourcesTool
  class WolframAlphaQueryRun
  class YahooFinanceNewsTool
  class YouSearchTool
  class YouTubeSearchTool
  class ZapierNLARunAction
  class ZapierNLAListActions
  class PythonREPLTool
  class PythonAstREPLTool
  class InvalidTool
  class ExceptionTool
  class DataForSeoAPISearchRun
  class DataForSeoAPISearchResults
  class GoldenQueryRun
  class GoogleFinanceQueryRun
  class GoogleJobsQueryRun
  class GoogleLensQueryRun
  class GoogleScholarQueryRun
  class GoogleTrendsQueryRun
  class Memorize

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145

我们可以挑几个简单看下：

  1. WikipediaQueryRun：用于向维基百科API发送查询并获取数据
  2. RequestsGetTool：使用HTTP GET方法从指定的URL获取数据。
  3. RequestsPostTool：使用HTTP POST方法向服务器发送数据。
  4. QuerySQLDataBaseTool：对数据库执行SQL查询并返回结果。
  5. ShellTool：执行shell命令。
  6. PythonREPLTool：执行Python命令

下面以WikipediaQueryRun为例子，简单解释说明下使用和原理。

基本使用如下：

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)
tool = WikipediaQueryRun(api_wrapper=api_wrapper)

tool.name
"""
wikipedia
"""

tool.description
"""
A wrapper around Wikipedia. Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects. Input should be a search query.
"""

tool.args
"""
{'query': {'title': 'Query', 'type': 'string'}}
"""

tool.return_direct
"""
False
"""

tool.run({"query": "langchain"})
"""
Page: LangChain\nSummary: LangChain is a framework designed to simplify the creation of applications 
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

在上面代码中，我们通过打印相关参数可以看到这个Tool的基本信息。最后通过tool.run方法执行此工具，搜索"langchain"得到了最终输出。

WikipediaAPIWrapper源码分析如下：

class WikipediaAPIWrapper(BaseModel):
    """
    wiki百科工具的具体调用逻辑，包括
    - 定义一些参数
    - 具体执行逻辑
    - 加载搜索结果为Document
    - ...
    """
    
    # 具体的调用客户端，通过pip install wikipedia后能拿到
    wiki_client: Any 
    # 搜索结果的数量
    top_k_results: int = 3
    # 搜索结果的语言
    lang: str = "en"
    # 用于配置元数据信息
    load_all_available_meta: bool = False
    # 单个Document的内容长度
    doc_content_chars_max: int = 4000

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """
        环境验证，也就是有没有安装wikipedia三方库
        不安装没法做请求
        """
        try:
            import wikipedia

            wikipedia.set_lang(values["lang"])
            values["wiki_client"] = wikipedia
        except ImportError:
            raise ImportError(
                "Could not import wikipedia python package. "
                "Please install it with `pip install wikipedia`."
            )
        return values

    def run(self, query: str) -> str:
        """
        具体的执行逻辑，本质还是做网络请求拿结果
        """
        page_titles = self.wiki_client.search(
            query[:WIKIPEDIA_MAX_QUERY_LENGTH], results=self.top_k_results
        )
        summaries = []
        for page_title in page_titles[: self.top_k_results]:
            if wiki_page := self._fetch_page(page_title):
                if summary := self._formatted_page_summary(page_title, wiki_page):
                    summaries.append(summary)
        if not summaries:
            return "No good Wikipedia Search Result was found"
        return "\n\n".join(summaries)[: self.doc_content_chars_max]

    @staticmethod
    def _formatted_page_summary(page_title: str, wiki_page: Any) -> Optional[str]:
        """
        对接过进行封装展示（相当于格式化）
        """
        return f"Page: {page_title}\nSummary: {wiki_page.summary}"

    def _page_to_document(self, page_title: str, wiki_page: Any) -> Document:
        """
        搜索结果转Document，方便后续LLM使用
        """
        main_meta = {
            "title": page_title,
            "summary": wiki_page.summary,
            "source": wiki_page.url,
        }
        add_meta = (
            {
                "categories": wiki_page.categories,
                "page_url": wiki_page.url,
                "image_urls": wiki_page.images,
                "related_titles": wiki_page.links,
                "parent_id": wiki_page.parent_id,
                "references": wiki_page.references,
                "revision_id": wiki_page.revision_id,
                "sections": wiki_page.sections,
            }
            if self.load_all_available_meta
            else {}
        )
        doc = Document(
            page_content=wiki_page.content[: self.doc_content_chars_max],
            metadata={
                **main_meta,
                **add_meta,
            },
        )
        return doc

    def _fetch_page(self, page: str) -> Optional[str]:
        """
        这里还是具体的调用逻辑
        """
        try:
            return self.wiki_client.page(title=page, auto_suggest=False)
        except (
            self.wiki_client.exceptions.PageError,
            self.wiki_client.exceptions.DisambiguationError,
        ):
            return None

    def load(self, query: str) -> List[Document]:
        """
        从输入直接到Document集合，内部还是先获取结果，然后再封装
        """
        return list(self.lazy_load(query))

    def lazy_load(self, query: str) -> Iterator[Document]:
        """
        从输入直接到Document集合，内部还是先获取结果，然后再封装
        load(self, query: str) 本质还是调用这个方法
        """
        page_titles = self.wiki_client.search(
            query[:WIKIPEDIA_MAX_QUERY_LENGTH], results=self.top_k_results
        )
        for page_title in page_titles[: self.top_k_results]:
            if wiki_page := self._fetch_page(page_title):
                if doc := self._page_to_document(page_title, wiki_page):
                    yield doc

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123

WikipediaQueryRun源码如下，本质还是对WikipediaAPIWrapper的调用：

class WikipediaQueryRun(BaseTool):
    """Tool that searches the Wikipedia API."""

    name: str = "wikipedia"
    description: str = (
        "A wrapper around Wikipedia. "
        "Useful for when you need to answer general questions about "
        "people, places, companies, facts, historical events, or other subjects. "
        "Input should be a search query."
    )
    api_wrapper: WikipediaAPIWrapper

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

源码分析到这，后续我们如果想自定义Tool，基本步骤也就出来了。怎么说呢，不难bro~

03 自定义Tool

有时候我们想“骚一下”，或者说LangChain内部预定义的Tools已经不能满足我们的使用需求了，这个时候自定义Tool的需求就来了。

一般情况下，自定义Tool需要定义以下几个组件：

  1. name （str） 是必需的，并且在提供给代理的一组工具中必须是唯一的*
  2. description （str） 是可选的，但建议使用，因为代理使用它来确定工具的使用情况*
  3. args_schema （Pydantic BaseModel） 是可选的，但推荐使用，可用于提供更多信息（例如，少量示例）或验证预期参数。*

@tool decorator

@tool 装饰器是定义自定义工具的最简单方法。默认情况下，装饰器使用函数名称作为工具名称，但可以通过传递字符串作为第一个参数来覆盖此名称。此外，装饰器将使用函数的文档字符串作为工具的描述 - 因此必须提供文档字符串。

基本使用如下：

@tool
def search(query: str) -> str:
    """Look up things online."""
    return "LangChain"

print(search.name)
"""
search
"""

print(search.description)
"""
search(query: str) -> str - Look up things online.
"""

print(search.args)
"""
{'query': {'title': 'Query', 'type': 'string'}}
"""
# =================================================================

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

print(multiply.name)
"""
multiply
"""

print(multiply.description)
"""
multiply(a: int, b: int) -> int - Multiply two numbers.
"""

print(multiply.args)
"""
{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

还可以通过将工具名称和 JSON 参数传递到工具装饰器中来自定义它们。

# 注意这里的命名空间，必须是langchain.pydantic_v1
from langchain.pydantic_v1 import BaseModel, Field

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


@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:
    """Look up things online."""
    return "LangChain"

print(search.name)
"""
search-tool
"""

print(search.description)
"""
search-tool(query: str) -> str - Look up things online.
"""

print(search.args)
"""
{'query': {'title': 'Query', 'description': 'should be a search query', 'type': 'string'}}
"""

print(search.return_direct)
"""
True
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

Subclass BaseTool

还可以通过对 BaseTool 类进行子类化来显式定义自定义工具，这种方式更灵活但是更复杂。

我们改造先上面的两个自定义Tool

from typing import Optional, Type
from langchain.pydantic_v1 import BaseModel, Field
from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)


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

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")
    
class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"
    args_schema: Type[BaseModel] = SearchInput
    
    def _run(
        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return "LangChain"
    
    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
    return_direct: bool = True

    def _run(
        self, a: int, b: int, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> int:
        """Use the tool."""
        return a * b

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

打印参数，看下输出

search = CustomSearchTool()
print(search.name)
"""
custom_search
"""

print(search.description)
"""
useful for when you need to answer questions about current events
"""

print(search.args)
"""
{'query': {'title': 'Query', 'description': 'should be a search query', 'type': 'string'}}
"""

multiply = CustomCalculatorTool()
print(multiply.name)
"""
Calculator
"""

print(multiply.description)
"""
useful for when you need to answer questions about math
"""

print(multiply.args)
"""
{'a': {'title': 'A', 'description': 'first number', 'type': 'integer'}, 'b': {'title': 'B', 'description': 'second number', 'type': 'integer'}}
"""

print(multiply.return_direct)
"""
True
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36

StructuredTool dataclass

还可以使用 StructuredTool 数据类。这种方法是前两种方法的混合。它比从 BaseTool 类继承更方便，但提供了比仅使用装饰器更多的功能。

示例代码如下：

from langchain.tools import BaseTool, StructuredTool, tool

def search_function(query: str):
    return "LangChain"


search = StructuredTool.from_function(
    func=search_function,
    name="Search",
    description="useful for when you need to answer questions about current events",
)

print(search.name)
"""
Search
"""

print(search.description)
"""
Search(query: str) - useful for when you need to answer questions about current events
"""

print(search.args)
"""
{'query': {'title': 'Query', 'type': 'string'}}
"""

# ===============================================================

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


calculator = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="multiply numbers",
    args_schema=CalculatorInput,
    return_direct=True,
)

print(calculator.name)
"""
Calculator
"""

print(calculator.description)
"""
Calculator(a: int, b: int) -> int - multiply numbers
"""

print(calculator.args)
"""
{'a': {'title': 'A', 'description': 'first number', 'type': 'integer'}, 'b': {'title': 'B', 'description': 'second number', 'type': 'integer'}}
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61

对于以上三种自定义Tool的方式，从打印结果来看基本一致，具体选择哪种方式还得看实际需求。这里我个人建议使用第一种装饰器的方式~

Handling Tool Errors

当工具遇到错误且未捕获异常时，代理将停止执行。如果希望代理继续执行，可以引发 ToolException 并相应地设置 handle_tool_error。

当抛出 ToolException 时，智能体不会停止工作，而是会根据工具的 handle_tool_error 变量处理异常，处理结果将作为观察结果返回给智能体，并以红色打印。

有点懵？看看代码理解一下~

from langchain_core.tools import ToolException
from langchain.tools import BaseTool, StructuredTool, tool

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

search = StructuredTool.from_function(
    func=search_tool1,
    name="Search_tool1",
    description="A bad tool",
)

search.run("test")

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

上面代码会直接抛异常报错，整个程序会停止工作。

我们将 handle_tool_error 设置成True，此时程序正常执行并输出错误信息。

search = StructuredTool.from_function(
    func=search_tool1,
    name="Search_tool1",
    description="A bad tool",
    handle_tool_error=True,
)

search.run("test") # The search tool1 is not available.

"""
[tool/start] [tool:Search_tool1] Entering Tool run with input:
"test"
[tool/end] [tool:Search_tool1] [1ms] Exiting Tool run with output:
"The search tool1 is not available."
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

我们还可以定义一种自定义方法来处理工具错误，如下：

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


search = StructuredTool.from_function(
    func=search_tool1,
    name="Search_tool1",
    description="A bad tool",
    handle_tool_error=_handle_error,
)

search.run("test") # The following errors occurred during tool execution:The search tool1 is not available.Please try another tool.

"""
[tool/start] [tool:Search_tool1] Entering Tool run with input:
"test"
[tool/end] [tool:Search_tool1] [1ms] Exiting Tool run with output:
"The following errors occurred during tool execution:The search tool1 is not available.Please try another tool."
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

实际开发中，建议在自定义Tool的时候加上错误处理。

04 Toolkits

顾名思义，这叫“工具包”。

工具包是旨在一起用于特定任务的工具集合。他们有方便的装载方法。所有工具包都公开一个 get_tools 方法，该方法返回工具列表。因此，您可以执行以下操作：

# Initialize a toolkit
toolkit = ExampleTookit(...)

# Get list of tools
tools = toolkit.get_tools()

# Create agent
agent = create_agent_method(llm, tools, prompt)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

LangChain内置了很多常用的工具包，如下：

_module_lookup = {
    "AINetworkToolkit": "langchain_community.agent_toolkits.ainetwork.toolkit",
    "AmadeusToolkit": "langchain_community.agent_toolkits.amadeus.toolkit",
    "AzureAiServicesToolkit": "langchain_community.agent_toolkits.azure_ai_services",
    "AzureCognitiveServicesToolkit": "langchain_community.agent_toolkits.azure_cognitive_services",  # noqa: E501
    "CogniswitchToolkit": "langchain_community.agent_toolkits.cogniswitch.toolkit",
    "ConneryToolkit": "langchain_community.agent_toolkits.connery",
    "FileManagementToolkit": "langchain_community.agent_toolkits.file_management.toolkit",  # noqa: E501
    "GmailToolkit": "langchain_community.agent_toolkits.gmail.toolkit",
    "JiraToolkit": "langchain_community.agent_toolkits.jira.toolkit",
    "JsonToolkit": "langchain_community.agent_toolkits.json.toolkit",
    "MultionToolkit": "langchain_community.agent_toolkits.multion.toolkit",
    "NLAToolkit": "langchain_community.agent_toolkits.nla.toolkit",
    "NasaToolkit": "langchain_community.agent_toolkits.nasa.toolkit",
    "O365Toolkit": "langchain_community.agent_toolkits.office365.toolkit",
    "OpenAPIToolkit": "langchain_community.agent_toolkits.openapi.toolkit",
    "PlayWrightBrowserToolkit": "langchain_community.agent_toolkits.playwright.toolkit",
    "PolygonToolkit": "langchain_community.agent_toolkits.polygon.toolkit",
    "PowerBIToolkit": "langchain_community.agent_toolkits.powerbi.toolkit",
    "SQLDatabaseToolkit": "langchain_community.agent_toolkits.sql.toolkit",
    "SlackToolkit": "langchain_community.agent_toolkits.slack.toolkit",
    "SparkSQLToolkit": "langchain_community.agent_toolkits.spark_sql.toolkit",
    "SteamToolkit": "langchain_community.agent_toolkits.steam.toolkit",
    "ZapierToolkit": "langchain_community.agent_toolkits.zapier.toolkit",
    "create_json_agent": "langchain_community.agent_toolkits.json.base",
    "create_openapi_agent": "langchain_community.agent_toolkits.openapi.base",
    "create_pbi_agent": "langchain_community.agent_toolkits.powerbi.base",
    "create_pbi_chat_agent": "langchain_community.agent_toolkits.powerbi.chat_base",
    "create_spark_sql_agent": "langchain_community.agent_toolkits.spark_sql.base",
    "create_sql_agent": "langchain_community.agent_toolkits.sql.base",
}

__all__ = [
    "create_xorbits_agent",
    "create_pandas_dataframe_agent",
    "create_spark_dataframe_agent",
    "create_python_agent",
    "create_csv_agent",
]

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39

这些工具包帮助我们快速方便的处理Json、CSV、Github、SQL、python等任务。

04 Tool as OpenAI Functions

专门适配OpenAI FunctionCall功能，只对gpt相关模型有效，目的是为了将我们定义的各种Tool转换成OpenAI Function。基本使用如下：

from langchain_community.tools import MoveFileTool
from langchain_core.messages import HumanMessage
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-3.5-turbo")

# 创建tool集合
tools = [MoveFileTool()]
# 将tool转换成openai支持的function格式
functions = [convert_to_openai_function(t) for t in tools]

functions[0]
"""
{'name': 'move_file',
 'description': 'Move or rename a file from one location to another',
 'parameters': {'type': 'object',
  'properties': {'source_path': {'description': 'Path of the file to move',
    'type': 'string'},
   'destination_path': {'description': 'New path for the moved file',
    'type': 'string'}},
  'required': ['source_path', 'destination_path']}}
"""

# 将消息和function传给模型执行，这里模型会自动调用tool
message = model.invoke(
    [HumanMessage(content="move file foo to bar")], functions=functions
)
"""
AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n  "source_path": "foo",\n  "destination_path": "bar"\n}', 'name': 'move_file'}})
"""

# 获取function_call,这里的内容都是MoveFileTool里面定义好的
message.additional_kwargs["function_call"]
"""
{'name': 'move_file',
 'arguments': '{\n  "source_path": "foo",\n  "destination_path": "bar"\n}'}
"""

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38

借助 OpenAI 聊天模型，我们还可以使用 bind_functions 自动绑定和转换类似函数的对象。

model_with_functions = model.bind_functions(tools)
model_with_functions.invoke([HumanMessage(content="move file foo to bar")])
"""
AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n  "source_path": "foo",\n  "destination_path": "bar"\n}', 'name': 'move_file'}})
"""

  • 1
  • 2
  • 3
  • 4
  • 5

或者，我们可以使用更新的 OpenAI API，它使用 tools 和 tool_choice 而不是 functions 和 function_call 通过使用 ChatOpenAI.bind_tools：

model_with_tools = model.bind_tools(tools)
model_with_tools.invoke([HumanMessage(content="move file foo to bar")])
"""
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_btkY3xV71cEVAOHnNa5qwo44', 'function': {'arguments': '{\n  "source_path": "foo",\n  "destination_path": "bar"\n}', 'name': 'move_file'}, 'type': 'function'}]})
"""

  • 1
  • 2
  • 3
  • 4
  • 5

05 总结

本篇文章主要是介绍Tool的基本配置，以及如何自定义Tool，并没有介绍如何使用Tool。

实际上，Tool一般会与Agent，也就是所谓的代理（也可以叫智能体）一起使用。这块内容下篇文章会介绍，感谢阅读。

以上就是本次 Tools 的全部内容，希望能给到你们一些小小的帮助。

如果能帮我点个免费的关注，那就是对我个人的最大的肯定。如果觉得写的还行，分享一下也是我生活的小确幸~

在这里插入图片描述

以上内容依据官方文档编写，官方地址：https://python.langchain.com/docs/modules/tools

Peace Guys~

在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/运维做开发/article/detail/875210
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号