LangGraph 学习笔记

State：图的"共享内存"

State 是整个图的数据载体。所有节点都能读写它。定义为一个 TypedDict：

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加模式：新消息不会覆盖旧消息
    search_results: str                        # 覆盖模式：新值直接替换旧值
    final_answer: str

两个关键概念：Reducer（归约器）

Node：图的执行单元

每个 Node 是一个 Python 函数，接收 State，返回要更新的字段：

def search_node(state: AgentState) -> dict:
    # 1. 读取 State
    query = state["messages"][-1].content

    # 2. 做处理...
    result = search_knowledge_base(query)

    # 3. 返回要更新的字段（不直接修改 State！）
    return {
        "search_results": result,
        "messages": [AIMessage(content=result)],
    }

State + Node 协同工作示意图

Edge（普通边）：固定路线

A 做完后必然去 B。就像工厂的固定传送带。

builder.add_edge("search", "answer")   # 搜索完 → 必然去回答
builder.add_edge("answer", END)        # 回答完 → 结束

Conditional Edge（条件边）：动态路由

A 做完后，根据 State 内容动态决定下一步去向。就像分拣机，检查标签后分流。

def router(state: AgentState) -> str:
    # 看 State 决定去哪——这就是"动态决策"
    if state.get("search_satisfied"):
        return "answer"    # 满意 → 去回答
    else:
        return "search"    # 不满意 → 重新搜索（循环！）

builder.add_conditional_edges(
    "search", router,
    {"answer": "answer", "search": "search"}
)

Memory 的两个层次

Checkpointer 使用

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 每次调用传入 thread_id（标识一次对话）
config = {"configurable": {"thread_id": "conversation_1"}}

# 第1轮
graph.invoke({"messages": [HumanMessage(content="研究AI趋势")]}, config)

# 第2轮（同一个 thread_id，Agent 记得上一轮的所有中间状态！）
graph.invoke({"messages": [HumanMessage(content="详细说说第2点")]}, config)

Demo 1：简单工作流（问题 → 搜索 → 回答）

这是 LangGraph 的 "Hello World"——最简单的线性工作流。

class Demo1State(TypedDict):
    messages: Annotated[list, add_messages]
    search_results: str

builder = StateGraph(Demo1State)
builder.add_node("search", search_node)
builder.add_node("answer", answer_node)
builder.add_edge(START, "search")
builder.add_edge("search", "answer")
builder.add_edge("answer", END)
graph = builder.compile()

手写版 vs LangGraph 版

def ask_question(query):
    results = search(query)      # 步骤1
    answer = llm.invoke(          # 步骤2
        f"基于{results}回答{query}")
    return answer

# 痛点：
# - 中间状态不可见
# - 加一步要改整个函数
# - 没有错误处理

graph = build_demo1()
result = graph.invoke({
    "messages": [HumanMessage(content=q)],
    "search_results": "",
})

# 优势：
# - graph.stream() 可看每一步
# - 加节点只需 add_node + add_edge
# - 框架提供错误边界

Demo 2：研究 Agent（拆解 → 搜索 → 总结 → 报告）

真实的研究任务需要多步骤协作——每步的输出是下一步的输入。

State 字段随流程"生长"

每个节点的输出被下一个节点读取——State 就是"交接单"。

与 Demo 1 的关键区别

Demo 3：多 Agent 协作系统

Supervisor 模式 + Reviewer 循环 + Memory —— 这是 LangChain 做不到的。

典型执行流程

关键设计决策

三个 Demo 的演进路径

什么时候用哪个？

两者配合使用（不是替代关系！）

# LangGraph 的 Node 里用 LangChain 的组件
def researcher_node(state):
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate

    llm = ChatOpenAI(...)                          # LangChain 组件
    prompt = ChatPromptTemplate.from_messages([...]) # LangChain 组件
    chain = prompt | llm | StrOutputParser()        # LangChain LCEL

    return {"research_findings": chain.invoke(state)}

# LangGraph 负责"什么时候调用谁"（流程控制）
# LangChain 负责"怎么调用"（组件实现）

生产环境五大注意事项

class State(TypedDict):
    revision_count: int

def supervisor(state):
    if state["revision_count"] > 3:
        return "FINISH"  # 强制结束

def safe_node(state):
    try:
        result = risky_op(state)
        return {"output": result}
    except Exception as e:
        return {"messages": [AIMessage(
            content=f"节点失败: {e}")]}

def search_with_timeout(state):
    signal.alarm(30)  # 30 秒超时
    try:
        result = search(state)
        signal.alarm(0)
        return result
    except TimeoutError:
        return {"search_results": "超时"}

5. 可观测性：stream() 模式

逐步观察每个节点的输入输出——这是手写 pipeline 做不到的：

for step in graph.stream(initial_state):
    for node_name, output in step.items():
        print(f"[{node_name}] {output}")

# 输出示例：
# [search] {'search_results': 'LangGraph 是...'}
# [answer] {'messages': [AIMessage('LangGraph 是一个构建...')]}

核心收获

配套资源