langgraph搭建智能体各个组件使用与总结

LangGraph 技术学习总结

基于leaning_langgraph.py的 LangGraph 学习笔记与实践总结

📚 目录

环境配置
核心概念
Agent 创建
记忆机制
MCP 集成
Graph 图结构
高级特性

1. 环境配置

1.1 基础依赖

importosimportpicklefromdatetimeimportdatetimefromlangchain_community.chat_modelsimportChatTongyifromlanggraph.prebuiltimportcreate_react_agentfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.store.memoryimportInMemoryStorefromlanggraph.graphimportStateGraph,MessagesState,START,END

1.2 模型配置

# 配置 DashScope API Keyos.environ["DASHSCOPE_API_KEY"]="your-api-key"# 创建百炼模型实例llm=ChatTongyi(model_name="qwen-max",# 选项: qwen-max, qwen-plus, qwen-turbotemperature=0.7,streaming=False)

2. 核心概念

2.1 Agent（智能体）

Agent 是 LangGraph 的核心组件，能够使用工具（tools）来完成任务。

创建基础 Agent:

agent=create_react_agent(model=llm,tools=[tool_function],checkpointer=checkpointer# 可选，用于记忆管理)

2.2 Tools（工具）

工具是 Agent 可调用的函数，需要带有详细的注释说明。

defcurrent_time():'''查询时间'''# 注释是必须的，根据注释分配工具time=datetime.today()time.strftime("%Y-%m-%d")returntime

2.3 State（状态）

State 是整个应用共享的状态，可以使用 TypedDict 或 Pydantic 定义。

fromtypingimportTypedDictclassOverallState(TypedDict):user_input:strmiddle_output1:strmiddle_output2:strgraph_output:str

2.4 Node（节点）

节点是具体的任务执行单元。

defnode_1(state:OverallState)->OverallState:return{"middle_output1":state["user_input"]+">进行到middle1"}

2.5 Edge（边）

边定义节点之间的依赖关系。

类型	说明	示例
普通边	单向连接	`graph.add_edge("node_1", "node_2")`
条件边	根据条件动态路由	`graph.add_conditional_edge(START, function_node)`

3. Agent 创建与使用

3.1 基础 Agent

fromlanggraph.prebuiltimportcreate_react_agent checkpointer=InMemorySaver()agent=create_react_agent(model=llm,tools=[current_time],checkpointer=checkpointer)config={"configurable":{"thread_id":"1"}}# 调用 Agentresult=agent.invoke({"messages":[{"role":"user","content":"现在几月几号了？"}]},config)print(result)

3.2 带 Hook 的 Agent

使用pre_model_hook对输入进行预处理。

fromlangmem.short_termimportSummarizationNode# 方式1: 总结式记忆summarization_node=SummarizationNode(token_counter=count_tokens_approximately,model=llm,max_tokens=384,max_summary_tokens=128,output_messages_key="llm_input_message")classState(AgentState):context:dict[str,Any]agent=create_react_agent(model=llm,tools=[current_time],checkpointer=checkpointer,pre_model_hook=summarization_node,state_schema=State)

# 方式2: 裁剪式记忆defpre_model_hook(state):result=trim_messages(state['messages'],strategy="last",token_counter=count_tokens_approximately,max_tokens=384,start_on="human",end_on=['human',"tool"])return{"llm_input_message":result}

4. 记忆机制

LangGraph 提供三种记忆管理方式：

4.1 Checkpointer（短期记忆）

使用InMemorySaver保存对话历史。

fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointer=InMemorySaver()agent=create_react_agent(model=llm,tools=[current_time],checkpointer=checkpointer)config={"configurable":{"thread_id":"1"}}# 第一次调用print(agent.invoke({"messages":[{"role":"user","content":"现在几月几号了？"}]},config))# 第二次调用（可记住上下文）print(agent.invoke({"messages":[{"role":"user","content":"那明天呢？"}]},config))

4.2 消息处理策略

策略	说明	适用场景
Summarization	总结历史记录	长对话需要保留关键信息
Trimming	删除旧信息	只需要最近上下文

4.3 Store（长期记忆）

使用InMemoryStore保存用户信息。

fromlanggraph.store.memoryimportInMemoryStorefromlanggraph.configimportget_storefromlangchain_core.runnablesimportRunnableConfig store=InMemoryStore()store.put(("user"),"id_123",{"name":"邓茗芳","age":"27"})defget_user_info(config:RunnableConfig):"""提供用户的个人信息的函数"""store=get_store()user_id=config["configurable"].get("user_id")user_info=store.get(("user"),user_id)returnstr(user_info.value)ifuser_infoelse"暂未查询到该用户的任何信息"agent=create_react_agent(model=llm,tools=[get_user_info],store=store)config={"configurable":{"user_id":"id_123"}}print(agent.invoke({"messages":[{"role":"user","content":"查用户年龄"}]},config))

5. MCP 集成

MCP (Model Context Protocol) 是一种用于连接 AI 模型与外部服务的协议。

5.1 MCP 两种传输方式

方式	说明	特点
SSE	基于 HTTP 的长连接协议	适合远程服务
stdio	本地执行应用程序获取结果	适合本地工具

5.2 使用 MCP 客户端（注释示例）

# from langchain_mcp_adapters.client import MultiServerMCPClient# import asyncio# async def create_use_mcp_agent():# client = MultiServerMCPClient({# "amap-maps": {# "command": "npx",# "args": ["-y", "@amap/amap-maps-mcp-server"],# "env": {"AMAP_MAPS_API_KEY": "your-api-key"},# "transport": "stdio"# }# })# tools = await client.get_tools()# agent = create_react_agent(model=llm, tools=tools)# response = await agent.ainvoke({# "messages": [{"role": "user", "content": "杭州新天地到欣景苑怎么走？"}]# })# return response# response = asyncio.run(create_use_mcp_agent())# print(response)

5.3 连接自定义 MCP 服务器

fromlangchain_mcp_adapters.toolsimportload_mcp_toolsfrommcpimportStdioServerParameters,stdio_client,ClientSessionimportmcp.typesastypes server_params=StdioServerParameters(command="python",args=["/www/learning_langchain/mcp_server.py"])asyncdefhandle_sample_message(message:types.CreateMessageRequestParams)->types.CreateMessageResult:returntypes.CreateMessageResult(role="assistant",content=types.TextContent(type="text",text="hello!"),model='qwen-plus',stopReason="endTurn")asyncdefuse_mcp_agent():asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write,sampling_callback=handle_sample_message)assession:awaitsession.initialize()tools=awaitsession.list_tools()print(tools)resource=awaitsession.list_resources()print(resource)result=awaitsession.call_tool("is_daka",{})print(result)returnresultprint(asyncio.run(use_mcp_agent()))

6. Graph 图结构

Graph 是有向无环图（DAG），用于描述任务之间的依赖关系。

6.1 构建基础 Graph

fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassOverallState(TypedDict):user_input:strmiddle_output1:strmiddle_output2:strgraph_output:strdefnode_1(state:OverallState)->OverallState:return{"middle_output1":state["user_input"]+">进行到middle1"}defnode_2(state:OverallState)->OverallState:return{"middle_output2":state["middle_output1"]+">进行到middle2"}defnode_3(state:OverallState)->OverallState:return{"graph_output":state["middle_output2"]+">结束"}graph=StateGraph(state_schema=OverallState)graph.add_node("node_1",node_1)graph.add_node("node_2",node_2)graph.add_node("node_3",node_3)graph.add_edge(START,"node_1")graph.add_edge("node_1","node_2")graph.add_edge("node_2","node_3")graph.add_edge("node_3",END)graph=graph.compile()print(graph.invoke({"user_input":"开始"}))

6.2 可视化 Graph

fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))

6.3 Graph 与大模型结合

defcall_model(state:MessagesState)->MessagesState:response=llm.invoke(state["messages"])return{"messages":response}builder=StateGraph(state_schema=MessagesState)builder.add_node("call_model",call_model)builder.add_edge(START,"call_model")builder.add_edge("call_model",END)builder=builder.compile()print(builder.invoke({"messages":[{"role":"user","content":"湖南省会城市是哪里？"}]}))

6.4 Node 高级特性

重试机制:

graph.add_node("node_1",node_1,retry=RetryPolicy(max_attempts=4))

缓存机制:

graph=graph.compile(InMemorycache())

7. 高级特性

7.1 条件边（Conditional Edge）

根据函数结果动态选择下一个节点。

defshould_continue(state):# 返回下一个节点的名称ifsome_condition:return"node_2"return"end"graph.add_conditional_edge("node_1",should_continue)

7.2 Time Travel（时间回溯）

由于大模型回答具有不确定性，发现某个步骤出问题时，可以手动指定从 Graph 的某个 Node 开始重演。

importuuidfromlanggraph.checkpoint.memoryimportInMemorySaverclassState(TypedDict):planning:strcity:strdefget_city(state:State):prompt="请帮我随机推荐一个热门旅游城市，只说出城市名称即可，严禁多于废话"res=llm.invoke(prompt)return{"city":res}deftravel_planning(state:State):prompt=f'请给我推荐一下{state["city"]}中的美食和三天的出行规划'res=llm.invoke(prompt)return{"planning":res}builder=StateGraph(State)builder.add_node("get_city",get_city)builder.add_node("travel_planning",travel_planning)builder.add_edge(START,"get_city")builder.add_edge("get_city","travel_planning")builder.add_edge("travel_planning",END)config={"configurable":{"thread_id":uuid.uuid4()}}checkpoints=InMemorySaver()builder=builder.compile(checkpointer=checkpoints)# 执行print(builder.invoke({"messages":[{"role":"user","content":"请给我推荐一下热门旅游城市"}]},config))# 获取历史状态state=list(builder.get_state_history(config))for_instate:print(_.next)print(_.config['configurable']['checkpoint_id'])# 修改状态并重新执行selected_state=state[1]new_config=builder.update_state(selected_state.config,values={"city":"西藏"})print(builder.invoke(None,new_config))

7.3 Human-in-the-Loop（人类监督）

使用interrupt()方法实现人类干预，决定是否执行工具。

# 示例（需具体实现）defhuman_review_node(state):# interrupt 等待人类确认result=interrupt({"question":"是否继续执行下一步？"})ifresult:return{"status":"approved"}else:return{"status":"rejected"}

8. 总结与最佳实践

8.1 核心组件对比

组件	作用	常用类
Agent	智能主体，调用工具	`create_react_agent`
Checkpointer	短期记忆，对话历史	`InMemorySaver`
Store	长期记忆，用户信息	`InMemoryStore`
Graph	任务流程图	`StateGraph`