LangChain的函数，工具和代理(五)：Tools Routing

关于langchain的函数、工具、代理系列的博客我之前已经写了四篇，还没有看过的朋友请先看一下，这样便于对后续博客内容的理解：

LangChain的函数，工具和代理(一)：OpenAI的函数调用

LangChain的函数，工具和代理(二)：LangChain的表达式语言(LCEL)

LangChain的函数，工具和代理(三)：LangChain中轻松实现OpenAI函数调用

LangChain的函数，工具和代理(四)：使用 OpenAI 函数进行标记(Tagging) & 提取(Extraction)

今天我们来学习Langchain中非常有用的工具“tools”,以及用来选择tools的方法“routing”,在之前的几篇博客中我们介绍了如何在langchain中实现openai的函数调用的功能，这里需要强调的是我们之前介绍的langchain的函数调用并非真正意义上的函数调用，而是让llm根据用户信息的上下文来返回被调用函数的参数，真正的函数调用还是需要手动编写函数调用的代码，那么今天我们来介绍一种让langchain实现真正意义上函数调用功能.

一、Tools

在langchain中，tools是代理(agent)用来与外界交互的接口。这里所谓的“与外界交换”是指让llm可以访问外部世界的功能，如执行互联网搜索，调用外部api等的能力。之前我们介绍了如何让llm来调用外部函数，也就是我们需要先定义好外部函数，然后生成一个openai能理解的该函数的描述对象，最后通过解析llm的返回值来获取调用参数，整个过程比较繁琐。但是有了langchain的tools以后，就可以大大简化整个函数调用的流程。

接下来在正式“抠腚”😀之前，先让我们做一些初始化的工作，如设置opai的api_key，这里我们需要说明一下，在我们项目的文件夹里会存放一个 .env的配置文件，我们将api_key放置在该文件中，我们在程序中会使用dotenv包来读取api_key，这样可以避免将api_key直接暴露在程序中：

#pip install -U python-dotenv
 
import os
import openai
 
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']

下面我们来定义一个函数用来模拟查询天气的函数search，它的逻辑简单只返回一个温度值：

#导入langchain的tool
from langchain.agents import tool

#添加tool装饰器
@tool
def search(query: str) -> str:
    """Search for weather online"""
    return "42f"

print(f"search.name:{search.name}")
print(f"search.description:{search.description}")
print(f"search.args:{search.args}")

这里我们在定义search函数时添加了一个@tool的装饰器，它表示search函数为一个tool，当search函数成为一个tool以后，它就具备了"run"的能力：

#执行函数调用
search.run("sf")

下面我们需要对search函数的输入参数做一个类型限制，所以我们需要创建一个pydantic类：

from pydantic import BaseModel, Field
class SearchInput(BaseModel):
    query: str = Field(description="Thing to search for")
        
@tool(args_schema=SearchInput)
def search(query: str) -> str:
    """Search for the weather online."""
    return "42f"

print(f"search.args:{search.args}")

这里我们给装饰器@tool附加了args_schema参数，它的值为SearchInput，这样就可以限制search函数的输入参数的类型了。

接下来我们需要定义一个真实的调用外部api获取天气温度的函数：

import requests
from pydantic import BaseModel, Field
import datetime

# Define the input schema
class OpenMeteoInput(BaseModel):
    latitude: float = Field(..., description="Latitude of the location to fetch weather data for")
    longitude: float = Field(..., description="Longitude of the location to fetch weather data for")

@tool(args_schema=OpenMeteoInput)
def get_current_temperature(latitude: float, longitude: float) -> str:
    """Fetch current temperature for given coordinates."""
    
    BASE_URL = "https://api.open-meteo.com/v1/forecast"
    
    # Parameters for the request
    params = {
        'latitude': latitude,
        'longitude': longitude,
        'hourly': 'temperature_2m',
        'forecast_days': 1,
    }

    # Make the request
    response = requests.get(BASE_URL, params=params)
    
    if response.status_code == 200:
        results = response.json()
    else:
        raise Exception(f"API Request failed with status code: {response.status_code}")

    current_utc_time = datetime.datetime.utcnow()
    time_list = [datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00')) for time_str in results['hourly']['time']]
    temperature_list = results['hourly']['temperature_2m']
    
    closest_time_index = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))
    current_temperature = temperature_list[closest_time_index]
    
    return f'The current temperature is {current_temperature}°C'

这里我们定义了一个调用外部api(https://api.open-meteo.com/v1/forecast)来获取指定经纬度坐标d地区的天气温度的函数get_current_temperature，该函数的输入参数时经纬度坐标，它们被pydantic类限制了数据类型。下面我们查看一下该函数的输入参数：

print(get_current_temperature.args)

下面我们实际调用一下该函数：

get_current_temperature.run({"latitude": 13, "longitude": 14})

从上面的返回结果中我们看到外部api返回了温度值为18.2度，这里需要说明的是，我们在定义该函数时输入参数为pydantic 类，因此在执行函数调用的时候，输入参数必须是字典的格式，否则会报错。下面我们再定义一个维基百科的查询函数：

#pip install wikipedia

import wikipedia
@tool
def search_wikipedia(query: str) -> str:
    """Run Wikipedia search and get page summaries."""
    page_titles = wikipedia.search(query)
    summaries = []
    for page_title in page_titles[: 3]:
        try:
            wiki_page =  wikipedia.page(title=page_title, auto_suggest=False)
            summaries.append(f"Page: {page_title}\nSummary: {wiki_page.summary}")
        except (
            self.wiki_client.exceptions.PageError,
            self.wiki_client.exceptions.DisambiguationError,
        ):
            pass
    if not summaries:
        return "No good Wikipedia Search Result was found"
    return "\n\n".join(summaries)

这里我们定义了一个维基百科的查询函数search_wikipedia，接下来我们可以直接调用该函数

search_wikipedia.run({"query": "langchain"})

这里我们看到 search_wikipedia函数通过维基百科查询到了关于langchain的相关信息。

二、Routing

前面我们已经有了两个实用的外部函数“get_current_temperature”和“search_wikipedia”，它们可以根据要求来返回天气温度和维基百科的相关信息，接下来我们要让llm在和用户对话的过程中自主判该调用哪一个函数，不过由于llm只能理解函数的描述信息，因此我们需要将这两个函数转换成openai能理解的函数描述变量,我们可以使用langchain的format_tool_to_openai_function方法将函数转换成描述变量比如像这样:

from langchain.tools.render import format_tool_to_openai_function

format_tool_to_openai_function(get_current_temperature)

format_tool_to_openai_function(search_wikipedia)

接下来我们来定义一个可以查询天气温度和维基百科的chain:

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

#创建函数描述变量
functions = [
    format_tool_to_openai_function(f) for f in [
        search_wikipedia, get_current_temperature
    ]
]

#定义llm
model = ChatOpenAI(temperature=0).bind(functions=functions)

#创建prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are helpful but sassy assistant"),
    ("user", "{input}"),
])

#定义chain
chain = prompt | model

#调用chain
chain.invoke({"input": "上海现在的天气怎么样？"})

这里我们看到llm并没有实际去调用函数，而是返回了函数调用的参数，并且llm通过上下文准确识别出需要调用哪个函数，并给出了函数的参数即经纬度坐标。我们再测试一下维基百科查询：

chain.invoke({"input": "langchain 是什么？"})

这里我们看到当llm需要调用函数时返回的消息AIMessage中content都是空的，这个content表示llm返回给用户的文本内容，这就意味着当llm觉得需要调用外部函数时llm不会回答用户任何文本信息，因此返回消息中content都是空的。下面我们给llm打个招呼，看看会返回什么信息：

chain.invoke({"input": "你好"})

这里我们看到当我们和llm打招呼时，llm返回的AIMessage消息中只包含了content内容，除此之外没有其他任何的内容，这说明此时llm意识到不需要调用任何外部函数。接下来我们给chain加上一个解析器组件：

from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser

chain = prompt | model | OpenAIFunctionsAgentOutputParser()

result = chain.invoke({"input": "上海现在的天气怎么样？"})
result

当我们给chain添加了一个解析器组件OpenAIFunctionsAgentOutputParser以后，llm返回了一个AgentActionMessageLog结构的消息体，通过该消息体我们可以轻松获取被调用的函数名和参数：

result.tool

result.tool_input

有了该消息体，我们还可以轻松调用外部函数：

get_current_temperature(result.tool_input)

下面我们测试一下和llm打招呼：

result = chain.invoke({"input": "你好"})
result

这里我们看到当我们和llm打招呼时候，llm返回的并不是之前的 AgentActionMessageLog结构的消息体，而是AgentFinish结构的消息体，这说明当llm需要或者不需要调用外部函数时它们返回的消息体类型是不一样的。

type(result)

当不需要调用函数时，我们可以使用return_values方法来获取llm返回的文本消息：

result.return_values

前面我们介绍的让llm调用外部函数的方法和之前博客中介绍的类似，都是让llm返回被调用函数的参数，并没有真正实现函数调用，要实现真正的函数调用还必须要手工去执行该函数，下面我们来介绍真正的让llm自动化调用函数的功能，不过我们先要创建一个route函数，我们要借助这个rute函数来让llm实现真正的函数调用：

from langchain.schema.agent import AgentFinish

def route(result):
    if isinstance(result, AgentFinish):
        return result.return_values['output']
    else:
        tools = {
            "search_wikipedia": search_wikipedia, 
            "get_current_temperature": get_current_temperature,
        }
        return tools[result.tool].run(result.tool_input)

这里需要说明的是route函数通过识别llm返回消息的类型，来判断是否需要调用函数，如何需要调用函数，则执行之前介绍的函数的run功能来实现真正的函数调用，接下来我们需要给chain添加一个route组件，让它可以实现真正的函数调用：

chain = prompt | model | OpenAIFunctionsAgentOutputParser() | route

result = chain.invoke({"input": "上海的天气现在怎么样？"})
result

这里我们看到当我们向llm询问上海的天气时候，llm调用了天气查询函数，并通过该函数返回了一个真实的温度值。

result = chain.invoke({"input": "langchain是什么？"})
result

从上面的结果中我们看到，当我们询问“langchain是什么”的时候,llm调用了维基百科搜索函数，并将搜索结果返回给了用户。下面我们测试一下不需要调用函数的场景：

result = chain.invoke({"input": "请问老鼠生病了可以吃老鼠药吗？"})
result

这里我们看到，当我们询问llm“老鼠生病了可以吃老鼠药吗”这个问题时，llm给出的结果是它自己判断的结果，并没有调用我们定义好的外部函数。

三、OpenAPI Specification

OpenAPI 规范 (OAS) 定义了一个与语言无关的标准 HTTP API 接口，允许人类和计算机发现和理解服务的功能，而无需访问源代码、文档或通过网络流量检查。下面我们再介绍一下langchain中如何将符合OpenAPI规范文本转换成openai的函数描述变量。下面我们有一个符合openAPI规范的函数描述文本，我们要将它转换成openai的函数描述变量：

text = """
{
  "openapi": "3.0.0",
  "info": {
    "version": "1.0.0",
    "title": "Swagger Petstore",
    "license": {
      "name": "MIT"
    }
  },
  "servers": [
    {
      "url": "http://petstore.swagger.io/v1"
    }
  ],
  "paths": {
    "/pets": {
      "get": {
        "summary": "List all pets",
        "operationId": "listPets",
        "tags": [
          "pets"
        ],
        "parameters": [
          {
            "name": "limit",
            "in": "query",
            "description": "How many items to return at one time (max 100)",
            "required": false,
            "schema": {
              "type": "integer",
              "maximum": 100,
              "format": "int32"
            }
          }
        ],
        "responses": {
          "200": {
            "description": "A paged array of pets",
            "headers": {
              "x-next": {
                "description": "A link to the next page of responses",
                "schema": {
                  "type": "string"
                }
              }
            },
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Pets"
                }
              }
            }
          },
          "default": {
            "description": "unexpected error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Error"
                }
              }
            }
          }
        }
      },
      "post": {
        "summary": "Create a pet",
        "operationId": "createPets",
        "tags": [
          "pets"
        ],
        "responses": {
          "201": {
            "description": "Null response"
          },
          "default": {
            "description": "unexpected error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Error"
                }
              }
            }
          }
        }
      }
    },
    "/pets/{petId}": {
      "get": {
        "summary": "Info for a specific pet",
        "operationId": "showPetById",
        "tags": [
          "pets"
        ],
        "parameters": [
          {
            "name": "petId",
            "in": "path",
            "required": true,
            "description": "The id of the pet to retrieve",
            "schema": {
              "type": "string"
            }
          }
        ],
        "responses": {
          "200": {
            "description": "Expected response to a valid request",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Pet"
                }
              }
            }
          },
          "default": {
            "description": "unexpected error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/Error"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "Pet": {
        "type": "object",
        "required": [
          "id",
          "name"
        ],
        "properties": {
          "id": {
            "type": "integer",
            "format": "int64"
          },
          "name": {
            "type": "string"
          },
          "tag": {
            "type": "string"
          }
        }
      },
      "Pets": {
        "type": "array",
        "maxItems": 100,
        "items": {
          "$ref": "#/components/schemas/Pet"
        }
      },
      "Error": {
        "type": "object",
        "required": [
          "code",
          "message"
        ],
        "properties": {
          "code": {
            "type": "integer",
            "format": "int32"
          },
          "message": {
            "type": "string"
          }
        }
      }
    }
  }
}
"""

该变量text 包含了三个函数的描述内容,接下来我们要将它转换成openai的函数描述变量：

from langchain.chains.openai_functions.openapi import openapi_spec_to_openai_fn
from langchain.utilities.openapi import OpenAPISpec

spec = OpenAPISpec.from_text(text)

pet_openai_functions, pet_callables = openapi_spec_to_openai_fn(spec)
pet_openai_functions

这里我们使用了openapi_spec_to_openai_fn方法轻松的将openapi的变换成了openai的函数描述变量其中包含了3个函数 listPets,createPets,showPetById，下面我们还实际使用一下该函数描述变量：

from langchain.chat_models import ChatOpenAI

model = ChatOpenAI(temperature=0).bind(functions=pet_openai_functions)

model.invoke("what are three pets names")

model.invoke("tell me about pet with id 42")

从上面的结果中我们看到llm能根据问题返回调用函数的参数，这说明符合openapi规范的文本也可以被正确转换成openai能识别的函数描述变量。

四、总结

今天我们学习了langchain的tools,routing组件的原理,并借助tool和routing让llm实现了真正意思上的函数调用，最后我们还介绍了openapi格式的函数说明文本如何通过langchain来转换成openai的函数描述变量，希望今天的内容对大家有所帮助。