基于LangGraph、Groq和Tavily打造可以调用外部搜索引擎工具的对话机器人（核心代码 万字详解）

一、python环境 & 相关库版本信息

代码运行在 conda 创建的python环境下，python和相关库的版本信息如下：

$ python --version
Python 3.12.3$ pip list | grep langchain
langchain                 0.3.15
langchain-community       0.3.15
langchain-core            0.3.31
langchain-groq            0.2.3
langchain-text-splitters  0.3.5
langgraph                 0.2.64
langgraph-checkpoint      2.0.10
langgraph-sdk             0.1.51
langsmith                 0.2.11

导入需要用的库：

import os
import json
from langchain_community.tools.tavily_search import TavilySearchResults
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_groq import ChatGroq
from langchain_core.messages import ToolMessage

(若不确定相关库是否有缺失，可以先运行，根据库缺失等报错信息进行 pip install)

安装 langchain ：pip install langchain

安装 langgraph ：pip install -U langgraph

安装 langchain-groq ：pip install langchain-groq

二、申请 API Key

本次代码运行需要用到三个不同的 API Key：

1. GROQ_API_KEY（用于借助groq云平台实现高效LLM推理）：申请页面
2. TAVILY_API_KEY（专为大型语言模型（LLMs）和检索增强生成（RAG）应用设计的搜索引擎，在本次代码示例中作为LLM可使用的Tool）：获取页面
3. LANGCHAIN_API_KEY（为了使用LangSmith需要用到，LangSmith可以帮助我们清晰直观地跟踪搭建的LangGraph的每一次状态变化过程）：申请页面

获取完以上 API Key 之后，可以将它们统一放在代码同级的 .env 文件中进行管理，并使用 python-dotenv 库在运行时加载这些环境变量。因为我使用该方法时有点问题，所以我还是直接把环境变量写在代码里了。

因为我们要使用LangSmith用于调试和跟踪，涉及到另外一个环境变量 LANGCHAIN_TRACING_V2，默认为 false，设置为 true 就可以开启跟踪功能。

os.environ["TAVILY_API_KEY"] = "TAVILY_API_KEY"
os.environ["LANGCHAIN_API_KEY"] = "LANGCHAIN_API_KEY"  # Get this from smith.langchain.com
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["GROQ_API_KEY"] = "GROQ_API_KEY"

三、Graph功能描述

Graph图如下所示：

我们主要想实现的功能就是，用户可以不断提问，LLM将根据用户提问内容判断是否需要使用作为外部搜索引擎的tavily API服务获取更多信息，若需要则结合这些信息对用户提问做出回答。

Graph处理逻辑流程描述

1. 用户输入问题；
2. chatbot 节点接收用户输入并更新 Graph 的状态信息；
3. 通过 conditional edge，根据相应的代码逻辑判断 Graph 的下一步操作，决定是进入 tools 节点调用工具，还是直接跳转至 END 节点结束流程；
4. 如果进入 tools 节点，则执行工具调用逻辑，通过调用 Tavily API 服务进行联网检索。在获取返回结果后，更新 Graph 的状态信息，然后通过与 chatbot 节点之间的有向边返回 chatbot；
5. chatbot 接收更新后的状态信息，作为 LLM 的输入进行推理。如果推理结果不需要再次调用工具，则通过 conditional edge 判断后直接进入 END 节点；否则继续执行，直到流程到达 END 节点，完成一次 Graph 的迭代。

四、核心代码详解

以下代码全部来自 Langgraph 官方教程文档 。

1. 定义用于保存graph状态的数据结构

首先，创建一个继承自 TypedDict 类的子类 State，且 State 类定义了一个 messages 键，用于定义描述graph状态的数据结构。messages 键的类型被指定为 list，意味着该键的值是一个列表，用于存储Graph状态变化时产生的消息。Annotated 则是用来为 messages 类型（列表）添加额外的元数据 add_messages，以此控制 messages 键的列表值的更新方式为追加而不是默认的覆盖。即当graph的状态，也就是消息列表被更新时，上一次的消息列表不会被新产生的消息列表所覆盖，而是会将新消息追加到上一次的消息列表中，以达到每一次graph流迭代时，所有产生的状态都会保存在消息列表中。

# 定义状态
class State(TypedDict):# Messages have the type "list". The `add_messages` function# in the annotation defines how this state key should be updated# (in this case, it appends messages to the list, rather than overwriting them)messages: Annotated[list, add_messages]

2. 定义 tool 节点

2.1 创建工具列表

# 创建tavily搜索引擎工具，max_results用来设置返回检索结果的数量
tool = TavilySearchResults(max_results=2)
# 放入列表中
tools = [tool]

2.2 定义tool节点类

该类的主要作用是执行上一个 AI 消息中请求的工具，并返回执行结果。

class BasicToolNode:"""A node that runs the tools requested in the last AIMessage."""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:tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])outputs.append(ToolMessage(content=json.dumps(tool_result),name=tool_call["name"],tool_call_id=tool_call["id"],))return {"messages": outputs}

更为详细的代码解释如下（ChatGPT生成）：

BasicToolNode 类是一个工具节点，它的主要作用是执行上一个 AI 消息中请求的工具，并返回执行结果。下面是对类中每个部分的详细解释：

类的总体作用：该类的核心功能是接收输入信息（通常是包含工具请求的消息），找到对应的工具，执行工具，然后将执行结果以消息的形式返回。它主要处理与工具相关的操作，通过调用不同的工具来完成任务。

__init__ 方法：

• 构造函数 __init__(self, tools: list) 用于初始化 BasicToolNode 实例。
• tools: list 是传递给该类的一个工具列表。每个工具都有一个 name 属性（工具的名称），并且可以通过 invoke() 方法被调用。
• self.tools_by_name 是一个字典，将工具名称与工具对象映射在一起，方便后续根据名称快速查找对应的工具。

self.tools_by_name = {tool.name: tool for tool in tools}

• 例如，如果有一个名为 calculator 的工具，它就会被存储为 {"calculator": calculator}，以便根据名称查找工具并调用。

__call__ 方法：

• 该方法允许将类实例当作函数来调用。接收一个字典 inputs 作为参数，执行工具请求，并返回工具执行结果。

具体流程如下：

一、获取消息：

if messages := inputs.get("messages", []):message = messages[-1]
else:raise ValueError("No message found in input")

• 首先检查 inputs 字典中是否有键 "messages"。如果存在，则提取 messages 列表的最后一条消息（即最近的一条 AIMessage）。
• 如果找不到消息，则抛出错误 ValueError("No message found in input")，防止后续操作出错。

二、执行工具请求：

for tool_call in message.tool_calls:tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])

• 该消息包含了 tool_calls，即消息中请求的工具列表。tool_calls 列表中的每一项代表一次工具调用，里面有工具名称 name 和调用参数 args。
• 根据工具名称 tool_call["name"]，从 self.tools_by_name 中找到相应的工具，并调用它的 invoke 方法，将调用参数 args 传递给工具。

三、保存工具执行结果：

outputs.append(ToolMessage(content=json.dumps(tool_result),name=tool_call["name"],tool_call_id=tool_call["id"],)
)

工具执行完成后，将结果存储到 outputs 列表中。每个执行结果通过 ToolMessage 封装，包含以下内容：

• content：工具的执行结果，使用 json.dumps() 将其转换为 JSON 格式。
• name：工具名称，用于标识是哪一个工具产生的结果。
• tool_call_id：工具调用的唯一 ID，用于追踪工具调用。

四、返回结果：

return {"messages": outputs}

最后，返回一个字典 {"messages": outputs}，其中 messages 是包含所有工具执行结果的列表。

总结：

• BasicToolNode 类的作用是接收带有工具请求的消息，提取出工具请求后调用相应的工具执行操作，并将执行结果打包为消息返回。
• 它可以被视为一个工具执行器，自动根据消息中的请求调用工具，并返回相应的结果。

2.3 创建 tool 节点类的对象

# define a tool node
tool_node = BasicToolNode(tools=tools)

3. 定义 chatbot 节点

3.1 创建绑定工具列表的LLM

使用的模型是 mixtral-8x7b-32768，如果想尝试其他模型，可以在 Groq 官网提供的模型列表 中进行选择。llm 调用的 bind_tools() 函数是可以将一组工具与 llm 进行绑定关联，以此扩展语言模型的功能，使llm 在对话或推理过程中，可以根据上下文选择合适的工具并调用它们。

llm = ChatGroq(temperature=0.8, model_name="mixtral-8x7b-32768")
llm_with_tools = llm.bind_tools(tools)

3.2 定义 chatbot 节点

chatbot 节点的定义比较简洁，就是一个调用LLM进行推理并将生成内容以State数据格式返回的函数。

def chatbot(state: State):llm_response = llm_with_tools.invoke(state["messages"])return {"messages": [llm_response]}

4. 定义 conditional edge

当graph流走到chatbot节点，并执行完成后，将根据 route_tools 函数作为conditional edge决定下一步去向。

def route_tools(state: State,
):"""Use in the conditional_edge to route to the ToolNode if the last messagehas tool calls. Otherwise, route to the end."""if isinstance(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}")if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:return "tools"return END

更为详细的代码解释如下（ChatGPT生成）：

route_tools 函数的作用是决定工作流的路由方向：当收到一条消息时，检查其中是否有工具调用（tool_calls）。如果有工具调用，路由到工具节点；否则，路由到结束（END）。

一、函数参数:

• state: State：传递进来的 state 是一个状态对象，通常包含有关当前对话或任务的上下文信息，尤其是消息历史等。这个 state 可以是一个 list 或者是一个包含 messages 的字典。

二、函数逻辑概述:

• 该函数首先尝试从 state 中提取最近的一条消息（ai_message）。
• 然后，检查该消息是否包含工具调用（tool_calls）。
• 如果发现有工具调用，则返回 "tools"，表示应该转到工具节点执行工具；如果没有工具调用，则返回 END，表示任务已完成或不需要调用工具。

三、详细解释:

1. 从 state 提取最近的消息：

if isinstance(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}")

• 首先检查 state 是否是一个列表。如果是列表，直接取 state 的最后一个元素作为最近的消息（ai_message = state[-1]）。
• 如果 state 不是列表，则假定它是一个字典，并尝试获取键 "messages" 对应的消息列表（messages := state.get("messages", [])），同样从中提取最后一条消息 messages[-1]。
• 如果 state 既不是列表，也没有包含消息列表，则抛出一个错误，提示找不到有效的消息。

2. 检查消息是否包含工具调用：

if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:return "tools"
return END

• 一旦成功提取到 ai_message，该部分逻辑检查这条消息中是否有 tool_calls 属性。
• 如果消息中包含 tool_calls 属性并且它的长度大于 0（即有工具请求），则函数返回 "tools"，表示将工作流引导至工具节点执行工具。
• 如果没有工具调用，函数返回 END，表示流程可以结束或不需要再进行工具调用。

route_tools 的主要功能是根据对话状态中的最后一条消息，检查是否存在工具调用，并动态路由到相应的节点。有工具调用时，路由到工具节点执行工具；没有工具调用时，路由到结束节点，表示任务不需要进一步的工具执行。

5. 创建并编译状态图

在定义好状态图所需的chatbot和tool这两个节点，以及路由到下一个节点的condition edge函数后，我们来构建Graph（代码注释由 ChatGPT 生成）。

# 创建一个状态图对象，初始状态为 State 类型
graph_builder = StateGraph(State)# 在状态图中添加一个 "chatbot" 节点
graph_builder.add_node("chatbot", chatbot)
# 在状态图中添加一个 "tools" 节点，表示调用工具的节点
graph_builder.add_node("tools", tool_node)# 定义初始边：状态图开始时，进入 "chatbot" 节点
graph_builder.add_edge(START, "chatbot")
# 在聊天机器人节点设置条件边，根据不同条件决定路由
graph_builder.add_conditional_edges("chatbot",      # 源节点：聊天机器人节点route_tools,    # 条件函数：用于判断是否需要调用工具{              # 条件函数的返回值与目标节点的映射"tools": "tools",  # 如果 route_tools 返回 "tools"，跳转到工具节点END: END           # 如果 route_tools 返回 END，表示结束流程}
)
# 定义一个边：当工具节点完成任务后，返回到 "chatbot" 节点，以决定下一步操作
graph_builder.add_edge("tools", "chatbot")# 编译状态图，将节点、边和条件函数转换成可执行的状态机
graph = graph_builder.compile()

6. 用户无限问答实现

最后我们来实现上一次构建并编译好的graph的运行。

6.1 定义推理函数

stream_graph_updates 函数的主要作用是基于用户输入，通过状态图（graph）进行推理并实时输出 AI 助手的对话回应，以实现一种流式对话交互的效果（代码注释由 ChatGPT 生成）。

def stream_graph_updates(user_input: str):"""根据用户输入更新对话状态，并实时流式输出 AI 助手的回应。参数:user_input (str): 用户输入的消息内容，作为对话的开始。返回:None: 通过打印助手的响应实时输出消息。"""# 调用状态图的 stream 方法，将用户输入传递给图进行处理，并获取对话事件的流式更新for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):# 遍历每个事件中的所有值。一个事件可能包含多个值（例如多个工具调用或消息）for value in event.values():# 取出最新一条助手的回复消息，并输出内容# messages 是一个列表，[-1] 表示列表中的最后一条消息（最新消息）# content 是消息的实际文本内容print("Assistant:", value["messages"][-1].content)

6.2 实现用户无限输入模式

我们可以让这个对话无限进行下去，只到用户输入 “quit”, “exit”, “q” 中的任何一个字符串以终止对话。

while True:try:user_input = input("User: ")if user_input.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakstream_graph_updates(user_input)except:# fallback if input() is not availableuser_input = "What do you know about LangGraph?"print("User: " + user_input)stream_graph_updates(user_input)break

五、效果演示

因为 stream_graph_updates 做了循环打印，所以每到一个节点产生的新的message都会被打印出来。

User: 英伟达现在的股价是多少？     
Assistant: 
Assistant: [{"url": "https://companiesmarketcap.com/nvidia/marketcap/", "content": "Market cap history of NVIDIA from 2001 to 2023\nEnd of year Market Cap\nEnd of Day market cap according to different sources\nOn Dec 16th, 2023 the market cap of NVIDIA was reported to be:\nMarket capitalization for similar companies or competitors\nThe market capitalization sometimes referred as Marketcap, is the value of a publicly listed company.\n The market capitalization, commonly called market cap, is the total market value of a publicly traded company's outstanding shares and is commonly used to measure how much a company is worth.\n Market capitalization of NVIDIA (NVDA)\nMarket cap: $1.207 Trillion\nAs of December 2023 NVIDIA has a market cap of $1.207 Trillion.\n CompaniesMarketCap is receiving financial compensation for Delta App installs.\nCompaniesMarketCap is not associated in any way with CoinMarketCap.com\nStock prices are delayed, the delay can range from a few minutes to several hours.\n In January 1999, Nvidia was included in the NASDAQ (NVDA) and delivered the ten millionth graphics chip in the same year."}, {"url": "https://stockanalysis.com/stocks/nvda/market-cap/", "content": "NVIDIA has a market cap or net worth of $3.37 trillion as of January 17, 2025. Its market cap has increased by 187.03% in one year. Market Cap 3.37T. Enterprise Value ... Market capitalization, also called net worth, is the total value of all of a company's outstanding shares. It is calculated by multiplying the stock price by the number of"}]
Assistant: Nvidia's market cap was reported to be $1.207 trillion on December 16th, 2023, according to CompaniesMarketCap. However, as of January 17, 2025, NVIDIA has a market cap or net worth of $3.37 trillion, according to StockAnalysis. Therefore, the market cap of NVIDIA has significantly increased by 187.03% in one year.

我们最终要看的是最后一个输出内容，这是 LLM 基于tavily搜索引擎工具拿到的返回内容生成的回复。(根据回复内容可以看出，tavily索引到的结果也不是那么实时)

Assistant: Nvidia's market cap was reported to be $1.207 trillion on December 16th, 2023, according to CompaniesMarketCap. However, as of January 17, 2025, NVIDIA has a market cap or net worth of $3.37 trillion, according to StockAnalysis. Therefore, the market cap of NVIDIA has significantly increased by 187.03% in one year.

相关阅读

[1] 又一AI搜索引擎开源了

以上就是本篇博客全部内容，大家一起玩起来吧！用 LangGraph 开发出更多有趣实用的LLM智能体。
PS：关于 LangSmith 的使用我将在后续的博客中介绍，大家感兴趣可以自行了解或关注follow后续更新。