示例代碼：

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  # 引入 ToolNodetools = [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)  # 解析 JSONprint('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'])# 構造 ToolInvocationaction = 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)# 構造 FunctionMessagefunction_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 Vegasaction = 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)  # 解析 JSONprint('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'}
  

    # 構造 ToolInvocationaction = 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”，同時打印出調用的是哪個工具。

    # 構造 FunctionMessagefunction_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(...)]}
    

  • 這樣你就可以看到整個流程是如何從用戶詢問，到模型判斷調用工具，再到實際查詢并返回結果的。

---

總結

整個代碼構建了一個基于狀態機的對話工作流，主要步驟如下：

1. 初始化：導入相關庫，并配置好一個天氣查詢工具和一個 OpenAI 聊天模型。
2. 工具綁定：將工具轉換并綁定到模型上，使得模型能夠在必要時調用外部 API（這里是天氣查詢）。
3. 定義節點函數：
   • function_1（agent 節點）：處理用戶輸入，生成回復，可能包含工具調用指令。
   • function_2（tool 節點）：解析 agent 的回復中是否包含工具調用信息，若有則提取參數并調用工具，最后將工具結果封裝成消息返回。
4. 構建工作流圖：設置節點之間的轉移規則，例如：如果 agent 的回復中包含工具調用信息，就跳轉到 tool 節點，否則結束；工具執行完后再回到 agent 節點。
5. 啟動工作流：傳入用戶問題（例如查詢拉斯維加斯溫度），依次執行 agent 和 tool 節點，最終打印出結果。

通俗舉例：
想象你和一個智能助手對話：

• 第一步：你問：“拉斯維加斯現在幾度？”
• 第二步：智能助手（agent）生成回復，并發現需要查詢天氣數據（包含一個工具調用指令）。
• 第三步：系統自動跳轉到工具執行部分（tool），根據指令調用天氣查詢 API，獲得當前溫度。
• 第四步：查詢結果返回給智能助手，然后你看到最終的回復，比如“拉斯維加斯當前溫度為 28°C”。

這種設計讓智能助手既能生成自然語言回復，又能在必要時調用外部工具獲取實時數據，從而提升回答的準確性和實用性。