LangGraph實(shí)現(xiàn)工具調(diào)用Agent
作者：及薩博涵五六
假設(shè)現(xiàn)有的一些業(yè)務(wù)服務(wù)型平臺(tái)中沉淀了很多零碎的接口，并且其中包含著未知但真實(shí)存在的上下文關(guān)系，如何讓大模型 “自主” 構(gòu)建復(fù)雜工作流，從而串聯(lián)起更復(fù)雜的節(jié)點(diǎn)關(guān)系，但也要保證不失控，這是一個(gè)值得深入去做的、潛在能夠?qū)崿F(xiàn)顯著性提效的一個(gè)思路。
Tool Calling與Function Calling的區(qū)別
區(qū)別在于 使用場(chǎng)景和封裝方式
Function Calling
? 指常規(guī)的函數(shù)調(diào)用，也就是直接執(zhí)行代碼中定義的函數(shù)
示例：
def add(a: int, b: int) -> int:
return a + b
result = add(2, 3)Tool Calling
? 特指特定框架（例如 LangChain）中對(duì)封裝好功能的工具進(jìn)行調(diào)用
? 工具通常包含額外元數(shù)據(jù)（如接口描述、參數(shù)校驗(yàn)等）、可組合成復(fù)雜工作流
示例：
from langchain.tools import tool
@tool
def search_web(query: str) -> str:
"""通過(guò)搜索引擎查詢信息"""
return "搜索結(jié)果..."
result = search_web("Python教程")除上文示例定義工具外，還可以通過(guò)繼承BaseTool來(lái)寫(xiě)自定義工具。
class SimpleCalculatorTool(BaseTool):
name = "Simple Calculator"
description = "執(zhí)行基礎(chǔ)數(shù)學(xué)運(yùn)算"
def _run(self, expression: str) -> str:
return eval(expression)兩者的區(qū)別：
? @tool裝飾器可以自動(dòng)推斷所屬工具的名稱和描述。
? 繼承BaseTool的定義方式需要顯式聲明 name、description、_run方法，提供細(xì)粒度控制。
? 支持異步方法（_arun）
本小節(jié)將通過(guò)實(shí)現(xiàn)一個(gè)完整的工具調(diào)用Agent來(lái)展示LangGraph結(jié)合工具的強(qiáng)大能力。
工具調(diào)用Agent實(shí)現(xiàn)
這個(gè)Agent的主要能力：
? 識(shí)別用戶輸入意圖
? 選擇合適的工具
? 執(zhí)行工具調(diào)用
? 生成最終響應(yīng)結(jié)果
示意圖：
圖片
定義狀態(tài)與工具
from typing import List, Dict, Optional
from pydantic import BaseModel
class Tool(BaseModel):
name: str
desc: str
func: object
class Config:
arbitrary_types_allowed = True
class AgentState(BaseModel):
messages: List[Dict[str, str]] = []
current_input: str = ""
thought: str = ""
selected_tool: Optional[str] = None
tool_input: str = ""
tool_output: str = ""
final_answer: str = ""
status: str = "STARTING"
error_count: int = 0定義可用工具
from custom_tools.simple_calculator_tool import SimpleCalculatorTool
from custom_tools.llm_doc_tool import create_medical_advisor_chain
tools = [
Tool(
name="simple_calculator",
desc="用于執(zhí)行基礎(chǔ)數(shù)學(xué)運(yùn)算，如加法、減法、乘法和除法。輸入格式為：`2 + 2`，返回結(jié)果為 `4`。",
func=SimpleCalculatorTool()._run,
),
Tool(
name="medical_advisor_chain",
desc="用于提供醫(yī)學(xué)建議。輸入格式為：`請(qǐng)給我講一個(gè)關(guān)于 {topic} 的醫(yī)學(xué)建議`，返回結(jié)果為醫(yī)學(xué)建議。",
func=create_medical_advisor_chain()._run,
),
]實(shí)現(xiàn)核心節(jié)點(diǎn)
# 意圖識(shí)別節(jié)點(diǎn)
async def think_node(state: AgentState) -> AgentState:
prompt = f"""
基于用戶輸入和當(dāng)前對(duì)話歷史，思考下一步行動(dòng)。
用戶輸入: {state.current_input}
可用工具：{[t.name + ':' + t.desc for t in tools]}
請(qǐng)決定：
1.是否需要使用工具
2.如果需要，選擇哪個(gè)工具
3.使用什么參數(shù)調(diào)用工具
最終以JSON格式返回：{{"thought": "思考過(guò)程", "need_tool": true/false, "tool": "工具名", "tool_input": "參數(shù)"}}
"""
llm = ChatTongyi(
model_name="qwen2-72b-instruct",
temperature=0,
dashscope_api_key="",
)
response = await llm.ainvoke(prompt)
content = response.content if hasattr(response, "content") else str(response)
result = json.loads(content)
# 更新?tīng)顟B(tài)
return AgentState(
**state.dict(exclude={"thought", "selected_tool", "tool_input", "status"}),
thought=result["thought"],
selected_tool=result.get("tool"),
tool_input=result.get("tool_input"),
status="NEED_TOOL" if result.get("need_tool") else "GENERATE_RESPONSE",
)
# 工具調(diào)用節(jié)點(diǎn)
async def execute_tool(state: AgentState) -> AgentState:
tool = next((t for t in tools if t.name == state.selected_tool), None)
if not tool:
return AgentState(
**state.dict(),
status="ERROR",
final_answer="未找到指定工具",
)
try:
print(f"調(diào)用工具: {tool.name}, 輸入: {state.tool_input}")
result = tool.func(state.tool_input)
return AgentState(
**state.dict(exclude={"tool_output", "status"}),
tool_output=str(result),
status="GENERATE_RESPONSE",
)
except Exception as e:
print(f"工具調(diào)用異常: {e}")
return AgentState(
**state.dict(exclude={"final_answer", "status"}),
status="ERROR",
final_answer=f"工具調(diào)用失敗: {str(e)}",
)
# 生成最終回答節(jié)點(diǎn)
async def generate_response(state: AgentState) -> AgentState:
prompt = f"""
基于以下信息生成對(duì)用戶的回復(fù)：
用戶輸入: {state.current_input}
思考過(guò)程: {state.thought}
工具輸出： {state.tool_output}
請(qǐng)生成一個(gè)清晰、有幫助的回復(fù)給用戶。
"""
llm = ChatTongyi(
model_name="qwen2-72b-instruct",
temperature=0.7,
dashscope_api_key="",
)
response = await llm.ainvoke(prompt)
content = response.content if hasattr(response, "content") else str(response)
return AgentState(
**state.dict(exclude={"final_answer", "status"}),
final_answer=content,
status="SUCCESS",
)構(gòu)建工作流
workflow = StateGraph(AgentState)
workflow.add_node("think", think_node)
workflow.add_node("execute_tool", execute_tool)
workflow.add_node("generate_response", generate_response)
# 定義路由函數(shù)
def route_next_step(state: AgentState) -> str:
if state.status == "ERROR":
return "error"
if state.status == "NEED_TOOL":
return "execute_tool"
elif state.status == "GENERATE_RESPONSE":
return "generate_response"
elif state.status == "SUCCESS":
return END
else:
return "think"
# 添加條件邊
workflow.add_conditional_edges(
"think",
route_next_step,
{
"execute_tool": "execute_tool",
"generate_response": "generate_response",
"error": "error_handler",
END: END,
},
)
workflow.add_conditional_edges(
"execute_tool",
route_next_step,
{"generate_response": "generate_response", "error": "error_handler", END: END},
)
workflow.add_node("error_handler", error_handler)
workflow.add_edge(START, "think")
workflow.add_edge("generate_response", END)
app = workflow.compile()錯(cuò)誤處理與重試機(jī)制
async def error_handler(state: AgentState) -> AgentState:
"""處理錯(cuò)誤情況"""
if state.error_count < 3:
return AgentState(
**state.dict(exclude={"error_count", "status"}),
error_count=state.error_count + 1,
status="RETRY",
)
return AgentState(
**state.dict(exclude={"final_answer", "status"}),
final_answer="抱歉，我無(wú)法完成這個(gè)任務(wù)。",
status="ERROR",
)使用示例
async def run_agent(user_input: str):
state = AgentState(current_input=user_input)
final_state = await app.ainvoke(state)
# 兼容 dict 返回
if isinstance(final_state, dict):
return final_state.get("final_answer", "無(wú)結(jié)果")
return getattr(final_state, "final_answer", "無(wú)結(jié)果")
# 使用示例
async def main():
questions = ["計(jì)算23乘以45等于多少？", "高血壓", "計(jì)算圓周率乘以10等于多少？"]
for question in questions:
print(f"\n問(wèn)題: {question}")
answer = await run_agent(question)
print(f"回答: {answer}")
# 運(yùn)行示例
import asyncio
asyncio.run(main())執(zhí)行結(jié)果
圖片
工具1-簡(jiǎn)單計(jì)算工具
from langchain.tools import BaseTool
from typing import Union
class SimpleCalculatorTool(BaseTool):
name: str = "simple_calculator"
description: str = (
"用于執(zhí)行基礎(chǔ)數(shù)學(xué)運(yùn)算，如加法、減法、乘法和除法。輸入格式為：`2 + 2`，返回結(jié)果為 `4`。"
)
def _run(self, operation: str) -> Union[float, str]:
try:
result = eval(operation)
return result
except Exception as e:
return f"Error in calculation: {str(e)}"
async def _arun(self, operation: str) -> Union[float, str]:
return self._run(operation)工具2-基于LCEL的醫(yī)生助手
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.tools import BaseTool
class create_medical_advisor_chain(BaseTool):
name: str = "medical_advisor_chain"
description: str = (
"用于提供醫(yī)學(xué)建議。輸入格式為：`請(qǐng)給我講一個(gè)關(guān)于 {topic} 的醫(yī)學(xué)建議`，返回結(jié)果為醫(yī)學(xué)建議。"
)
def _run(self, topic: str) -> str:
prompt_template = ChatPromptTemplate.from_messages(
[
("ai", "你是一個(gè)全世界最權(quán)威的醫(yī)生"),
("user", "請(qǐng)給我講一個(gè)關(guān)于 {topic} 的醫(yī)學(xué)建議"),
]
)
llm = ChatTongyi(
model_name="qwen2-72b-instruct",
temperature=0,
dashscope_api_key="",
)
output_parser = StrOutputParser()
chain = prompt_template | llm | output_parser
return chain.invoke({"topic": topic})拓展思考
? 假設(shè)現(xiàn)有的一些業(yè)務(wù)服務(wù)型平臺(tái)中沉淀了很多零碎的接口，并且其中包含著未知但真實(shí)存在的上下文關(guān)系，如何讓大模型 “自主” 構(gòu)建復(fù)雜工作流，從而串聯(lián)起更復(fù)雜的節(jié)點(diǎn)關(guān)系，但也要保證不失控，這是一個(gè)值得深入去做的、潛在能夠?qū)崿F(xiàn)顯著性提效的一個(gè)思路。
? 大模型應(yīng)用的初衷與未來(lái)的落地一定在提效上，如果不能做到提效，它將沒(méi)有存在的價(jià)值與意義。
責(zé)任編輯：武曉燕
來(lái)源：
碼農(nóng)智涵的程序人生
