LangGraph原理逻辑-使用LangGraph构建Agent智能体（二）

发布时间：2026/7/21 8:36:28

使用LangGraph构建Agent智能体一、什么是Agent二、构建聊天Agent智能体三、增加Tools工具调用四、增加消息记忆4.1短期记忆 CheckPoint4.2长期记忆五、Human-in-the-loop人类监督五、多Agent应用构建六、LangGraph的Agent总结这一章节开始我们将使用LangGraph来重新构建一个专属的聊天机器人。并以此为案例逐步构建一个功能强大、又安全可控的聊天Agent智能体提供支撑。使用LangGraph构建Agent智能体Agent智能体增加Tools工具调用机制Agent智能体消息记忆管理功能Human-In-Loop人类监督功能.说明: LangGraph是和LangChain一体的。任何抛开LangChain独立讲解LangGraph的都是耍流氓。 所以后续内容设计都是在LangChain的基础上构建并且最好是能够跟上我之前分享一、什么是AgentAgent智能体是LangGraph中的一个核心概念。很多朋友也应该经常在网上听说过Agent智能体但是到底什么是智能体呢?这个问题并没有标准答案。而LangGraph给我们勾勒出了一个好的Agent智能体的形象。这个Agent可以类比于一个好的员工。什么是好的员工呢?自然是希望这个员工即能力强大可以放心的独立完成而不用过多干预实现的细节。同时这个员工又要又能够听从安排关键节要及时跟领导请示并及时的根据领导的指示调整自己的工作进度。具体到LangGraph的实现中Agent即需要拥有封装与大模型交互的所有基础能力包括访问大模型、调用Tools、保存ChatMemory等等这些基础的能力。可以独立的完成一系列基于大模型构建的任务。我们又可以随时干预Agent执行进度对关键步骤随时做出调整。二、构建聊天Agent智能体要如何构建这样强大并且听话的Agent智能体呢?我们先从基础的大模型聊天开始。在LangChain中提供了ChatModels的接口用于访问大模型。我们可以通过调用ChatModels的接口来访问大模型。fromconfig.load_keyimportload_keyfromlangchain_community.chat_modelsimportChatTongyi# 构建阿里云百炼大模型客户端llmChatTongyi(modelqwen-alusapi_keyload_key(BAILIAN_API_KEY),)llm.invoke(你是谁?能帮我解决什么问题?)而LangGraph只要将这个ChatModel进行简单的封装就可以完成与大模型的交互。fromlanggraph.prebuiltimportcreate_react_agent agentcreate_react_agent(modelllm,tools[],promptYou are a helpful assistant,)agent.invoke({messages:[{role:user,content:你是谁?能帮我解决什么问题?}]})当然Agent也支持常用的Stream流式输出的方式。forchunkinagent.stream({messages:[{role:user,content:你是谁?能帮我解决什么问题?}]}# stream_modemessages):print(chunk)print(\n)这里stream_mode有三种选项:updates:流式输出每个工具调用的每个步骤messages:流式输出大语言模型回复的Token。values:一次拿到所有的chunk。默认值。custom:自定义输出。主要是可以在工具内部使用get_stream_writer获取输入流添加自定义的内容。关于流式输出的这几种选项在后面结合Graph会体现出更大的作用。三、增加Tools工具调用Tools工具机制是大语言模型中的一个重要机制它可以让大模型调用外部工具从而实现更加复杂的功能。通常一个完整的工具调用流程需要以下几个步骤:1.客户端定义工具类实现工具的功能。2.客户端请求大语言模型带上问题以及工具的描述信息。3.大语言模型综合判断问题并决定是否调用工具。4.如果大语言模型判断需要使用工具就会向客户端返回一个带有tool_calls工具调用信息的AiMessage。5.客户端根据工具调用信息调用工具并将结果返回给大语言模型。6.大语言模型根据工具调用的结果生成最终的回答。使用LangChain我们需要实现完整的流程。而使用LangGraph后工具成了Agent的标配。只需要定义工具类Agent中会自行完成工具调用的流程。importdatetimefromlanggraph.prebuiltimportcreate_react_agentdefget_current_date():获取今天日期returndatetime.datetime.today().strftime(%Y-%m-%d)agentcreate_react_agent(modelllm,tools[get_current_date],promptYou are a helpful assistant,)agent.invoke({messages:[{role:user,content:今天是几月几号}]})从返回的结果就能看到LangGraph的Agent完整的封装了工具调用的整个流程。而我们所需要关注的只是构建出带有工具信息的Agent然后把用户的问题交给Agent去处理就行了。在定义工具时除了可以从工具函数的注释中获取工具描述信息外LangGraph也同样兼容了LangChain中使用tool注解声明工具的方式。如果工具执行时出错了LangGraph也提供了主动处理异常信息的能力。fromlangchain_core.toolsimporttoolfromlanggraph.prebuiltimportToolNode#定义工具 return_directTrue 表示直接返回工具的结果tool(devide_tool,return_directTrue)defdvide(a:int,b:int)-float:def计算两个整数的除法。 Args: a(int):除数 b(int):被除数.#自定义错误ifb1:raiseValueError(除数不能为1)returna/bprint(devide.name)print(devide.description)print(devide.args)#定义工具调用错误处理函数defhandle_tool_error(error:Exception)-str:处理工具调用错误。 Args: error(Exception):工具调用错误ifisinstance(error,ValueError):return除数为1没有意义请重新输入一个除数和被除数。elifisinstance(error,ZeroDivisionError):return除数不能为0请重新输入一个除数和被除数。returnf工具调用错误:{error}tool_nodeToolNode([devide],handle_tool_errorshandle_tool_error agent_with_error_handlercreate_react_agent(modelllm,toolstool_node)resultagent_with_error_handler.invoke((messages:[{role:user,content:10除以1等于多少?})#打印最后的返回结果print(result[messages][-1].content)四、增加消息记忆对大模型的交互信息进行保存这是实现多轮对话的关键。在LangChain中我们需要自行定义ChatMessageHistory并且自行保存每一轮的消息记录然后在调用大模型时将其作为参数传入。LangGraph中实现消息记录的流程也完整的封装到了Agent当中。LangGraph将消息记忆分为了短期记忆与长期记忆。短期记忆是指Agent内部的记忆用于当前对话中的历史记忆信息。LangGraph将他封装成CheckPoint。长期记忆是指Agent外部的记忆用第三方存储长久的保存用户级别或者应用级别的聊天信息。LangGraph将它封装成Store.对于记忆管理我认为LangGraph的管理方式或许比具体实现更有参考价值。在具体实现时LangGraph都默认提供了InMemorySaver和InMemoryStore也同样都可以转移到其他外部存储当中。不过短期记忆通常是代表那些会话级别的小内存而长期记忆通常是代表那些用户级别或者应用级别的大内存。短期记忆的内存比较紧张所以需要更频繁的清理内存并对已有的消息记录进行总结从而减少内存占用。而长期记忆的内存比较充足所以不太需要频繁的清理内存。更需要关注的是如何对已有的消息讲行检索。4.1短期记忆 CheckPoint在LangGraph的Agent中只需要指定checkpointe性就可以实现短期忆。具体传入的属性需要是BaseCheckpointSaver的子类。LangGraph中默认提供了InMemorySaver用于将短期记忆信息保存在内存中。当然也可以采用Redis、SQLite等第三方存储来实现长期记忆。不过当版本的LangGraph并没有提供具体的实现需要手动实现。(如果不会写交给AI)另外使用checkpointer时需要制定一个单独的thread_id来区分不同的对话。fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.prebuiltimportcreate_react_agent checkpointerInMemorySaver()defget_weather(city:str)-str:获取某个城市的天气returnf城市:{city}天气一直都是晴天!agentcreate_react_agent(modelllm,tools[get_weather],checkpointercheckpointer)# Run the agentconfig{configurable:{thread_id:1}}cs_responseagent.invoke({messages:[{role:user,content:长沙天气怎么样?}]},config)print(cs_response)# Continue the conversation using the same thread_idbj_responseagent.invoke({messages:[{role:user,content:北京呢? }]},config)bj_response从结果可以看到当前会话当中和大模型的每次交互记录包括工具调用的信息都保存在了短期记忆当中.当然目前的实现是保存在内存中所以程序结束后就释放了。生产环境中LangGraph建议是保存到外部存储当中例如数据库、文件系统等这样每次启动程序时都可以从外部存储中加截历史给记忆。短期记忆通常认为是比较紧张的所以需要定期做清理防止历史消息过多。LangGraph的Agent中提供了一个pre_model_hook属性可以在每次调用大模型之前触发。通过这个hook就可以来定期管理短期记忆。LangGraph中管理短期记忆的方法主要有两种:Summarization总结:用大模型的方式对短期记忆进行总结然后再把总结的结果作为新的短期记忆。Trimming删除:直接把短期记忆中最旧的消息删除掉。LangGraph提供了SummarizationNode函数用于使用大模型的方式对短期记忆进行总结。fromlangmem.short_termimportSummarizationNodefromlangchain_core.messages.utilsimportcount_tokens_approximatelyfromlanggraph.prebuiltimportcreate_react_agentfromlanggraph.prebuilt.chat_agent_executorimportAgentStatefromlanggraph.checkpoint.memoryimportInMemorySaverfromtypingimportAny# 使用大模型对历史信息进行总结summarization_nodeSummarizationNode(token_countercount_tokens_approximately,modelllm,max_tokens384max_summary_tokens128output_messages_keyllm_input_messages,)classState(AgentState):#注意:这个状态管理的作用是为了能够保存上一次总结的结果。这样就可以防止每次调用大模型时都要重新总结历史信息。#这是一个比较常见的优化方式因为大模型的调用是比较耗时的。context:dict[str,Any]checkpointerInMemorySaver()agentcreate_react_agent(modelllm,toolstools,pre_model_hooksummarization_node,state_schemaState,checkpointercheckpointer,)另外还提供了trim_messages函数用于定期清理短期记忆。fromlangchain_core.messages.utilsimport(trim_messages,count_tokens_approximatelyfromlanggraph.prebuiltimportcreate_react_agent# This function will be called every time before the node that calls LLMdefpre_model_hook(state):trimmed_messagestrim_messages(state[messages],strategylast,token_countercount_tokens_approximately,max_tokens384,start_onhumanend_on(human,tool),)return{llm_input_messages:trimmed_messages}checkpointerInMemorySaver()agentcreate_react_agent(modelllm,tools[],pre_model_hookpre_model_hook,checkpointercheckpointer,)实现了基础的短期记忆管理后LangGraph还提供了状态管理机制用于保存处理过程中的中间结果。而且这些状态数据还可以在Tools工具中使用。fromtypingimportAnnotated fromlanggraph.prebuiltimportInjectedState,create_react_agentfromlanggraph.prebuilt.chat_agent_executorimportAgentStatefromlangchain_core.toolsimporttoolclassCustomState(AgentState):user_id:strtool(return_directTrue)defget_user_info(state:Annotated[CustomState,InjectedState])-str:查询用户信息.user_idstate[user_id]returnuser_123用户的姓名 楼兰.ifuser_iduser_123else未知用户agentcreate_react_agent(modelllm,tools[get_user_info],state_schemaCustomState,)agent.invoke({messages:查询用户信息user_id:user_123})4.2长期记忆长期记忆通常认为是比较充足的记忆空间因此使用时可以比短期记忆更加粗犷不太需要实时关注内存空间大小。至于使用方式和短期记忆差不太多。主要是通过Agent的store属性指定一个实现类就可以了。与短期记忆最大的区别在于短期记忆通过thread_id来区分不同的对话而长期记忆则通过namespace来区分不同的命名空间。fromlangchain_core.runnablesimportRunnableConfigfromlanggraph.configimportget_storefromlanggraph.prebuiltimportcreate_react_agentfromlanggraph.store.memoryimportInMemoryStorefromlangchain_core.toolsimporttool#定义长期存储storeInMemoryStore()# 添加一些测试数据。users是命名空间user_123是key后面的JSON数据是valuestore.put((users,),user_123,{name:楼兰,age:33,})#定义工具tool(return_directTrue)defget_user_info(config:RunnableConfig)-str:查找用户信息获取长期存储。获取到了后这个存储组件可读也可写storeget_store()# store.put(# (users,),# user_456,# {# name”:“楼兰# age:33,# }#)# 获取配置中的用户IDuser_idconfig[configurable].get(user_id)user_infostore.get((users,),user_id)returnstr(user_info.value)ifuser_infoelseUnknown useragentcreate_react_agent(modelllm,tools[get_user_info],storestore)# Run the agentagent.invoke((messages:[(role:user,content:查找用户信息}]},config{(configurable:(user_id:user_123}})五、Human-in-the-loop人类监督这也是LangGraph的Agent中非常核心的一个功能。在Agent的工作过程中有一个问题是非常致命的。就是Agent可以添加Tools工具但是要不要调用工具却完全是由Agent自己决定的。这就会导致Agent在面对一些问题时可能会出现错误的判断。为了解决这个问题LangGraph提供了Human-in-the-loop的功能。在Agert进行工具调用的过程中允许用户进行监督。这就需要中断当前的执行任务等待用户输入后再重新恢复任务。在实现时LangGraph提供了interruput()方法添加人类监督。监督时需要中断当前任务所以通常是和stream流式方法配合使用。fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterruptfromlanggraph.prebuiltimportcreate_react_agentfromlangchain_core.toolsimporttool# An example of a sensitive tool that requires human review / approvaltool(return_directTrue)defbook_hotel(hotel_name:str):预定宾馆responseinterrupt(f正准备执行book_hotel工具预定宾馆,相关参数名:{{hotel_name:{hotel_name}}}.请选择OK表示同意或者选择edit提出补充意见.ifresponse[type]OK:passelifresponse[type]edit:hotel_nameresponse[args][hotel_name]else:raiseValueError(fUnknown response type:{response[type]})returnf成功在{hotel_name}预定了一个房间.checkpointerInMemorySaver()agentcreate_react_agent(modelllm,tools[book_hotel],checkpointercheckpointer,)config{configurable:{thread_id:1}}forchunkinagent.stream({messages:[{role:user,content:帮我在图灵宾馆预定一个房间}]},config):print(chunk)print(\n)执行完成后会在book_hotel执行过程中输出一个Interrupt响应表示当前正在等待用户输入确认。接下来可以通过Agent提交一个Command请求来继续完成之前的任务。需要注意的是在这个示例中Agent只会一直等待用户输入。如果等待时间过长后续请求就无法恢复了。fromlanggraph.typesimportCommandforchunkinagent.stream(Command(resume{type:OK}),# Command(resume{type:edit,args:{hotel_name:三号宾馆}}),config):print(chunk)print(chunk[tools][messages][-1].content)print(\n)五、多Agent应用构建通常我们希望一个Agent能够专注干好一件事情。但是如果是面对一些复杂的任务我们可能需要多个Agent协同完成。例如一种典型的多Agent系统会是这样的:由一个Supervisor Agent对任务进行分发。然后交由另一个Agent来处理具体的事情。这样这多个Agent就可以成为一个同时处理多个任务的系统。LangGraph中单独提供个一个langgraph-supervisor依赖库来实现这种类型的多Agent系统。当然LangGraph的核心是用Graph图的方式来实现多Agent协作。所以这个库更多是作为基础了解。# 安装langgraph-supervisor依赖库pip install langgraph-supervisor--upgradefromlanggraph.prebuiltimportcreate_react_agentfromlanggraph_supervisorimportcreate_supervisordefbook_hotel(hotel_name:str):Book a hotelreturnfSuccessfully booked a stay at{hotel_name}.defbook_flight(from_airport:str,to_airport:str):Book a flightreturnfSuccessfully booked a flight from (from_airport} to{to_airport}.flight_assistantcreate_react_agent(modelllm,tools[book_flight],promptYou are a flight booking assistant,nameflight_assistanthotel_assistantcreate_react_agent(modelllm,tools[book_hotel],promptYou are a hotel booking assistant,namehotel_assistantsupervisorcreate_supervisor(agents[flight_assistant,hotel_assistant],modelllm,prompt(You manage a hotel booking assistant and a,flight booking assistant. Assign work to them.).compile()forchunkinsupervisor.stream({messages:[{role:usercontent:book a flight from B0S to JFK and a stay at McKittrick Hotel}]}):print(chunk)print(\n)六、LangGraph的Agent总结Agent这是LangGraph后续构建Graph图的基础。但其实Agent并不是LangGraph框架当中独有的。甚至Agent并不是一种技术而是我们设想的一种理想的大模型工作模式。那么到底什么是Agent?或者说AI行业心目中理想的Agent应该是什么样子呢?LangGraph实际上给我们提供了-种理解。而这个理解或许比具体实现更重要。这一章节我们重点在演练LangGraph中Agent的功能。我们介绍了LangGr:aph的Agent功能以及如何使用LangGraph的Agent功能来构建一个简单的应用。对于LangGraph框架来说由于有了LangChain作为支撑Agent智能体或许并不是他的重点。后续LangGraph使用Graph图的方式协调管理多个Agent或许是更大的价值所在。使用LangGraph构建Agent智能体Agent智能体增加Tools工具调用机制Agent智能体消息记忆管理功能Human-In-Loop人类监督功能但是LangGraph对于Agent的功能封装却给大模型应用落地提供了非常好的勺思想指导。通过Agent我们不再需要关注应用的实现细节而是可以更专注于应用的功能设计。而Agent绝不仅仅只是LangGraph所需要构建的即便脱离LangGraph框架如何构建一个能力强大又听话懂事的Agent或许是我们后续都需要思考的问题。