Skip to content

大语言模型函数调用与Agent开发 ​

大语言模型工具上手与定义 ​

  1. 学习掌握LangChain构建自定义工具的3种方案,以及不同方案之间的差异性。
  2. 学习掌握LangChain内置工具与工具包的快速上手,实现利用OpenAI+DallE实现自然语言绘图功能。
  3. 学习掌握封装基于第三方API及内置工具的工具,涵盖了高德天气预报、谷歌搜索工具。

大语言模型函数调用原理及应用 ​

  1. 学习掌握大语言模型函数调用的作用与应用,了解不支持函数回调的模型兼容方案及对应的注意事项。
  2. 学习掌握利用大语言模型函数调用实现数据格式化与规范提取技巧,以及LangChain底层实现。
  3. 学习掌握如何使用多模态数据调用工具,为AI应用的多模态输入输出构建基础。

Agent智能体初识与开发 ​

  1. 深入了解什么是Agent智能体以及智能体的组成,以及智能体与大语言模型函数回调的关联。
  2. 学习掌握使用LangChain传统Agent组件实现ReACT架构的智能体,让大语言模型拥有搜索+天气预报查询功能。
  3. 学习LangChain内置封装的Agent组件并快速上手,了解LangChain传统Agent组件的缺陷与不可控性。

LLM函数调用使用技巧与应用场景 ​

01.函数调用的本质与组成 ​

在概率架构下,大语言模型的回复是非结构化且不一致的,所以将大语言模型集成到其他系统中,是存在一定风险的,例如下游系统中想要JSON数据,但是大语言模型返回了一串文本,会让整个流程崩溃。但是如果能更好控制响应格式,就可以轻松地将响应下游集成到其他系统中(虽然通过prompt也可以实现,但是不稳定)。

其次,大语言模型的训练需要大量的计算资源和时间,因此它们的知识库通常是在某个知识点之前的数据集上训练的,例如,GPT-3.5和GPT-4的知识截至2023年10月,这意味着它们无法提供此后的新消息或事件,为保持时效性,需定期重训模型,但成本高昂且费时。

而函数调用(Function calling)的使用就是为了克服大语言模型的上述缺点,LLM中的函数调用架构其实也非常简单,本质上是对输入prompt的自动拆解,利用拆解出来的参数和自动识别出的函数名称,调用外部资源,然后将返回的外部资源添加到prompt之中,再次调用LLMs,从而得到返回结果。

简单来说,函数回调的作用就是可以更可靠地从大语言模型中获取结构化的数据。并且目前绝大部分大参数的模型都支持函数调用,而且除了系统消息、人类消息、AI消息之外,部分大模型还可以添加函数/工具消息,即将函数的生成结果单独作为消息传递给大模型,让大模型基于工具消息生成对应的内容,例如GPT模型。

相关接口文档 ​

  1. OpenAI 大语言模型聊天接口文档: https://platform.openai.com/docs/api-reference/chat/create
  2. 百度文心言函数调用接口文档: https://cloud.baidu.com/doc/WENXINWORKSHOP/s/5ltxyqupn

原生支持函数调用的大模型参数特点 ​

对于原生支持函数调用大语言模型而言,一般来说会提供多一个除了prompt之外的参数,这个参数用于描述函数调用的相关参数(并且目前绝大部分LLM函数参数描述格式都保持一致,参考GPT)。

参数涵盖了函数名字、函数描述、函数参数、函数调用模型等,这样大语言模型才能知道可以调用哪些函数。

GPT模型执行函数调用示例 ​

python
from openai import OpenAI
client=OpenAI()
tools=[
    {
        "type": "function",
        "function":{
            "name":"get_current_weather",
            "description":"Get the current weather in a given location",
            "parameters": {
                "type":"object",
                "properties":{
                    "location": {
                        "type":"string",
                        "description":"The city and state,eg. san Francisco,CA",
                    },
                    "unit": {
                        "type":"string",
                        "enum": ["celsius","fahrenheit"]
                    }
                },
                "required":["location"],
            }
        }
    }
]
messages=[{"role":"user","content":"what's the weather like in Boston today?"}]
completion = client.chat.completions.create(
    model="gpt-3.5-turbo-16k",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)
print(completion)

上述代码定义了get_current_weather函数,用于查询指定城市天气,包含location(必填)和unit两个参数。

当提问波士顿天气时,大模型会自动检测并调用函数,返回结果将包含tool_calls参数,示例返回数据:

json
{
  "id": "chatcmp1-abc123",
  "object": "chat.completion",
  "created": 1699896916,
  "choices": [
    {
      "model": "gpt-4o-mini",
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_abc123",
            "type": "function",
            "function": {
              "arguments": "{\"location\": \"Boston, MA\"}",
              "name": "get_current_weather"
            }
          }
        ]
      },
      "logprobs": null,
      "finish_reason": "tool_calls"
    }
  ],
  "usage": {
    "prompt_tokens": 82,
    "completion_tokens": 17,
    "total_tokens": 99
  }
}

关键注意点 ​

大语言模型本身的函数调用并不会直接调用预定义的函数,而是仅仅生成需要调用的函数的调用参数而已,具体调用函数的动作,需要在应用代码中实现。

函数调用的组成:大语言模型(支持函数调用)+预定Prompt+函数/工具参数列表+本地调用代码。

02.函数调用的想象空间 ​

在函数调用的加持下,让大语言模型作为大脑,接入到上下游系统中,可实现万物智能化,典型应用场景示例:

  1. 用户对着微信说:给我每个女性好友发送一条情真切意的拜年消息,还要带一点小幽默。
  2. 用户对着富途牛牛:人工智能相关股票,市盈率最低的是哪一个?最近交易量如何?都有那些机构持有?
  3. 用户对着京东说:我想买一台65寸的电视,不要日货,价格在5000元左右。

上述需求仅需预设好特定的函数/工具,将参数描述传递给大模型,大模型从prompt中抽取参数后,执行本地函数调用即可实现。

此外,函数调用还可应用于self-RAG优化策略,让LLM自行判断是否需要检索内容,设置tool_choice为auto即可让LLM自主决策。

更进阶的,可让LLM将需求拆分为多个子任务/函数,由不同LLM分工调用不同工具,循环检测任务完成度,最终整合结果输出,这也是AutoGPT的设计架构基础,是基于LLM构建通用人工智能的重要路径。

LangChain中的工具与工具包 ​

01.LangChain内置工具 ​

函数调用需要定义好函数/工具并传递相关信息给大模型,LangChain中封装了大量开箱即用的预设工具,涵盖搜索、图像生成、百科、视频信息提取、代码执行、数据分析等领域。

内置工具相关链接 ​

LangChain内置工具列表: https://imooc-langchain.shortvar.com/docs/integrations/tools/

工具核心特性 ​

  1. 所有工具都是BaseTool的子类,且是Runnable可运行组件;
  2. 支持通过invoke、batch和stream等方法进行调用;
  3. 可通过name、description、args、return_direct属性获取工具相关信息。

DuckDuckGo搜索工具使用示例 ​

  1. 安装依赖库
bash
pip install -u duckduckgo-search
  1. 示例代码
python
from langchain_community.tools import DuckDuckGoSearchRun
search = DuckDuckGoSearchRun()
print("工具名字:",search.name)
print("工具描述:",search.description)
print("工具参数:",search.args)
print("是否直接返回:",search.return_direct)
print(search.run("LangChain目前最新版本是什么?"))
  1. 输出内容示例
工具名字:duckduckgo_search
工具描述:A wrapper around DuckDuckGo Search. Useful for when you need to answer questions about current events. Input should be a search query.
工具参数:{'query':{'title':'query','description':'search query to look up', 'type':'string'}}
是否直接返回:False
LangChain基准是一个python包和相关数据集,用于促进不同认知架构的实验和基准测试。每个基准任务都针对常见LLM应用程序中的关键功能,例如基于检索的问答、提取、代理工具使用等。对于我们的第一个基准,我们发布了一个关于LangChainpython文档的问答...

转换为OpenAI工具格式 ​

可使用convert_to_openai_tool()辅助函数将LangChain工具转换为符合OpenAI规范的工具参数形式,示例:

python
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.utils.function_calling import convert_to_openai_tool
search = DuckDuckGoSearchRun()
print(convert_to_openai_tool(search))

输出结果:

json
{
  "type": "function",
  "function": {
    "name": "duckduckgo_search",
    "description": "A wrapper around DuckDuckGo Search. Useful for when you need to answer questions about current events. Input should be a search query.",
    "parameters": {
      "type": "object",
      "properties": {
        "query": { "description": "search query to look up", "type": "string" }
      },
      "required": ["query"]
    }
  }
}

02.LangChain内置工具包 ​

工具包是一组设计用于一起执行特定任务的工具集,具有便捷加载方法get_tools(),调用该方法可返回工具列表,无需逐个加载工具。

内置工具包相关链接 ​

LangChain内置工具包列表: https://imooc-langchain.shortvar.com/docs/integrations/toolkits/

工具包特点 ​

工具包一般由同一个服务提供商的一系列工具组成,例如AzureAI工具包包含5个工具,分别用于图像分析、文档智能、语音转文字、文字转语音、医疗实体提取。

AzureAI工具包使用示例 ​

  1. 安装依赖包
bash
pip install -U azure-ai-formrecognizer azure-cognitiveservices-speech azure-ai-textanalytics azure-ai-vision-imageanalysis
  1. 配置环境变量
python
import os
os.environ ["OPENAI_API_KEY"] = "sk-"
os.environ["AZURE_AI_SERVICES_KEY"]=""
os.environ["AZURE_AI_SERVICES_ENDPOINT"]=""
os.environ["AZURE_AI_SERVICES_REGION"]=""
  1. 加载工具包
python
from langchain_community.agent_toolkits import AzureAIServicesToolkit
toolkit = AzureAIServicesToolkit()
print([tool.name for tool in toolkit.get_tools()])
  1. 输出内容
['azure_ai_services_document_intelligence',
'azure_ai_services_image_analysis',
'azure_ai_services_speech_to_text',
'azure_ai_services_text_to_speech',
'azure_ai_services_text_analytics_for_health']

工具包使用注意点 ​

  1. 工具包包含多组工具,一般很少单独拆分使用,通常用于Agent或函数调用中;
  2. 实际开发中,更多是手动实例化工具进行组装,而非直接使用工具包。

创建自定义工具的3种技巧与使用场景 ​

LangChain内置工具不一定适配业务场景,实际开发中常需要自定义工具,LangChain提供3种构建方式:@tool装饰器、structuredtool.from_function()类方法、BaseTool子类,不同方式有不同优缺点和应用场景。

01.@tool装饰器 ​

@tool装饰器是定义工具的最简单方式,可快速将普通函数改造成大语言模型工具,适合原有函数、参数简单的场景。

核心特性 ​

  1. 默认使用函数名称作为工具名称,可通过传递字符串覆盖;
  2. 使用函数的文档字符串作为工具的描述,被装饰函数必须提供文档字符串;
  3. 不支持同时拥有同步和异步方法,需单独装饰。

基础使用示例 ​

python
from langchain_core.tools import tool
@tool
def multiply(a:int,b:int)->int:
    """将传递的两个数字相乘"""
    return a*b
# 打印工具信息
print("名称:",multiply.name)
print("描述:",multiply.description)
print("参数:",multiply.args)
print("直接返回:",multiply.return_direct)
# 调用工具
print(multiply.invoke({"a": 2,"b":8}))

输出结果:

名称: multiply
描述: 将传递的两个数字相乘
参数: {'a':{'title':'A','type':'integer'},'b':{'title':'B','type':'integer'}}
直接返回: False
16

自定义配置示例 ​

可传递工具名、参数schema、是否直接返回等配置,示例:

python
from langchain_core.tools import tool
from langchain_core.pydantic_v1 import BaseModel, Field
class CalculatorInput (BaseModel):
    a:int=Field(description="第一个数字")
    b:int=Field(description="第二个数字")
@tool("multiply_tool", args_schema=CalculatorInput, return_direct=True)
def multiply(a: int,b:int)->int:
    """将传递的两个数字相乘"""
    return a*b
# 打印工具信息
print("名称:",multiply.name)
print("描述:",multiply.description)
print("直接返回:",multiply.return_direct)
print("参数:",multiply.args)
# 调用工具
print(multiply.invoke({"a": 2,"b": 8}))

输出结果:

名称: multiply_tool
描述: 将传递的两个数字相乘
直接返回: True
参数: {'a':{'title':'A','description':'第一个数字','type':'integer'},'b':{'title':'B','description':'第二个数字','type':'integer'}}
16

解析Google-style文档字符串 ​

设置parse_docstring=True,可从Google风格的函数注释中提取参数解释,示例:

python
@tool(parse_docstring=True)
def foo(bar: str, baz: int) -> str:
    """The foo.
    Args:
        bar: The bar.
        baz: The baz.
    """
    return bar
print(foo.args)

输出结果:

{'bar':{'title':'Bar','description':'The bar.','type':'string'},'baz':{'title':'Baz','description':'The baz.','type':'integer'}}

异步函数装饰示例 ​

python
from langchain_core.tools import tool
@tool
async def amultiply(a:int,b:int)->int:
    """Multiply two numbers."""
    return a*b

02. StructuredTool 类方法 ​

使用structuredtool.from_function()类方法创建工具,支持更多配置项,可同时定义同步和异步方法,适合已有函数且需要同步/异步兼容的场景。

核心特性 ​

  1. 配置灵活,支持自定义工具名、描述、参数schema、是否直接返回;
  2. 可同时指定同步函数func和异步协程coroutine;
  3. 无需额外大量代码,适配性强。

使用示例 ​

python
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.tools import StructuredTool
class CalculatorInput(BaseModel):
    a:int= Field(description="第一个数字")
    b:int=Field(description="第二个数字")
# 同步函数
def multiply(a:int,b:int)->int:
    """将传递的两个数字相乘"""
    return a*b
# 异步函数
async def amultiply(a:int,b:int)->int:
    """将传递的两个数字相乘"""
    return a*b
# 创建工具
calculator = StructuredTool.from_function(
    coroutine=amultiply,
    name="multiply_tool",
    func=multiply,
    description="用于将传递的两个整型相乘",
    return_direct=True,
    args_schema=CalculatorInput,
)
# 打印工具信息
print("名称:",calculator.name)
print("描述:",calculator.description)
print("参数:", calculator.args)
print("直接返回:",calculator.return_direct)
# 调用工具
print(calculator.invoke({"a": 2, "b": 8}))

输出结果:

名称: multiply_tool
描述: 用于将传递的两个整型相乘
参数: {'a':{'title':'A','description':'第一个数字','type':'integer'},'b':{'title':'B','description':'第二个数字','type':'integer'}}
直接返回: True
16

03.BaseTool子类 ​

LangChain中所有工具本质都是BaseTool子类,@tool和StructuredTool.from_function()创建的工具底层也会包装为StructuredTool(BaseTool子类)。适合无现成代码,需要完全自定义工具逻辑的场景。

实现方式 ​

继承BaseTool基类,实现_run()方法(核心业务逻辑),可自定义工具名、描述、参数schema。

使用示例 ​

python
from typing import Any,Type
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.tools import BaseTool
# 定义参数schema
class CalculatorInput (BaseModel):
    a:int=Field(description="第一个数字")
    b:int=Field(description="第二个数字")
# 自定义工具类
class MultiplyTool(BaseTool):
    """乘法计算工具"""
    name="multiply_tool"
    description="将传递的两个数字相乘"
    args_schema: Type[BaseModel]= CalculatorInput
    # 实现核心运行方法
    def _run(self,a:int,b:int)->Any:
        return a*b
# 实例化工具
calculator = MultiplyTool()
# 打印工具信息
print("名称:",calculator.name)
print("描述:",calculator.description)
print("参数:",calculator.args)
print("直接返回:",calculator.return_direct)
# 调用工具
print(calculator.invoke({"a": 2, "b": 8}))

输出结果:

名称: multiply_tool
描述: 将传递的两个数字相乘
参数: {'a':{'title':'A','description':'第一个数字','type':'integer'},'b':{'title':'B','description':'第二个数字','type':'integer'}}
直接返回: False
16

高德天气预报查询插件的集成与编写 ​

01.高德城市与天气预报接口 ​

高德Web服务API提供大量HTTP接口,涵盖地理/逆地理编码、路径规划、天气查询等,开发/调试有足量免费调用额度,商用需付费。

相关链接 ​

  1. 高德Web服务API链接: https://lbs.amap.com/api/webservice/summary
  2. 高德开放平台控制台: https://console.amap.com/dev/index

前置准备 ​

  1. 实名认证高德开放平台并创建应用,获取API秘钥;
  2. 将秘钥配置到环境变量:GAODE_API_KEY=00b0****e6e7。

核心接口 ​

实现天气查询需要两个接口配合,因为天气预报接口仅支持城市编码入参:

  1. 行政区域查询接口:根据城市名获取城市编码
    https://restapi.amap.com/v3/config/district?keywords=<城市名>&subdistrict=0&key=<用户key>
  2. 天气预报查询接口:根据城市编码获取天气信息
    https://restapi.amap.com/v3/weather/weatherInfo?city=<城市编码>&key=<用户key>

运行流程 ​

原始问题 → LLM提取城市名 → 行政区域查询获取城市编码 → 天气预报接口查询天气 → LLM综合生成 → 最终答案

02.天气预报查询工具实现 ​

无现成代码,采用BaseTool子类方式实现,工具接收city参数,依次调用行政区域查询和天气预报接口,返回天气信息。

完整代码示例 ​

python
import json
import os
from typing import Type,Any
import dotenv
from langchain_core.pydantic_v1 import Field, BaseModel
import requests
from langchain_core.tools import BaseTool
dotenv.load_dotenv()
# 定义参数schema
class GaodeWeatherArgsSchema(BaseModel):
    city:str=Field(description="需要查询天气预报的目标城市,例如:广州")
# 自定义天气工具类
class GaodeWeatherTool(BaseTool):
    """根据传入的城市名查询天气"""
    name="gaode_weather"
    description="当你想询问天气或与天气相关的问题时的工具。"
    args_schema: Type[BaseModel]= GaodeWeatherArgsSchema
    def _run(self, *args: Any, **kwargs: Any)-> str:
        """运行工具获取对应城市的天气预报"""
        try:
            # 1.获取高德API秘钥
            gaode_api_key = os.getenv("GAODE_API_KEY")
            if not gaode_api_key:
                return f"高德开放平台API秘钥未配置"
            # 2.提取城市名,调用行政区域查询接口获取城市编码
            city = kwargs.get("city", "")
            session = requests.session()
            api_domain = "https://restapi.amap.com/v3"
            city_response = session.request(
                method="GET",
                url=f"{api_domain}/config/district?keywords={city}&subdistrict=0&extensions=all&key={gaode_api_key}",
                headers={"Content-Type": "application/json; charset=utf-8"},
            )
            city_response.raise_for_status()
            city_data = city_response.json()
            # 3.提取行政编码,调用天气预报接口
            if city_data.get("info") == "OK":
                if len(city_data.get("districts"))>0:
                    ad_code = city_data["districts"][0]["adcode"]
                    weather_response = session.request(
                        method="GET",
                        url=f"{api_domain}/weather/weatherInfo?city={ad_code}&extensions=all&key={gaode_api_key}&output=json",
                        headers={"Content-Type": "application/json;charset=utf-8"},
                    )
                    weather_response.raise_for_status()
                    weather_data = weather_response.json()
                    if weather_data.get("info")== "OK":
                        return json.dumps (weather_data)
            session.close()
            return f"获取{kwargs.get('city')}天气预报信息失败"
        except Exception as e:
            return f"获取{kwargs.get('city')}天气预报信息失败"
# 实例化并调用工具
gaode_weather = GaodeWeatherTool()
print(gaode_weather.invoke({"city": "广州"}))

输出结果示例 ​

json
{
  "status": "1",
  "count": "1",
  "info": "OK",
  "infocode": "10000",
  "forecasts": [
    {
      "city": "广州市",
      "adcode": "440100",
      "province": "广东",
      "reporttime": "2024-08-12 15:30:27",
      "casts": [
        {
          "date": "2024-08-12",
          "week": "1",
          "dayweather": "中雨",
          "nightweather": "雨",
          "daytemp": "34",
          "nighttemp": "25",
          "daywind": "北",
          "nightwind": "北",
          "daypower": "1-3",
          "nightpower": "1-3",
          "daytemp_float": "34.0",
          "nighttemp_float": "25.0"
        },
        {
          "date": "2024-08-13",
          "week": "2",
          "dayweather": "中雨",
          "nightweather": "雨",
          "daytemp": "33",
          "nighttemp": "25",
          "daywind": "北",
          "nightwind": "北",
          "daypower": "1-3",
          "nightpower": "1-3",
          "daytemp_float": "33.0",
          "nighttemp_float": "25.0"
        }
      ]
    }
  ]
}

谷歌实时信息搜索插件的集成与编写 ​

01.Serper谷歌搜索 ​

谷歌官方搜索API体验差、访问慢,选择第三方提供商Serper,其在后台模拟浏览器操作,检索内容更全面、响应速度更快。

相关链接 ​

  1. LangChain Serper谷歌搜索集成: https://imooc-langchain.shortvar.com/docs/integrations/tools/google_serper/
  2. Serper官网: https://serper.dev/
  3. LangChain SerpAPI谷歌搜索集成: https://imooc-langchain.shortar.com/docs/integrations/tools/serpapi/
  4. SerpAPI官网: https://serpapi.com/

前置准备 ​

  1. 用邮箱/Git/Google账号注册Serper并创建API秘钥;
  2. 将秘钥配置到环境变量:SERPER_API_KEY=9b5bd82f638b1e51a9bf14802a0a2799286b194e。

LangChain内置相关工具 ​

LangChain已封装Serper搜索的两个内置工具,区别如下:

  1. GoogleSerperRun:仅返回搜索文本内容;
  2. GoogleSerperResults:返回搜索文本+网页链接等元数据。

注意点 ​

LangChain封装的GoogleSerperRun工具无中文参数说明,工具描述为英文,需二次封装适配中文场景;Serper与SerpAPI名称、功能相近,易混淆,注意区分。

02.谷歌实时信息搜索工具实现 ​

对GoogleSerperRun进行二次封装,无需重新定义工具类,仅在实例化时自定义名称、描述、参数schema即可。

完整代码示例 ​

python
import dotenv
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.tools import GoogleSerperRun
from langchain_core.pydantic_v1 import BaseModel, Field
# 定义参数schema
class GoogleSerperArgsSchema(BaseModel):
    query:str=Field(description="执行谷歌搜索的查询语句")
dotenv.load_dotenv()
# 实例化并二次封装工具
google_serper= GoogleSerperRun(
    name="google_serper",
    description=(
        "一个低成本的谷歌搜索API."
        "当你需要回答有关时事的问题时,可以调用该工具。"
        "该工具的输入是搜索查询语句。"
    ),
    args_schema=GoogleSerperArgsSchema,
    api_wrapper=GoogleSerperAPIWrapper(),
)
# 调用工具
print(google_serper.invoke("马拉松的世界记录是多少?"))

输出结果示例 ​

2004年1月1日,国际田联宣布,马拉松开始拥有世界纪录。2023年10月8日晚(北京时间),在芝加哥马拉松比赛中,肯尼亚人基普图在芝加哥马拉松比赛中以2小时00分35秒的成绩打破基普乔格保持的世界纪录。

ChatModel使用函数调用的技巧与流程 ​

01.bind()与bind_tools() ​

OpenAI原生函数调用方式 ​

在GPT模型中,通过在chat.completions.create中传递tools和tool_choice参数实现函数调用,示例:

python
completion= client.chat.completions.create(
    model="gpt-3.5-turbo-16k",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

LangChain中的绑定方式 ​

  1. 基础方式:使用convert_to_openai_tool将LangChain工具转为OpenAI格式,再通过bind()函数传递tools和tool_choice参数,完成模型与工具的绑定;
  2. 快捷方式:LangChain为支持函数调用的大模型提供bind_tools()函数,可直接完成工具绑定,简化操作。

bind_tools()核心代码逻辑 ​

python
def bind_tools(
    self,
    tools: Sequence[Union[Dict[str, Any], Type[BaseModel], callable, BaseTool]],
    tool_choice: Optional[Union[dict, str,Literal["auto","none","required","any"],bool]]=None,
    **kwargs: Any
) -> Runnable[LanguageModelInput,BaseMessage]:
    formatted_tools = [convert_to_openai_tool(tool) for tool in tools]
    if tool_choice:
        # 工具选择参数格式处理逻辑
        pass
    kwargs["tool_choice"] = tool_choice
    return super().bind(tools=formatted_tools,** kwargs)

02.LLM绑定天气预报与谷歌实时搜索 ​

将高德天气预报和谷歌Serper搜索工具绑定到GPT模型,模型调用工具后,将工具结果附加到历史消息,再次调用模型生成自然语言答案。

完整代码示例 ​

python
import json
import os
from typing import Type, Any
import dotenv
import requests
from langchain_community.tools import GoogleSerperRun
from langchain_community.utilities import GoogleSerperAPIwrapper
from langchain_core.messages import ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import Field, BaseModel
from langchain_core.runnables import RunnablePassthrough
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
# 1.定义参数schema
class GaodeWeatherArgsSchema(BaseModel):
    city:str=Field(description="需要查询天气预报的目标城市,例如:广州")
class GoogleSerperArgsSchema(BaseModel):
    query:str=Field(description="执行谷歌搜索的查询语句")
# 2.实现高德天气工具
class GaodeWeatherTool(BaseTool):
    """根据传入的城市名查询天气"""
    name="gaode_weather"
    description="当你想询问天气或与天气相关的问题时的工具。"
    args_schema: Type[BaseModel] = GaodeWeatherArgsSchema
    def _run(self, *args: Any, **kwargs: Any)-> str:
        try:
            """运行工具获取对应城市的天气预报"""
            gaode_api_key = os.getenv("GAODE_API_KEY")
            if not gaode_api_key:
                return f"高德开放平台API秘未配置"
            city = kwargs.get("city", "")
            session = requests.session()
            api_domain = "https://restapi.amap.com/v3"
            # 行政区域查询
            city_response = session.request(
                method="GET",
                url=f"{api_domain}/config/district?keywords={city}&subdistrict=0&extensions=all&key={gaode_api_key}",
                headers={"Content-Type": "application/json; charset=utf-8"},
            )
            city_response.raise_for_status()
            city_data = city_response.json()
            # 天气预报查询
            if city_data.get("info")== "ok":
                if len(city_data.get("districts"))>0:
                    ad_code = city_data["districts"][0]["adcode"]
                    weather_response = session.request(
                        method="GET",
                        url=f"{api_domain}/weather/weatherInfo?city={ad_code}&extensions=all&key={gaode_api_key}&output=json",
                        headers={"Content-Type": "application/json;charset=utf-8"},
                    )
                    weather_data = weather_response.json()
                    weather_response.raise_for_status()
                    if weather_data.get("info")=="OK":
                        return json.dumps (weather_data)
            session.close()
            return f"获取{kwargs.get('city')}天气预报信息失败"
        except Exception as e:
            return f"获取{kwargs.get('city')}天气预报信息失败"
# 3.定义工具列表并实例化
gaode_weather = GaodeWeatherTool()
google_serper = GoogleSerperRun(
    name="google_serper",
    description=(
        "一个低成本的谷歌搜索API。"
        "该工具的输入是搜索查询语句。 "
        "当你需要回答有关时事的问题时,可以调用该工具。"
    ),
    args_schema=GoogleSerperArgsSchema,
    api_wrapper=GoogleSerperAPIWrapper(),
)
tool_dict={
    gaode_weather.name: gaode_weather,
    google_serper.name:google_serper,
}
tools = [tool for tool in tool_dict.values()]
# 4.创建Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system","你是由OpenAI开发的聊天机器人,可以帮助用户回答问题,必要时刻请用工具帮助用户解答"),
    ("human","{query}"),
])
# 5.创建大语言模型并绑定工具
llm = ChatOpenAI (model="gpt-3.5-turbo-16k", temperature=0)
llm_with_tool = llm.bind_tools(tools=tools)
# 6.创建链应用
chain = {"query": RunnablePassthrough} | prompt | llm_with_tool
# 7.执行查询并解析结果
query="广州现在天气怎样,有什么适合穿的衣服呢"
resp = chain.invoke(query)
tool_calls= resp.tool_calls
# 8.判断是否调用工具
if len(tool_calls)<=0:
    print("生成内容:",resp.content)
else:
    # 组合历史消息
    messages = prompt.invoke (query).to_messages()
    messages.append(resp)
    # 遍历执行工具调用
    for tool_call in tool_calls:
        tool= tool_dict.get(tool_call.get("name"))
        print("正在执行工具:",tool.name)
        id=tool_call.get("id")
        content = tool.invoke(tool_call.get("args"))
        print("工具输出:",content)
        # 添加工具消息到历史
        messages.append(ToolMessage(
            content=content,
            tool_call_id=id,
        ))
    # 模型基于工具结果生成最终答案
    print("输出内容:",llm.invoke (messages))

输出结果示例 ​

正在执行工具: gaode_weather
工具输出: {"status":"1","count":"1","info":"OK","infocode":"10000","forecasts": [{"city": "广州市","adcode": "440100","province": "广东","reporttime": "2024-08-12 18:30:27","casts": [{"date": "2024-08-12","week":"1","dayweather":"中雨","nightweather":"中雨","daytemp": "34","nighttemp": "25","daywind": "北", "nightwind": "北", "daypower": "1-3", "nightpower": "1-3","daytemp_float": "34.0", "nighttemp_float": "25.0"}]}]}
输出内容: content='广州目前的天气情况是中雨,白天气温为34摄氏度,夜间气温为25摄氏度,根据天气情况,建议您穿着轻薄透气的衣物,同时带上一把伞以备不时之需。请注意保持身体健康,避免长时间暴露在雨中.'

关键说明 ​

上述代码实际已构建出基础的Agent智能体,具备实时信息搜索和天气预报查询能力;但通过LCEL表达式结合判断、循环、工具调用后,代码维护性有所下降,LangChain提供了传统Agent和LangGraph两种封装方案解决该问题。

不支持函数调用的大语言模型解决技巧 ​

市面上部分大模型(尤其是开源模型)不支持原生函数调用,可通过编写特定Prompt让模型按指定格式输出函数调用参数,实现间接的函数调用功能。

01.Prompt实现函数调用功能思路 ​

核心原理 ​

将函数调用的相关参数(工具名、描述、参数格式)嵌入到Prompt中传递给大模型,告知模型需按指定格式(如JSON)输出函数调用参数,然后在代码中解析模型输出:若解析到有效参数则调用对应工具,否则直接返回模型回答。

运行流程 ​

工具描述+格式要求 → 嵌入Prompt → 大模型输出 → 解析输出 → 调用工具/直接返回 → 整合结果

方案缺陷 ​

  1. 大模型输出具有不确定性,无法保证每次都正确输出参数格式,应用易出错;
  2. 复杂函数的参数描述会让Prompt过长,占用大量模型上下文;
  3. 难以实现多工具同时调用,多工具调用会导致模型输出混乱,对模型能力要求极高。

LangChain辅助函数 ​

LangChain提供工具转文本描述的辅助函数,快速生成Prompt所需的工具信息:

  1. render_text_description_and_args():转换为带参数结构信息的工具描述;
  2. render_text_description():仅转换为工具基础介绍(无参数信息)。

02.Prompt函数调用示例 ​

基于高德天气和谷歌Serper搜索工具,通过Prompt指定格式,结合JsonOutputParser解析模型输出,实现函数调用,兼容不支持原生函数调用的大模型。

核心代码片段(关键部分) ​

python
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.tools import render_text_description_and_args
from typing import TypedDict, Dict, Optional
# 定义工具调用请求格式
class ToolCallRequest(TypedDict):
    """工具调用请求字典"""
    name:str
    arguments: Dict[str, Any]
# 定义工具执行函数
def invoke_tool(
    tool_call_request: ToolCallRequest,
    config: Optional[RunnableConfig]=None,
):
    """执行工具调用的函数"""
    tool_name_to_tool={tool.name: tool for tool in tools}
    name=tool_call_request["name"]
    requested_tool = tool_name_to_tool[name]
    return requested_tool.invoke(tool_call_request["arguments"], config=config)
# 定义系统Prompt,包含工具描述和格式要求
system_prompt="""您是一个由OpenAI开发的聊天机器人,可以访问以下一组工具。
以下是每个工具的名称和描述:
{rendered_tools}
根据用户输入,返回要使用的工具的名称和输入。
将您的响应作为有name和arguments键的JSON块返回,arguments应该是一个字典,其键对应于参数名称,值对应与请求的值。"""
# 创建Prompt并填充工具描述
prompt = ChatPromptTemplate.from_messages([
    ("system",system_prompt),
    ("human","{query}"),
]).partial(rendered_tools=render_text_description_and_args(tools))
# 创建链应用:Prompt → LLM → JSON解析 → 工具调用
llm = ChatOpenAI (model="gpt-3.5-turbo-16k",temperature=0)
chain=(
    {"query":RunnablePassthrough}
    | prompt
    | llm
    | JsonOutputParser()
    | RunnablePassthrough.assign(output=invoke_tool)
)
# 调用链
print(chain.invoke("马拉松世界记录是多少?"))

输出结果示例 ​

{'name':'google_serper','arguments':{'query':'马拉松世界记录是多少'},'output':'2004年1月1日,国际田联宣布,马拉松开始拥有世界纪录。2023年10月8日(北京时间),在芝加哥马拉松比赛中,肯尼亚人基图姆在芝加哥马拉松比赛中以2小时00分35秒的成绩打破基乔格保持的世界纪录。'}

生产环境注意点 ​

该方案依赖JsonOutputParser解析模型输出,若模型未按JSON格式输出(如闲聊问题“你好,你是?”),会导致链应用崩溃。生产环境中应尽可能使用支持原生函数调用的大模型,避免程序变得脆弱。

Talk is cheap, show me the code!