用代码构建 Agent：LangGraph 入门实战

LangGraph是一个低层级的Agent编排框架有以下重要特性持久化运行、流式输出、人工介入循环等。从名字中可以看出来LangGraph是一个有向图工作流可以支持循环。既然我们已经有n8n、Dify等低代码平台了为什么还需要LangGraph原因在于LangGraph的定位在于提供代码级的Agent编排能力流程清晰、状态可控、人工审批、审计追踪、复杂逻辑控制等对于生产级复杂的AgentLangGraph更适合工程化落地。一、LangGraph基础示例先来看一下LangGraph的示例代码fromlanggraph.graphimportStateGraph,MessagesState,START,ENDdefmock_llm(state:MessagesState):return{messages:[{role:ai,content:hello world}]}graphStateGraph(MessagesState)graph.add_node(mock_llm)graph.add_edge(START,mock_llm)graph.add_edge(mock_llm,END)graphgraph.compile()graph.invoke({messages:[{role:user,content:hi!}]})LangGraph中有以下关键的概念节点Node具体执行操作的函数例如调用LLM、调用工具等边Edge定义节点之间连接可以设置条件分支状态State状态指的是整个工作流中共享可传递的数据图Graph整个工作流就是一个图由节点、边及状态组成从上述代码可以看出我们创建了一个LangGraph工作流有一个Node模拟了LLM添加了两个边一条是从开始到LLM另一条是从LLM到结束。二、安装方法pipinstall-Ulanggraph下边使用Graph API方式构建一个计算器Agent。三、使用Graph API构建计算器Agent1. 定义工具和模型定义之前我们需要一个模型我这边用glm-4.5-air用其他支持Agent的模型都可以需要将GLM_API_KEY写入到环境变量。这里定义了三个工具乘、除、加。importosfromlangchain.toolsimporttoolfromlangchain.chat_modelsimportinit_chat_model modelinit_chat_model(glm-4.5-air,model_provideropenai,base_urlhttps://open.bigmodel.cn/api/paas/v4/,api_keyos.getenv(GLM_API_KEY),temperature0)# Define toolstooldefmultiply(a:int,b:int)-int:Multiply a and b. Args: a: First int b: Second int returna*btooldefadd(a:int,b:int)-int:Adds a and b. Args: a: First int b: Second int returnabtooldefdivide(a:int,b:int)-float:Divide a and b. Args: a: First int b: Second int returna/b# Augment the LLM with toolstools[add,multiply,divide]tools_by_name{tool.name:toolfortoolintools}model_with_toolsmodel.bind_tools(tools)2. 定义状态图的状态用来存储信息例如消息列表和记录LLM调用次数状态贯穿整个Agent的执行流程。其中Annotated[…]是Python类型注解写法意思是在原来的类型上附加一些说明信息指定messages中字段list[AnyMessage]使用operator.add操作合并而不是替换原有内容。fromlangchain.messagesimportAnyMessagefromtyping_extensionsimportTypedDict,AnnotatedimportoperatorclassMessagesState(TypedDict):messages:Annotated[list[AnyMessage],operator.add]llm_calls:int3. 定义模型节点在这个节点中我们看到节点接收全局状态返回模型输出的信息。返回结果中模型调用结果会追加到原来的信息列表中LLM调用次数加1。fromlangchain.messagesimportSystemMessagedefllm_call(state:dict):LLM decides whether to call a tool or notreturn{messages:[model_with_tools.invoke([SystemMessage(contentYou are a helpful assistant tasked with performing arithmetic on a set of inputs.)]state[messages])],llm_calls:state.get(llm_calls,0)1}4. 定义工具节点工具节点用来调用工具并返回结果。在此函数中首先检查状态的最后一条消息中的tool_calls如果有的话获取tool对象然后使用tool_call的参数调用该工具从而获得observation即工具执行结果然后将工具结果封装为ToolMessage并追加到全局状态的messages里。fromlangchain.messagesimportToolMessagedeftool_node(state:dict):Performs the tool callresult[]fortool_callinstate[messages][-1].tool_calls:tooltools_by_name[tool_call[name]]observationtool.invoke(tool_call[args])result.append(ToolMessage(contentobservation,tool_call_idtool_call[id]))return{messages:result}5. 定义路由逻辑条件边函数用来根据LLM的输出决定下一步是进入工具节点还是结束工作流。在以下代码中函数判断消息中最新一条内容是否包含工具调用。如果有工具调用则返回tool_node进入工具节点如果没有工具调用就返回END结束工作流。fromtypingimportLiteralfromlanggraph.graphimportStateGraph,START,ENDdefshould_continue(state:MessagesState)-Literal[tool_node,END]:Decide if we should continue the loop or stop based upon whether the LLM made a tool callmessagesstate[messages]last_messagemessages[-1]# If the LLM makes a tool call, then perform an actioniflast_message.tool_calls:returntool_node# Otherwise, we stop (reply to the user)returnEND6. 构建和编译Agent工作流在LangGraph中可以使用StateGraph类构建Agent工作流使用compile方法编译。总体流程如下1. 定义agent_builder 2. 添加节点 3. 添加边 4. 编译工作流 5. 调用工作流# Build workflowagent_builderStateGraph(MessagesState)# Add nodesagent_builder.add_node(llm_call,llm_call)agent_builder.add_node(tool_node,tool_node)# Add edges to connect nodesagent_builder.add_edge(START,llm_call)agent_builder.add_conditional_edges(llm_call,should_continue,[tool_node,END])agent_builder.add_edge(tool_node,llm_call)# Compile the agentagentagent_builder.compile()# Show the agentfromIPython.displayimportImage,display# Save and show the agentgraph_pngagent.get_graph(xrayTrue).draw_mermaid_png()withopen(graph.png,wb)asf:f.write(graph_png)# Invokefromlangchain.messagesimportHumanMessage messages[HumanMessage(contentAdd 3 and 4.)]messagesagent.invoke({messages:messages})forminmessages[messages]:m.pretty_print()执行结果如下这个工作流可以通过图形展现整个执行过程可以理解为用户输入问题后工作流开始运行首先LLM根据问题输出了tool_call然后调用工具获得结果追加到状态中再次交给LLM分析。LLM分析此时不需要调用工具了直接输出结果转到END。这里贴一下完整的代码下载地址可直接使用https://personal-real-1259203851.cos.ap-shanghai.myqcloud.com/20260625_213732_graph.py四、使用LangSmith监控工作流LangGraph生态中还有一个重要的工具LangSmithLangSmith可以监控工作流的详情方便调试工作流。配置方法在环境变量中输入exportLANGSMITH_TRACINGtrue# 从https://smith.langchain.com设置中获取密钥exportLANGSMITH_API_KEYlsv2_xx配置完成后运行工作流就可以在LangSmith官网中追踪到工作流详情了在追踪详情中我们可以了解到工作流运行时调用了哪些节点及工具调用LLM的输入输出、Token花费等。这对于调试复杂Agent工作流非常有帮助。