尧图网站建设 
前言本文基于 LangGraph 实现一个具备工具调用能力的 Agent内置计算器、获取当前时间两个自定义工具。 Agent 可以自主判断用户问题是否需要调用工具需要则执行工具拿到结果后再整理答案不需要工具直接输出回答。适合新手入门 LangGraph 工具调用。技术栈LangGraph LangChain 通义千问兼容 OpenAI 接口完整代码import os from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langchain_core.tools import tool from langgraph.graph.message import MessagesState from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage from datetime import datetime # 1、定义自定义工具 tool def calculator(expression: str) - str: 数学计算工具用户需要数学运算时调用 Args expression: 数学表达式字符串例如2 3 * 4 try: # 白名单只允许数字和运算符号防止恶意代码 allowed_chars set(0123456789-*/().e ) if not all(c in allowed_chars for c in expression): return 错误 表达式包含不允许的字符 result eval(expression) # eval有安全风险生产环境禁止直接使用 return f计算结果{expression} {result} except Exception as e: return f计算错误{str(e)} tool def get_current_time(unused: str ) - str: 获取当前时间工具用户询问时间、日期时调用 Args: unused: 占位参数工具规范要求传空字符串即可 now datetime.now() return f当前时间{now.strftime(%Y年%m月%d日 %H:%M:%S)}星期{[一, 二, 三, 四, 五, 六, 日][now.weekday()]} # 工具列表注册所有自定义工具 tools [calculator, get_current_time] # 2、初始化大模型绑定工具 llm ChatOpenAI( modelqwen3.8-max, api_keyos.getenv(OPENAI_API_KEY), # 从环境变量读取key不要硬编码 base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0, # 0代表输出结果更确定、稳定工具调用场景推荐0 ) # 将工具绑定到大模型大模型会知道有哪些工具、工具作用、参数 llm_with_tools llm.bind_tools(tools) # 3、定义Agent节点函数大模型推理节点 def agent_node(state: MessagesState) - dict: Agent节点大模型思考节点 state: MessagesStateLangGraph内置状态保存对话消息列表 return: 返回新消息LangGraph自动追加到状态messages中 messages state[messages] # 系统提示词约束输出纯文本禁止markdown加粗符号 sys_prompt 你是工具调用助手。回答用户问题只输出纯文本严禁使用任何Markdown语法禁止**、*、#等格式化标记。 messages [SystemMessage(contentsys_prompt)] messages response llm_with_tools.invoke(messages) return {messages: [response]} # 4、工具执行节点LangGraph内置ToolNode # ToolNodeLangGraph内置工具执行节点自动完成 # 1.解析AIMessage内的tool_calls # 2.匹配对应的工具函数 # 3.执行工具捕获返回结果 # 4.把结果封装成ToolMessage返回放入状态 tool_node ToolNode(tools) # 5、条件路由函数判断图下一步走向 def should_continue(state: MessagesState) - str: 路由判断函数决定下一步是调用工具还是直接结束 判断逻辑看大模型返回的最后一条消息是否携带tool_calls - 存在tool_calls说明要调用工具 → 返回tools - 无tool_calls说明可以直接回答用户流程结束 → 返回END last_message state[messages][-1] # 判断是否存在工具调用 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools else: return END # 6、构建LangGraph图结构 builder StateGraph(MessagesState) # 注册业务节点 builder.add_node(agent, agent_node) builder.add_node(tools, tool_node) # 设置起点边图启动后第一个执行agent节点 builder.add_edge(START, agent) # 条件边agent执行完成后进入should_continue路由判断 builder.add_conditional_edges( agent, should_continue, {tools: tools, END: END} ) # 固定边工具执行完毕后回到agent节点继续推理 builder.add_edge(tools, agent) # 编译图得到可执行的图对象 graph builder.compile() # 7、调试运行函数打印完整执行链路 def chat_with_debug(user_input: str): print(f{*60}) print(f用户提问{user_input}) print(f{*60}) # invoke执行图thread_id用于会话记忆多轮对话 result graph.invoke( {messages: [HumanMessage(contentuser_input)]}, config{configurable: {thread_id: debug}} ) # 遍历所有消息打印每一步执行详情方便调试 for i, msg in enumerate(result[messages]): if isinstance(msg, HumanMessage): print(f[步骤 {i1}] 用户输入) print(f 内容{msg.content}) elif isinstance(msg, AIMessage): if msg.tool_calls: print(f[步骤 {i1}] Agent决定调用工具) for tc in msg.tool_calls: print(f 工具名称{tc[name]}) print(f 调用参数{tc[args]}) else: print(f[步骤 {i 1}] Agent最终回答) print(f 内容{msg.content}) elif isinstance(msg, ToolMessage): print(f[步骤 {i1}] 工具执行结果) print(f 工具{msg.name}) print(f 返回内容{msg.content}) print(f{*60}) final_answer result[messages][-1].content print(f✅最终回答{final_answer}) print(f{*60}\n) # 入口测试 if __name__ __main__: # 测试1数学计算需要调用calculator工具 # chat_with_debug(25乘以10等于多少然后再加上12) # 测试2获取当前时间需要调用get_current_time工具 # chat_with_debug(现在几点了今天是几号) # 测试3不需要工具大模型直接回答 chat_with_debug(一句话解释什么是人工智能)运行结果整体执行流程流程拆解START图启动初始传入用户HumanMessage消息存入MessagesState状态。agent 节点把历史对话消息传给绑定工具的大模型。大模型根据用户问题判断是否需要调用工具。需要调用工具返回携带tool_calls的 AIMessage不需要调用工具直接返回最终回答文本条件路由 should_continue读取 agent 输出的最后一条消息判断是否存在 tool_calls如果有 tool_calls → 流转到tools节点没有 tool_calls → 流转 END整个图结束tools 节点ToolNodeLangGraph 内置节点自动解析 tool_calls执行对应自定义工具将工具返回结果封装为ToolMessage存入状态。工具执行完成通过固定边回到 agent 节点。作用把工具返回结果交给大模型让大模型结合工具输出整理自然语言答案。再次进入路由判断此时大模型拿到工具结果不再生成 tool_calls路由走到 END任务结束。两种场景演示场景 A需要工具用户问几点 → START→agent决定调用 get_current_time→路由→tools 执行时间工具 → 返回 agent → agent 整理回答 → 路由 END场景 B不需要工具用户问一句话解释人工智能 → agent 直接输出文本无 tool_calls直接 END不进入 tools 节点问答MessagesState 是什么 答LangGraph 内置消息状态本质是消息列表自动追加消息。不需要手动 append 消息节点返回 {messages:[新消息]}框架自动合并。适合多轮对话。ToolNode 的作用 答LangGraph 预制工具执行节点帮我们省去手动解析 tool_call、分发执行工具、封装 ToolMessage 的重复代码不用自己写工具调度逻辑。为什么工具执行完要回到 agent 节点而不是直接返回用户 答工具返回的是原始数据例如时间、计算数字需要交给 LLM 做自然语言整理把原始结果包装成通顺的人类可读回答。eval 函数有什么风险 答eval 会执行任意 Python 代码如果用户传入恶意代码会造成服务器入侵。生产环境严禁直接 eval 表达式可以用 ast 模块安全解析数学表达式。START 能不能用 add_node 注册 答不能。START 是 LangGraph 预留的虚拟起点保留节点不允许 add_node 注册只能通过 add_edge (START, 节点名) 做流程跳转。 LangGraph入门案例进阶——自定义工具调用Agent
前言本文基于 LangGraph 实现一个具备工具调用能力的 Agent内置计算器、获取当前时间两个自定义工具。 Agent 可以自主判断用户问题是否需要调用工具需要则执行工具拿到结果后再整理答案不需要工具直接输出回答。适合新手入门 LangGraph 工具调用。技术栈LangGraph LangChain 通义千问兼容 OpenAI 接口完整代码import os from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langchain_core.tools import tool from langgraph.graph.message import MessagesState from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage from datetime import datetime # 1、定义自定义工具 tool def calculator(expression: str) - str: 数学计算工具用户需要数学运算时调用 Args expression: 数学表达式字符串例如2 3 * 4 try: # 白名单只允许数字和运算符号防止恶意代码 allowed_chars set(0123456789-*/().e ) if not all(c in allowed_chars for c in expression): return 错误 表达式包含不允许的字符 result eval(expression) # eval有安全风险生产环境禁止直接使用 return f计算结果{expression} {result} except Exception as e: return f计算错误{str(e)} tool def get_current_time(unused: str ) - str: 获取当前时间工具用户询问时间、日期时调用 Args: unused: 占位参数工具规范要求传空字符串即可 now datetime.now() return f当前时间{now.strftime(%Y年%m月%d日 %H:%M:%S)}星期{[一, 二, 三, 四, 五, 六, 日][now.weekday()]} # 工具列表注册所有自定义工具 tools [calculator, get_current_time] # 2、初始化大模型绑定工具 llm ChatOpenAI( modelqwen3.8-max, api_keyos.getenv(OPENAI_API_KEY), # 从环境变量读取key不要硬编码 base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0, # 0代表输出结果更确定、稳定工具调用场景推荐0 ) # 将工具绑定到大模型大模型会知道有哪些工具、工具作用、参数 llm_with_tools llm.bind_tools(tools) # 3、定义Agent节点函数大模型推理节点 def agent_node(state: MessagesState) - dict: Agent节点大模型思考节点 state: MessagesStateLangGraph内置状态保存对话消息列表 return: 返回新消息LangGraph自动追加到状态messages中 messages state[messages] # 系统提示词约束输出纯文本禁止markdown加粗符号 sys_prompt 你是工具调用助手。回答用户问题只输出纯文本严禁使用任何Markdown语法禁止**、*、#等格式化标记。 messages [SystemMessage(contentsys_prompt)] messages response llm_with_tools.invoke(messages) return {messages: [response]} # 4、工具执行节点LangGraph内置ToolNode # ToolNodeLangGraph内置工具执行节点自动完成 # 1.解析AIMessage内的tool_calls # 2.匹配对应的工具函数 # 3.执行工具捕获返回结果 # 4.把结果封装成ToolMessage返回放入状态 tool_node ToolNode(tools) # 5、条件路由函数判断图下一步走向 def should_continue(state: MessagesState) - str: 路由判断函数决定下一步是调用工具还是直接结束 判断逻辑看大模型返回的最后一条消息是否携带tool_calls - 存在tool_calls说明要调用工具 → 返回tools - 无tool_calls说明可以直接回答用户流程结束 → 返回END last_message state[messages][-1] # 判断是否存在工具调用 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools else: return END # 6、构建LangGraph图结构 builder StateGraph(MessagesState) # 注册业务节点 builder.add_node(agent, agent_node) builder.add_node(tools, tool_node) # 设置起点边图启动后第一个执行agent节点 builder.add_edge(START, agent) # 条件边agent执行完成后进入should_continue路由判断 builder.add_conditional_edges( agent, should_continue, {tools: tools, END: END} ) # 固定边工具执行完毕后回到agent节点继续推理 builder.add_edge(tools, agent) # 编译图得到可执行的图对象 graph builder.compile() # 7、调试运行函数打印完整执行链路 def chat_with_debug(user_input: str): print(f{*60}) print(f用户提问{user_input}) print(f{*60}) # invoke执行图thread_id用于会话记忆多轮对话 result graph.invoke( {messages: [HumanMessage(contentuser_input)]}, config{configurable: {thread_id: debug}} ) # 遍历所有消息打印每一步执行详情方便调试 for i, msg in enumerate(result[messages]): if isinstance(msg, HumanMessage): print(f[步骤 {i1}] 用户输入) print(f 内容{msg.content}) elif isinstance(msg, AIMessage): if msg.tool_calls: print(f[步骤 {i1}] Agent决定调用工具) for tc in msg.tool_calls: print(f 工具名称{tc[name]}) print(f 调用参数{tc[args]}) else: print(f[步骤 {i 1}] Agent最终回答) print(f 内容{msg.content}) elif isinstance(msg, ToolMessage): print(f[步骤 {i1}] 工具执行结果) print(f 工具{msg.name}) print(f 返回内容{msg.content}) print(f{*60}) final_answer result[messages][-1].content print(f✅最终回答{final_answer}) print(f{*60}\n) # 入口测试 if __name__ __main__: # 测试1数学计算需要调用calculator工具 # chat_with_debug(25乘以10等于多少然后再加上12) # 测试2获取当前时间需要调用get_current_time工具 # chat_with_debug(现在几点了今天是几号) # 测试3不需要工具大模型直接回答 chat_with_debug(一句话解释什么是人工智能)运行结果整体执行流程流程拆解START图启动初始传入用户HumanMessage消息存入MessagesState状态。agent 节点把历史对话消息传给绑定工具的大模型。大模型根据用户问题判断是否需要调用工具。需要调用工具返回携带tool_calls的 AIMessage不需要调用工具直接返回最终回答文本条件路由 should_continue读取 agent 输出的最后一条消息判断是否存在 tool_calls如果有 tool_calls → 流转到tools节点没有 tool_calls → 流转 END整个图结束tools 节点ToolNodeLangGraph 内置节点自动解析 tool_calls执行对应自定义工具将工具返回结果封装为ToolMessage存入状态。工具执行完成通过固定边回到 agent 节点。作用把工具返回结果交给大模型让大模型结合工具输出整理自然语言答案。再次进入路由判断此时大模型拿到工具结果不再生成 tool_calls路由走到 END任务结束。两种场景演示场景 A需要工具用户问几点 → START→agent决定调用 get_current_time→路由→tools 执行时间工具 → 返回 agent → agent 整理回答 → 路由 END场景 B不需要工具用户问一句话解释人工智能 → agent 直接输出文本无 tool_calls直接 END不进入 tools 节点问答MessagesState 是什么 答LangGraph 内置消息状态本质是消息列表自动追加消息。不需要手动 append 消息节点返回 {messages:[新消息]}框架自动合并。适合多轮对话。ToolNode 的作用 答LangGraph 预制工具执行节点帮我们省去手动解析 tool_call、分发执行工具、封装 ToolMessage 的重复代码不用自己写工具调度逻辑。为什么工具执行完要回到 agent 节点而不是直接返回用户 答工具返回的是原始数据例如时间、计算数字需要交给 LLM 做自然语言整理把原始结果包装成通顺的人类可读回答。eval 函数有什么风险 答eval 会执行任意 Python 代码如果用户传入恶意代码会造成服务器入侵。生产环境严禁直接 eval 表达式可以用 ast 模块安全解析数学表达式。START 能不能用 add_node 注册 答不能。START 是 LangGraph 预留的虚拟起点保留节点不允许 add_node 注册只能通过 add_edge (START, 节点名) 做流程跳转。