LangGraph--基础学习（工具调用）
本节将详细学习大模型是怎么调用工具的，为什么可以调用工具等等，手写一个工具调用，后续可以通过mcp自己调用即可，没必要自己写，但是学习过程中需要手写，通常怎么使用第三方工具调用呢？
import os
from langchain_community.tools.tavily_search import TavilySearchResults def _set_env(var: str, value): if not os.environ.get(var): os.environ[var] = value _set_env("TAVILY_API_KEY", "tvly-xxxx") tool = TavilySearchResults(max_results=4)
tool.invoke("苏超谁赢了？")[{'title': '美团、京东、饿了么组团借势苏超，这把谁赢了？','url': 'https://www.digitaling.com/articles/1369073.html','content': '\n\n# 美团、京东、饿了么组团借势苏超，这把谁赢了？\n\n\n\n\n\n扫描,分享朋友圈\n\n#### **美团、**京东、**饿了么，** **这相爱（**或许爱得不多？**）相杀的三巨头，又！又！又！把火烧到了苏超的绿茵场。**\n\n就是那个真·草根真·热血、球迷互喷、“没有假球，只有世仇”的江苏省城市足球联赛！\n\n从电商到外卖，从补贴大战到谐音梗对轰，这三家如今在球场上短兵相接，狭路相逢，商业宿命感直接拉满。 [...] ****\n\n\n\n**宿迁 \n赢了干杯，输了也干杯！ \n霸王敬酒，洋河大曲喝到美。**\n\n\n\n\n\n**苏州 \n江苏的苏是苏州的苏。 \n大闸蟹的蟹，是阳澄湖的蟹。** [...] 打开数英App \n进入【我的】右上角扫一扫登录\n\n\n\n其他登录\n\n选择收藏夹\n\n\n\n上传封面\n\n建议封面尺寸为400x400像素，\n\n仅支持JPG/PNG静态图片\n\n**资料更新成功！**\n\n[查看人才库](https://www.digitaling.com/jobs/talent)[取消](javascript:void(0);)\n\n![]()\n','score': 0.7625837},{'title': '面对面|敬一丹：“苏超”不管谁赢江苏都赢了 - 扬子晚报','url': 'https://www.yangtse.com/news/bzxc/202506/t20250619_224004.html','content': '... 苏超”，敬一丹表示：“江苏的苏超太引人注目，太欢乐了，我觉得这是一个特别值得琢磨的事，现在不管是哪个城市赢，我觉得江苏都赢了。” 扬子晚报|紫牛','score': 0.7395922}]大模型如果只依赖自身的知识，那么很难流行起来，也不会解决太多的问题，因为大模型自身的训练数据是有限的，像 agent 系统之所以能够流行，主要依赖大模型的决策能力，以前我们如果写程序，需要程序员按照产品经理的产品设计，设置好代码的运行逻辑依次的按照产品逻辑进行执行，比如用户询问北京的天气，那么会通过一定的代码比如关键词匹配，正则表达式或者NLP相关的工具来判断出用户的意图，还有分析出想要查询的地区是北京，然后调用相应的工具获取数据，过程中会有很大的出错概率，如参数分析的不准确，工具调用的不正确等，现在有了大模型，我们可以利用大模型的意图分析能力准确的得到需要调用哪个工具以及调用参数。
LangGraph 中可以利用LangChain 中的tool call 功能来实现工具的调用，本文先介绍一下在LangChain 中如何使用 tool call， 然后再使用LangGraph 工作流中调用工具。
上面是工具自身的调用，并不需要大模型，就像是普通的 python 函数调用，只是这里的函数需要按照一定的规则进行定义。然后使用工具的invoke 方法来真正的调用工具。
二、自定义工具¶
由于langchain_community 中的工具很多我们是用不了的，所以这里我们来封装一个自定义工具，使用高德的天气查询接口来做一个天气查询的工具。高德的天气查询接口文档请查阅 基础 API 文档-开发指南-Web服务 API | 高德地图API
2.1 工具类的定义¶
参考 TavilySearchResults 工具的实现以及 LangChain 官方的 tools 文档 How to create tools | 🦜️🔗 LangChain , 我这里使用继承自BaseTool 的方式来实现。
主要的流程包括
- 继承
langchain_core.tools.BaseTool类 - 定义几个属性，
- name: 工具的名字。
- description: 工具的描述，这里的描述很重要，越详细越准确越好，后面的大模型是需要根据这里的描述来判断要调用哪个工具。
- args_schema: 参数的定义，
pydantic.BaseModel类型，这里也需要对函数调用的参数进行详细的描述，越准确越好。 - 除了上面几个属性，如果工具还需要自己单独的属性，也需要在类里定义，如这个天气api 需要的api_key。
- 定义
_run方法，实现真正的业务调用逻辑， 如果需要使用异步调用，则需要再定义_arun方法。
根据上面的规则，定义如下的查询天气 tool。
import json
from typing import Type
import requests
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field class GaoDeWeatherInput(BaseModel): city: str = Field(description="要查询天气的城市名称") class GaoDeWeather(BaseTool): """ 高德天气查询 """ name: str = "高德天气查询" description: str = "高德天气查询，输入城市名，返回该城市的天气情况，例如：北京" args_schema: Type[BaseModel] = GaoDeWeatherInput api_key: strdef _run(self, city): s = requests.session() api_domain = 'https://restapi.amap.com/v3' url = f"{api_domain}/config/district?keywords={city}"f"&subdistrict=0&extensions=base&key={self.api_key}" headers = {"Content-Type": "application/json; charset=utf-8"} city_response = s.request(method='GET', headers=headers, url=url) City_data = city_response.json() if city_response.status_code == 200 and City_data.get('info') == 'OK': if len(City_data.get('districts')) > 0: CityCode = City_data['districts'][0]['adcode'] weather_url = f"{api_domain}/weather/weatherInfo?city={CityCode}&extensions=all&key={self.api_key}" weatherInfo_response = s.request(method='GET', url=weather_url) weatherInfo_data = weatherInfo_response.json() if weatherInfo_response.status_code == 200 and weatherInfo_data.get('info') == 'OK': contents = [] if len(weatherInfo_data.get('forecasts')) > 0: for item in weatherInfo_data['forecasts'][0]['casts']: content = {} content['date'] = item.get('date') content['week'] = item.get('week') content['dayweather'] = item.get('dayweather') content['daytemp_float'] = item.get('daytemp_float') content['daywind'] = item.get('daywind') content['nightweather'] = item.get('nightweather') content['nighttemp_float'] = item.get('nighttemp_float') contents.append(content) s.close() return contents else: return "没有查询到该城市的天气信息" weather_tool = GaoDeWeather(api_key='xxxx') result = weather_tool.invoke("安徽")
print(json.dumps(result, ensure_ascii=False, indent=4))[{"date": "2025-06-21","week": "6","dayweather": "大雨","daytemp_float": "23.0","daywind": "东南","nightweather": "小雨","nighttemp_float": "18.0"},{"date": "2025-06-22","week": "7","dayweather": "多云","daytemp_float": "25.0","daywind": "东南","nightweather": "多云","nighttemp_float": "18.0"},{"date": "2025-06-23","week": "1","dayweather": "多云","daytemp_float": "26.0","daywind": "东","nightweather": "阴","nighttemp_float": "19.0"},{"date": "2025-06-24","week": "2","dayweather": "晴","daytemp_float": "28.0","daywind": "东","nightweather": "阴","nighttemp_float": "20.0"}
]由于高德的天气接口需要传入城市的cityCode, 并不是城市名称，所以先要获取一下城市的citycode，再根据 citycode 获取气象数据。
调用上面自定义的天气查询接口得到如上输出
该形式可以针对任何你想调用的api，其实就是研究官方的web服务接口，解析数据就行了
上面的工具可以很好的运行，也成功获取到了天气数据，当我们在调用工具的时候，需要输入城市名的，但是用户在交互过程中，很难知道用户是怎么问的，如用户可能问
- 北京明天的天气如何？
- 上海后天会下雨吗？
- 我在浙江，晚上出门需要带雨伞吗？
当遇到这些问题时，传统的业务流程很难做到准确的分析调用，这时我们就需要使用大模型来分析规划调用哪个工具，以及调用工具时的参数。
接下来我们看一下如何使用大模型来调用。
weather_tool = GaoDeWeather(api_key='xxxx')
tools = [weather_tool]
# 定义一个工具名称和工具实例的字典，用于之后的工具调用
tools_by_name = {tool.name: tool for tool in tools}
# 将工具与大模型进行绑定
llm_with_tools = llm.bind_tools(tools) # 还没有引入langgraph，所以需要人工调用工具，后面使用引入langgraph加入节点和边就可以直接调用了
messages = [HumanMessage(content="安徽明天的天气如何?")]
result = llm_with_tools.invoke(messages)
messages.append(result)
outputs = []
if result.tool_calls: for tool_call in result.tool_calls: tool_name = tool_call.get("name") if tool_name in tools_by_name: tool = tools_by_name[tool_name] tool_result = tool.invoke(tool_call.get("args")) tool_message = ToolMessage( content=json.dumps(tool_result, ensure_ascii=False), name=tool_call["name"], tool_call_id=tool_call["id"], ) messages.append(tool_message) outputs.append(tool_result) final_result = llm_with_tools.invoke(messages) print(final_result.content) outputs.append(final_result)city = 安徽
[ToolMessage(content='[{"date": "2025-06-21", "week": "6", "dayweather": "大雨", "daytemp_float": "23.0", "daywind": "东南", "nightweather": "小雨", "nighttemp_float": "18.0"}, {"date": "2025-06-22", "week": "7", "dayweather": "多云", "daytemp_float": "25.0", "daywind": "东南", "nightweather": "多云", "nighttemp_float": "18.0"}, {"date": "2025-06-23", "week": "1", "dayweather": "多云", "daytemp_float": "26.0", "daywind": "东", "nightweather": "阴", "nighttemp_float": "19.0"}, {"date": "2025-06-24", "week": "2", "dayweather": "晴", "daytemp_float": "28.0", "daywind": "东", "nightweather": "阴", "nighttemp_float": "20.0"}]', name='get_current_weather', tool_call_id='call_0_a9c4ef6b-7548-4dc2-927b-9dce4e258029')]上面的代码当第一次使用 result = llm_with_tools.invoke("北京明天的天气如何?") 调用大模型时，此时大模型如果判断出要使用某个工具时，那么会返回如下的数据
注意到返回的数据中有个 tool_calls 字段，这是一个列表，因为有时候大模型可能分析出要调用多个工具，但是大多数情况下只有一个工具，并将调用这个工具的参数放到 args 里返回了。如上面的调用，告诉客户端，需要调用一个叫 高德天气查询 的工具，参数为 {"city": "安徽"} , 当客户端拿到大模型的分析输出时，就会使用 tool.invoke(tool_call.get("args")) 来真正的进行工具调用。当得到工具的输出以后，将结果保存在 outputs 列表中，此时如果我们直接将工具的输出返回给用户，只是一堆json 数据，并没有真正回答用户的问题，所以这里还需要再次调用大模型，将用户的问题(北京明天的天气如何?)和工具的输出再次传给大模型，让大模型进行总结回答。
这里需要将之前的用户消息和ai返回的消息以及调用工具返回的消息放到一个列表里，最后再统一调用一次大模型。
上面的代码后面的if就是为了拿到数据，调用工具，拿到内容后再调用大模型，让其总结返回结果：
- 初始化一个messages 来存放消息
result = llm_with_tools.invoke(messages)为第一次调用大模型，用于让大模型分析要调用哪个工具以及调用工具时的参数。tool_result = tool.invoke(tool_call.get("args"))为真正的调用工具，得到工具的输出。tool_message = ToolMessage(content=json.dumps(tool_result, ensure_ascii=False))为定义tool_message， 之后使用messages.append(tool_message)将工具的输出也放到消息列表里final_result = llm_with_tools.invoke(messages)为最终调用大模型，根据用户的提问以及大模型的输出做最终的解答
得到的输出为：
city = 安徽
安徽明天的天气情况如下：- **白天天气**: 多云
- **白天温度**: 25°C
- **白天风向**: 东南风
- **夜间天气**: 多云
- **夜间温度**: 18°C请注意天气变化，合理安排出行！四、LangGraph 调用工具¶
上面铺垫了这么多，终于进入了主题，如何在LangGraph 中调用工具？ 我们来回顾总结一下上面langchain 进行工具调用的流程
- 定义工具，写清楚工具的描述和参数的描述信息
- 初始化大模型，并将工具绑定到大模型中
- 拿到用户的提问，进行第一次的大模型调用，得到大模型分析出来的需要调用哪个工具以及对应的参数
- 根据大模型返回的工具信息真正的执行工具调用
- 将之前的用户问题，AI的回答以及工具的输出再次发给大模型，得到最终的输出结果
这里有一些注意的地方，首先，大模型并不是每次都能返回工具调用的参数，如果用户的问题和工具一毛钱关系都没有，则大模型也是不会返回工具调用信息的，如用户问 "iphone 16 多少钱"， 这个问题很明显和这个天气查询工具一点关系都没有，则大模型会直接回答问题。其次，大模型有时会返回多个工具调用，需要依次进行工具调用，虽然多数情况下只会返回一个工具调用。
清楚了LangChain 的工具调用，我们使用 LangGraph 来实现一下 。
- 初始化一个带有工具调用的大模型client
- 添加一个大模型调用节点(node)，用于和大模型进行交互
- 当需要进行工具调用时，进行工具调用，此时需要一个条件路由函数
- 如果需要调用工具，则将工具调用的结果和之前的用户消息大模型消息再次请求一次大模型，得到的结果再次进行是否需要工具调用判断，也就是每次进行大模型调用都需要判断一下是否还需要工具调用
- 如果不需要进行工具调用，则直接将大模型生成的恢复返回给用户，结束工作流
- 整个graph 过程中的消息放到State 中进行节点之间的共享
得到的流程如下
参考这张图，chatbot 节点有两条虚线分别连接 tools 节点和 END 节点，虚线表示的是条件边，对应代码中的 add_conditional_edges，后面可能会走哪条分支，实线是确定的走向，对应代码为add_edge, 如 tools 节点之后确定要走chatbot 节点。
这里有一个很巧妙的设计，上面说过，工具调用结束以后，需要再次调用大模型对问题进行总结回答，这里的大模型调用没有在工具函数中定义，而是通过 LangGraph 的普通边来实现。
上面的代码定义了两个运行节点，一个是chatbot, 使用 chatbot 函数定义，另外一个是使用BasicToolNode 类定义的类节点，如果不明白可以查看前面的文章。
import json
from typing import Type, TypedDict, Annotated
import requests
from langchain_core.messages import ToolMessage, HumanMessage, AIMessage
from langchain_core.tools import BaseTool
from langgraph.graph import add_messages, StateGraph, START, END
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI class State(TypedDict): messages: Annotated[list, add_messages] # 上面的代码都创建了，我这边注释了
# class GaoDeWeatherInput(BaseModel):
# city: str = Field(description="要查询天气的城市名称") # class GaoDeWeather(BaseTool):
# """
# 高德天气查询,和上面的代码一样
# """
# ...weather_tool = GaoDeWeather(api_key='XX') # llm = ChatOpenAI(
# model_name="qwen-turbo",
# temperature=0.7,
# max_tokens=1024,
# top_p=1,
# openai_api_key="sk-xxxx",
# openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
# ) tools = [weather_tool]
llm_with_tools = llm.bind_tools(tools) # 大模型执行节点
def chatbot(state: State): return {"messages": [llm_with_tools.invoke(state["messages"])]} # 工具执行节点
class BasicToolNode: """A node that runs the tools requested in the last AIMessage.""" # 这里是我们自定定义的适配langgraph的工具节点，后续可以直接使用langgraph的节点对象即可def __init__(self, tools: list) -> None: self.tools_by_name = {tool.name: tool for tool in tools} # 拿出工具名称def __call__(self, inputs: State): messages = inputs.get("messages", []) # 拿出数据if messages: message = messages[-1] # 拿出最新的数据else: raise ValueError("No message found in input") outputs = [] for tool_call in message.tool_calls: tool_result = self.tools_by_name[tool_call["name"]].invoke( tool_call["args"] ) # 把工具函数和参数绑定进行执行，得到结果json化，然后存储outputs.append( ToolMessage( content=json.dumps(tool_result), name=tool_call["name"], tool_call_id=tool_call["id"], ) ) return {"messages": outputs} # 定义一个条件节点
def condition_tools(state: State): ai_message = state["messages"][-1] if ai_message.tool_calls: return "tools" return END # 构建 graph
graph_builder = StateGraph(State) # 添加节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", BasicToolNode(tools)) # 添加边
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", condition_tools,{"tools": "tools", # 条件返回的字符串 → 目标节点END: END,})
graph_builder.add_edge("tools", "chatbot") app = graph_builder.compile() app.get_graph().draw_mermaid_png(output_file_path="graph_tool.png") inputs = {"messages": [HumanMessage(content="安徽合肥后天的天气")]} result = app.invoke(inputs)
print(result.get("messages")[-1].content)city = 合肥
安徽合肥后天的天气情况如下：- **日期**: 2025-06-23（星期一）
- **白天天气**: 多云
- **白天温度**: 26°C
- **白天风向**: 东风
- **夜间天气**: 阴
- **夜间温度**: 19°C请注意天气变化，合理安排出行！总结：
学习是个循序渐进的过程，不要指望一下子就学会了，多写代码，多思考，虽然现在可以直接让大模型写就行了，但是前提你需要理解这个，否则后面有问题都不知道咋回事