当前位置：首页 > article >正文

大语言模型智能助手——既能生成自然语言回复，又能在必要时调用外部工具获取实时数据

article 2025/3/6 22:52:26

示例代码：

import json
from langgraph.graph import Graph, END,StateGraph
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_community.tools.openweathermap import OpenWeatherMapQueryRun
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.prebuilt import ToolInvocation, ToolNode,ToolExecutor  # 引入 ToolNode

tools = [OpenWeatherMapQueryRun()]

model = ChatOpenAI(
            openai_api_base=OPENAI_API_BASE,
            openai_api_key=OPENAI_API_KEY,
            temperature=0)

functions = [convert_to_openai_function(t) for t in tools]

model = model.bind_tools(functions)

def function_1(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

tool_executor = ToolExecutor(tools)

def function_2(state):
    messages = state['messages']
    last_message = messages[-1]  # 取最后一条消息，获取要发送给工具的查询
    print('last_message===\n',last_message)

    # 确保 tool_calls 存在且为非空列表
    tool_calls = last_message.additional_kwargs.get("tool_calls", [])

    if tool_calls:
        function_data = tool_calls[0].get("function", {})  # 获取 function 字典
        arguments_str = function_data.get("arguments", "{}")  # 获取 arguments JSON 字符串
        parsed_tool_input = json.loads(arguments_str)  # 解析 JSON
        print('parsed_tool_input===\n', parsed_tool_input)
        tool_call_id=tool_calls[0]["id"]
    else:
        print("Warning: tool_calls is empty.")
    
    print('function_data===\n',function_data,function_data["name"])
    print('parsed_tool_input===\n',parsed_tool_input,parsed_tool_input['location'])
    # 构造 ToolInvocation
    action = ToolInvocation(
        tool=function_data["name"],
        tool_input=parsed_tool_input['location'],
    )

    # 使用 tool_executor 处理请求
    response = tool_executor.invoke(action)
    print('response===\n',response,'\n',action.tool)
    # 构造 FunctionMessage
    function_message = ToolMessage(response, tool_call_id=tool_call_id)

    # 返回消息列表
    return {"messages": [function_message]}

def where_to_go(state):
    messages = state['messages']
    last_message = messages[-1]
    
    if "tool_calls" in last_message.additional_kwargs:
        return "continue"
    else:
        return "end"
    

# from langgraph.graph import Graph, END
# workflow = Graph()

# Or you could import StateGraph and pass AgentState to it
workflow = StateGraph(AgentState)

workflow.add_node("agent", function_1)
workflow.add_node("tool", function_2)

# The conditional edge requires the following info below.
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
# Next, we pass in the function that will determine which node is called next, in our case where_to_go().

workflow.add_conditional_edges("agent", where_to_go,{   # Based on the return from where_to_go
                                                        # If return is "continue" then we call the tool node.
                                                        "continue": "tool",
                                                        # Otherwise we finish. END is a special node marking that the graph should finish.
                                                        "end": END
                                                    }
)

# We now add a normal edge from `tools` to `agent`.
# This means that if `tool` is called, then it has to call the 'agent' next. 
workflow.add_edge('tool', 'agent')

# Basically, agent node has the option to call a tool node based on a condition, 
# whereas tool node must call the agent in all cases based on this setup.

workflow.set_entry_point("agent")

app = workflow.compile()


inputs = {"messages": [HumanMessage(content="what is the temperature in las vegas?")]} # what is the temperature in las vegas
result = app.invoke(inputs)
print('type result=====\n\n\n',type(result))
print('result=====\n\n\n',result)

输出：

last_message===
 content='' additional_kwargs={'tool_calls': [{'id': 'chatcmpl-c8tdWcPD6h68XZXYEX3I1lIDLPZn6', 'function': {'arguments': '{"location":"Las Vegas"}', 'name': 'open_weather_map'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 16, 'prompt_tokens': 80, 'total_tokens': 96, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-53715c56-b036-4721-a09e-376a950fbf5c-0' tool_calls=[{'name': 'open_weather_map', 'args': {'location': 'Las Vegas'}, 'id': 'chatcmpl-c8tdWcPD6h68XZXYEX3I1lIDLPZn6', 'type': 'tool_call'}] usage_metadata={'input_tokens': 80, 'output_tokens': 16, 'total_tokens': 96, 'input_token_details': {}, 'output_token_details': {}}
parsed_tool_input===
 {'location': 'Las Vegas'}
function_data===
 {'arguments': '{"location":"Las Vegas"}', 'name': 'open_weather_map'} open_weather_map
parsed_tool_input===
 {'location': 'Las Vegas'} Las Vegas

  action = ToolInvocation(
response===
 In Las Vegas, the current weather is as follows:
Detailed status: clear sky
Wind speed: 5.36 m/s, direction: 0°
Humidity: 35%
Temperature: 
  - Current: 13.53°C
  - High: 14.51°C
  - Low: 10.88°C
  - Feels like: 11.85°C
Rain: {}
Heat index: None
Cloud cover: 0% 
 open_weather_map
type result=====


 <class 'langgraph.pregel.io.AddableValuesDict'>
result=====


 {'messages': [HumanMessage(content='what is the temperature in las vegas?', additional_kwargs={}, response_metadata={}), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'chatcmpl-c8tdWcPD6h68XZXYEX3I1lIDLPZn6', 'function': {'arguments': '{"location":"Las Vegas"}', 'name': 'open_weather_map'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 16, 'prompt_tokens': 80, 'total_tokens': 96, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-53715c56-b036-4721-a09e-376a950fbf5c-0', tool_calls=[{'name': 'open_weather_map', 'args': {'location': 'Las Vegas'}, 'id': 'chatcmpl-c8tdWcPD6h68XZXYEX3I1lIDLPZn6', 'type': 'tool_call'}], usage_metadata={'input_tokens': 80, 'output_tokens': 16, 'total_tokens': 96, 'input_token_details': {}, 'output_token_details': {}}), ToolMessage(content='In Las Vegas, the current weather is as follows:\nDetailed status: clear sky\nWind speed: 5.36 m/s, direction: 0°\nHumidity: 35%\nTemperature: \n  - Current: 13.53°C\n  - High: 14.51°C\n  - Low: 10.88°C\n  - Feels like: 11.85°C\nRain: {}\nHeat index: None\nCloud cover: 0%', tool_call_id='chatcmpl-c8tdWcPD6h68XZXYEX3I1lIDLPZn6'), AIMessage(content='The current temperature in Las Vegas is 13.53°C.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 203, 'total_tokens': 217, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-d115a20e-e9fd-4ac4-bda3-2c2fbc433dad-0', usage_metadata={'input_tokens': 203, 'output_tokens': 14, 'total_tokens': 217, 'input_token_details': {}, 'output_token_details': {}})]}

补充代码输出最后一个消息：

# 取出 messages 列表中第一个消息对象
first_msg = result["messages"][-1]

# 获取第一个消息的 content 属性
first_content = first_msg.content

# 打印第一个消息的内容
print(first_content)

输出：

The current temperature in Las Vegas is 13.53°C.

下面解释这段代码的含义，并用通俗的语言和例子来说明每个部分是如何工作的。

1. 导入依赖库

import json

作用：引入 Python 内置的 json 模块，用于将 Python 数据结构（如字典）转换为 JSON 格式的字符串，或将 JSON 字符串解析回 Python 数据结构。
举例：比如，当你从 API 得到返回的 JSON 字符串 {"location": "las vegas"}，你可以使用 json.loads() 将它转换成字典，然后取出 "location" 对应的值。

from langgraph.graph import Graph, END, StateGraph

作用：从 langgraph.graph 模块中导入了三个组件：
- Graph：构建工作流图的基础类。
- END：常量，用于标记工作流的结束节点。
- StateGraph：基于状态的工作流图，用来描述多个步骤之间的转移。
举例：你可以把整个对话过程看作一张流程图，每一步（节点）负责处理一部分逻辑，而 END 就表示对话结束。

from langchain_core.utils.function_calling import convert_to_openai_function

作用：引入一个函数 convert_to_openai_function，用来将工具对象转换成 OpenAI 可调用的函数格式。
举例：假如你有一个天气查询工具，通过转换后，语言模型就可以“知道”如何调用这个工具来查询天气信息。

from langchain_community.tools.openweathermap import OpenWeatherMapQueryRun

作用：导入 OpenWeatherMapQueryRun 类，这是一个用于调用 OpenWeatherMap API 查询天气数据的工具。
举例：当你需要知道某个城市的当前温度时，就会使用这个工具发送请求，获取实际的天气数据。

from langchain_core.messages import HumanMessage, ToolMessage

作用：导入两种消息类型：
- HumanMessage：表示用户输入的消息。
- ToolMessage：表示工具调用返回的消息。
举例：当用户发送“what is the temperature in las vegas?”时，会封装成一个 HumanMessage；当工具返回温度结果时，会封装成一个 ToolMessage。

from langgraph.prebuilt import ToolInvocation, ToolExecutor  # 引入 ToolNode

作用：从 langgraph.prebuilt 导入几个用于工具调用的类：
- ToolInvocation：用于构造一个工具调用的描述（包括工具名称和输入参数）。
- ToolExecutor：执行工具调用的执行器。
举例：当语言模型决定需要查询天气时，会生成一个 ToolInvocation 对象，然后 ToolExecutor 根据这个对象去实际调用 OpenWeatherMap API。

2. 配置工具和模型

tools = [OpenWeatherMapQueryRun()]

作用：创建一个列表，里面放了一个 OpenWeatherMapQueryRun 的实例。这样我们就有一个可以查询天气的工具。
举例：假设你需要查询“las vegas”的天气，后续就会调用这个工具来获取数据。

model = ChatOpenAI(
            openai_api_base=OPENAI_API_BASE,
            openai_api_key=OPENAI_API_KEY,
            temperature=0)

作用：初始化一个聊天模型 ChatOpenAI（这里需要你事先定义或导入这个类）。传入 API 的基础地址、密钥，并设置 temperature 为 0 表示生成结果将更确定、不会有随机性。
举例：当你向这个模型发送询问消息时（比如“las vegas 的温度是多少？”），模型会根据提示生成回复或决定是否调用工具查询。

functions = [convert_to_openai_function(t) for t in tools]

作用：遍历之前定义的 tools 列表，将每个工具对象转换成 OpenAI 可调用的函数格式，保存到 functions 列表中。
举例：转换后的函数可以嵌入到模型生成的函数调用建议中，比如生成一段 JSON 指示“调用 OpenWeatherMapQueryRun 查询天气”。

model = model.bind_tools(functions)

作用：将 转换好的工具函数 绑定到模型上，使得在对话过程中模型可以直接调用这些工具。
举例：当用户问天气问题时，模型就能自动调用相应的工具，而不需要人工干预。

3. 定义“agent”节点的处理函数

def function_1(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

作用：这是工作流中的“agent”节点对应的函数，主要逻辑如下：
- 从输入状态中提取消息列表（state['messages']）。
- 调用模型的 invoke 方法，根据当前的消息生成一个回复（response）。
- 返回一个包含回复的字典，其格式符合工作流要求，即 {"messages": [response]}。
举例：如果初始状态中只有一个用户消息 “what is the temperature in las vegas?”，那么这个函数会将消息传给模型，模型可能生成一个包含工具调用指令的回复，例如提示需要调用天气查询工具。

4. 创建工具执行器

tool_executor = ToolExecutor(tools)

作用：初始化一个 ToolExecutor 实例，传入之前定义的 tools 列表。它负责实际执行工具调用。
举例：当系统需要查询天气时，会利用 tool_executor 发送请求给 OpenWeatherMap API，并返回查询结果。

5. 定义“tool”节点的处理函数

def function_2(state):
    messages = state['messages']
    last_message = messages[-1]  # 取最后一条消息，获取要发送给工具的查询
    print('last_message===\n', last_message)

作用：这是工作流中“tool”节点对应的函数，用于处理工具调用。第一步，它获取状态中的所有消息，并取最后一条消息，因为这条消息可能包含模型生成的工具调用信息。
举例：假设前面 agent 节点生成的回复中包含了一个指令，比如“调用天气查询工具查询 ‘las vegas’”，那么这条消息就是最后一条消息。

    # 确保 tool_calls 存在且为非空列表
    tool_calls = last_message.additional_kwargs.get("tool_calls", [])

作用：尝试从最后一条消息的 additional_kwargs 字段中获取 tool_calls 列表。这个列表包含了模型建议调用的工具信息。
举例：如果模型生成的回复中包含工具调用的说明，则 tool_calls 列表里会有相应的字典；如果没有，则返回空列表。

    if tool_calls:
        function_data = tool_calls[0].get("function", {})  # 获取 function 字典
        arguments_str = function_data.get("arguments", "{}")  # 获取 arguments JSON 字符串
        parsed_tool_input = json.loads(arguments_str)  # 解析 JSON
        print('parsed_tool_input===\n', parsed_tool_input)
        tool_call_id = tool_calls[0]["id"]
    else:
        print("Warning: tool_calls is empty.")

作用：
- 判断 tool_calls 是否有内容。
- 如果有，从第一个工具调用中提取 function 字典，其中包含工具名称及其参数（以 JSON 字符串形式存储）。
- 使用 json.loads 将 JSON 字符串解析为 Python 字典，得到工具调用的输入参数。
- 同时获取该工具调用的唯一标识符 id。
举例：如果模型建议调用天气工具，并给出参数 {"location": "las vegas"}，那么解析后 parsed_tool_input 就是字典 {"location": "las vegas"}，function_data["name"] 为 "open_weather_map"。

    print('function_data===\n', function_data, function_data["name"])
    print('parsed_tool_input===\n', parsed_tool_input, parsed_tool_input['location'])

作用：打印出获取的工具调用数据和解析后的输入参数，方便调试。这里分别显示了工具的名称和输入参数（如位置）。

举例：控制台中可能输出：

function_data===
{'name': 'open_weather_map', 'arguments': '{"location": "las vegas"}'} 
parsed_tool_input===
{'location': 'las vegas'}

    # 构造 ToolInvocation
    action = ToolInvocation(
        tool=function_data["name"],
        tool_input=parsed_tool_input['location'],
    )

作用：利用获取到的工具名称和输入参数构造一个 ToolInvocation 对象，用于描述需要调用哪个工具以及传入的参数。
举例：生成的 action 对象表示“调用 open_weather_map工具，并传入参数 ‘las vegas’”。

    # 使用 tool_executor 处理请求
    response = tool_executor.invoke(action)
    print('response===\n', response, '\n', action.tool)

作用：
- 使用 tool_executor 执行前面构造的 ToolInvocation。
- 将工具返回的结果保存在 response 中，并打印输出调试信息。
举例：如果 API 返回当前温度为 28°C，则 response 可能就是 “28°C”，同时打印出调用的是哪个工具。

    # 构造 FunctionMessage
    function_message = ToolMessage(response, tool_call_id=tool_call_id)

作用：将工具调用返回的结果包装成一个 ToolMessage 对象，同时关联上之前获取的工具调用 ID，这样后续流程可以识别是哪个工具调用的结果。
举例：function_message 对象中会包含返回的温度数据和工具调用的标识，用于和 agent 的对话衔接。

    # 返回消息列表
    return {"messages": [function_message]}

作用：最终返回一个字典，键为 "messages"，值是包含工具调用结果的消息列表，符合工作流的状态格式。

6. 定义决策函数：确定下一步走向

def where_to_go(state):
    messages = state['messages']
    last_message = messages[-1]
    
    if "tool_calls" in last_message.additional_kwargs:
        return "continue"
    else:
        return "end"

作用：
- 该函数用于决定工作流的下一步走向。
- 检查最后一条消息的 additional_kwargs 中是否包含 tool_calls 字段。
- 如果存在，返回字符串 "continue"，表示 agent 节点后续需要调用工具节点；否则返回 "end"，表示整个工作流结束。
举例：假设 agent 节点生成的回复中包含工具调用信息（即 additional_kwargs 中有 tool_calls），则函数返回 "continue"，工作流会跳转到工具节点去执行工具调用。

7. 构建工作流图

# from langgraph.graph import Graph, END
# workflow = Graph()

# Or you could import StateGraph and pass AgentState to it
workflow = StateGraph(AgentState)

作用：
- 这里展示了两种构建工作流的方法。注释部分给出了另一种可能的做法。
- 实际代码使用 StateGraph 来构建一个基于状态的工作流图。AgentState（虽然未在代码中定义）表示对话状态的数据结构。
举例：可以把整个对话流程看成一个状态机，每个节点（agent 或 tool）根据当前状态执行对应的函数。

workflow.add_node("agent", function_1)
workflow.add_node("tool", function_2)

作用：
- 将两个节点添加到工作流图中：
  - "agent" 节点关联 function_1，负责生成模型回复。
  - "tool" 节点关联 function_2，负责处理工具调用。
举例：当工作流启动时，会先执行 "agent" 节点中的逻辑；如果生成的回复需要调用工具，就会跳转到 "tool" 节点。

workflow.add_conditional_edges("agent", where_to_go, {
    "continue": "tool",
    "end": END
})

作用：
- 为 "agent" 节点添加基于条件的边（转移规则）。
- 这里使用 where_to_go 函数决定下一步：
  - 如果返回 "continue"，则转向 "tool" 节点。
  - 如果返回 "end"，则工作流结束（用特殊常量 END 表示）。
举例：如果 agent 节点生成的消息中包含工具调用信息，则 where_to_go 返回 "continue"，工作流会执行 function_2；否则工作流结束，返回最终结果。

workflow.add_edge('tool', 'agent')

作用：
- 添加一条普通的边，表示无论如何当 "tool" 节点执行完后，都要回到 "agent" 节点。
举例：工具调用执行完后，系统会把工具返回的结果传递给 agent 继续处理后续对话或生成新的回复。

workflow.set_entry_point("agent")

作用：
- 设置工作流的入口节点为 "agent"，即工作流启动时第一个执行的节点。
举例：当整个应用启动时，会首先调用 function_1 来处理用户输入。

app = workflow.compile()

作用：
- 编译（构建）工作流图，生成一个可执行的工作流应用 app。
举例：此时 app 已经包含了所有节点和边的定义，可以根据输入状态开始整个对话流程。

8. 传入输入并执行工作流

inputs = {"messages": [HumanMessage(content="what is the temperature in las vegas?")]} # what is the temperature in las vegas

作用：
- 定义了工作流的初始输入状态，是一个字典，键为 "messages"，值为包含一个 HumanMessage 对象的列表。
- 这个 HumanMessage 的内容为 “what is the temperature in las vegas?”，即用户询问拉斯维加斯的温度。
举例：当用户在聊天窗口输入这个问题后，系统会将其包装成 HumanMessage 对象传入工作流。

result = app.invoke(inputs)
print('type result=====\n\n\n', type(result))
print('result=====\n\n\n', result)

作用：
- 使用编译好的工作流 app 调用 invoke 方法，并传入初始输入。
- 将执行结果保存到 result 中，并打印结果的类型和内容。
举例：
- 假设工作流执行完后返回了一个消息列表，其中可能包含从 OpenWeatherMap 查询到的温度信息，控制台就会输出类似：
```
type result=====
<class 'dict'>

result=====
{'messages': [ToolMessage(...)]}
```
- 这样你就可以看到整个流程是如何从用户询问，到模型判断调用工具，再到实际查询并返回结果的。

总结

整个代码构建了一个基于状态机的对话工作流，主要步骤如下：

初始化：导入相关库，并配置好一个天气查询工具和一个 OpenAI 聊天模型。
工具绑定：将工具转换并绑定到模型上，使得模型能够在必要时调用外部 API（这里是天气查询）。
定义节点函数：
- function_1（agent 节点）：处理用户输入，生成回复，可能包含工具调用指令。
- function_2（tool 节点）：解析 agent 的回复中是否包含工具调用信息，若有则提取参数并调用工具，最后将工具结果封装成消息返回。
构建工作流图：设置节点之间的转移规则，例如：如果 agent 的回复中包含工具调用信息，就跳转到 tool 节点，否则结束；工具执行完后再回到 agent 节点。
启动工作流：传入用户问题（例如查询拉斯维加斯温度），依次执行 agent 和 tool 节点，最终打印出结果。

通俗举例：
想象你和一个智能助手对话：