langchain从入门到精通（四十）——函数调用技巧与流程
1. ChatModel使用函数调用的技巧与流程
1.1 bind()与bind_tools()
在 OpenAI 的 GPT 模型如何添加工具的描述参数，即在实例化的时候传递多一个 tools 参数，如下：
completion = client.chat.completions.create(model="gpt-3.5-turbo-16k",messages=messages,tools=tools,tool_choice="auto"
)
其实除了上面的这些参数，我们还可以传递 temperature 温度参数，这些参数本质上都是模型生成内容时传递的 运行时参数，所以在 LangChain 中，我们可以使用 convert_to_openai_tool 将自定义工具转换成符合 GPT 模型的参数格式，使用 .bind() 函数来传递对应的tools和 tool_choice，从而完成对大语言模型函数的绑定。
运行流程如下：
不过由于上述的步骤太过于常见，所以 LangChain 团队单独对支持 函数调用 的大语言模型添加了 .bind_tools() 函数，可以快捷地完成此操作，例如 GPT 模型的 .bind_tools() 函数的核心代码如下:
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:if isinstance(tool_choice, str):# tool_choice is a tool/function nameif tool_choice not in ("auto", "none", "any", "required"):tool_choice = {"type": "function","function": {"name": tool_choice},}# 'any' is not natively supported by OpenAI API.# We support 'any' since other models use this instead of 'required'.if tool_choice == "any":tool_choice = "required"elif isinstance(tool_choice, bool):tool_choice = "required"elif isinstance(tool_choice, dict):tool_names = [formatted_tool["function"]["name"]for formatted_tool in formatted_tools]if not any(tool_name == tool_choice["function"]["name"]for tool_name in tool_names):raise ValueError(f"Tool choice {tool_choice} was specified, but the only "f"provided tools were {tool_names}.")else:raise ValueError(f"Unrecognized tool_choice type. Expected str, bool or dict. "f"Received: {tool_choice}")kwargs["tool_choice"] = tool_choice
return super().bind(tools=formatted_tools, **kwargs)
1.2 LLM绑定天气预报与谷歌实时搜索
天气预报 与 谷歌实时搜索 接入到 GPT 模型中，其实只需要创建好自定义工具后，将自定义工具组装成列表，然后调用 .bind_tools() 函数即可完成 LLM 对函数的绑定，当大语言模型返回的内容携带 函数调用参数 时，可以通过 .tool_calls 属性来获取对应的信息。
例如：将 GPT 模型绑定 天气预报 与 谷歌实时，并且当执行工具调用时，将 工具结果 附加到历史消息列表中，再次传递给大语言模型，让其生成对应的内容，完整代码如下
import json
import os
from typing import Type, Anyimport 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 ChatOpenAIdotenv.load_dotenv()class GaodeWeatherArgsSchema(BaseModel):city: str = Field(description="需要查询天气预报的目标城市，例如：广州")class GoogleSerperArgsSchema(BaseModel):query: str = Field(description="执行谷歌搜索的查询语句")class GaodeWeatherTool(BaseTool):"""根据传入的城市名查询天气"""name = "gaode_weather"description = "当你想询问天气或与天气相关的问题时的工具。"args_schema: Type[BaseModel] = GaodeWeatherArgsSchemadef _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')}天气预报信息失败"# 4.整合天气预报信息并返回except Exception as e:return f"获取{kwargs.get('city')}天气预报信息失败"# 1.定义工具列表
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()]# 2.创建Prompt
prompt = ChatPromptTemplate.from_messages([("system", "你是由OpenAI开发的聊天机器人，可以帮助用户回答问题，必要时刻请调用工具帮助用户解答"),("human", "{query}"),
])# 3.创建大语言模型并绑定工具
llm = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0)
llm_with_tool = llm.bind_tools(tools=tools)# 4.创建链应用
chain = {"query": RunnablePassthrough()} | prompt | llm_with_tool# 5.解析输出
query = "广州现在天气怎样，有什么适合穿的衣服呢"
resp = chain.invoke(query)
tool_calls = resp.tool_calls# 6.判断是工具调用还是正常输出结果
if len(tool_calls) <= 0:print("生成内容:", resp.content)
else:# 7.将历史的系统消息、人类消息、AI消息组合messages = prompt.invoke(query).to_messages()messages.append(resp)# 8.循环遍历所有工具调用信息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))
使用 LECL 表达式构建的链应用是非常优雅的，但是加入了 判断、循环、工具调用 等模块后，维护起来也相对吃力，不过这个问题很快就会解决，在 LangChain 中针对 Agent 应用的创建，目前有两种封装好的策略，一种使用传统 Agent，另外一种使用 LangGraph。
2. 不支持函数调用的大语言模型解决技巧
2.1 Prompt实现函数调用功能思路
由于市面上的大语言模型众多，并不是所有的大语言模型都支持 函数调用 这个功能的，还存在相当多的大模型并不支持（特别是开源模型），对于这类模型，其实也可以通过编写特定的 Prompt，即可让模型调用适当的工具，原理其实就是让大语言模型尽可能按照特定的格式进行输出（例如输出函数的调用参数 JSON 数据）。
所以在这种解决方案中，就需要将 函数调用 的相关参数嵌入到 prompt 中一起传递给大语言模型，并且告知大语言模型生成参数的响应格式，接下来针对大语言模型输出的内容进行解析，如果解析到参数则调用函数，否则正常输出。
该解决方案虽然可以临时为大语言模型添加 函数调用 功能，但是缺陷也非常大：
1. 大模型输出的不确定性，并不是每一次都可以正确调用函数的，应用很容易出错。
2. 如果函数参数很复杂，产生的 prompt 可能会非常长，占用大模型的上下文。
3. 没法一次性调用多个工具，或者一次性调用多个工具，会让大语言模型的输出变得异常混乱，对模型考验能力大。
在 LangChain 中，除了封装了 convert_to_openai_tool() 工具快速将工具转换成 GPT 模型的函数参数，还可以使用 render_text_description_and_args() 或者 render_text_description() 快速将工具转换成 描述文本，使用技巧一模一样，其中前者转换的描述带参数结构信息，后者仅为函数基础介绍。
Prompt函数调用示例,将示例修改成通过 prompt 填充函数参数，并让大语言模型按照特定的规则生成相应的数据（可以使用任意的大语言模型来完成这个示例），完整代码如下：
import json
import os
from typing import Type, Any, TypedDict, Dict, Optionalimport dotenv
import requests
from langchain_community.tools import GoogleSerperRun
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import Field, BaseModel
from langchain_core.runnables import RunnablePassthrough, RunnableConfig
from langchain_core.tools import BaseTool, render_text_description_and_args
from langchain_openai import ChatOpenAIdotenv.load_dotenv()class GaodeWeatherArgsSchema(BaseModel):city: str = Field(description="需要查询天气预报的目标城市，例如：广州")class GoogleSerperArgsSchema(BaseModel):query: str = Field(description="执行谷歌搜索的查询语句")class GaodeWeatherTool(BaseTool):"""根据传入的城市名查询天气"""name = "gaode_weather"description = "当你想询问天气或与天气相关的问题时的工具。"args_schema: Type[BaseModel] = GaodeWeatherArgsSchemadef _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')}天气预报信息失败"# 4.整合天气预报信息并返回except Exception as e:return f"获取{kwargs.get('city')}天气预报信息失败"class ToolCallRequest(TypedDict):"""工具调用请求字典"""name: strarguments: Dict[str, Any]def invoke_tool(tool_call_request: ToolCallRequest, config: Optional[RunnableConfig] = None,
):"""我们可以使用的执行工具调用的函数。:param tool_call_request: 一个包含键名和参数的字典，名称必须与现有工具的名称匹配，参数是该工具的参数。:param config: 这是LangChain使用的包含回调、元数据等信息的配置信息。:return: 请求工具的输出内容。"""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)# 1.定义工具列表
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()]system_prompt = """您是一个由OpenAI开发的聊天机器人，可以访问以下一组工具。
以下是每个工具的名称和描述：{rendered_tools}根据用户输入，返回要使用的工具的名称和输入。
将您的响应作为具有`name`和`arguments`键的JSON块返回。
`arguments`应该是一个字典，其中键对应于参数名称，值对应与请求的值。"""
prompt = ChatPromptTemplate.from_messages([("system", system_prompt),("human", "{query}"),
]).partial(rendered_tools=render_text_description_and_args(tools))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 信息，如果某一次回复大语言模型输出了 正常字符串，例如提问 “你好，你是？”，这个时候链应用就崩溃了，所以在生产环境中，涉及到 函数调用、路由逻辑 等需要规范化数据的内容，尽可能使用支持 函数调用 的大语言模型，避免程序变得很脆弱。
3. 函数调用快速提取结构化数据使用技巧
3.1 结构化输出策略与选择
让 LLM 返回结构化输出非常重要，这是因为 LLM 应用程序的输出通常用于下游程序，这些应用程序需要特定的参数，目前常见的几种让 LLM 结构化输出的策略有：
1. Prompt：通过 prompt 让 LLM 输出特定结构的内容，兼容所有 LLM，但是输出不稳定。
2. 函数/工具调用：让 LLM 绑定函数，并设置 选择模式 为强制，让 LLM 强制调用函数，从而获取结构化输出数据。
3. JSON模式 ：对于支持 JSON 模式输出的 LLM，还可以通过设置输出结构为 JSON模式，从而获取结构化数据。
其中后两种输出模式会更稳定一些，在 LangChain 中为后两种方法封装了 .with_structured_output() 方法，这也是获取结构化输出的最简单和最可靠的方法，在 .with_structured_output() 的底层会使用 LLM 原生的 函数调用 或 JSON模式。
该方法接受一个 BaseModel子类 作为输入，该子类需要指定输出属性的名称、描述和类型，该方法返回的是一个 Runnable 可运行对象，但并不是输出字符串或者 AI 消息，而是输出与给定模式对应的对象，其中模式可以指定为 JSON 模式（返回一个字典）或 Pydantic 类（返回一个 Pydantic 对象）。
另外检测一个 LLM 是否支持 .with_structured_output() 可以通过查看源码或者在 LangChain 的大语言模型 高级功能 列表中查看，链接：https://imooc-langchain.shortvar.com/docs/integrations/chat/
使用这节课所学习的知识来完成一个 QA问答数据 的提取，代码示例如下:
import dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAIdotenv.load_dotenv()class QAExtra(BaseModel):"""一个问答键值对工具，传递对应的假设性问题+答案"""question: str = Field(description="假设性问题")answer: str = Field(description="假设性问题对应的答案")llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
structured_llm = llm.with_structured_output(QAExtra)prompt = ChatPromptTemplate.from_messages([("system", "请从用户传递的query中提取出假设性的问题+答案。"),("human", "{query}")
])chain = {"query": RunnablePassthrough()} | prompt | structured_llmprint(chain.invoke("我叫慕小课，我喜欢打篮球，游泳"))
输出内容：
question='你喜欢打篮球吗？' answer='是的，我喜欢打篮球。'
在 .with_structured_output() 的底层会优先使用 函数调用 模式，如果 LLM 支持 JSON模式，还可以在函数内传递多一个参数 method="json_mode"，即使用大语言模型的 JSON输出模式 来操作，修改代码如下：
import dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAIdotenv.load_dotenv()class QAExtra(BaseModel):"""一个问答键值对工具，传递对应的假设性问题+答案"""question: str = Field(description="假设性问题")answer: str = Field(description="假设性问题对应的答案")llm = ChatOpenAI(model="gpt-4o")
structured_llm = llm.with_structured_output(QAExtra, method="json_mode")prompt = ChatPromptTemplate.from_messages([("system", "请从用户传递的query中提取出假设性的问题+答案。响应格式为JSON，并携带`question`和`answer`两个字段。"),("human", "{query}")
])chain = {"query": RunnablePassthrough()} | prompt | structured_llmprint(chain.invoke("我叫慕小课，我喜欢打篮球，游泳。"))
输出内容：
question='你叫什么名字？你喜欢做什么？' answer='我叫慕小课，我喜欢打篮球，游泳。'
在上面的代码中修改了几个部分：
- gpt-4o模型：在 OpenAI 提供的模型中，并不是所有模型都支持 JSON 模式的，需要查看文档确认。
- prompt提示模板：在 JSON 模式中，需要在 Prompt 中告知 LLM 输出 JSON 的结构信息，要不然会报错。
不过因为支持 JSON 模式的大语言模型较少，所以使用起来限制也比较多，还会对原始的 Prompt 产生干扰，所以尽可能使用 函数调用 的形式来结构化数据，会更稳定。
[!IMPORTANT]
在创建 BaseModel 子类时，类的名称、文档字符串以及参数的名称和提供的描述也非常重要，因为在该函数的底层大部分使用的都是函数调用，只有添加上这些信息后，才可以将对应的 LLM 绑定函数调用，获取正确的结果。
3.2 .with_structured_output() 源码解析
在 .with_structured_output() 底层，会通过传递的 method 来执行不同的 运行时参数绑定，例如使用 函数调用 亦或者是 JSON模式 输出，核心代码:
def with_structured_output(self,schema: Optional[_DictOrPydanticClass] = None,*,method: Literal["function_calling", "json_mode"] = "function_calling",include_raw: bool = False,**kwargs: Any,
) -> Runnable[LanguageModelInput, _DictOrPydantic]:if kwargs:raise ValueError(f"Received unsupported arguments {kwargs}")is_pydantic_schema = _is_pydantic_class(schema)if method == "function_calling":if schema is None:raise ValueError("schema must be specified when method is 'function_calling'. ""Received None.")llm = self.bind_tools([schema], tool_choice="any")if is_pydantic_schema:output_parser: OutputParserLike = PydanticToolsParser(tools=[schema], first_tool_only=True)else:key_name = convert_to_openai_tool(schema)["function"]["name"]output_parser = JsonOutputKeyToolsParser(key_name=key_name, first_tool_only=True)elif method == "json_mode":llm = self.bind(response_format={"type": "json_object"})output_parser = (PydanticOutputParser(pydantic_object=schema)if is_pydantic_schemaelse JsonOutputParser())else:raise ValueError(f"Unrecognized method argument. Expected one of 'function_calling' or "f"'json_mode'. Received: '{method}'")if include_raw:parser_assign = RunnablePassthrough.assign(parsed=itemgetter("raw") | output_parser, parsing_error=lambda _: None)parser_none = RunnablePassthrough.assign(parsed=lambda _: None)parser_with_fallback = parser_assign.with_fallbacks([parser_none], exception_key="parsing_error")return RunnableMap(raw=llm) | parser_with_fallbackelse:return llm | output_parser
所以哪怕不使用 LangChain，该思路也非常值得借鉴，在需要使用 LLM 获取结构化数据时，只需要构建一个 虚假的函数，让 LLM 绑定该函数，并且设置 tool_choice="any" 即强制调用所有函数，这样就可以在最大程度上确保 LLM 能稳定地输出结构化数据
4. 函数调用出错捕获提升程序健壮性
4.1 函数调用出错处理提升程序健壮性
在执行函数调用的过程中，错误几乎是不可避免的，无论是大语言模型错误地传递了参数，还是工具内部抛出的错误，如果错误不做任何处理操作，都会让程序变得异常脆弱，所以可以考虑在链中构建错误处理/捕获来减轻这些故障出现的概率。
try/except 捕获工具错误
首先我们先来构建一个 复杂工具，并让 LLM 尝试调用这个工具，并故意引发一些错误，示例代码如下:
import dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAIdotenv.load_dotenv()@tool
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:"""使用复杂工具进行复杂计算操作"""return int_arg * float_arg# 1.创建大语言模型并绑定工具
llm = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0)
llm_with_tools = llm.bind_tools([complex_tool])# 2.创建链并执行工具
chain = llm_with_tools | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool# 3.调用链
print(chain.invoke("使用复杂工具，对应参数为5和2.1"))
输出内容如下
Traceback (most recent call last):File "D:\Project\llmops\llmops-api\study\41-工具调用\1.错误捕获.py", line 29, in <module>print(chain.invoke("使用复杂工具，对应参数为5和2.1"))File "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\runnables\base.py", line 2875, in invokeinput = step.invoke(input, config)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\tools.py", line 427, in invokereturn self.run(tool_input, **kwargs)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\tools.py", line 615, in runraise error_to_raiseFile "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\tools.py", line 578, in runtool_args, tool_kwargs = self._to_args_and_kwargs(tool_input)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\tools.py", line 501, in _to_args_and_kwargstool_input = self._parse_input(tool_input)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\langchain_core\tools.py", line 454, in _parse_inputresult = input_args.parse_obj(tool_input)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\pydantic\v1\main.py", line 526, in parse_objreturn cls(**obj)File "D:\Project\llmops\llmops-api\venv\lib\site-packages\pydantic\v1\main.py", line 341, in __init__raise validation_error
pydantic.v1.error_wrappers.ValidationError: 1 validation error for complex_toolSchema
dict_argfield required (type=value_error.missing)
在上面的示例中，因为工具的描述并不清晰，大语言模型只针对前 2 个参数生成了对应的数值，第 3 个参数并没有生成，所以引发了 pydantic 数据校验错误，对于这类错误，通用的处理方法是在工具调用步骤上使用 try/except 进行错误的捕获，并在出错时返回游泳的提示信息，修正代码部分如下
def try_except_tool(tool_args: dict, config: RunnableConfig) -> Any:try:return complex_tool.invoke(tool_args, config)except Exception as e:return f"调用工具时使用以下参数:\n\n{tool_args}\n\n引发了以下错误:\n\n{type(e)}: {e}"chain = llm_with_tools | (lambda msg: msg.tool_calls[0]["args"]) | try_except_tool
输出内容
调用工具时使用以下参数:
{'int_arg': 5, 'float_arg': 2.1}
引发了以下错误:
<class 'pydantic.v1.error_wrappers.ValidationError'>: 1 validation error for complex_toolSchema
dict_argfield required (type=value_error.missing)
在这种模式下，如果将 工具返回的错误消息 重新返回给 LLM 时，LLM 会重新判断并继续执行 函数调用，从而将参数补全，让程序正常执行，也是最推荐的一种错误处理方案，即在 工具内部 进行错误的捕获，并将错误信息独立返回。
4.2 回退与重试处理
如果 Runnable 可运行组件出错，可以执行 回退 和 重试 两种策略，在函数调用中也可以使用这两种策略来处理，例如：
1. 在函数调用参数生成错误时，可以考虑回退到一个更好的模型，例如平时使用 gpt-3.5-turbo-16k 模型，在出错时，回退到 gpt-4o 模型上进行重新试验。
2. 亦或者是触发重试机制，并且在重试的时候，携带上错误信息，让 LLM 强大的自然语言处理功能进行自我纠正。
使用更好的模型进行回退，操作起来非常简单，只需要定义一个更强大的模型，并创建一条新链，然后使用 Runnable可运行组件 的 .with_fallbacks() 即可绑定对应的回退策略，示例代码如下:
import dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAIdotenv.load_dotenv()@tool
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:"""使用复杂工具进行复杂计算操作"""return int_arg * float_arg# 1.创建大语言模型并绑定工具
llm = ChatOpenAI(model="gpt-3.5-turbo-16k").bind_tools([complex_tool])
better_llm = ChatOpenAI(model="gpt-4o").bind_tools([complex_tool], tool_choice="complex_tool",
)# 2.创建链并执行工具
better_chain = better_llm | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool
chain = (llm | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool).with_fallbacks([better_chain])# 3.调用链
print(chain.invoke("使用复杂工具，对应参数为5和2.1"))
回退策略 并不一定是万能的，有的时候哪怕使用了参数更大的模型，生成的 函数调用 参数依旧不符合规范，这种情况就要考虑下工具的相关描述、Prompt 编写得是否存在问题。另外一种优化策略是 携带错误信息的重试策略，让 LLM 纠正其行为，只需要在抛出错误时，将错误信息一起携带给 LLM，让其重新操作即可，示例代码如下:
from typing import Anyimport dotenv
from langchain_core.messages import ToolCall, AIMessage, ToolMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langchain_openai import ChatOpenAIdotenv.load_dotenv()class CustomToolException(Exception):"""自定义的工具错误异常"""def __init__(self, tool_call: ToolCall, exception: Exception) -> None:super().__init__()self.tool_call = tool_callself.exception = exception@tool
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:"""使用复杂工具进行复杂计算操作"""return int_arg * float_argdef tool_custom_exception(msg: AIMessage, config: RunnableConfig) -> Any:try:return complex_tool.invoke(msg.tool_calls[0]["args"], config)except Exception as e:raise CustomToolException(msg.tool_calls[0], e)def exception_to_messages(inputs: dict) -> dict:# 1.从输入中提取错误信息exception = inputs.pop("exception")# 2.将历史消息添加到原始输入中，以便模型直到它在上一次工具调用中犯了一个错误messages = [AIMessage(content="", tool_calls=[exception.tool_call]),ToolMessage(tool_call_id=exception.tool_call["id"], content=str(exception.exception)),HumanMessage(content="最后一次工具调用引发了异常，请尝试使用更正的参数再次调用该工具，不要重复犯错。")]inputs["last_output"] = messagesreturn inputs# 1.创建prompt，并预留占位符，用于存储错误输出信息
prompt = ChatPromptTemplate.from_messages([("human", "{query}"),("placeholder", "{last_output}"),
])# 2.创建大语言模型并绑定工具
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools=[complex_tool])# 3.创建链并执行工具
chain = prompt | llm | tool_custom_exception
self_correcting_chain = chain.with_fallbacks([exception_to_messages | chain], exception_key="exception",
)# 4.调用自我纠正链完成任务
print(self_correcting_chain.invoke({"query": "使用复杂工具，对应参数为5和2.1"}))
输出内容
5. 多模态LLM执行函数调用的技巧
目前市面上的 LLM 除了能接收文本数据，还出现了不少多模态 LLM，这些 LLM 不仅能接收文本，还能接收 图像、音频 等内容，并且这类 多模态LLM 也支持 函数调用 功能，要想使用这些模型调用工具，只需要按照正常的方式将工具绑定到模型上，在传递 消息 给大语言模型时，按照 多模态LLM 的特定规则即可。
5.1 GPT-4o 多模态输入
OpenAI 提供的 GPT-4o 模型是一个多模态的模型（输入多模态），在传递 消息列表 时，可以在 人类消息 中添加 图片地址，这样即可将特定的图片上传给 GPT-4o 模型进行识别，从而实现多模态输入。
• GPT-4o 多模态输入文档链接：https://platform.openai.com/docs/api-reference/chat/create
官方提供原生示例如下
from openai import OpenAIclient = OpenAI()response = client.chat.completions.create(model="gpt-4o",messages=[{"role": "user","content": [{"type": "text", "text": "What's in this image?"},{"type": "image_url","image_url": {"url": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"},},],}],max_tokens=300,
)print(response.choices[0])
在 LangChain 中，如果消息也是多模态的，只需要按照特定的模型对应的 消息结构 历史创建 LangChain消息 即可，例如在上面的示例中，人类消息 传递了一个列表，包含了 文本 和 图片链接，转换成 LangChain消息/提示模板 如下
prompt = ChatPromptTemplate.from_messages([("human", [{"type": "text", "text": "{query}"},{"type": "image_url", "image_url": {"url": "{image_url}"}}])
])print(prompt.invoke({"query": "这张图片所在的地址的天气怎么样","image_url": "https://img1.baidu.com/it/u=644490943,1781886584&fm=253&fmt=auto&app=138&f=JPEG"
}).to_messages())
使用技巧和普通的提示模板没有差异，只是将原本的 字符串 替换成了 列表字典 的格式，输出内容
[HumanMessage(content=[{'type': 'text', 'text': '这张图片所在的地址的天气怎么样'}, {'type': 'image_url', 'image_url': {'url': 'https://img1.baidu.com/it/u=644490943,1781886584&fm=253&fmt=auto&app=138&f=JPEG'}}])]
这里我们以 GPT-4O + 天气预报查询工具 + 多模态输入 三个组件构建一个 LLM 应用，让该 LLM 应用能识别上传图片城市所在的天气信息，运行流程如下
import json
import os
from typing import Type, Anyimport dotenv
import requests
from langchain_core.output_parsers import StrOutputParser
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 ChatOpenAIdotenv.load_dotenv()class GaodeWeatherArgsSchema(BaseModel):city: str = Field(description="需要查询天气预报的目标城市，例如：广州")class GaodeWeatherTool(BaseTool):"""根据传入的城市名查询天气"""name = "gaode_weather"description = "当你想询问天气或与天气相关的问题时的工具。"args_schema: Type[BaseModel] = GaodeWeatherArgsSchemadef _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')}天气预报信息失败"# 4.整合天气预报信息并返回except Exception as e:return f"获取{kwargs.get('city')}天气预报信息失败"# 1.构建prompt
prompt = ChatPromptTemplate.from_messages([("human", [{"type": "text", "text": "请获取下上传图片对应城市的天气信息。"},{"type": "image_url", "image_url": {"url": "{image_url}"}}]),
])
weather_prompt = ChatPromptTemplate.from_template("""请整理下传递的城市的天气预报信息，并以用户友好的方式输出。<weather>
{weather}
</weather>""")# 2.构建LLM并绑定工具
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools=[GaodeWeatherTool()], tool_choice="gaode_weather")# 3.创建链应用并执行
chain = ({"weather": ({"image_url": RunnablePassthrough()}| prompt| llm_with_tools| (lambda msg: msg.tool_calls[0]["args"])| GaodeWeatherTool())} | weather_prompt | llm | StrOutputParser()
)print(chain.invoke("https://imooc-langchain.shortvar.com/guangzhou.jpg"))
输出内容：
以下是广州未来几天的天气预报信息：
### 广州市天气预报**发布时间：2024年7月11日 12:00**#### 2024年7月11日 (星期四)
- **白天天气：** 大雨
- **夜间天气：** 大雨
- **白天温度：** 33°C
- **夜间温度：** 25°C
- **白天风向：** 西南
- **夜间风向：** 西南
- **风力等级：** 1-3级#### 2024年7月12日 (星期五)
- **白天天气：** 大雨
- **夜间天气：** 中雨-大雨
- **白天温度：** 33°C
- **夜间温度：** 24°C
- **白天风向：** 北
- **夜间风向：** 北
- **风力等级：** 1-3级#### 2024年7月13日 (星期六)
- **白天天气：** 中雨-大雨
- **夜间天气：** 中雨-大雨
- **白天温度：** 32°C
- **夜间温度：** 25°C
- **白天风向：** 西南
- **夜间风向：** 西南
- **风力等级：** 1-3级#### 2024年7月14日 (星期日)
- **白天天气：** 中雨-大雨
- **夜间天气：** 大雨
- **白天温度：** 33°C
- **夜间温度：** 25°C
- **白天风向：** 北
- **夜间风向：** 北
- **风力等级：** 1-3级
请注意天气变化，做好防雨准备。
5.2 GPT-4o 接入 DALL·E 文生图
除了输入的多模态，大语言模型的输出其实也可以通过曲线救国的方式实现 多模态输出，有使用过 ChatGPT-Plus 的小伙伴肯定了解，当我们和 ChatGPT 对话时，可以让 ChatGPT 帮忙生成对应的图片，该功能底层本质上也是通过 函数调用 来实现的，运行原理非常简单：
1. 创建一个根据 文本 生成 图片 工具，例如可以使用 DALLE，工具的调用参数为生成图片的 Prompt。
2. 将工具绑定到 LLM 上，并预设特定的 Prompt，让 LLM 将用户输入的 query 转换成绘图的 Prompt。
3. 调用工具，将 Prompt 生成图片，并将 工具消息、AI消息 叠加到历史消息中，再次提问，获得最终答复。
运行流程和刚刚我们创建的 图片获取天气预报应用 非常接近，如下
实现代码如下：
import dotenv
from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool
from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper
from langchain_openai import ChatOpenAIdotenv.load_dotenv()dalle = OpenAIDALLEImageGenerationTool(name="openai_dalle",api_wrapper=DallEAPIWrapper(model="dall-e-3")
)
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools=[dalle], tool_choice="openai_dalle")chain = llm_with_tools | (lambda msg: msg.tool_calls[0]["args"]) | dalleprint(chain.invoke("帮我绘制一幅老爷爷的图片"))