2024-08 |hengshiousheu
摘要
這篇文章探討了 LangChain 和 LangGraph 這兩個強大的工具,它們能夠賦予 AI 模型呼叫外部程式碼的能力,進而擴展其功能並實現更智能的交互。文章首先介紹了「工具」的概念,並說明其如何充當 AI 模型與外部世界溝通的橋樑。接著,文章深入解析了在 LangChain 中創建工具的各種方法,從簡單的函數裝飾器到更靈活的 StructuredTool,並強調錯誤處理在確保 AI 應用穩健性的重要性。文章最後引入 LangGraph,展示如何使用 ToolNode 來整合外部工具,並通過一個簡單的聊天機器人範例說明了其實際應用。透過本文的探討,我們可以學習到如何利用 LangChain 和 LangGraph 打造功能更強大、互動更自然的 AI 應用。
前言
在人工智慧和自然語言處理領域中,LangChain 和 LangGraph 是兩個備受矚目的工具。本文將深入探討如何讓 AI 模型呼叫您的程式,擴展其功能並實現更智能的交互。
1. Tools 的本質:AI 代理的功能擴展器
Tools(工具)是專門設計給 AI 模型使用的功能單元。它們接受模型生成的輸入,並將輸出回傳給模型使用。當您希望模型能夠控制您程式碼的某些部分,或是呼叫外部 API 時,Tools 就顯得尤為重要。
1.1 Tools 的核心組成
一個典型的 Tool 通常包含以下幾個關鍵元素:
- 工具名稱(name)
- 工具功能描述(description)
- 定義工具輸入的 JSON 結構(JSON schema)
- 執行功能的函式(可選擇性地包含其非同步版本)
當 Tool 綁定到模型時,其名稱、描述和 JSON 結構會作為上下文提供給模型。給定一系列工具和一組指令,模型可以請求使用特定輸入呼叫一個或多個工具。
1.2 Tools 的實際應用場景
讓我們來看一個簡單的例子,假設我們要創建一個查詢台灣天氣的工具:
@tool
def query_taiwan_weather(city: str, date: Optional[str] = None) -> str:
"""
查詢台灣特定城市的天氣情況。
參數:
city (str): 要查詢天氣的城市名稱,例如 "台北"、"高雄" 等。
date (str, 可選): 要查詢的日期,格式為 "YYYY-MM-DD"。如果不提供,則查詢當天天氣。
返回:
str: 包含天氣信息的字符串。
"""
# 這裡應該是實際的天氣 API 調用邏輯
# 為了示例,我們返回一個模擬的結果
return f"{city} 在 {date or '今天'} 的天氣晴朗,溫度 25°C,適合外出活動。"
讓我們看看這個工具的 JSON Schema:
rich.print(query_taiwan_weather.args_schema.schema())

現在,我們來使用這個工具,並查看返回的訊息結構:
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo-0125")
llm_with_weather = llm.bind_tools([query_taiwan_weather])
response = llm_with_weather.invoke("請告訴我台北明天的天氣如何?")
rich.print(response)

🌟 亮點:通過使用 Tools,我們可以大大擴展 AI 模型的能力,使其能夠執行更加複雜和具體的任務,同時保持靈活性和可控性。
2. LangChain Tools 創建指南
現在我們已經了解了 Tools 在 LangChain 中的角色,讓我們深入探討如何創建這些 Tools。當您在構建一個 agent(代理)時,提供一系列它可以使用的 Tools 是至關重要的。
2.1 Tool 的關鍵屬性
以下是構成 Tool 的主要屬性:
屬性 | 型別 | 說明 |
---|---|---|
name | 字串 | 必須在提供給 LLM 或代理的一組工具中是唯一的。 |
description | 字串 | 描述這個工具的功能。作為 LLM 或代理的上下文使用。 |
args_schema | Pydantic BaseModel | 選擇性但建議使用,可用於提供更多資訊或對預期參數進行驗證。 |
return_direct | 布林值 | 僅與代理相關。當設為 True 時,在調用給定工具後,代理會停止並直接將結果返回給使用者。 |
2.2 創建 Tools 的方法
LangChain 支持以下幾種創建 Tools 的方法:
- 從函式創建:這是最簡單的方法,適用於大多數使用場景。
- 從 LangChain Runnables 創建:適用於更複雜的邏輯。
- 通過繼承 BaseTool 類:這是最靈活的方法,提供了最大程度的控制,但需要更多的代碼和努力。
💡 專業提示:模型的表現會因 Tools 的名稱、描述和 JSON 結構的選擇而有所不同。精心設計這些元素可以顯著提升模型的效能。
2.3 從函數創建 Tools:簡單而強大
2.3.1 @tool 裝飾器:簡化 LangChain 自訂工具的創建
@tool
裝飾器提供了一種極為便捷的方式來定義和創建自訂工具。這個裝飾器的設計理念是讓開發者能夠快速將普通的 Python 函式轉變為 LangChain 可識別和使用的工具。
主要特點:
- 預設使用函式名稱作為工具名稱
- 使用函式的文件字串(docstring)作為工具描述
- 可以解析函式的型別註解
📌 注意事項: 確保為每個使用 @tool 裝飾器的函式提供清晰、詳細的文件字串,以便 LLM 或代理能夠正確使用該工具。
2.3.1.1 基本用法示範
以查詢台灣城市天氣為例,展示 @tool 的基本用法:
from langchain.tools import tool
@tool
def get_taiwan_weather(city: str) -> str:
"""查詢台灣特定城市的天氣狀況。"""
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
# 檢視工具的相關屬性
print(f"name: {get_taiwan_weather.name}")
print(f"description: {get_taiwan_weather.description}")
print(f"args: {get_taiwan_weather.args}")
# 使用這個工具
result = get_taiwan_weather.run("台北")
print(f"呼叫工具結果{result}")

在這個例子中:
- 我們使用 @tool 裝飾了 get_taiwan_weather 函式,將其轉換為 LangChain 工具。
- 函式的文件字串成為了工具的描述。
- 函式的參數 city: str 被自動解析為工具的輸入參數。
- 我們可以直接呼叫這個函式來使用工具,就像呼叫普通的 Python 函式一樣。
2.3.1.2 非同步實現:提高效能
對於可能需要等待外部 API 回應的情況,我們可以使用非同步版本的工具:
import asyncio
@tool
async def get_taiwan_weather_async(city: str) -> str:
"""非同步查詢台灣特定城市的天氣狀況。"""
await asyncio.sleep(1) # 模擬 API 調用
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
result = await get_taiwan_weather_async.ainvoke({"city": "高雄"})
print(result)

這個非同步版本的工具:
- 使用 async def 定義函式,允許非阻塞操作。
- 無法也不能使用同步的
invoke
來呼叫async
異步工具
🚀 效能提示:非同步工具在處理需要等待外部資源(如網路請求)的場景中特別有用,可以提高整體應用的效能。
2.3.1.3 自訂工具名稱和參數結構:精確控制
有時,我們可能想要更精確地控制工具的名稱和參數結構。這可以通過向 @tool 裝飾器傳遞參數來實現:
from langchain.pydantic_v1 import BaseModel, Field
class TravelPlanInput(BaseModel):
city: str = Field(description="欲訪問的台灣城市")
days: int = Field(description="旅遊天數")
budget: int = Field(description="預算(新台幣)")
@tool("台灣旅遊規劃器", args_schema=TravelPlanInput, return_direct=True)
def plan_taiwan_trip(city: str, days: int, budget: int) -> str:
"""根據指定的城市、天數和預算規劃台灣旅遊行程。"""
per_day_budget = budget // days
if city == "台北":
return f"台北{days}天行程:每天預算{per_day_budget}元。建議景點:101大樓、故宮博物院、象山。別忘了品嚐鼎泰豐的小籠包!"
elif city == "台中":
return f"台中{days}天行程:每天預算{per_day_budget}元。建議景點:高美濕地、彩虹眷村、逢甲夜市。記得嘗試太陽餅!"
elif city == "高雄":
return f"高雄{days}天行程:每天預算{per_day_budget}元。建議景點:蓮池潭、駁二藝術特區、旗津海岸。不要錯過壽山動物園!"
else:
return f"抱歉,目前沒有{city}的旅遊規劃資訊。"
print(plan_taiwan_trip.name)
print(plan_taiwan_trip.description)
print(plan_taiwan_trip.args)
print(plan_taiwan_trip.return_direct)
# 使用這個工具
result = plan_taiwan_trip.run({"city": "台北", "days": 3, "budget": 10000})
print(result)

在這個例子中:
- 使用 .run() 方法:當使用 @tool 裝飾器創建工具時,使用 .run() 方法來執行工具,而不是直接調用函數。
- 使用 @tool(“台灣旅遊規劃器”, …) 來自訂工具名稱。
- 參數傳遞:定義了一個 TravelPlanInput 類來精確指定工具的輸入結構。
- 返回結果:return_direct=True 表示這個工具的輸出應該直接返回給用戶,而不需要進一步處理。
這種方法允許我們對工具的結構和行為有更精細的控制,特別適用於複雜的工具或需要特定輸入驗證的場景。
2.3.2 StructuredTool:更靈活的工具創建方式
StructuredTool.from_function
類方法提供了比 @tool
裝飾器更多的配置選項,同時不需要太多額外的程式碼。這個部分將展示如何使用 StructuredTool.from_function
方法來配置和使用工具,同時融入台灣的夜市文化元素。
2.3.2.1 基本使用
首先,讓我們創建一個簡單的夜市小吃價格計算器:
from langchain_core.tools import StructuredTool
def calculate_night_market_price(item: str, quantity: int) -> str:
"""計算台灣夜市小吃的總價。"""
prices = {
"蚵仔煎": 60,
"臭豆腐": 40,
"珍珠奶茶": 50,
"鹽酥雞": 70,
"大腸包小腸": 55
}
if item not in prices:
return f"抱歉,我們沒有 {item} 的價格信息。"
total = prices[item] * quantity
return f"{quantity} 份 {item} 的總價是 {total} 元新台幣。"
async def acalculate_night_market_price(item: str, quantity: int) -> str:
"""非同步計算台灣夜市小吃的總價。"""
return calculate_night_market_price(item, quantity)
night_market_calculator = StructuredTool.from_function(
func=calculate_night_market_price,
coroutine=acalculate_night_market_price
)
print(night_market_calculator.invoke({"item": "臭豆腐", "quantity": 2}))
print(await night_market_calculator.ainvoke({"item": "珍珠奶茶", "quantity": 3}))

StructuredTool 的一個主要優勢是它能夠同時支援同步(sync)和非同步(async)函數。
- func: 指定同步函數 calculate_night_market_price
- coroutine: 指定非同步函數 acalculate_night_market_price
這樣配置允許工具根據調用方式(invoke 或 ainvoke)自動選擇適當的函數。
💡 專業提示:使用 StructuredTool 可以讓您的工具更加靈活,能夠應對不同的執行環境和需求。
2.3.2.2 進階配置:精細化工具定義
現在,讓我們看看如何進一步配置 StructuredTool
:
from langchain.pydantic_v1 import BaseModel, Field
class NightMarketInput(BaseModel):
item: str = Field(description="夜市小吃名稱,例如:蚵仔煎、臭豆腐、珍珠奶茶等")
quantity: int = Field(description="購買數量")
def calculate_night_market_price(item: str, quantity: int) -> str:
"""計算台灣夜市小吃的總價。"""
prices = {
"蚵仔煎": 60,
"臭豆腐": 40,
"珍珠奶茶": 50,
"鹽酥雞": 70,
"大腸包小腸": 55
}
if item not in prices:
return f"抱歉,我們沒有 {item} 的價格信息。"
total = prices[item] * quantity
return f"{quantity} 份 {item} 的總價是 {total} 元新台幣。"
night_market_calculator = StructuredTool.from_function(
func=calculate_night_market_price,
name="台灣夜市小吃計價器",
description="計算台灣夜市常見小吃的總價",
args_schema=NightMarketInput,
return_direct=True
)
print(night_market_calculator.invoke({"item": "蚵仔煎", "quantity": 2}))
print(night_market_calculator.name)
print(night_market_calculator.description)
print(night_market_calculator.args)

進階配置中,可以清楚看到以下特性:
- 定義輸入模型:
- NightMarketInput 使用 BaseModel 和 Field 定義了輸入參數的結構和描述。
- 這有助於確保輸入的正確性,並為使用者提供清晰的指引。
- 自定義工具名稱和描述:
- 通過 name 和 description 參數,我們可以為工具提供更具描述性的標識。
- 指定參數結構:
- args_schema 參數允許我們使用預先定義的 Pydantic 模型來規範輸入格式。
- 直接返回結果:
- return_direct=True 表示這個工具的輸出應該直接返回給用戶。
2.3.2.3 StructuredTool 的使用場景與優勢
- 結構化輸入:通過定義明確的輸入模型,可以確保傳入的參數符合預期格式。
- 彈性配置:相比 @tool 裝飾器,提供了更多的配置選項。
- 同步與非同步支援:可以同時定義同步和非同步版本的函數。
- 易於整合:可以輕鬆地與 LangChain 的其他組件(如 Agents)整合。
🌟 亮點:StructuredTool 為開發者提供了更大的靈活性,特別適合需要精確控制輸入輸出結構的複雜應用場景。
3. 工具錯誤處理:打造穩健的 AI 應用
在使用 LangChain 的工具(Tools)時,建立一個錯誤處理策略是非常重要的,特別是當這些工具與代理(Agents)一起使用時。適當的錯誤處理可以讓代理從錯誤中恢復並繼續執行。讓我們深入探討如何在 LangChain 中實現工具的錯誤處理。
3.1 基本策略:ToolException 和 handle_tool_error
一個簡單的錯誤處理策略是在工具內部拋出 ToolException
,並使用 handle_tool_error
來指定錯誤處理方式。當指定了錯誤處理器時,異常會被捕獲,錯誤處理器將決定從工具返回什麼輸出。
3.2 設置錯誤處理:三種方式
您可以將 handle_tool_error
設置為以下幾種方式:
- True:使用預設的錯誤處理行為
- 字串:始終返回指定的字串
- 函數:自定義錯誤處理邏輯
⚠️ 重要提示:僅僅拋出
ToolException
是不夠的。您需要先設置工具的 handle_tool_error,因為其預設值為 False。
3.3 使用範例:天氣查詢工具
讓我們通過一個查詢天氣的例子來說明這些錯誤處理方法:
def get_weather(city: str) -> int:
"""查詢指定城市的天氣。"""
raise ToolException(f"錯誤:找不到名為 {city} 的城市。")
3.3.1 範例 1:預設錯誤處理行為
from langchain_core.tools import ToolException
from langchain.tools import StructuredTool
# 範例 1:預設錯誤處理行為
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error=True,
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("預設錯誤處理:", result)

預設錯誤處理:
- 設置 handle_tool_error=True 會直接返回錯誤訊息。
- 適用於簡單的錯誤提示場景。
3.3.2 範例 2:使用自定義字串
# 範例 2:使用自定義字串
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error="抱歉,找不到該城市,但那裡的溫度應該高於絕對零度!",
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("自定義字串錯誤處理:", result)

自定義字串錯誤處理:
- 設置
handle_tool_error
為一個字串,會在發生錯誤時始終返回該字串。 - 適用於想要提供固定、友善錯誤訊息的情況。
3.3.3 範例 3:使用自定義函數
# 範例 3:使用自定義函數
def custom_error_handler(error: ToolException) -> str:
return f"工具執行期間發生以下錯誤:`{error.args[0]}`"
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error=custom_error_handler,
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("自定義函數錯誤處理:", result)

自定義函數錯誤處理:
- 定義一個接受 ToolException 作為參數並返回字串的函數。
- 最靈活的方法,允許根據錯誤的具體情況進行不同的處理。
🛠️ 最佳實踐:根據您的應用需求選擇適當的錯誤處理方法。對於簡單情況,使用預設或自定義字串可能就足夠了。對於需要更複雜邏輯的情況,自定義函數提供了最大的靈活性。
4.初探 LangGraph 如何善用 ToolNode
在前面的章節中,我們深入探討了 LangChain 中工具(Tool)的各個方面,包括創建方法和異常處理等。現在,讓我們將焦點轉向 LangGraph,看看如何在這個強大的框架中運用這些知識。本文將通過一個簡單而實用的例子,逐步展示 LangGraph 的操作方法。
4.1 整合搜索工具
為了增強聊天機器人的能力,我們將整合一個網絡搜索工具。這使得機器人能夠查找相關信息,提供更準確的回答。
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode, tools_condition
tool = TavilySearchResults(max_results=2)
tools = [tool]
tool_node = ToolNode(tools=[tool])
tool.invoke("高雄必比登推薦餐廳?")

ToolNode 封裝了我們定義的工具(在這個例子中是網路搜尋工具)。它允許聊天機器人在需要時調用外部資源來增強其回應能力
4.2 添加至 Graph 當中
將所有組件組合成一個完整的圖形結構。這定義了聊天機器人的工作流程。
直接把剛剛建立的 tool_node
放盡 Graph 當中。
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
同時也添加條件邊,讓 chatbot 節點輸出結果中帶有 tool_calls
前往指定節點
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)

4.3 與聊天機器人互動
最後,我們創建一個簡單的交互界面,允許用戶與聊天機器人對話。
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": ("user", user_input)}):
for value in event.values():
# print("Assistant:", value["messages"][-1].content)
value["messages"][-1].pretty_print()
這個循環允許用戶輸入消息,聊天機器人會處理這些輸入並給出回應。用戶可以隨時輸入 “quit”、”exit” 或 “q” 來結束對話。

5. 結語
通過本文的探討,我們深入了解了 LangChain 和 LangGraph 中 Tools 的重要性和實現方法。從基本的 @tool 裝飾器到更為靈活的 StructuredTool,再到錯誤處理策略,這些知識將幫助您構建更加強大和可靠的 AI 應用。
記住,選擇合適的工具創建方法和錯誤處理策略,不僅可以提高您的 AI 模型的功能性,還能大大提升其穩定性和用戶體驗。在實際應用中,根據具體需求靈活運用這些技術,將使您的 AI 解決方案更加出色。
最後,隨著 AI 技術的不斷發展,保持學習和實踐的態度至關重要。希望本文能為您的 AI 開發之旅提供有價值的見解和指導。