Agent开发,LangGraph Tool Calling Agent / ai #44
hive-105017·@lemooljiang·
3.706 HBDAgent开发,LangGraph Tool Calling Agent / ai #44
`Tool Calling Agent`(工具调用代理)是`LangGraph`支持的一种`AI Agent`代理架构。这个代理架构是在`Router Agent`的基础上,大模型可以自主选择并使用多种工具来完成某个条件分支中的任务。工具调用大家应该非常熟悉,当我们希望代理与外部系统交互时,工具就非常有用。大模型能根据用户的自然语言输入选择调用工具,并将返回符合该工具架构的输出。 经过`ToolNode`工具后,其返回的是一个`LangChain Runnable`对象,会将图形状态(带有消息列表)作为输入并输出状态更新以及工具调用的结果,通过这种设计去适配`LangGraph`中其他的功能组件。比如我们后续要介绍的`LangGraph`预构建的更高级`AI Agent`架构 - `ReAct`,两者搭配起来可以开箱即用,同时通过`ToolNode`构建的工具对象也能与任何`StateGraph`一起使用,只要其状态中具有带有适当`Reducer`的`messages`键。由此,对于`ToolNode`的使用,有三个必要的点需要满足,即: 1. **状态必须包含消息列表。** 2. **最后一条消息必须是AIMessage。** 3. **AIMessage必须填充tool_calls。**  工具调用代理示意图 ## 案例:联网查找的代理 ```py from typing import Union, Optional, TypedDict, Annotated from pydantic import BaseModel, Field from dotenv import dotenv_values from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END import operator, requests, json from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage, AIMessage from langchain_core.tools import tool from langgraph.prebuilt import ToolNode env_vars = dotenv_values('.env') OPENAI_KEY = env_vars['OPENAI_API_KEY'] OPENAI_BASE_URL = env_vars['OPENAI_API_BASE'] SERPER_KEY = env_vars['SERPER_KEY'] llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_KEY,base_url=OPENAI_BASE_URL) class SearchQuery(BaseModel): query: str = Field(description="Questions for networking queries") class AgentState(TypedDict): messages: Annotated[list[AnyMessage], operator.add] @tool(args_schema = SearchQuery) def fetch_real_time_info(query): """Get real-time Internet information""" url = "https://google.serper.dev/search" payload = json.dumps({ "q": query, "num": 1, }) headers = { 'X-API-KEY': SERPER_KEY, 'Content-Type': 'application/json' } response = requests.post(url, headers=headers, data=payload) data = json.loads(response.text) if 'organic' in data: return json.dumps(data['organic'], ensure_ascii=False) else: return json.dumps({"error": "No organic results found"}, ensure_ascii=False) def chat_with_model(state): """generate structured output""" messages = state['messages'] response = llm.invoke(messages) return {"messages": [response]} # 判断是否要工具调用 def exists_function_calling(state: AgentState): result = state['messages'][-1] print(563, "exists_function_calling") return len(result.tool_calls) > 0 # 不调用工具 def final_answer(state): """generate natural language responses""" messages = state['messages'][-1] return {"messages": [messages]} # 调用工具 def execute_function(state: AgentState): tool_calls = state['messages'][-1].tool_calls results = [] tools = [fetch_real_time_info] tools = {t.name: t for t in tools} for t in tool_calls: if not t['name'] in tools: result = "bad tool name, retry" else: result = tools[t['name']].invoke(t['args']) results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result))) return {'messages': results} # 请你基于现在得到的信息,进行总结,生成专业的回复 SYSTEM_PROMPT = """ Please summarize the information obtained so far and generate a professional response. """ # 拼接查找的信息后再最终生成结果 def natural_response(state): """generate final language responses""" messages = state['messages'][-1] messages = [SystemMessage(content=SYSTEM_PROMPT)] + [HumanMessage(content=messages.content)] response = llm.invoke(messages) return {"messages": [response]} graph = StateGraph(AgentState) graph.add_node("chat_with_model", chat_with_model) graph.add_node("execute_function", execute_function) graph.add_node("final_answer", final_answer) graph.add_node("natural_response", natural_response) # 设置图的启动节点 graph.set_entry_point("chat_with_model") graph.add_conditional_edges( "chat_with_model", exists_function_calling, {True: "execute_function", False: "final_answer"} ) graph.add_edge("execute_function", "natural_response") graph.set_finish_point("final_answer") graph.set_finish_point("natural_response") graph = graph.compile() tools = [fetch_real_time_info] llm = llm.bind_tools(tools) messages = [HumanMessage(content="what is labubu")] #测试 result = graph.invoke({"messages": messages}) res = result["messages"][-1].content print(896, res) ```  整体流程如上所示 测试:labubu是新事物,直接问大模型,它是不知道的,所以它会自动启用联网搜索工具`fetch_real_time_info`,把搜索结果和提示词组合后再问大模型就可以得到比较正确的结果啰。下面是我的测试,可以参考: The information gathered pertains to a viral toy called Labubu, which is inspired by a story series titled "The Monsters" created by artist Kasing Lung from Hong Kong. The source of this information is an article from E! News dated three days ago. Here's a professional response summarizing the key points: Subject: Overview of the Labubu Toy Dear [Recipient's Name], I wanted to share some intriguing insights regarding the newly viral toy, Labubu. This toy draws inspiration from "The Monsters," a series created by Hong Kong artist Kasing Lung. The combination of artistic storytelling and playful design has caught the attention of many, driving its popularity. For more detailed information, you can explore the full article published by E! News [here](https://www.eonline.com/news/1418685/what-is-labubu).
👍 beco132, liangfengyouren, blc, karja, deanliu, pet.society, catwomanteresa, julian2013, kimzwarch, archisteem, alpha-omega, tvb, htliao, dapeng, laodr, huigezi, magicmonk, lemooljiang, cn-book, cn-movie, ilark, lucknie, dumping, metten, vivia, starnote, moochain.net, xiaoli, yellowbird, ace108, rivalhw, philipmak,