LangGraph学习笔记

> 学习时间：2026-04-10
> 课程来源：DeepLearning.AI - AI Agents in LangGraph
> 授课教师：Harrison Chase (LangChain创始人) + Rotem Weiss (Tavily创始人)
> 学习背景：已学ReAct论文、Generative Agents论文、EntroCamp课程

---

一、课程概要

1.1 课程信息

项目	内容
课程名称	AI Agents in LangGraph
时长	1小时32分钟
难度	中级
视频数量	9个
代码示例	6个
教师	Harrison Chase (LangChain创始人) + Rotem Weiss (Tavily创始人)

1.2 课程核心收获

1. 从零构建Agent：理解LLM与周围代码的任务分工
2. LangGraph实现：用状态图重构Agent，理解组件和流程
3. Agentic Search：用Tavily获取多答案格式的检索结果
4. 持久化机制：跨Thread状态管理、对话切换、时间旅行调试
5. Human-in-the-loop：人工介入机制
6. 论文写作Agent实战：复制研究人员的工作流程

1.3 课程大纲

1. Introduction (6分钟)
Build an Agent from Scratch (12分钟) - 从零构建Agent
LangGraph Components (19分钟) - 核心组件详解
Agentic Search Tools (5分钟) - 代理搜索工具
Persistence and Streaming (9分钟) - 持久化和流式输出
Human in the loop (14分钟) - 人工介入
Essay Writer (18分钟) - 论文写作Agent实战
LangChain Resources (2分钟)
Conclusion (4分钟)

---

二、LangGraph核心概念

2.1 三大核心组件

LangGraph将Agent工作流建模为图，通过三个关键组件定义行为：

┌─────────────────────────────────────────────────────────────┐
│                        LangGraph 架构                         │
├─────────────────────────────────────────────────────────────┤
│  State（状态）                                               │
│  ├─ 共享数据结构，表示应用当前快照                            │
│  ├─ 通常是 TypedDict 或 Pydantic BaseModel                  │
│  └─ 包含模式和reducer函数                                   │
├─────────────────────────────────────────────────────────────┤
│  Nodes（节点）                                               │
│  ├─ Python函数，编码Agent逻辑                               │
│  ├─ 接收当前State作为输入                                   │
│  ├─ 执行计算或副作用，返回更新后的State                       │
│  └─ 可以包含LLM，也可以是普通Python代码                       │
├─────────────────────────────────────────────────────────────┤
│  Edges（边）                                                 │
│  ├─ Python函数，根据当前State决定下一步执行哪个Node           │
│  ├─ 条件分支：动态路由                                       │
│  └─ 固定转换：顺序执行                                       │
└─────────────────────────────────────────────────────────────┘

核心原则：
- Nodes做工作（do the work）
- Edges指示下一步（tell what to do next）
- Nodes和Edges都是Python函数，可以包含LLM或普通代码

2.2 StateGraph详解

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
定义AgentState - TypedDict + Reducer
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    # messages使用operator.add作为reducer
    # 意味着每次更新都追加到列表，而不是覆盖
    
StateGraph核心流程
workflow = StateGraph(AgentState)
1. 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)
2. 设置入口点
workflow.set_entry_point("agent")
3. 添加条件边（基于LLM输出决定路由）
workflow.add_conditional_edges(
    "agent",
    should_continue,  # 函数决定下一步
    {
        "continue": "tools",
        "end": END
    }
)
4. 添加普通边（固定路由）
workflow.add_edge("tools", "agent")
5. 编译图
app = workflow.compile()

2.3 状态归约器（Reducer）

归约器决定节点更新如何应用到State。每个State键都有独立的reducer函数：

from typing import Annotated
import operator
默认reducer：覆盖（无reducer时默认行为）
class DefaultState(TypedDict):
    value: str  # 每次更新都覆盖
追加reducer：添加到列表
class AppendState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]  
    # 每次更新都追加到列表
    
覆盖reducer：显式指定
class OverrideState(TypedDict):
    counter: Annotated[int, lambda x, y: y]  # 覆盖为最新值

2.4 超步（Super-step）与消息传递

LangGraph底层算法受Google Pregel系统启发，使用消息传递定义通用程序：

超步 = 图节点的一次迭代
┌──────────────────────────────────────────────────────────┐
│                    Super-step 执行模型                     │
├──────────────────────────────────────────────────────────┤
│                                                          │
│   [Start] → [Node A] → [Node B] → [Node C] → [End]       │
│                                                          │
│   Step 0: Input                                          │
│   Step 1: Node A 执行                                    │
│   Step 2: Node B 执行                                    │
│   Step 3: Node C 执行                                    │
│   Step 4: End                                            │
│                                                          │
│   每个step结束时保存checkpoint                            │
└──────────────────────────────────────────────────────────┘

节点状态机：
- inactive：初始状态，无消息时进入
- active：收到消息时进入，执行函数
- halt：执行完毕，投票停止

---

三、为什么Agent需要循环？

3.1 循环 vs DAG

传统LCEL（LangChain Expression Language）只支持线性链（DAG），无法支持需要迭代的Agent场景：

特性	LCEL (DAG)	LangGraph (循环)
流程	线性顺序	可循环迭代
适用场景	简单RAG、生成任务	Agent推理、工具调用
状态管理	无	有
人工介入	无	支持
错误恢复	无	支持checkpoint

3.2 Agent循环的本质

Agent需要循环的核心原因：LLM无法一次性完成复杂任务

┌─────────────────────────────────────────────────────────────┐
│                    ReAct Agent 循环                          │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│    ┌─────────┐                                               │
│    │  START  │                                               │
│    └────┬────┘                                               │
│         │                                                    │
│         ▼                                                    │
│    ┌─────────┐    Thought     ┌──────────┐                   │
│    │  Agent  │ ────────────→ │ 分析思考  │                   │
│    └────┬────┘               └────┬─────┘                   │
│         │                          │                         │
│         │ Action                   │                         │
│         ▼                          ▼                         │
│    ┌─────────┐               ┌──────────┐                   │
│    │  Tools  │               │ 执行工具  │                   │
│    └────┬────┘               └────┬─────┘                   │
│         │                          │                         │
│         │ Observation              │                         │
│         ▼                          ▼                         │
│    ┌─────────┐ ◀────────────── ┌───────┐                   │
│    │  Agent  │                 │ 结果？  │                   │
│    └────┬────┘                 └───┬───┘                    │
│         │                          │                         │
│         │ No                       │ Yes                     │
│         └──────────┬───────────────┘                        │
│                    ▼                                         │
│              ┌─────────┐                                      │
│              │   END   │                                      │
│              └─────────┘                                      │
└─────────────────────────────────────────────────────────────┘

循环终止条件：

1. LLM决定完成（无tool_calls）
2. 达到递归限制（recursion_limit）
3. 遇到错误（Error Handling）
4. 人工介入（Human-in-the-loop）

---

四、持久化机制（Persistence）

4.1 Checkpointer核心概念

from langgraph.checkpoint.memory import InMemorySaver
创建内存检查点（生产环境用Redis/Postgres/SQLite）
checkpointer = InMemorySaver()
编译时传入checkpointer
app = workflow.compile(checkpointer=checkpointer)
调用时指定thread_id
config = {"configurable": {"thread_id": "user_123_session_1"}}
首次调用
app.invoke(inputs, config)
后续调用 - 自动恢复之前状态
app.invoke(inputs, config)

4.2 Checkpoint数据结构

Checkpoint = {
    "values": {
        # 当前状态
        "messages": [...],
        "user_query": "...",
    },
    "metadata": {
        "thread_id": "user_123_session_1",
        "thread_ts": "2024-01-01T12:00:00.000Z",
        "parent_ts": "2024-01-01T11:59:00.000Z",  # 用于时间旅行
    }
}

4.3 持久化的四大价值

功能	描述	应用场景
Memory	对话记忆	多轮对话，上下文保持
Time Travel	时间旅行调试	回溯任意checkpoint重新执行
Fault Tolerance	容错恢复	节点失败从上一个checkpoint恢复
Human-in-the-loop	人工介入	查看/修改状态后恢复执行

4.4 时间旅行调试实战

# 查看所有历史checkpoint
checkpoints = app.get_state_history(config)
从特定checkpoint恢复
specific_config = {
    "configurable": {
        "thread_id": "user_123",
        "checkpoint_id": "specific-checkpoint-id"
    }
}
app.invoke(inputs, specific_config)
Fork：从某checkpoint创建新分支
app.update_state(config, {"messages": [...]})

---

五、Human-in-the-loop机制

5.1 Interrupt核心概念

from langgraph.types import interrupt, Command
def human_review_node(state):
    """人工审核节点"""
    # 中断执行，显示信息给人工审核
    user_input = interrupt({
        "action": "book_hotel",
        "args": {"hotel_name": "McKittrick Hotel"},
        "message": "请确认是否预订此酒店？"
    })
    
    if user_input["approved"]:
        return {"status": "approved"}
    else:
        return {"status": "rejected", "reason": user_input["reason"]}

5.2 Command恢复机制

from langgraph.types import Command
用户批准
app.invoke(
    Command(resume={"type": "accept"}),
    config
)
用户修改参数
app.invoke(
    Command(resume={"type": "edit", "args": {"hotel_name": "Hilton"}}),
    config
)
用户拒绝
app.invoke(
    Command(resume={"type": "reject", "reason": "太贵了"}),
    config
)

5.3 HITL决策类型

决策类型	描述	args
approve	直接批准	None
edit	修改参数后批准	修改后的args
reject	拒绝操作	reason字符串
response	跳过工具，直接响应	响应内容字符串
ignore	忽略工具调用	None

---

六、与论文的关联分析

6.1 LangGraph与ReAct论文

ReAct核心机制：Thought-Action-Observation循环

Paper: "ReAct: Synergizing Reasoning and Acting in Language Models"
       (Yao et al., 2023)
LangGraph实现：
┌─────────────────────────────────────────────────────────────┐
│  ReAct理论                  LangGraph实现                   │
├─────────────────────────────────────────────────────────────┤
│  Thought (思考)          →  Agent Node (LLM调用)            │
│  Action (行动)           →  Tools Node (工具执行)            │
│  Observation (观察)      →  返回Agent继续循环                │
│                                                              │
│  条件判断：Should continue?                                  │
│  → add_conditional_edges + should_continue函数               │
│                                                              │
│  工具绑定：OpenAI function calling                           │
│  → model.bind_tools(tools)                                  │
└─────────────────────────────────────────────────────────────┘

代码对应：

# ReAct的should_continue函数（LangGraph版）
def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    
    # ReAct: "If the Agent has tool_calls, continue"
    if getattr(last_message, "tool_calls", None):
        return "continue"  # → 工具节点
    return "end"  # → 结束
对比ReAct论文伪代码：
while env.needs_more_steps():
    thought = model.think(context)
    if "action" in thought:
        action = parse_action(thought)
        obs = env.execute(action)
        context += [thought, action, obs]
    else:
        break

6.2 LangGraph与Generative Agents论文

Generative Agents核心组件：记忆流-反思-规划

Paper: "Generative Agents: Interactive Simulacra of Human Behavior"
       (Park et al., Stanford, 2023)
LangGraph实现映射：
┌─────────────────────────────────────────────────────────────┐
│  Generative Agents        LangGraph实现                       │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  Memory Stream            State + Checkpointer               │
│  ├─ 自然语言存储          ├─ messages: Annotated[list, add]  │
│  ├─ 检索评分              └─ 持久化到checkpoint               │
│                                                              │
│  Reflection              Dedicated Node + LLM               │
│  ├─ 综合生成推断          ├─ reflection_node                 │
│  └─ 高层次洞察            └─ 使用LLM生成反思                  │
│                                                              │
│  Planning                 State + Conditional Edges          │
│  ├─ 生成行动序列          ├─ state.plans                     │
│  └─ 动态调整              └─ 条件边动态路由                   │
│                                                              │
│  Reactivity              Interrupt + Human-in-the-loop       │
│  ├─ 感知环境变化          ├─ interrupt()                     │
│  └─ 实时调整              └─ Command恢复                     │
└─────────────────────────────────────────────────────────────┘

看宝AI五层记忆的LangGraph实现：

# 看宝AI的AgentState
class KanbaoAgentState(TypedDict):
    # 记忆层（Generative Agents的Memory Stream）
    session_memory: Annotated[list, operator.add]      # 会话记忆
    profile_memory: dict                              # 画像记忆
    situational_memory: list                          # 情景记忆
    knowledge_memory: list                            # 知识记忆
    growth_memory: list                              # 成长痕迹
    
    # 循环状态（Generative Agents的Planning + Reactivity）
    current_trigger: Optional[CareTrigger]            # 当前触发点
    care_queue: list                                 # 关心队列
    loop_count: int                                  # 循环计数
    feedback_type: Optional[str]                      # 反馈类型
    
    # Human-in-the-loop
    needs_approval: bool                             # 需要人工确认
    user_preference_override: Optional[dict]         # 用户偏好覆盖

6.3 论文→LangGraph→看宝AI的完整链条

┌─────────────────────────────────────────────────────────────────┐
│                      理论到实践的完整链条                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────────┐                                           │
│  │ ReAct论文         │ ← Thought-Action-Observation循环          │
│  │ (2023)           │   LangGraph: add_conditional_edges         │
│  └────────┬─────────┘                                           │
│           │                                                       │
│           ▼                                                       │
│  ┌──────────────────┐                                           │
│  │ Generative Agents │ ← 记忆流-反思-规划                          │
│  │ (Stanford 2023)  │   LangGraph: State + Checkpointer          │
│  └────────┬─────────┘                                           │
│           │                                                       │
│           ▼                                                       │
│  ┌──────────────────┐                                           │
│  │ LangGraph框架      │ ← StateGraph + Nodes + Edges              │
│  │ (DeepLearning.AI) │   持久化 + Human-in-the-loop               │
│  └────────┬─────────┘                                           │
│           │                                                       │
│           ▼                                                       │
│  ┌──────────────────┐                                           │
│  │ 看宝AI主动关心循环  │ ← 五状态循环（等待→感知→判断→行动→反馈）     │
│  │                   │   五层记忆驱动 + 防打扰机制                 │
│  └──────────────────┘                                           │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

---

七、可运行代码示例

7.1 基础ReAct Agent

#!/usr/bin/env python3
"""
LangGraph ReAct Agent - 基础版
支持工具调用、自动推理循环、持久化
"""
import operator
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt, Command
============ 1. 定义Agent状态 ============
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
============ 2. 定义工具 ============
def get_weather(city: str) -> str:
    """查询天气"""
    weather_data = {
        "北京": "28°C，晴朗",
        "上海": "26°C，多云",
        "广州": "32°C，晴天",
        "深圳": "31°C，阴天",
    }
    return weather_data.get(city, f"未找到{city}天气数据")
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"
tools = [get_weather, calculate]
============ 3. 构建Graph ============
def build_react_agent():
    # 初始化LLM并绑定工具
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    llm_with_tools = llm.bind_tools(tools)
    
    # 判断是否继续
    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]
        if getattr(last_message, "tool_calls", None):
            return "continue"
        return END
    
    # Agent节点：调用LLM
    def call_model(state: AgentState):
        messages = state["messages"]
        response = llm_with_tools.invoke(messages)
        return {"messages": [response]}
    
    # 构建工作流
    workflow = StateGraph(AgentState)
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", ToolNode(tools))
    
    workflow.set_entry_point("agent")
    workflow.add_conditional_edges("agent", should_continue, {
        "continue": "tools",
        END: END
    })
    workflow.add_edge("tools", "agent")
    
    return workflow.compile()
============ 4. 运行 ============
if __name__ == "__main__":
    app = build_react_agent()
    
    # 带持久化的运行
    checkpointer = InMemorySaver()
    app = app.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "test_1"}}
    
    # 测试多轮对话
    query = "北京天气怎么样？帮我算一下26度空调一天耗电多少度"
    
    for chunk in app.stream(
        {"messages": [HumanMessage(content=query)]},
        config
    ):
        for node, update in chunk.items():
            print(f"\n--- {node} ---")
            if "messages" in update:
                last = update["messages"][-1]
                if hasattr(last, "tool_calls") and last.tool_calls:
                    for tc in last.tool_calls:
                        print(f"调用工具: {tc['name']} 参数: {tc['args']}")

7.2 Human-in-the-loop Agent

#!/usr/bin/env python3
"""
LangGraph Human-in-the-loop Agent
支持人工审核关键决策
"""
from typing import TypedDict, Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt, Command
class HITLState(TypedDict):
    messages: Annotated[list, operator.add]
    pending_action: dict | None
需要人工审核的工具
def book_hotel(hotel_name: str):
    """预订酒店 - 需要人工审核"""
    response = interrupt({
        "action": "book_hotel",
        "hotel_name": hotel_name,
        "warning": "此操作会产生实际费用，请确认"
    })
    
    if response["type"] == "accept":
        return f"已成功预订: {hotel_name}"
    elif response["type"] == "edit":
        return f"已预订: {response['args']['hotel_name']}"
    else:
        return "预订已取消"
def search_hotels(location: str) -> str:
    """搜索酒店"""
    return f"找到3家酒店: 希尔顿、万豪、四季"
tools = [search_hotels, book_hotel]
def build_hitl_agent():
    llm = ChatOpenAI(model="gpt-3.5-turbo").bind_tools(tools)
    
    def should_interrupt(state):
        """检查是否需要人工审核"""
        last_msg = state["messages"][-1]
        if tool_calls := getattr(last_msg, "tool_calls", None):
            for tc in tool_calls:
                if tc["name"] == "book_hotel":
                    return "human_review"
        return "continue"
    
    def agent_node(state):
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    
    workflow = StateGraph(HITLState)
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", ToolNode(tools))
    workflow.add_node("human_review", human_review_node)
    
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges("agent", should_interrupt, {
        "human_review": "human_review",
        "continue": "tools"
    })
    workflow.add_conditional_edges("human_review", 
        lambda x: "end" if x.get("status") == "cancelled" else "agent",
        {
            "agent": "agent",
            "end": END
        }
    )
    workflow.add_edge("tools", "agent")
    
    return workflow.compile(checkpointer=InMemorySaver())
运行示例
def run_example():
    app = build_hitl_agent()
    config = {"configurable": {"thread_id": "hotel_booking"}}
    
    # 启动预订流程
    for chunk in app.stream(
        {"messages": [HumanMessage(content="帮我预订北京希尔顿酒店")]},
        config
    ):
        print(chunk)
        
        # 检测中断
        if "__interrupt__" in chunk:
            print("\n=== 等待人工审核 ===")
            
            # 人工批准
            app.invoke(
                Command(resume={"type": "accept"}),
                config
            )
if __name__ == "__main__":
    run_example()

7.3 看宝AI主动关心循环（LangGraph版）

#!/usr/bin/env python3
"""
看宝AI主动关心循环 - LangGraph实现
基于ReAct理论 + Generative Agents记忆系统
"""
from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage
import operator
from enum import Enum
from datetime import datetime
============ 1. 定义类型 ============
class CareType(Enum):
    HEALTH_CHECK = "health_check"      # 健康检查
    GROWTH_MILESTONE = "growth"       # 成长里程碑
    SITUATIONAL = "situational"       # 情景关心
    LONG_ABSENCE = "absence"          # 长期未互动
    TOPIC_CONTINUE = "topic"          # 话题延续
class FeedbackType(Enum):
    POSITIVE = "positive"             # 积极回应
    NEUTRAL = "neutral"               # 中性回应
    NEGATIVE = "negative"             # 消极回应
    STOP = "stop"                     # 要求停止
============ 2. 定义状态 ============
class CareState(TypedDict):
    # 记忆层
    session_memory: Annotated[list, operator.add]      # 会话记忆
    profile_memory: dict                                 # 画像记忆
    situational_memory: list                            # 情景记忆
    growth_memory: list                                 # 成长痕迹
    
    # 循环控制
    current_trigger: Optional[dict]                      # 当前触发点
    care_queue: list                                    # 关心队列
    loop_count: int                                    # 循环次数
    feedback_type: Optional[str]                        # 反馈类型
    continue_tracking: bool                             # 是否继续追踪
    
    # 行动输出
    care_content: Optional[str]                         # 生成的关心内容
    care_type: Optional[str]                            # 关心类型
============ 3. 定义节点 ============
def wait_node(state: CareState) -> CareState:
    """等待状态：监控时间和事件"""
    # 检查关心队列
    if state.get("care_queue") and not state.get("current_trigger"):
        state["current_trigger"] = state["care_queue"].pop(0)
        state["loop_count"] = 0
    return state
def sense_node(state: CareState) -> CareState:
    """感知状态：从记忆中提取信号"""
    trigger = state.get("current_trigger")
    if not trigger:
        return state
    
    care_type = trigger.get("care_type")
    
    # 根据关心类型从对应记忆层提取信息
    if care_type == CareType.HEALTH_CHECK.value:
        context = _extract_health_context(state)
    elif care_type == CareType.GROWTH_MILESTONE.value:
        context = _extract_growth_context(state)
    elif care_type == CareType.SITUATIONAL.value:
        context = _extract_situational_context(state)
    else:
        context = {}
    
    state["current_trigger"]["context"] = context
    return state
def decide_node(state: CareState) -> CareState:
    """判断状态：决策是否关心及关心方式"""
    trigger = state.get("current_trigger", {})
    context = trigger.get("context", {})
    
    # 简化判断：直接生成关心内容
    care_content = _generate_care_content(trigger)
    state["care_content"] = care_content
    state["care_type"] = trigger.get("care_type")
    
    return state
def act_node(state: CareState) -> CareState:
    """行动状态：生成自然关心内容"""
    # care_content已在decide_node生成
    return state
def feedback_node(state: CareState) -> CareState:
    """反馈状态：处理用户回应"""
    # 模拟反馈处理
    # 实际使用时从用户回复中提取
    feedback = state.get("feedback_type", FeedbackType.POSITIVE.value)
    
    # 判断是否继续追踪
    if feedback == FeedbackType.STOP.value:
        state["continue_tracking"] = False
    elif feedback == FeedbackType.POSITIVE.value:
        state["continue_tracking"] = True
    else:
        state["continue_tracking"] = False
    
    # 记录到会话记忆
    state["session_memory"].append({
        "type": "care_feedback",
        "care_content": state.get("care_content"),
        "feedback": feedback,
        "timestamp": datetime.now().isoformat()
    })
    
    return state
============ 4. 辅助函数 ============
def _extract_health_context(state: CareState) -> dict:
    """从情景记忆中提取健康相关上下文"""
    recent_events = state.get("situational_memory", [])
    health_events = [e for e in recent_events if e.get("type") == "health"]
    return {"recent_health": health_events[-1] if health_events else None}
def _extract_growth_context(state: CareState) -> dict:
    """从成长记忆中提取里程碑上下文"""
    growth = state.get("growth_memory", [])
    return {"recent_milestones": growth[-3:]}
def _extract_situational_context(state: CareState) -> dict:
    """从情景记忆中提取上下文"""
    return {"situational": state.get("situational_memory", [])[-1]}
def _generate_care_content(trigger: dict) -> str:
    """生成自然的关心内容"""
    care_type = trigger.get("care_type")
    context = trigger.get("context", {})
    
    templates = {
        CareType.HEALTH_CHECK.value: [
            "昨天说宝宝有点发烧，今天怎么样了？",
            "宝宝咳嗽好点了吗？",
        ],
        CareType.GROWTH_MILESTONE.value: [
            "宝宝最近有什么新变化吗？",
            "到了认人的阶段了吗？",
        ],
        CareType.LONG_ABSENCE.value: [
            "这几天没见你上线，宝宝一切都好吗？",
        ],
    }
    
    return templates.get(care_type, ["宝宝怎么样？"])[0]
============ 5. 构建Graph ============
def build_care_graph():
    from langgraph.graph import StateGraph, END
    
    workflow = StateGraph(CareState)
    
    # 添加节点
    workflow.add_node("wait", wait_node)
    workflow.add_node("sense", sense_node)
    workflow.add_node("decide", decide_node)
    workflow.add_node("act", act_node)
    workflow.add_node("feedback", feedback_node)
    
    # 设置入口
    workflow.add_edge("__start__", "wait")
    
    # 条件边：wait -> sense 或 wait -> 结束
    def wait_condition(state: CareState) -> str:
        if state.get("current_trigger"):
            return "sense"
        return END
    
    workflow.add_conditional_edges("wait", wait_condition, {
        "sense": "sense",
        END: END
    })
    
    # 普通边
    workflow.add_edge("sense", "decide")
    workflow.add_edge("decide", "act")
    workflow.add_edge("act", "feedback")
    
    # 条件边：feedback -> wait 或 feedback -> 结束
    def feedback_condition(state: CareState) -> str:
        if state.get("continue_tracking"):
            return "wait"
        return END
    
    workflow.add_conditional_edges("feedback", feedback_condition, {
        "wait": "wait",
        END: END
    })
    
    return workflow.compile()
============ 6. 运行示例 ============
def run_care_example():
    app = build_care_graph()
    
    # 初始化状态
    initial_state = {
        "session_memory": [],
        "profile_memory": {"baby_age": "8个月", "parent_style": "新手爸妈"},
        "situational_memory": [
            {"type": "health", "content": "宝宝昨天有点发烧", "date": "2024-01-01"}
        ],
        "growth_memory": [
            {"milestone": "第一次翻身", "date": "2024-01-15"},
            {"milestone": "会说妈妈", "date": "2024-02-01"}
        ],
        "care_queue": [
            {"care_type": CareType.HEALTH_CHECK.value, "trigger": "health_reminder"}
        ],
        "current_trigger": None,
        "loop_count": 0,
        "feedback_type": FeedbackType.POSITIVE.value,
        "continue_tracking": False,
        "care_content": None,
        "care_type": None
    }
    
    # 执行
    result = app.invoke(initial_state)
    
    print("=== 主动关心结果 ===")
    print(f"关心内容: {result.get('care_content')}")
    print(f"关心类型: {result.get('care_type')}")
    print(f"循环次数: {result.get('loop_count')}")
    print(f"会话记忆: {result.get('session_memory')}")
if __name__ == "__main__":
    run_care_example()

---

八、对常思杨架构的启发

8.1 当前架构回顾

当前看宝AI架构：
├── 主动关心循环 (LangGraph)  ← 五状态循环
├── 多角色专家团队 (CrewAI)   ← 层级协作
├── 育儿知识库 (LlamaIndex)  ← RAG检索
└── 五层记忆系统 (自研)       ← 持久化

8.2 LangGraph带来的新思路

改进点	当前实现	LangGraph增强
状态管理	手动管理dict	StateGraph自动归约
循环控制	手动条件判断	add_conditional_edges
持久化	文件系统	Checkpointer支持多后端
调试	print调试	Time Travel回溯
人工介入	无	interrupt + Command

8.3 具体架构改进建议

建议1：采用StateGraph重构主动关心循环

# 当前实现（简化版状态机）
def run_simple_care(user_id, trigger_type, trigger_reason):
    # 手动状态管理，容易出错
    state = {"phase": "wait", "trigger": None, ...}
    while True:
        if state["phase"] == "wait":
            # 等待逻辑
            pass
        elif state["phase"] == "sense":
            # 感知逻辑
            pass
        # ...
        
建议实现（StateGraph）
workflow = StateGraph(CareState)
workflow.add_node("wait", wait_node)
workflow.add_node("sense", sense_node)
workflow.add_node("decide", decide_node)
workflow.add_node("act", act_node)
workflow.add_node("feedback", feedback_node)
条件边自动管理状态流转

优势：
- 状态流转由图自动管理，减少手動bug
- 支持checkpoint，恢复执行
- 支持time-travel调试
- 支持interrupt人工介入

建议2：引入Checkpointer实现真正的持久化

# 当前：文件系统存储
with open("memory.json", "w") as f:
    json.dump(memory, f)
建议：Checkpointer + 多后端支持
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
开发环境
checkpointer = InMemorySaver()
生产环境
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
编译时传入
app = workflow.compile(checkpointer=checkpointer)
调用时指定thread_id
config = {"configurable": {"thread_id": user_id}}
app.invoke(inputs, config)

优势：
- 自动checkpoint，无需手动保存
- 支持多用户并发
- 支持数据库持久化
- 支持状态回溯

建议3：引入Human-in-the-loop审核关心内容

def care_approval_node(state: CareState) -> CareState:
    """关心内容审核节点"""
    care_content = state.get("care_content")
    
    # 高风险关心需要审核
    if state.get("care_type") == "health_check":
        response = interrupt({
            "care_content": care_content,
            "warning": "健康类关心，请确认发送"
        })
        
        if response["type"] == "reject":
            state["care_content"] = None
            state["care_type"] = None
    
    return state

适用场景：
- 健康相关关心（避免误导）
- 敏感话题（避免不当表达）
- 首次关心某类事件

8.4 LangGraph学习路径建议

LangGraph学习路径：
├── 基础：StateGraph + Nodes + Edges
├── 进阶：Conditional Edges + Reducers
├── 持久化：Checkpointer + Thread
├── 调试：Time Travel + Stream
├── 人工介入：Interrupt + Command
└── 生产：Postgres/SQLite + LangSmith

---

九、实践项目建议

9.1 项目1：看宝AI LangGraph重构

目标：用StateGraph重构主动关心循环

步骤：

1. 定义CareState，包含五层记忆和循环状态
2. 实现5个节点：wait, sense, decide, act, feedback
3. 配置条件边管理循环流转
4. 集成Checkpointer
5. 添加interrupt用于人工审核

预期产出：./实践项目/看宝AI-LangGraph重构/

9.2 项目2：多Agent协作框架

目标：用LangGraph实现看宝AI的多Agent协作

架构：

┌─────────────────────────────────────────────────────┐
│              看宝AI Multi-Agent架构                   │
├─────────────────────────────────────────────────────┤
│                                                      │
│  ┌──────────────┐                                  │
│  │ Orchestrator │ ← 协调器（StateGraph）            │
│  └──────┬───────┘                                  │
│         │                                           │
│    ┌────┴────┬────────────┐                        │
│    ▼         ▼            ▼                        │
│ ┌──────┐ ┌──────┐   ┌────────┐                   │
│ │ Care │ │RAG   │   │ Memory │                   │
│ │Agent │ │Agent │   │ Agent  │                   │
│ └──┬───┘ └──┬───┘   └───┬────┘                   │
│    │         │           │                         │
│    └─────────┴───────────┘                         │
│              │                                      │
│              ▼                                      │
│      ┌──────────────┐                              │
│      │  Memory      │ ← 五层记忆                    │
│      └──────────────┘                              │
│                                                      │
└─────────────────────────────────────────────────────┘

9.3 项目3：智能育儿助手Demo

目标：完整展示LangGraph Agent能力

功能：

1. 多轮对话（Checkpointer持久化）
2. 主动关心（主动关心循环）
3. 知识问答（RAG检索）
4. 人工审核（interrupt）
5. 时间旅行调试（checkpoint回溯）

---

十、关键术语表

术语	英文	解释
状态图	StateGraph	LangGraph核心图类，参数化用户定义的状态
节点	Node	执行逻辑的函数，接收State输入，返回State更新
边	Edge	定义节点间的路由逻辑
超步	Super-step	图节点的一次迭代
归约器	Reducer	定义节点更新如何应用到State的函数
检查点	Checkpoint	某时刻State的快照
线程	Thread	唯一ID，关联一系列Checkpoint
人工介入	Human-in-the-loop	人工审核/修改Agent状态
中断	Interrupt	暂停图执行，等待人工输入
命令	Command	恢复中断执行的指令

---

十一、参考资料

1. 课程：AI Agents in LangGraph, DeepLearning.AI

- https://www.deeplearning.ai/short-courses/ai-agents-in-langgraph/

1. 官方文档：LangGraph Documentation

- https://langchain-ai.github.io/langgraph/

1. 论文：

- ReAct: Synergizing Reasoning and Acting in Language Models (Yao et al., 2023)
- Generative Agents: Interactive Simulacra of Human Behavior (Park et al., Stanford, 2023)

1. GitHub仓库：

- https://github.com/langchain-ai/langgraph
- https://github.com/pguso/langgraph-course

---

十二、行为准则更新

基于LangGraph学习，更新以下行为准则：

准则1：状态管理优先使用StateGraph

- 涉及多步骤、有循环的流程，优先用StateGraph
- 避免手动状态管理，减少bug
- 显式定义状态流转，避免隐式逻辑

准则2：持久化优先使用Checkpointer

- Checkpointer自动checkpoint，比手动文件保存可靠
- 多用户场景必须用thread_id隔离
- 生产环境用Postgres/SQLite，开发环境用InMemorySaver

准则3：复杂调试使用Time Travel

- 遇到循环bug，用get_state_history回溯
- 确认checkpoint后用update_state修正
- 避免重启服务丢失状态

准则4：敏感操作引入Human-in-the-loop

- 健康、财经等敏感操作用interrupt人工审核
- interrupt配合Checkpointer实现可靠的暂停/恢复
- Command支持批准、修改、拒绝三种响应

---

> 学习完成时间：2026-04-10
> 下一步：结合看宝AI项目进行LangGraph重构实践