企业官网建设流程全解析

这一章节开始，我们将使用LangGraph来重新构建一个专属的聊天机器人。并以此为案例，逐步构建一个功能强大、又安全可控的聊天Agent智能体提供支撑。

• 使用LangGraph构建Agent智能体
• Agent智能体增加Tools工具调用机制
• Agent智能体消息记忆管理功能
• Human-In-Loop人类监督功能.

说明: LangGraph是和LangChain一体的。任何抛开LangChain独立讲解LangGraph的，都是耍流氓。 所以，后续内容设计都是在LangChain的基础上构建，并且最好是能够跟上我之前分享

一、什么是Agent？

Agent智能体，是LangGraph中的一个核心概念。很多朋友也应该经常在网上听说过Agent智能体，但是，到底什么是智能体呢?
这个问题并没有标准答案。而LangGraph给我们勾勒出了一个好的Agent智能体的形象。这个Agent，可以类比于一个好的员工。什么是好的员工呢?自然是希望这个员工即能力强大，可以放心的独立完成而不用过多干预实现的细节。同时，这个员工又要又能够听从安排，关键节要及时跟领导请示，并及时的根据领导的指示调整自己的工作进度。

具体到LangGraph的实现中，Agent即需要拥有封装与大模型交互的所有基础能力，包括访问大模型、调用Tools、保存ChatMemory等等这些基础的能力。可以独立的完成一系列基于大模型构建的任务。
我们又可以随时干预Agent执行进度，对关键步骤随时做出调整。

二、构建聊天Agent智能体

要如何构建这样强大并且听话的Agent智能体呢?我们先从基础的大模型聊天开始。
在LangChain中，提供了ChatModels的接口，用于访问大模型。我们可以通过调用ChatModels的接口来访问大模型。

fromconfig.load_keyimportload_keyfromlangchain_community.chat_modelsimportChatTongyi# 构建阿里云百炼大模型客户端llm=ChatTongyi(model="qwen-alus"api_key=load_key("BAILIAN_API_KEY"),)llm.invoke("你是谁?能帮我解决什么问题?")

而LangGraph只要将这个ChatModel进行简单的封装，就可以完成与大模型的交互。

fromlanggraph.prebuiltimportcreate_react_agent agent=create_react_agent(model=llm,tools=[],prompt="You are a helpful assistant",)agent.invoke({"messages":[{"role":"user","content":"你是谁?能帮我解决什么问题?"}]})

当然，Agent也支持常用的Stream流式输出的方式。

forchunkinagent.stream({"messages":[{"role":"user","content":"你是谁?能帮我解决什么问题?"}]}# stream_mode="messages"):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")agent=create_react_agent(model=llm,tools=[get_current_date],prompt="You are a helpful assistant",)agent.invoke({"messages":[{"role":"user","content":"今天是几月几号"}]})

从返回的结果就能看到，LangGraph的Agent完整的封装了工具调用的整个流程。而我们所需要关注的，只是构建出带有工具信息的Agent，然后把用户的问题交给Agent去处理就行了。
在定义工具时，除了可以从工具函数的注释中获取工具描述信息外，LangGraph也同样兼容了LangChain中使用@tool注解声明工具的方式。如果工具执行时出错了，LangGraph也提供了主动处理异常信息的能力。

fromlangchain_core.toolsimporttoolfromlanggraph.prebuiltimportToolNode#定义工具 return_direct=True 表示直接返回工具的结果@tool("devide_tool",return_direct=True)defdvide(a:int,b:int)->float:def"""计算两个整数的除法。 Args: a(int):除数 b(int):被除数."""#自定义错误ifb==1: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_node=ToolNode([devide],handle_tool_errors=handle_tool_error agent_with_error_handler=create_react_agent(model=llm,tools=tool_node)result=agent_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 checkpointer=InMemorySaver()defget_weather(city:str)->str:"""获取某个城市的天气"""returnf"城市:{city}，天气一直都是晴天!"agent=create_react_agent(model=llm,tools=[get_weather],checkpointer=checkpointer)# Run the agentconfig={"configurable":{"thread_id":"1"}}cs_response=agent.invoke({"messages":[{"role":"user","content":"长沙天气怎么样?"}]},config)print(cs_response)# Continue the conversation using the same thread_idbj_response=agent.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_node=SummarizationNode(token_counter=count_tokens_approximately,model=llm,max_tokens=384max_summary_tokens=128output_messages_key="llm_input_messages",)classState(AgentState):#注意:这个状态管理的作用是为了能够保存上一次总结的结果。这样就可以防止每次调用大模型时，都要重新总结历史信息。#这是一个比较常见的优化方式，因为大模型的调用是比较耗时的。context:dict[str,Any]checkpointer=InMemorySaver()agent=create_react_agent(model=llm,tools=tools,pre_model_hook=summarization_node,state_schema=State,checkpointer=checkpointer,)

另外，还提供了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_messages=trim_messages(state["messages"],strategy="last",token_counter=count_tokens_approximately,max_tokens=384,start_on="human"end_on=("human","tool"),)return{"llm_input_messages":trimmed_messages}checkpointer=InMemorySaver()agent=create_react_agent(model=llm,tools=[],pre_model_hook=pre_model_hook,checkpointer=checkpointer,)

实现了基础的短期记忆管理后，LangGraph还提供了状态管理机制，用于保存处理过程中的中间结果。而且，这些状态数据，还可以在Tools工具中使用。

fromtypingimportAnnotated fromlanggraph.prebuiltimportInjectedState,create_react_agentfromlanggraph.prebuilt.chat_agent_executorimportAgentStatefromlangchain_core.toolsimporttoolclassCustomState(AgentState):user_id:str@tool(return_direct=True)defget_user_info(state:Annotated[CustomState,InjectedState])->str:"""查询用户信息."""user_id=state["user_id"]return"user_123用户的姓名 楼兰."ifuser_id=="user_123"else"未知用户"agent=create_react_agent(model=llm,tools=[get_user_info],state_schema=CustomState,)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#定义长期存储store=InMemoryStore()# 添加一些测试数据。users是命名空间，user_123是key，后面的JSON数据是valuestore.put(("users",),"user_123",{"name":"楼兰","age":"33",})#定义工具@tool(return_direct=True)defget_user_info(config:RunnableConfig)->str:"""查找用户信息"""获取长期存储。获取到了后，这个存储组件可读也可写store=get_store()# store.put(# ("users",),# "user_456",# {# "name”:“楼兰"，# "age":"33",# }#)# 获取配置中的用户IDuser_id=config["configurable"].get("user_id")user_info=store.get(("users",),user_id)returnstr(user_info.value)ifuser_infoelse"Unknown user"agent=create_react_agent(model=llm,tools=[get_user_info],store=store)# 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 / approval@tool(return_direct=True)defbook_hotel(hotel_name:str):"""预定宾馆"""response=interrupt(f"正准备执行'book_hotel'工具预定宾馆,相关参数名:{{'hotel_name':{hotel_name}}}.""请选择OK，表示同意，或者选择edit，提出补充意见."ifresponse["type"]=="OK":passelifresponse["type"]=="edit":hotel_name=response["args"]["hotel_name"]else:raiseValueError(f"Unknown response type:{response['type']}")returnf"成功在{hotel_name}预定了一个房间."checkpointer=InMemorySaver()agent=create_react_agent(model=llm,tools=[book_hotel],checkpointer=checkpointer,)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--upgrade

fromlanggraph.prebuiltimportcreate_react_agentfromlanggraph_supervisorimportcreate_supervisordefbook_hotel(hotel_name:str):"""Book a hotel"""returnf"Successfully booked a stay at{hotel_name}."defbook_flight(from_airport:str,to_airport:str):"""Book a flight"""returnf"Successfully booked a flight from (from_airport} to{to_airport}."flight_assistant=create_react_agent(model=llm,tools=[book_flight],prompt="You are a flight booking assistant",name="flight_assistant"hotel_assistant=create_react_agent(model=llm,tools=[book_hotel],prompt="You are a hotel booking assistant",name="hotel_assistant"supervisor=create_supervisor(agents=[flight_assistant,hotel_assistant],model=llm,prompt=("You manage a hotel booking assistant and a","flight booking assistant. Assign work to them.").compile()forchunkinsupervisor.stream({"messages":[{"role":"user""content":"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，或许是我们后续都需要思考的问题。