利用 Tool-Calling 增強 LangGraph
- Tool-Calling 原理 ==================
上一節的內容中，我們藉助 LangGraph 框架、基於 DeepSeek 模型的驅動、通過多個節點的協作，快速搭建起了一個智能編輯 Agent，並且幫我們生成了一篇完整的文章，整體流程還是比較清晰的。
但是，不知道大家有沒有發現這樣一個問題：Agent 實際生成文章的內容是比較陳舊的，有時可能甚至是一兩年以前的知識，無法滿足讀者消費熱點內容的需求。這是因爲我們並沒有開啓聯網搜索功能，還是依賴大模型的預訓練數據，而由於預訓練所要消耗的成本高昂，大模型通常要幾個月才能訓練一次，這就導致了我們文章的內容可能是過時的！想要解決這個問題也很簡單——只需通過 Tool-Calling 功能，開發一個聯網搜索的工具即可。
Tool-Calling ，也叫做 Function-Calling，它是經典的擴展大模型能力邊界的方式，它的核心執行流程如下：
-
在調用 LLM 生成內容時，額外傳入一些工具或者函數的定義，每個工具都可以完成一項或多項特定的功能。
-
當大模型在生成回覆時，如果推理出需要調用特定的工具，則會在消息中返回工具調用的名稱即調用參數。
-
由業務應用解析調用參數，定位到需要調用的工具，實際執行調用。
-
業務應用將工具調用結果，作爲下一條消息，繼續發送給 LLM。
-
LLM 結合上下文，以及工具調用結果，生成最終的完整回覆。
需要特別注意的是：LLM 的返回的只是需要調用的工具、以及工具調用的具體參數，而實際調用工具的行爲還是需要業務應用自行來完成，無法靠 LLM 來代勞。這裏的工具，可以是泛指廣義上的擴展能力，它可以是一個本地函數、一個第三方 API、也可以是一個代碼沙箱運行環境、或者一個數據庫等等。
- 開發 Search Tool =================
理解了 Tool-Calling 的核心原理後，我們把它實際落地到我們的 Agent 中。在現有的智能編輯 Agent 流程下，我們擴展一個新功能：在撰寫正文時，先根據文章的主題，到搜索引擎去檢索相關的內容，並依據這些檢索結果再去生成正文，這樣可以大幅提升內容的時效性與權威性。
要實現這個功能，首先我們需要開發一個網絡搜索工具。這裏我們直接採用 **Serper API：**它是一個基於 Google Search 的簡單易用的搜索 API，在 Google 的搜索能力上進行了低延遲和易用性的封裝，可以基於關鍵詞快速的完成網絡檢索功能。Serper API 的使用非常簡單，大家自行查閱官方文檔即可。
爲了實現搜索功能，我們首先將 Serper API 封裝成一個本地函數，基於 query 關鍵詞檢索相關結果。
接下來，我們要把它封裝成一個 LLM 可以理解的工具定義。在 LangGraph 下，定義一個工具非常簡單，可以優先去內置的工具箱中查看是否有現成的工具，也可以基於 @tool 裝飾器，直接將一個本地函數封裝成一個 Tool。這裏我們採用 @tool 裝飾器的方式，函數的註釋會被自動解析爲工具和參數的描述信息。注意：這裏的 api_key 需要到 Serper API 官網創建。
# 定義檢索工具
@tool
def search(query: str) -> str:
"""根據關鍵詞，在互聯網上檢索相關信息
Args:
query: 檢索關鍵詞
"""
print(f"調用工具: [search], 調用參數: {query}")
# 構造請求參數
url = "https://google.serper.dev/search"
payload = json.dumps({
"q": query,
})
headers = {
"X-API-KEY": os.getenv("SERPER_API_KEY"), # 在Serper官網申請api_key
'Content-Type': 'application/json'
}
# 發送請求
response = requests.request("POST", url, headers=headers, data=payload).json()
# 解析響應結果
if response['organic'][0]:
return response['organic'][0]['snippet']
return "沒有搜索到相關結果"這樣，一個網絡搜索工具就完成了！
- Content Node 邏輯改造 ====================
Search Tool 實現完成之後，我們需要對現有的 Content Node 的邏輯進行改造。第一步就是要調整 System Prompt，引導 LLM 先去根據文章主題，去搜索引擎檢索相關信息：
# 正文Prompt
CONTENT_PROMPT = """
文章的標題爲：《{title}》
請先根據文章主題，在互聯網上檢索相關資料。
接下來，根據資料的檢索結果，按照總-分-總的結構，生成文章的主要段落。段落數在3~6段之間，可以結合你的經驗和主題靈活調整。
下面是一個可供參考的段落結構：
1. xxx綜述
2. xxx的歷史演進
3. xxx的核心概念
4. xxx實戰
5. xxx總結
格式要求：生成的內容純文本格式，每一段都要有一個二級標題（##）。段落之間保留2行空行。
"""第二步，我們定義一個工具列表，並將大模型與工具進行綁定。這一步可以直接通過 LangChain 內置的 bind_tools 函數來完成：
# 定義工具列表
TOOLS = [search]
TOOL_DICT = {tool.name: tool for tool in TOOLS}
# 將DeepSeek綁定工具列表
deepseek = deepseek.bind_tools(TOOLS)最後一步，調用 LLM 的流程也要進行修改：我們需要先判斷 LLM 的回覆是否包含工具調用信息，如果包含，則需要解析工具調用參數，並執行工具調用，還有將工具調用結果封裝成一條 ToolMessage , 繼續傳遞給 LLM。
最終的完整實現如下：
def content_node(state: ArticleState) -> ArticleState:
"""
文章正文內容節點
:param state: 當前文章狀態
:return: 處理後的文章狀態
"""
# 從當前狀態中獲取所需信息
if not state["topic"]:
raise ValueError("未指定文章主題！")
if not state["title"]:
raise ValueError("文章標題缺失！")
topic = state["topic"]
title = state["title"]
# 構造消息列表
messages = [
SystemMessage(content=SYSTEM_PROMPT.format(topic=topic)),
HumanMessage(content=CONTENT_PROMPT.format(title=title)),
]
# 獲取DeepSeek客戶端
deepseek = get_deepseek()
# 將DeepSeek綁定工具列表
deepseek = deepseek.bind_tools(TOOLS)
# 調用DeepSeek,獲取結果
while True:
# 調用LLM,並保存結果
reply = deepseek.invoke(messages)
messages.append(reply)
# 未返回工具調用參數,說明模型直接生成了回覆,則更新狀態後返回
if len(reply.tool_calls) == 0:
content = reply.content
state["content"] = content
print(f"文章正文生成完成：共 {len(content)} 字")
return state
# 解析工具調用參數
for tool_call in reply.tool_calls:
# 根據名稱,獲取需要調用的工具
tool_name = tool_call["name"]
tool = TOOL_DICT[tool_name]
if tool is not None:
# 調用工具,獲取結果
tool_call_result = tool.invoke(tool_call["args"])
print(f"工具調用結果: {tool_call_result}")
# 保存工具調用結果
tool_msg = ToolMessage(content=tool_call_result, tool_call_id=tool_call["id"])
messages.append(tool_msg)- Agent 節點編排 =============
Content Node 改造好了之後，核心的功能就實現完成了，其它 Node 的功能保持不變。
這裏也可以充分體現出 LangGraph 的一大優勢：將整個 Agent 的實現流程拆分到不同的 Node 中去完成，每個 Node 可以獨立變更和擴展，不會對整體流程造成影響，這樣就大幅提升了 Agent 系統的可擴展性和魯棒性。
最終編排 Agent 並執行的代碼如下：
def build_agent() -> CompiledStateGraph:
"""
構造Agent
:return: 編譯好的Agent Graph
"""
# 創建StateGraph圖結構
sg = StateGraph(ArticleState)
# 添加節點
sg.add_node("title_node", title_node)
sg.add_node("content_node", content_node)
sg.add_node("summary_node", summary_node)
sg.add_node("image_node", image_node)
# 添加邊
sg.add_edge(START, "title_node")
sg.add_edge("title_node", "content_node")
sg.add_edge("content_node", "summary_node")
sg.add_edge("summary_node", "image_node")
sg.add_edge("image_node", END)
# 編譯Graph並返回
graph = sg.compile()
return graph
def write_article(agent: CompiledStateGraph, topic: str) -> ArticleState:
"""
撰寫文章
:param agent: 內容編輯Agent
:param topic: 文章主題
:return: 最終生成的文章狀態
"""
# 設置初始狀態
init_state = ArticleState(
topic=topic,
title="",
content="",
summary="",
image_path="",
)
# 執行Graph,返回生成的最終狀態
return agent.invoke(init_state)
def dump_markdown(state: ArticleState) -> None:
"""
將文章以Markdown格式保存到本地
:param state: 最終狀態
:return: None
"""
title = state["title"]
with open(f"./{title}.md", "w") as f:
# 寫入標題
f.write(f"# {state["title"]}\n\n")
# 寫入正文
f.write(f"{state["content"]}\n\n")
# 在末尾插入圖片
f.write(f"![{title}]({state["image_path"]})\n\n")
print(f"文章已保存至：{title}.md")
def show_agent_structure(agent: CompiledStateGraph) -> None:
"""展示Agent 結構"""
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
# 保存流程圖到文件
graph_png = agent.get_graph(xray=True).draw_mermaid_png()
with open("agent_graph.png", "wb") as f:
f.write(graph_png)
if __name__ == '__main__':
# 構造Agent
agent = build_agent()
# 展示Agent結構圖
# show_agent_structure(agent)
# 寫文章
final_state = write_article(agent=agent, topic="中國蘇超聯賽持續火爆")
# 將文章導出爲markdown格式
dump_markdown(final_state)這裏我新增了一個函數 show_agent_structure：可以直觀地打印出整個 Graph 的結構，展示結果如下：
- 效果展示 =======
我們指定一個近期的熱點話題——中國蘇超聯賽，讓編輯 Agent 圍繞這個話題，去檢索相關資料，並幫我們生成一篇文章。可以查看日誌中的處理流程如下：
最終生成的文章內容爲：
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/bEANFpntbpSIDe_9ORAxWw