832 lines
29 KiB
Plaintext
832 lines
29 KiB
Plaintext
{
|
||
"cells": [
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"# 19 循环与记忆\n",
|
||
"\n",
|
||
"## 学习目标\n",
|
||
"1. 理解LangGraph中循环和持久化记忆的实现方式\n",
|
||
"2. 掌握MemorySaver的使用，实现对话状态持久化\n",
|
||
"3. 能够构建支持多轮交互的智能体流程"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 1. 为什么需要循环和记忆\n",
|
||
"\n",
|
||
"在之前的课程中，我们构建的图流程都是**单次执行**的：\n",
|
||
"\n",
|
||
"```\n",
|
||
"START -> node_a -> node_b -> END\n",
|
||
"```\n",
|
||
"\n",
|
||
"这种流程跑一次就结束了。但真实场景中，很多需求需要**多次循环**和**记忆状态**：\n",
|
||
"\n",
|
||
"- **聊天机器人**：需要记住之前说过的话，才能进行多轮对话\n",
|
||
"- **任务分解**：可能需要反复调用工具，直到任务完成\n",
|
||
"- **信息收集**：需要逐步收集用户的信息，直到完整\n",
|
||
"- **多用户系统**：每个用户的对话历史需要独立保存\n",
|
||
"\n",
|
||
"简单来说：\n",
|
||
"\n",
|
||
"- **循环**：让流程可以重复执行某些节点\n",
|
||
"- **记忆**：让状态可以在多次执行之间保持"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 2. 循环的实现方式\n",
|
||
"\n",
|
||
"在LangGraph中，循环主要通过**条件边**来实现。关键思路是：\n",
|
||
"\n",
|
||
"1. 执行某个节点\n",
|
||
"2. 判断是否需要继续循环\n",
|
||
"3. 如果需要，回到之前的节点重新执行\n",
|
||
"4. 如果不需要，走向结束\n",
|
||
"\n",
|
||
"流程图如下：\n",
|
||
"\n",
|
||
"```\n",
|
||
"START -> work_node -> 判断节点\n",
|
||
" |\n",
|
||
" ┌─────────┴─────────┐\n",
|
||
" ▼ ▼\n",
|
||
" 继续循环 结束\n",
|
||
" | |\n",
|
||
" └───────► work_node ◄──┘\n",
|
||
"```\n",
|
||
"\n",
|
||
"用条件边实现就是：`判断节点` 根据状态返回 `'loop'` 或 `'end'`，然后：\n",
|
||
"\n",
|
||
"- `'loop'` → 回到 `work_node`\n",
|
||
"- `'end'` → 走到 `END`"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 3. 第一个例子：简单的计数器循环\n",
|
||
"\n",
|
||
"我们用一个计数器来演示循环的基本实现。\n",
|
||
"\n",
|
||
"流程目标：\n",
|
||
"- 从0开始计数\n",
|
||
"- 每次加1\n",
|
||
"- 当计数达到5时停止\n",
|
||
"- 每次循环都打印当前计数值"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"\n",
|
||
"class CounterState(TypedDict):\n",
|
||
" count: int\n",
|
||
"\n",
|
||
"def increment_node(state: CounterState):\n",
|
||
" new_count = state['count'] + 1\n",
|
||
" print(f'当前计数：{new_count}')\n",
|
||
" return {'count': new_count}\n",
|
||
"\n",
|
||
"def should_continue(state: CounterState):\n",
|
||
" if state['count'] < 5:\n",
|
||
" return 'loop'\n",
|
||
" return 'end'\n",
|
||
"\n",
|
||
"builder = StateGraph(CounterState)\n",
|
||
"builder.add_node('increment', increment_node)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'increment')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'increment',\n",
|
||
" should_continue,\n",
|
||
" {\n",
|
||
" 'loop': 'increment',\n",
|
||
" 'end': END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"graph = builder.compile()\n",
|
||
"result = graph.invoke({'count': 0})\n",
|
||
"print(result)"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"1. **State 定义**：`CounterState` 只包含一个字段 `count`，用于存储当前计数值。\n",
|
||
"\n",
|
||
"2. **increment_node**：每次执行时，从状态中取出 `count`，加1后放回状态。同时打印当前计数值。\n",
|
||
"\n",
|
||
"3. **should_continue**：路由函数，判断是否继续循环。如果 `count < 5` 返回 `'loop'`，否则返回 `'end'`。\n",
|
||
"\n",
|
||
"4. **条件边设置**：\n",
|
||
" - 来源节点是 `'increment'`\n",
|
||
" - 路由函数是 `should_continue`\n",
|
||
" - 如果返回 `'loop'`，就回到 `'increment'` 节点（形成循环）\n",
|
||
" - 如果返回 `'end'`，就走到 `END`\n",
|
||
"\n",
|
||
"5. **执行**：初始状态是 `{'count': 0}`，图会自动循环执行，直到 `count` 达到5。\n",
|
||
"\n",
|
||
"运行结果显示计数从1到5，说明循环成功执行了5次。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 4. 状态累积：消息历史的保存\n",
|
||
"\n",
|
||
"循环的核心价值在于**状态累积**。最常见的场景是保存对话历史。\n",
|
||
"\n",
|
||
"在聊天机器人中，每次用户输入和AI回复都需要保存下来，这样AI才能理解上下文。\n",
|
||
"\n",
|
||
"我们来构建一个简单的对话系统，每次循环都会把消息追加到历史中。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"\n",
|
||
"class ChatState(TypedDict):\n",
|
||
" messages: list\n",
|
||
" current_user_input: str\n",
|
||
"\n",
|
||
"def process_message(state: ChatState):\n",
|
||
" user_input = state['current_user_input']\n",
|
||
" \n",
|
||
" responses = {\n",
|
||
" '你好': '你好！我是一个AI助手。',\n",
|
||
" '我叫张三': '你好张三！很高兴认识你。',\n",
|
||
" '我今天心情很好': '太好了！祝你有美好的一天！'\n",
|
||
" }\n",
|
||
" \n",
|
||
" assistant_reply = responses.get(user_input, '抱歉，我不太理解。')\n",
|
||
"\n",
|
||
" new_message = {'user': user_input, 'assistant': assistant_reply}\n",
|
||
" \n",
|
||
" return {\n",
|
||
" 'messages': state['messages'] + [new_message],\n",
|
||
" 'current_user_input': ''\n",
|
||
" }\n",
|
||
"\n",
|
||
"def route(state: ChatState):\n",
|
||
" if state['current_user_input']:\n",
|
||
" return 'process'\n",
|
||
" return END\n",
|
||
"\n",
|
||
"builder = StateGraph(ChatState)\n",
|
||
"builder.add_node('process', process_message)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'process')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'process',\n",
|
||
" route,\n",
|
||
" {\n",
|
||
" 'process': 'process',\n",
|
||
" END: END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"graph = builder.compile()\n",
|
||
"\n",
|
||
"state = {'messages': [], 'current_user_input': '你好'}\n",
|
||
"state = graph.invoke(state)\n",
|
||
"\n",
|
||
"state['current_user_input'] = '我叫张三'\n",
|
||
"state = graph.invoke(state)\n",
|
||
"\n",
|
||
"state['current_user_input'] = '我今天心情很好'\n",
|
||
"state = graph.invoke(state)\n",
|
||
"\n",
|
||
"print(state['messages'])"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"1. **State 定义**：`ChatState` 包含两个字段：\n",
|
||
" - `messages`：保存所有对话历史的列表\n",
|
||
" - `current_user_input`：当前用户输入\n",
|
||
"\n",
|
||
"2. **process_message**：处理消息的核心逻辑：\n",
|
||
" - 从状态中获取当前用户输入\n",
|
||
" - 根据预设的规则生成助手回复\n",
|
||
" - 将新消息追加到 `messages` 列表中（关键！）\n",
|
||
" - 清空 `current_user_input`\n",
|
||
"\n",
|
||
"3. **route**：路由函数判断是否有新输入需要处理。\n",
|
||
"\n",
|
||
"4. **执行流程**：\n",
|
||
" - 第一次调用：输入'你好'，消息列表变成 `[{'user': '你好', ...}]`\n",
|
||
" - 第二次调用：输入'我叫张三'，消息列表变成两条记录\n",
|
||
" - 第三次调用：输入'我今天心情很好'，消息列表变成三条记录\n",
|
||
"\n",
|
||
"运行结果显示消息历史被正确累积，这就是状态持久化的基础。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 5. MemorySaver：持久化状态存储\n",
|
||
"\n",
|
||
"上面的例子中，我们手动传递状态。但在真实应用中，状态需要：\n",
|
||
"\n",
|
||
"- 在多次请求之间保持\n",
|
||
"- 支持多个用户同时使用（会话隔离）\n",
|
||
"- 重启后不丢失\n",
|
||
"\n",
|
||
"LangGraph 提供了 `MemorySaver` 来解决这个问题。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"from langgraph.checkpoint.memory import MemorySaver\n",
|
||
"\n",
|
||
"class ChatState(TypedDict):\n",
|
||
" messages: list\n",
|
||
" current_user_input: str\n",
|
||
"\n",
|
||
"def process_message(state: ChatState):\n",
|
||
" user_input = state['current_user_input']\n",
|
||
" \n",
|
||
" responses = {\n",
|
||
" '你好': '你好！我是一个AI助手。',\n",
|
||
" '我叫张三': '你好张三！很高兴认识你。',\n",
|
||
" '我今天心情很好': '太好了！祝你有美好的一天！'\n",
|
||
" }\n",
|
||
" \n",
|
||
" assistant_reply = responses.get(user_input, '抱歉，我不太理解。')\n",
|
||
"\n",
|
||
" new_message = {'user': user_input, 'assistant': assistant_reply}\n",
|
||
" \n",
|
||
" return {\n",
|
||
" 'messages': state['messages'] + [new_message],\n",
|
||
" 'current_user_input': ''\n",
|
||
" }\n",
|
||
"\n",
|
||
"def route(state: ChatState):\n",
|
||
" if state['current_user_input']:\n",
|
||
" return 'process'\n",
|
||
" return END\n",
|
||
"\n",
|
||
"builder = StateGraph(ChatState)\n",
|
||
"builder.add_node('process', process_message)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'process')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'process',\n",
|
||
" route,\n",
|
||
" {\n",
|
||
" 'process': 'process',\n",
|
||
" END: END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"memory = MemorySaver()\n",
|
||
"graph = builder.compile(checkpointer=memory)\n",
|
||
"\n",
|
||
"config = {'configurable': {'thread_id': 'session_1'}}\n",
|
||
"\n",
|
||
"graph.invoke({'messages': [], 'current_user_input': '你好'}, config)\n",
|
||
"graph.invoke({'current_user_input': '我叫张三'}, config)\n",
|
||
"graph.invoke({'current_user_input': '我今天心情很好'}, config)\n",
|
||
"\n",
|
||
"final_state = graph.get_state(config)\n",
|
||
"print(final_state.values['messages'])"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"1. **导入 MemorySaver**：从 `langgraph.checkpoint.memory` 导入。\n",
|
||
"\n",
|
||
"2. **创建 MemorySaver 实例**：`memory = MemorySaver()`\n",
|
||
"\n",
|
||
"3. **编译时传入 checkpointer**：`graph = builder.compile(checkpointer=memory)`\n",
|
||
"\n",
|
||
"4. **配置会话ID**：\n",
|
||
" ```python\n",
|
||
" config = {'configurable': {'thread_id': 'session_1'}}\n",
|
||
" ```\n",
|
||
" `thread_id` 是会话的唯一标识，不同用户用不同的 `thread_id`。\n",
|
||
"\n",
|
||
"5. **多次调用**：每次调用只传入新的输入，不需要手动传递状态。\n",
|
||
"\n",
|
||
"6. **获取状态**：`graph.get_state(config)` 可以获取指定会话的当前状态。\n",
|
||
"\n",
|
||
"关键区别：之前需要手动传递 `state`，现在只需传入 `config`，MemorySaver 会自动保存和恢复状态。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 6. 多会话隔离\n",
|
||
"\n",
|
||
"MemorySaver 的一个重要特性是**会话隔离**。不同的 `thread_id` 会维护独立的状态。\n",
|
||
"\n",
|
||
"下面的例子演示两个用户同时使用同一个图，但各自的对话历史完全独立。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"from langgraph.checkpoint.memory import MemorySaver\n",
|
||
"\n",
|
||
"class ChatState(TypedDict):\n",
|
||
" messages: list\n",
|
||
" current_user_input: str\n",
|
||
"\n",
|
||
"def process_message(state: ChatState):\n",
|
||
" user_input = state['current_user_input']\n",
|
||
" \n",
|
||
" responses = {\n",
|
||
" '你好': '你好！我是一个AI助手。',\n",
|
||
" '我是Alice': '你好Alice！很高兴认识你。',\n",
|
||
" '我是Bob': '你好Bob！很高兴认识你。'\n",
|
||
" }\n",
|
||
" \n",
|
||
" assistant_reply = responses.get(user_input, '抱歉，我不太理解。')\n",
|
||
"\n",
|
||
" new_message = {'user': user_input, 'assistant': assistant_reply}\n",
|
||
" \n",
|
||
" return {\n",
|
||
" 'messages': state['messages'] + [new_message],\n",
|
||
" 'current_user_input': ''\n",
|
||
" }\n",
|
||
"\n",
|
||
"def route(state: ChatState):\n",
|
||
" if state['current_user_input']:\n",
|
||
" return 'process'\n",
|
||
" return END\n",
|
||
"\n",
|
||
"builder = StateGraph(ChatState)\n",
|
||
"builder.add_node('process', process_message)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'process')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'process',\n",
|
||
" route,\n",
|
||
" {\n",
|
||
" 'process': 'process',\n",
|
||
" END: END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"memory = MemorySaver()\n",
|
||
"graph = builder.compile(checkpointer=memory)\n",
|
||
"\n",
|
||
"config_a = {'configurable': {'thread_id': 'user_a'}}\n",
|
||
"config_b = {'configurable': {'thread_id': 'user_b'}}\n",
|
||
"\n",
|
||
"graph.invoke({'messages': [], 'current_user_input': '你好'}, config_a)\n",
|
||
"graph.invoke({'current_user_input': '我是Alice'}, config_a)\n",
|
||
"\n",
|
||
"graph.invoke({'messages': [], 'current_user_input': 'Hi'}, config_b)\n",
|
||
"graph.invoke({'current_user_input': '我是Bob'}, config_b)\n",
|
||
"\n",
|
||
"state_a = graph.get_state(config_a)\n",
|
||
"state_b = graph.get_state(config_b)\n",
|
||
"\n",
|
||
"print('用户A的对话历史:')\n",
|
||
"print(state_a.values['messages'])\n",
|
||
"print()\n",
|
||
"print('用户B的对话历史:')\n",
|
||
"print(state_b.values['messages'])"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"1. **创建两个配置**：\n",
|
||
" - `config_a` 使用 `thread_id: 'user_a'`\n",
|
||
" - `config_b` 使用 `thread_id: 'user_b'`\n",
|
||
"\n",
|
||
"2. **分别调用**：\n",
|
||
" - 用户A发送了'你好'和'我是Alice'\n",
|
||
" - 用户B发送了'Hi'和'我是Bob'\n",
|
||
"\n",
|
||
"3. **获取各自状态**：\n",
|
||
" - 用户A的消息历史只有自己的两条消息\n",
|
||
" - 用户B的消息历史只有自己的两条消息\n",
|
||
"\n",
|
||
"运行结果清楚地展示了会话隔离：两个用户的对话历史完全独立，互不干扰。\n",
|
||
"\n",
|
||
"这就是多用户聊天系统的核心原理。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 7. 使用标准消息类型管理对话\n",
|
||
"\n",
|
||
"在实际应用中，我们通常使用 LangChain 的标准消息类型（如 `HumanMessage`、`AIMessage`）来管理对话。\n",
|
||
"\n",
|
||
"为了让 MemorySaver 能够自动追加消息而不是覆盖，我们需要使用 `add_messages` reducer。\n",
|
||
"\n",
|
||
"这种方式的特点：\n",
|
||
"- 使用标准的消息类型，便于与LangChain组件集成\n",
|
||
"- 自动追加消息，避免手动拼接\n",
|
||
"- 方便与 LLM 集成"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from typing import Annotated\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"from langgraph.graph.message import add_messages\n",
|
||
"from langgraph.checkpoint.memory import MemorySaver\n",
|
||
"from langchain_core.messages import HumanMessage, AIMessage\n",
|
||
"\n",
|
||
"class ChatState(TypedDict):\n",
|
||
" messages: Annotated[list, add_messages]\n",
|
||
"\n",
|
||
"def process_message(state: ChatState):\n",
|
||
" last_message = state['messages'][-1]\n",
|
||
" \n",
|
||
" if isinstance(last_message, HumanMessage):\n",
|
||
" user_input = last_message.content\n",
|
||
" \n",
|
||
" responses = {\n",
|
||
" '你好': '你好！我是一个AI助手。',\n",
|
||
" '我叫张三': '你好张三！很高兴认识你。',\n",
|
||
" }\n",
|
||
" \n",
|
||
" assistant_reply = responses.get(user_input, '抱歉，我不太理解。')\n",
|
||
" \n",
|
||
" return {'messages': [AIMessage(content=assistant_reply)]}\n",
|
||
" \n",
|
||
" return state\n",
|
||
"\n",
|
||
"def route(state: ChatState):\n",
|
||
" last_message = state['messages'][-1]\n",
|
||
" if isinstance(last_message, HumanMessage):\n",
|
||
" return 'process'\n",
|
||
" return END\n",
|
||
"\n",
|
||
"builder = StateGraph(ChatState)\n",
|
||
"builder.add_node('process', process_message)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'process')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'process',\n",
|
||
" route,\n",
|
||
" {\n",
|
||
" 'process': 'process',\n",
|
||
" END: END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"memory = MemorySaver()\n",
|
||
"graph = builder.compile(checkpointer=memory)\n",
|
||
"\n",
|
||
"config = {'configurable': {'thread_id': 'session_with_messages'}}\n",
|
||
"\n",
|
||
"graph.invoke({'messages': [HumanMessage(content='你好')]}, config)\n",
|
||
"graph.invoke({'messages': [HumanMessage(content='我叫张三')]}, config)\n",
|
||
"graph.invoke({'messages': [HumanMessage(content='今天天气怎么样')]}, config)\n",
|
||
"\n",
|
||
"final_state = graph.get_state(config)\n",
|
||
"for msg in final_state.values['messages']:\n",
|
||
" if isinstance(msg, HumanMessage):\n",
|
||
" print(f'Human: {msg.content}')\n",
|
||
" elif isinstance(msg, AIMessage):\n",
|
||
" print(f'AI: {msg.content}')"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"1. **导入 add_messages reducer**：`from langgraph.graph.message import add_messages`\n",
|
||
" - reducer 用于控制状态字段的更新方式\n",
|
||
" - `add_messages` 会自动将新消息追加到列表中，而不是覆盖\n",
|
||
"\n",
|
||
"2. **定义 ChatState**：使用 `Annotated[list, add_messages]` 定义消息字段\n",
|
||
" - `Annotated` 是 Python 的类型提示工具，用于添加元数据\n",
|
||
" - 这里告诉 LangGraph 对 `messages` 字段使用 `add_messages` reducer\n",
|
||
"\n",
|
||
"3. **导入消息类型**：\n",
|
||
" - `HumanMessage`：用户消息\n",
|
||
" - `AIMessage`：AI回复\n",
|
||
"\n",
|
||
"4. **process_message**：\n",
|
||
" - 获取最后一条消息\n",
|
||
" - 判断是否是用户消息\n",
|
||
" - 如果是，生成回复并返回 `{'messages': [AIMessage(...)]}`\n",
|
||
" - 由于使用了 `add_messages`，返回的新消息会自动追加到历史中\n",
|
||
"\n",
|
||
"5. **route**：\n",
|
||
" - 如果最后一条是用户消息，继续处理\n",
|
||
" - 如果是AI消息，说明已经处理过了，结束\n",
|
||
"\n",
|
||
"6. **调用方式**：每次传入 `{'messages': [HumanMessage(content=...)]}`\n",
|
||
" - MemorySaver 会自动保存状态\n",
|
||
" - `add_messages` reducer 会自动追加新消息\n",
|
||
"\n",
|
||
"这种方式的优点：\n",
|
||
"- 使用标准的消息类型，便于与LangChain组件集成\n",
|
||
"- 自动追加消息，避免手动拼接\n",
|
||
"- 后续可以直接将消息列表传给LLM"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 8. 综合案例：带工具调用的循环智能体\n",
|
||
"\n",
|
||
"现在我们把循环、记忆和工具调用结合起来，构建一个完整的智能体。\n",
|
||
"\n",
|
||
"流程目标：\n",
|
||
"- 用户询问关于天气或时间的问题\n",
|
||
"- 智能体判断是否需要调用工具\n",
|
||
"- 如果需要，调用工具获取信息\n",
|
||
"- 如果不需要，直接回答\n",
|
||
"- 循环直到任务完成\n",
|
||
"- 使用MemorySaver保存对话历史"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"from typing_extensions import TypedDict\n",
|
||
"from langgraph.graph import StateGraph, START, END\n",
|
||
"from langgraph.checkpoint.memory import MemorySaver\n",
|
||
"from langchain_core.messages import HumanMessage, AIMessage, ToolMessage\n",
|
||
"\n",
|
||
"class AgentState(TypedDict):\n",
|
||
" messages: list\n",
|
||
" next_tool: str\n",
|
||
"\n",
|
||
"def decide_tool(state: AgentState):\n",
|
||
" last_message = state['messages'][-1]\n",
|
||
" user_input = last_message.content\n",
|
||
" \n",
|
||
" print(f'用户问了：{user_input}')\n",
|
||
" \n",
|
||
" if '几点' in user_input or '时间' in user_input:\n",
|
||
" print('需要调用工具：get_time')\n",
|
||
" return {'next_tool': 'get_time'}\n",
|
||
" elif '天气' in user_input:\n",
|
||
" print('需要调用工具：get_weather')\n",
|
||
" return {'next_tool': 'get_weather'}\n",
|
||
" else:\n",
|
||
" print('不需要调用工具，直接回答')\n",
|
||
" return {'next_tool': None}\n",
|
||
"\n",
|
||
"def call_tool(state: AgentState):\n",
|
||
" tool_name = state['next_tool']\n",
|
||
" \n",
|
||
" if tool_name == 'get_time':\n",
|
||
" result = '2024-01-15 14:30:00'\n",
|
||
" elif tool_name == 'get_weather':\n",
|
||
" result = '晴天，25度'\n",
|
||
" else:\n",
|
||
" result = '未知工具'\n",
|
||
" \n",
|
||
" print(f'调用工具 {tool_name}，结果：{result}')\n",
|
||
" \n",
|
||
" return {\n",
|
||
" 'messages': state['messages'] + [ToolMessage(content=result, tool_call_id='1')],\n",
|
||
" 'next_tool': None\n",
|
||
" }\n",
|
||
"\n",
|
||
"def summarize(state: AgentState):\n",
|
||
" last_message = state['messages'][-1]\n",
|
||
" \n",
|
||
" if isinstance(last_message, ToolMessage):\n",
|
||
" tool_result = last_message.content\n",
|
||
" reply = f'总结回答：{tool_result}'\n",
|
||
" else:\n",
|
||
" reply = '抱歉，我不太理解你的问题。'\n",
|
||
" \n",
|
||
" print(reply)\n",
|
||
" \n",
|
||
" return {\n",
|
||
" 'messages': state['messages'] + [AIMessage(content=reply)]\n",
|
||
" }\n",
|
||
"\n",
|
||
"def route_after_decide(state: AgentState):\n",
|
||
" if state['next_tool']:\n",
|
||
" return 'call_tool'\n",
|
||
" return 'summarize'\n",
|
||
"\n",
|
||
"def route_after_summarize(state: AgentState):\n",
|
||
" last_message = state['messages'][-1]\n",
|
||
" if isinstance(last_message, HumanMessage):\n",
|
||
" return 'decide_tool'\n",
|
||
" return END\n",
|
||
"\n",
|
||
"builder = StateGraph(AgentState)\n",
|
||
"builder.add_node('decide_tool', decide_tool)\n",
|
||
"builder.add_node('call_tool', call_tool)\n",
|
||
"builder.add_node('summarize', summarize)\n",
|
||
"\n",
|
||
"builder.add_edge(START, 'decide_tool')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'decide_tool',\n",
|
||
" route_after_decide,\n",
|
||
" {\n",
|
||
" 'call_tool': 'call_tool',\n",
|
||
" 'summarize': 'summarize'\n",
|
||
" }\n",
|
||
")\n",
|
||
"builder.add_edge('call_tool', 'summarize')\n",
|
||
"builder.add_conditional_edges(\n",
|
||
" 'summarize',\n",
|
||
" route_after_summarize,\n",
|
||
" {\n",
|
||
" 'decide_tool': 'decide_tool',\n",
|
||
" END: END\n",
|
||
" }\n",
|
||
")\n",
|
||
"\n",
|
||
"memory = MemorySaver()\n",
|
||
"graph = builder.compile(checkpointer=memory)\n",
|
||
"\n",
|
||
"config = {'configurable': {'thread_id': 'tool_agent_session'}}\n",
|
||
"\n",
|
||
"graph.invoke({'messages': [HumanMessage(content='现在几点了')], 'next_tool': None}, config)\n",
|
||
"\n",
|
||
"current_state = graph.get_state(config)\n",
|
||
"new_messages = current_state.values['messages'] + [HumanMessage(content='今天天气怎么样')]\n",
|
||
"graph.invoke({'messages': new_messages, 'next_tool': None}, config)\n",
|
||
"\n",
|
||
"final_state = graph.get_state(config)\n",
|
||
"print()\n",
|
||
"print('完整对话历史:')\n",
|
||
"for msg in final_state.values['messages']:\n",
|
||
" if isinstance(msg, HumanMessage):\n",
|
||
" print(f'Human: {msg.content}')\n",
|
||
" elif isinstance(msg, AIMessage):\n",
|
||
" print(f'AI: {msg.content}')"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"### 代码解释\n",
|
||
"\n",
|
||
"这个案例展示了一个完整的智能体流程：\n",
|
||
"\n",
|
||
"1. **AgentState**：使用 TypedDict 定义状态，包含：\n",
|
||
" - `messages`：消息列表\n",
|
||
" - `next_tool`：下一个要调用的工具\n",
|
||
"\n",
|
||
"2. **decide_tool**：判断是否需要调用工具\n",
|
||
" - 如果用户问时间，返回 `{'next_tool': 'get_time'}`\n",
|
||
" - 如果用户问天气，返回 `{'next_tool': 'get_weather'}`\n",
|
||
" - 如果不需要工具，返回 `{'next_tool': None}`\n",
|
||
"\n",
|
||
"3. **call_tool**：执行工具调用\n",
|
||
" - 根据 `next_tool` 调用相应工具\n",
|
||
" - 将结果追加到 `messages` 列表\n",
|
||
"\n",
|
||
"4. **summarize**：总结回答\n",
|
||
" - 如果最后是工具消息，提取结果并生成回答\n",
|
||
" - 将回答追加到 `messages` 列表\n",
|
||
"\n",
|
||
"5. **route_after_decide**：decide_tool 之后的路由\n",
|
||
" - 如果 `next_tool` 有值，进入 `call_tool`\n",
|
||
" - 如果 `next_tool` 为 None，进入 `summarize`\n",
|
||
"\n",
|
||
"6. **route_after_summarize**：summarize 之后的路由\n",
|
||
" - 如果最后是用户消息，回到 `decide_tool`\n",
|
||
" - 否则结束\n",
|
||
"\n",
|
||
"流程图：\n",
|
||
"\n",
|
||
"```\n",
|
||
"START -> decide_tool -> [条件边] -> call_tool -> summarize\n",
|
||
" ^ |\n",
|
||
" | |\n",
|
||
" +--------------+\n",
|
||
"```\n",
|
||
"\n",
|
||
"这个流程可以处理多轮对话，每轮都可以调用工具并保存历史。"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 9. 总结\n",
|
||
"\n",
|
||
"### 核心知识点\n",
|
||
"\n",
|
||
"1. **循环实现**：通过条件边让流程回到之前的节点，形成循环\n",
|
||
"2. **状态累积**：在节点中不断追加数据到状态字段（如消息列表）\n",
|
||
"3. **MemorySaver**：自动保存和恢复状态，无需手动传递\n",
|
||
"4. **会话隔离**：通过 `thread_id` 区分不同用户的会话\n",
|
||
"5. **add_messages reducer**：使用 `Annotated[list, add_messages]` 实现消息自动追加\n",
|
||
"\n",
|
||
"### 关键API\n",
|
||
"\n",
|
||
"| API | 作用 |\n",
|
||
"| --- | --- |\n",
|
||
"| `MemorySaver()` | 创建内存状态存储 |\n",
|
||
"| `builder.compile(checkpointer=memory)` | 编译时启用状态持久化 |\n",
|
||
"| `config = {'configurable': {'thread_id': ...}}` | 设置会话ID |\n",
|
||
"| `graph.get_state(config)` | 获取指定会话的状态 |\n",
|
||
"| `add_messages` | 自动追加消息的 reducer |\n",
|
||
"\n",
|
||
"### 实践要点\n",
|
||
"\n",
|
||
"- 循环的关键是路由函数返回某个已存在的节点名\n",
|
||
"- 状态累积需要在节点中返回新的状态值\n",
|
||
"- 使用 `Annotated[list, add_messages]` 可以让 MemorySaver 自动追加消息\n",
|
||
"- MemorySaver 适合开发和测试，生产环境可使用其他存储后端\n",
|
||
"- 会话隔离通过 `thread_id` 实现，必须为每个用户分配唯一ID"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 练习\n",
|
||
"\n",
|
||
"1. 修改计数器示例，让它从10递减到0\n",
|
||
"2. 扩展对话示例，添加更多回复规则\n",
|
||
"3. 创建一个多轮问答系统，支持用户追问\n",
|
||
"4. 尝试使用不同的 `thread_id` 模拟三个用户同时对话"
|
||
]
|
||
}
|
||
],
|
||
"metadata": {
|
||
"kernelspec": {
|
||
"display_name": ".venv",
|
||
"language": "python",
|
||
"name": "python3"
|
||
},
|
||
"language_info": {
|
||
"codemirror_mode": {
|
||
"name": "ipython",
|
||
"version": 3
|
||
},
|
||
"file_extension": ".py",
|
||
"mimetype": "text/x-python",
|
||
"name": "python",
|
||
"nbconvert_exporter": "python",
|
||
"pygments_lexer": "ipython3",
|
||
"version": "3.14.4"
|
||
}
|
||
},
|
||
"nbformat": 4,
|
||
"nbformat_minor": 4
|
||
}
|