LangGraph工具调用基础

LLM 在处理实时性问题、数学计算等方面往往表现不佳。为了弥补这些不足,一个合格的 Agent 应当具备调用外部工具的能力。

本文将从最常见的 Web 搜索工具 入手,带你快速上手 LangGraph 的工具调用。这里我们使用 TavilySearch(https://app.tavily.com) 作为示例。

在上一篇构建聊天机器人的基础上,只需少量新增代码,即可完成工具调用。LangGraph 已经帮我们封装了大部分复杂逻辑。

示例代码

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from pydantic import SecretStr
from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")

# 定义工具
tool = TavilySearch(tavily_api_key="你的tavily apikey", max_results=2)
tools = [tool]
# 工具绑定到模型
llm_with_tools = llm.bind_tools(tools)

# 定义图状态
classState(TypedDict):
    messages: Annotated[list, add_messages]  # 此处维护完整的消息历史

graph = StateGraph(State)

defchatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
# 使用LangGraph提供的工具节点
tool_node = ToolNode(tools=tools)

graph.add_node("chatbot", chatbot)
# 添加工具节点
graph.add_node("tools", tool_node)
# 添加工具 条件分支
graph.add_conditional_edges(
"chatbot",
    tools_condition,
)
graph.add_edge("tools", "chatbot")
graph.add_edge(START, "chatbot")

app = graph.compile()

if __name__ == "__main__":
    messages = []
whileTrue:
        user_input = input("👨‍💻: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Exiting...")
break
        messages.append({"role": "user", "content": user_input})
        response = app.invoke({"messages": messages})
        messages = response["messages"]
print(f'🤖: {response["messages"][-1].content}')

效果如下

LangGraph自定义工具与工具节点

LangChain为我们预构建了很多工具,常用的有

  • 搜索类: Bing, SerpAPI, Tavily
  • 代码执行类: Python REPL, Node.js REPL
  • 数据库类: SQL, MongoDB, Redis
  • Web 数据类: Scraping and browsing
  • 其他 API: OpenWeatherMap, NewsAPI 等

不过,在真实的企业开发场景中,更常见的做法是开发 自定义工具,以满足个性化需求。

下面演示一个基于 BaseTool 的自定义 Tavily 搜索工具,以及如何在 LangGraph 中接入。

示例代码

import json
from langchain_core.messages import ToolMessage
from langgraph.constants import END
from typing import Annotated, Type, Optional
from langchain_core.callbacks import CallbackManagerForToolRun, AsyncCallbackManagerForToolRun

from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from pydantic import SecretStr, BaseModel, Field
from langchain_core.tools import BaseTool
from tavily import TavilyClient, AsyncTavilyClient

# 自定义工具部分
classTavilySearchInput(BaseModel):
    query: str = Field(description=("搜索查询"))

classTavilySearchTool(BaseTool):
    name: str = "tavily_search"
    description: str = """一个针对全面、准确和可信的结果进行了优化的搜索引擎。
当需要回答有关时事的问题时很有用。
输入应该是搜索查询。"""
    args_schema: Type[BaseModel] = TavilySearchInput
# return_direct: bool = True

def_run(
        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
) -> int:
        client = TavilyClient()
        search_r = client.search(query=query, max_results=2)
return search_r

asyncdef_arun(
        self,
        query: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
) -> int:
        client = AsyncTavilyClient()
        search_r = await client.search(query=query, max_results=2)
return search_r

# 自定义的工具执行节点
classToolNode:
def__init__(self, tools: list) -> None:
self.tools_by_name = {tool.name: tool for tool in tools}

def__call__(self, inputs: dict):
if messages := inputs.get("messages", []):
            message = messages[-1]
else:
raise ValueError("No message found in input")
        outputs = []
for tool_call in message.tool_calls:
print(f'正在执行工具 {tool_call["name"]},参数 {tool_call["args"]}')
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
print(f'工具{tool_call["name"]}, 执行结果{json.dumps(tool_result, ensure_ascii=False)}')
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result, ensure_ascii=False),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
return {"messages": outputs}

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")
# 定义工具
tool = TavilySearchTool()
tools = [tool]
# 工具绑定到模型
llm_with_tools = llm.bind_tools(tools)
# 定义图状态
classState(TypedDict):
    messages: Annotated[list, add_messages]  # 此处维护完整的消息历史
graph = StateGraph(State)
defchatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 工具路由
defroute_tools(state: State):
ifisinstance(state, list):
        ai_message = state[-1]
elif messages := state.get("messages", []):
        ai_message = messages[-1]
else:
raise ValueError(f"No messages found in input state to tool_edge: {state}")
ifhasattr(ai_message, "tool_calls") andlen(ai_message.tool_calls) > 0:
return"tools"
return END

# 使用LangGraph提供的工具节点
tool_node = ToolNode(tools=tools)
graph.add_node("chatbot", chatbot)
# 添加工具节点
graph.add_node("tools", tool_node)
# 添加工具条件边
graph.add_conditional_edges(
"chatbot",
    route_tools,
    {"tools": "tools", END: END},
)
graph.add_edge("tools", "chatbot")
graph.add_edge(START, "chatbot")
app = graph.compile()

if __name__ == "__main__":
    messages = []
whileTrue:
        user_input = input("👨‍💻: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Exiting...")
break
        messages.append({"role": "user", "content": user_input})
        response = app.invoke({"messages": messages})
        messages = response["messages"]
print(f'🤖: {response["messages"][-1].content}')

效果如下

除了继承 BaseTool 的方式,LangChain 还支持通过 @tool 装饰器 来快速定义工具。相比之下,这种方法更简洁,适合轻量级工具

from typing import Annotated
from tavily import TavilyClient
from langchain_core.tools import tool

@tool("tavily_search")
deftavily_search_tool(query: Annotated[str, "搜索查询"]):
"""一个针对全面、准确和可信的结果进行了优化的搜索引擎。当需要回答有关时事的问题时很有用。输入应该是搜索查询。"""
    client = TavilyClient()
    search_r = client.search(query=query)
return search_r

tavily_search_tool.invoke({"query": "北京2025年8月27日天气怎么样?"})

LangGraph调用MCP

除了自定义工具之外,LangGraph 还支持 MCP(Model Context Protocol),这使得工具的复用和扩展性更强。

要在 LangGraph 中使用 MCP,需要额外安装依赖包:

pip install langchain-mcp-adapters

编写一个简单的MCP Server

💡

关于MCP Server编写,参考之前的文章

深入探索 MCP:构建你的第一个 AI Agent 后端服务

依旧是以Tavily搜索为例

import json
from mcp.server.fastmcp import FastMCP
from tavily import TavilyClient
mcp = FastMCP("search")

@mcp.tool()
asyncdeftavily_search(query: str) -> str:
"""一个针对全面、准确和可信的结果进行了优化的搜索引擎。当需要回答有关时事的问题时很有用。输入应该是搜索查询。"""
    client = TavilyClient()
    search_r = client.search(query=query, max_results=2)
return json.dumps(search_r, ensure_ascii=False)

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

LangGrpah+MCP完整示例代码

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import ToolNode, tools_condition
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from pydantic import SecretStr

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")
# 配置MCP Server
client = MultiServerMCPClient(
    {
"search": {
"url": "http://localhost:8000/mcp/",
"transport": "streamable_http",
        }
    }
)

classState(TypedDict):
    messages: Annotated[list, add_messages]  # 此处维护完整的消息历史
graph = StateGraph(State)

asyncdefmain():
    tools = await client.get_tools()
# 工具绑定到模型
    llm_with_tools = llm.bind_tools(tools)
defchatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
    graph = StateGraph(State)
    graph.add_node(chatbot)
    graph.add_node(ToolNode(tools))
    graph.add_edge(START, "chatbot")
    graph.add_conditional_edges(
"chatbot",
        tools_condition,
    )
    graph.add_edge("tools", "chatbot")
    app = graph.compile()
    messages = []
whileTrue:
        user_input = input("👨‍💻: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Exiting...")
break
        messages.append({"role": "user", "content": user_input})
        response = await app.ainvoke({"messages": messages})
        messages = response["messages"]
print(f'🤖: {response["messages"][-1].content}')

asyncio.run(main())

效果如下

以上所有的LangGraph图结构

想入门 AI 大模型却找不到清晰方向?备考大厂 AI 岗还在四处搜集零散资料?别再浪费时间啦!2025 年 AI 大模型全套学习资料已整理完毕,从学习路线到面试真题,从工具教程到行业报告,一站式覆盖你的所有需求,现在全部免费分享

👇👇扫码免费领取全部内容👇👇

一、学习必备:100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT,帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例?大模型电子书?这份资料帮你站在 “行业高度” 学 AI

1. 100+本大模型方向电子书

在这里插入图片描述

2. 26 份行业研究报告:覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容,涵盖:

  • 职业趋势:《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》;
  • 商业落地:《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》;
  • 领域细分:《AGI 在金融领域的应用报告》《AI GC 实践案例集》;
  • 行业监测:《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT:听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会,包含百度、腾讯、字节等企业的一线实践:

在这里插入图片描述

  • 安全方向:《端侧大模型的安全建设》《大模型驱动安全升级(腾讯代码安全实践)》;
  • 产品与创新:《大模型产品如何创新与创收》《AI 时代的新范式:构建 AI 产品》;
  • 多模态与 Agent:《Step-Video 开源模型(视频生成进展)》《Agentic RAG 的现在与未来》;
  • 工程落地:《从原型到生产:AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看:大厂 AI 岗面试 “弹药库”,300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗?这份面试资料帮你提前 “押题”,拒绝临场慌!

1. 107 道大厂面经:覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景,包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题,每道题都附带思路解析

2. 102 道 AI 大模型真题:直击大模型核心考点

针对大模型专属考题,从概念到实践全面覆盖,帮你理清底层逻辑:

3. 97 道 LLMs 真题:聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案,比如让很多人头疼的 “复读机问题”:


三、路线必明: AI 大模型学习路线图,1 张图理清核心内容

刚接触 AI 大模型,不知道该从哪学起?这份「AI大模型 学习路线图」直接帮你划重点,不用再盲目摸索!

在这里插入图片描述

路线图涵盖 5 大核心板块,从基础到进阶层层递进:一步步带你从入门到进阶,从理论到实战。

img

L1阶段:启航篇丨极速破界AI新时代

L1阶段:了解大模型的基础知识,以及大模型在各个行业的应用和分析,学习理解大模型的核心原理、关键技术以及大模型应用场景。

img

L2阶段:攻坚篇丨RAG开发实战工坊

L2阶段:AI大模型RAG应用开发工程,主要学习RAG检索增强生成:包括Naive RAG、Advanced-RAG以及RAG性能评估,还有GraphRAG在内的多个RAG热门项目的分析。

img

L3阶段:跃迁篇丨Agent智能体架构设计

L3阶段:大模型Agent应用架构进阶实现,主要学习LangChain、 LIamaIndex框架,也会学习到AutoGPT、 MetaGPT等多Agent系统,打造Agent智能体。

img

L4阶段:精进篇丨模型微调与私有化部署

L4阶段:大模型的微调和私有化部署,更加深入的探讨Transformer架构,学习大模型的微调技术,利用DeepSpeed、Lamam Factory等工具快速进行模型微调,并通过Ollama、vLLM等推理部署框架,实现模型的快速部署。

img

L5阶段:专题集丨特训篇 【录播课】

img
四、资料领取:全套内容免费抱走,学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型,还是有基础想冲刺大厂、了解行业趋势,这份资料都能满足你!
现在只需按照提示操作,就能免费领取:

👇👇扫码免费领取全部内容👇👇

2025 年想抓住 AI 大模型的风口?别犹豫,这份免费资料就是你的 “起跑线”!

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐